Add variant-group endpoints and cross-model queue creation (#671, #2570)

Adds /library/variant-groups for declaring that several sliced files are
the same job for different printers, and a variants payload on queue
creation that turns such a set into one queue item with a candidate per
file.

The candidate set is validated as a set: one file per printer model, each
file sliced for the model it is offered as, and at least one model with
an active printer. A cross-model item deliberately holds no file of its
own, because print_queue.library_file_id is ON DELETE CASCADE and would
destroy the whole job when a single alternative is deleted.

Fixes internal printer-model codes never being resolved on queue create
and update: normalize_printer_model returns unknown input unchanged, so
the or-chain never reached the code map and a "C13" target matched no
printer and waited forever.

Skips candidates whose file is trashed or missing. Library deletes are
soft, and SQLite runs with PRAGMA foreign_keys off, so neither case is
covered by the schema; the hard-delete paths now also drop the rows.

Adds library_files.variant_target_model so a user can say which printer
a file without slicer metadata is for, kept out of file_metadata so the
assertion is never mistaken for parsed data.

backend/app/api/routes/library.py

@@ -4692,6 +4692,9 @@ async def delete_file(
                 abs_thumb_path.unlink()
             except OSError as e:
                 logger.warning("Failed to delete thumbnail from disk: %s", e)
+        from backend.app.services.library_trash import delete_dependent_variants
+
+        await delete_dependent_variants(db, [file.id])
         await db.delete(file)
         await db.commit()
         return {"status": "success", "message": "File deleted", "trashed": False}

backend/app/api/routes/library_variants.py

@@ -0,0 +1,425 @@
+"""Variant groups — one job, several sliced files (#671 / #2570).
+
+A user with more than one printer model slices the same job once per model. The
+files are unrelated as far as the library is concerned: different names,
+different metadata, often uploaded separately after being sliced in Bambu Studio.
+A variant group is the user telling Bambuddy that they are interchangeable.
+
+Two features consume that statement from opposite ends:
+
+* the print queue picks the printer and needs the matching file (#671)
+* the File Manager's print action has the printer already and needs the same
+  match (#2570)
+
+The group itself stores no model information. Each member's target model comes
+from its own ``sliced_for_model``, parsed out of the 3MF, so a group can never
+disagree with the files in it. A legacy file that declares no model may name one
+explicitly, because there is nothing else to go on.
+
+Invariants enforced here rather than in the database, because they are about
+meaning rather than shape:
+
+* **Two members minimum.** A group of one expresses no choice. Removing members
+  down to one dissolves the group rather than leaving a stub that does nothing.
+* **One member per model.** Two files sliced for the same printer are not
+  alternatives — the resolver would have no basis to prefer one, so an
+  arbitrary pick would look like a bug the first time the wrong quality preset
+  came out.
+* **Members must be sliced and must resolve to a model.** An unsliced .3mf can
+  never be dispatched, so it cannot be a candidate.
+* **A file belongs to at most one group**, which the schema already guarantees;
+  this layer turns the resulting overwrite into an explicit 409.
+
+Permissions follow library_tags.py: mutations need LIBRARY_UPDATE_ALL /
+LIBRARY_UPDATE_OWN, reads need LIBRARY_READ_ALL / LIBRARY_READ_OWN, and an
+``*_OWN`` caller only ever sees or touches files they created.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.auth import require_ownership_permission
+from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.models.library import FileVariantGroup, LibraryFile
+from backend.app.models.user import User
+from backend.app.schemas.library import (
+    VariantGroupCreate,
+    VariantGroupMemberRequest,
+    VariantGroupMemberResponse,
+    VariantGroupResponse,
+    VariantGroupUpdate,
+)
+from backend.app.utils.printer_models import normalize_printer_model, normalize_printer_model_id
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/library/variant-groups", tags=["library-variants"])
+
+# File types that can actually be sent to a printer. A source .3mf or an .stl
+# has no G-code and no sliced_for_model, so it is never a dispatch candidate.
+_PRINTABLE_TYPES = ("gcode.3mf", "gcode")
+
+
+def normalize_model_name(raw: str | None) -> str | None:
+    """Normalize any spelling of a printer model to its short name.
+
+    Internal codes are resolved **first**. ``normalize_printer_model`` returns
+    unknown input unchanged rather than None, so an ``x or y`` chain in the other
+    order never reaches the code map and leaves "O1C" as "O1C" — which then
+    matches no printer row and leaves the job waiting forever. Running the code
+    map first is a no-op for every non-code input.
+    """
+    if not raw:
+        return None
+    return normalize_printer_model(normalize_printer_model_id(raw) or raw) or raw
+
+
+def resolve_variant_model(lib_file: LibraryFile, explicit: str | None = None) -> str | None:
+    """Normalized model a file will be dispatched to, or None if unknowable.
+
+    Precedence: the caller's explicit choice for this request, then the durable
+    override stored on the file, then what the 3MF itself declares. The override
+    exists because a file imported before Bambuddy parsed ``sliced_for_model``
+    declares nothing, and without a way to say so it could never be grouped.
+    It is kept separate from ``file_metadata`` so a user's assertion is never
+    mistaken for something parsed out of the file.
+    """
+    raw = explicit or lib_file.variant_target_model or (lib_file.file_metadata or {}).get("sliced_for_model")
+    return normalize_model_name(raw)
+
+
+async def _load_files(
+    db: AsyncSession,
+    file_ids: list[int],
+    user: User | None,
+    can_access_all: bool,
+) -> dict[int, LibraryFile]:
+    """Fetch the caller's visible, untrashed files by id."""
+    query = LibraryFile.active().where(LibraryFile.id.in_(file_ids))
+    if user is not None and not can_access_all:
+        query = query.where(LibraryFile.created_by_id == user.id)
+    rows = (await db.execute(query)).scalars().all()
+    return {f.id: f for f in rows}
+
+
+def _validate_member(lib_file: LibraryFile, explicit_model: str | None) -> str:
+    """Return the member's model, or raise the reason it cannot be one."""
+    if lib_file.file_type not in _PRINTABLE_TYPES:
+        raise HTTPException(
+            400,
+            f"{lib_file.filename} is not a sliced file — only sliced output can be a print variant",
+        )
+    model = resolve_variant_model(lib_file, explicit_model)
+    if not model:
+        raise HTTPException(
+            400,
+            f"{lib_file.filename} does not say which printer it was sliced for — set its target model explicitly",
+        )
+    if explicit_model:
+        # Persist the user's answer, normalized. The group stores no model data
+        # of its own, so without this the choice would last exactly one request
+        # and the member would read back with no model at all.
+        lib_file.variant_target_model = model
+    return model
+
+
+async def _group_response(db: AsyncSession, group: FileVariantGroup) -> VariantGroupResponse:
+    members = (
+        (
+            await db.execute(
+                LibraryFile.active()
+                .where(LibraryFile.variant_group_id == group.id)
+                .order_by(LibraryFile.variant_position, LibraryFile.id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    return VariantGroupResponse(
+        id=group.id,
+        name=group.name,
+        members=[
+            VariantGroupMemberResponse(
+                library_file_id=f.id,
+                filename=f.filename,
+                # Members were validated on the way in, but a file whose metadata
+                # was rewritten since then should not blow up a read.
+                target_model=resolve_variant_model(f) or "",
+                position=f.variant_position,
+            )
+            for f in members
+        ],
+    )
+
+
+async def _get_group_or_404(db: AsyncSession, group_id: int) -> FileVariantGroup:
+    group = (await db.execute(select(FileVariantGroup).where(FileVariantGroup.id == group_id))).scalar_one_or_none()
+    if not group:
+        raise HTTPException(404, "Variant group not found")
+    return group
+
+
+async def _dissolve_if_too_small(db: AsyncSession, group: FileVariantGroup) -> bool:
+    """Delete the group when fewer than two members remain.
+
+    A one-member group is not a choice, and leaving one behind would let the
+    queue create a cross-model item with a single candidate that silently
+    behaves like an ordinary job. Returns True when the group was dissolved.
+    """
+    remaining = (await db.execute(select(LibraryFile).where(LibraryFile.variant_group_id == group.id))).scalars().all()
+    if len(remaining) >= 2:
+        return False
+    for lib_file in remaining:
+        lib_file.variant_group_id = None
+        lib_file.variant_position = 0
+    await db.delete(group)
+    return True
+
+
+@router.post("", response_model=VariantGroupResponse, status_code=201)
+@router.post("/", response_model=VariantGroupResponse, status_code=201)
+async def create_variant_group(
+    payload: VariantGroupCreate,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_UPDATE_ALL,
+            Permission.LIBRARY_UPDATE_OWN,
+        )
+    ),
+) -> VariantGroupResponse:
+    """Group files as variants of one job, in priority order."""
+    user, can_update_all = auth_result
+
+    file_ids = [m.library_file_id for m in payload.members]
+    if len(set(file_ids)) != len(file_ids):
+        raise HTTPException(400, "The same file cannot appear twice in a variant group")
+
+    files = await _load_files(db, file_ids, user, can_update_all)
+    missing = [fid for fid in file_ids if fid not in files]
+    if missing:
+        raise HTTPException(404, f"Library file not found: {missing[0]}")
+
+    already_grouped = [files[fid].filename for fid in file_ids if files[fid].variant_group_id is not None]
+    if already_grouped:
+        raise HTTPException(409, f"{already_grouped[0]} already belongs to a variant group")
+
+    models: dict[str, str] = {}
+    for member in payload.members:
+        lib_file = files[member.library_file_id]
+        model = _validate_member(lib_file, member.target_model)
+        if model in models:
+            raise HTTPException(
+                400,
+                f"{lib_file.filename} and {models[model]} are both sliced for {model} — "
+                "variants must target different printers",
+            )
+        models[model] = lib_file.filename
+
+    group = FileVariantGroup(
+        name=payload.name or files[file_ids[0]].filename,
+        created_by_id=user.id if user else None,
+    )
+    db.add(group)
+    await db.flush()
+
+    for position, fid in enumerate(file_ids):
+        files[fid].variant_group_id = group.id
+        files[fid].variant_position = position
+
+    await db.commit()
+    logger.info("Created variant group %s with %d members", group.id, len(file_ids))
+    return await _group_response(db, group)
+
+
+@router.get("/by-file/{file_id}", response_model=VariantGroupResponse)
+async def get_group_for_file(
+    file_id: int,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
+) -> VariantGroupResponse:
+    """The group a file belongs to.
+
+    Both consumers start from a file rather than a group id: the print modal
+    knows which file the user clicked, and the queue-create flow knows which
+    file was selected.
+    """
+    user, can_read_all = auth_result
+    files = await _load_files(db, [file_id], user, can_read_all)
+    lib_file = files.get(file_id)
+    if not lib_file:
+        raise HTTPException(404, "Library file not found")
+    if lib_file.variant_group_id is None:
+        raise HTTPException(404, "File is not part of a variant group")
+    return await _group_response(db, await _get_group_or_404(db, lib_file.variant_group_id))
+
+
+@router.get("/{group_id}", response_model=VariantGroupResponse)
+async def get_variant_group(
+    group_id: int,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_READ_ALL,
+            Permission.LIBRARY_READ_OWN,
+        )
+    ),
+) -> VariantGroupResponse:
+    return await _group_response(db, await _get_group_or_404(db, group_id))
+
+
+@router.patch("/{group_id}", response_model=VariantGroupResponse)
+async def update_variant_group(
+    group_id: int,
+    payload: VariantGroupUpdate,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_UPDATE_ALL,
+            Permission.LIBRARY_UPDATE_OWN,
+        )
+    ),
+) -> VariantGroupResponse:
+    """Rename the group, re-order its members, or both.
+
+    Re-ordering is how the user says which printer they would rather have when
+    both are free, so it must be an explicit full ordering — a partial list
+    would leave the rest in an order nobody chose.
+    """
+    user, can_update_all = auth_result
+    group = await _get_group_or_404(db, group_id)
+
+    if payload.name is not None:
+        group.name = payload.name
+
+    if payload.member_file_ids is not None:
+        current = (
+            (await db.execute(select(LibraryFile).where(LibraryFile.variant_group_id == group.id))).scalars().all()
+        )
+        if set(payload.member_file_ids) != {f.id for f in current}:
+            raise HTTPException(400, "member_file_ids must list exactly the group's current members")
+        files = await _load_files(db, payload.member_file_ids, user, can_update_all)
+        if len(files) != len(payload.member_file_ids):
+            raise HTTPException(404, "Library file not found")
+        for position, fid in enumerate(payload.member_file_ids):
+            files[fid].variant_position = position
+
+    await db.commit()
+    return await _group_response(db, group)
+
+
+@router.post("/{group_id}/members", response_model=VariantGroupResponse)
+async def add_variant_group_member(
+    payload: VariantGroupMemberRequest,
+    group_id: int,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_UPDATE_ALL,
+            Permission.LIBRARY_UPDATE_OWN,
+        )
+    ),
+) -> VariantGroupResponse:
+    """Attach another slice to an existing group.
+
+    This is the common real case: the H2S version was queued last week, the H2C
+    version was sliced today.
+    """
+    user, can_update_all = auth_result
+    group = await _get_group_or_404(db, group_id)
+
+    files = await _load_files(db, [payload.library_file_id], user, can_update_all)
+    lib_file = files.get(payload.library_file_id)
+    if not lib_file:
+        raise HTTPException(404, "Library file not found")
+    if lib_file.variant_group_id == group.id:
+        raise HTTPException(409, f"{lib_file.filename} is already in this group")
+    if lib_file.variant_group_id is not None:
+        raise HTTPException(409, f"{lib_file.filename} already belongs to a variant group")
+
+    model = _validate_member(lib_file, payload.target_model)
+
+    existing = (
+        (
+            await db.execute(
+                LibraryFile.active()
+                .where(LibraryFile.variant_group_id == group.id)
+                .order_by(LibraryFile.variant_position, LibraryFile.id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    for other in existing:
+        if resolve_variant_model(other) == model:
+            raise HTTPException(
+                400,
+                f"{lib_file.filename} and {other.filename} are both sliced for {model} — "
+                "variants must target different printers",
+            )
+
+    lib_file.variant_group_id = group.id
+    lib_file.variant_position = len(existing)
+    await db.commit()
+    return await _group_response(db, group)
+
+
+@router.delete("/{group_id}/members/{file_id}", response_model=None, status_code=204)
+async def remove_variant_group_member(
+    group_id: int,
+    file_id: int,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_UPDATE_ALL,
+            Permission.LIBRARY_UPDATE_OWN,
+        )
+    ),
+) -> None:
+    """Drop one file out of a group; the file itself is untouched."""
+    user, can_update_all = auth_result
+    group = await _get_group_or_404(db, group_id)
+
+    files = await _load_files(db, [file_id], user, can_update_all)
+    lib_file = files.get(file_id)
+    if not lib_file or lib_file.variant_group_id != group.id:
+        raise HTTPException(404, "File is not a member of this group")
+
+    lib_file.variant_group_id = None
+    lib_file.variant_position = 0
+    await db.flush()
+    await _dissolve_if_too_small(db, group)
+    await db.commit()
+
+
+@router.delete("/{group_id}", response_model=None, status_code=204)
+async def delete_variant_group(
+    group_id: int,
+    db: AsyncSession = Depends(get_db),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.LIBRARY_UPDATE_ALL,
+            Permission.LIBRARY_UPDATE_OWN,
+        )
+    ),
+) -> None:
+    """Ungroup the files. The files themselves are kept — every one of them is
+    independently printable, which is the whole reason they were grouped."""
+    group = await _get_group_or_404(db, group_id)
+    members = (await db.execute(select(LibraryFile).where(LibraryFile.variant_group_id == group.id))).scalars().all()
+    for lib_file in members:
+        lib_file.variant_group_id = None
+        lib_file.variant_position = 0
+    await db.delete(group)
+    await db.commit()

backend/app/api/routes/print_queue.py

@@ -12,6 +12,7 @@ from sqlalchemy import and_, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
+from backend.app.api.routes.library_variants import normalize_model_name, resolve_variant_model
 from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_ownership_permission
 from backend.app.core.config import settings
 from backend.app.core.database import get_db
@@ -19,7 +20,7 @@ from backend.app.core.permissions import Permission
 from backend.app.models.archive import PrintArchive
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_batch import PrintBatch
-from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
 from backend.app.models.printer import Printer
 from backend.app.models.project import Project
 from backend.app.models.user import User
@@ -33,14 +34,13 @@ from backend.app.schemas.print_queue import (
     PrintQueueItemResponse,
     PrintQueueItemUpdate,
     PrintQueueReorder,
+    QueueVariantCreate,
 )
 from backend.app.services.filament_deficit import compute_deficit_for_queue_item
 from backend.app.services.filament_requirements import overrides_for_plate
 from backend.app.services.notification_service import notification_service
 from backend.app.utils.printer_models import (
     is_gcode_compatible,
-    normalize_printer_model,
-    normalize_printer_model_id,
 )
 from backend.app.utils.threemf_tools import (
     extract_plate_metadata_from_3mf,
@@ -382,6 +382,146 @@ async def list_queue(
     return [_enrich_response(item) for item in items]
 
 
+async def _resolve_queue_variants(
+    db: AsyncSession,
+    specs: list[QueueVariantCreate],
+    current_user: User | None,
+) -> list[tuple[QueueVariantCreate, LibraryFile, str]]:
+    """Validate a cross-model candidate set and pair each file with its model (#671).
+
+    Validated as a set, not file by file, because the failure modes are about the
+    set: two candidates for the same printer give the resolver no basis to choose,
+    and a set where nothing can ever run is a job that waits forever.
+
+    At least one candidate must have an active printer — the rest may not, which
+    is deliberate. Grouping the H2C slice before the H2C arrives is a reasonable
+    thing to do, and refusing the whole queue action over it would be worse than
+    letting that candidate simply never match.
+    """
+    file_ids = [s.library_file_id for s in specs]
+    if len(set(file_ids)) != len(file_ids):
+        raise HTTPException(400, "The same file cannot be listed twice as a variant")
+
+    rows = (await db.execute(LibraryFile.active().where(LibraryFile.id.in_(file_ids)))).scalars().all()
+    by_id = {f.id: f for f in rows}
+
+    resolved: list[tuple[QueueVariantCreate, LibraryFile, str]] = []
+    seen_models: dict[str, str] = {}
+    any_active_printer = False
+
+    for spec in specs:
+        library_file = by_id.get(spec.library_file_id)
+        # Same IDOR posture as the single-file path: a file the caller cannot read
+        # is reported as missing rather than forbidden.
+        if not library_file or (
+            current_user
+            and not current_user.has_permission(Permission.LIBRARY_READ_ALL.value)
+            and library_file.created_by_id != current_user.id
+        ):
+            raise HTTPException(404, f"Library file not found: {spec.library_file_id}")
+
+        from backend.app.utils.filename import InvalidFilenameError, validate_print_filename
+
+        try:
+            validate_print_filename(library_file.filename)
+        except InvalidFilenameError as e:
+            raise HTTPException(400, str(e)) from e
+
+        model = resolve_variant_model(library_file, spec.target_model)
+        if not model:
+            raise HTTPException(
+                400,
+                f"{library_file.filename} does not say which printer it was sliced for — "
+                "set its target model explicitly",
+            )
+
+        # Cross-model safety gate (#2578), per candidate. A set is only as safe as
+        # its worst member, and model-based dispatch has no human in the loop.
+        sliced_for = (library_file.file_metadata or {}).get("sliced_for_model")
+        if not is_gcode_compatible(sliced_for, model):
+            raise HTTPException(
+                400,
+                f"{library_file.filename} was sliced for {sliced_for} and cannot be dispatched to {model} printers",
+            )
+
+        if model in seen_models:
+            raise HTTPException(
+                400,
+                f"{library_file.filename} and {seen_models[model]} are both for {model} — "
+                "variants must target different printers",
+            )
+        seen_models[model] = library_file.filename
+
+        has_printer = (
+            (
+                await db.execute(
+                    select(Printer).where(Printer.model == model).where(Printer.is_active == True)  # noqa: E712
+                )
+            )
+            .scalars()
+            .first()
+        )
+        any_active_printer = any_active_printer or bool(has_printer)
+
+        resolved.append((spec, library_file, model))
+
+    if not any_active_printer:
+        raise HTTPException(400, f"No active printers for any of: {', '.join(seen_models)}")
+
+    return resolved
+
+
+def _variant_values(
+    spec: QueueVariantCreate,
+    library_file: LibraryFile,
+    model: str,
+    position: int,
+) -> dict:
+    """Column values for one candidate, extracted from its own 3MF.
+
+    Each candidate is a different slice, so its filament requirements and print
+    time come from its own file rather than being inherited from the item.
+
+    Returns values rather than a row so a quantity>1 batch can build one row per
+    copy without re-opening the 3MF for each.
+    """
+    lib_path = Path(library_file.file_path)
+    file_path = lib_path if lib_path.is_absolute() else settings.base_dir / library_file.file_path
+
+    required_types = None
+    filament_overrides_json = None
+    print_time = (library_file.file_metadata or {}).get("print_time_seconds")
+
+    if file_path.exists():
+        types = _extract_filament_types_from_3mf(file_path, spec.plate_id)
+        if types:
+            required_types = json.dumps(types)
+        if spec.plate_id:
+            plate_time = _extract_print_time_from_3mf(file_path, spec.plate_id)
+            if plate_time is not None:
+                print_time = plate_time
+        if spec.filament_overrides:
+            plate_overrides = overrides_for_plate(spec.filament_overrides, file_path, spec.plate_id)
+            if plate_overrides:
+                filament_overrides_json = json.dumps(plate_overrides)
+                override_types = sorted({o["type"] for o in plate_overrides if "type" in o})
+                if override_types:
+                    existing = set(json.loads(required_types)) if required_types else set()
+                    required_types = json.dumps(sorted(existing | set(override_types)))
+
+    return {
+        "position": position,
+        "library_file_id": library_file.id,
+        "target_model": model,
+        "plate_id": spec.plate_id,
+        "ams_mapping": json.dumps(spec.ams_mapping) if spec.ams_mapping else None,
+        "nozzle_mapping": json.dumps(spec.nozzle_mapping) if spec.nozzle_mapping else None,
+        "filament_overrides": filament_overrides_json,
+        "required_filament_types": required_types,
+        "print_time_seconds": print_time,
+    }
+
+
 @router.post("/", response_model=PrintQueueItemResponse)
 async def add_to_queue(
     data: PrintQueueItemCreate,
@@ -389,17 +529,40 @@ async def add_to_queue(
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_CREATE),
 ):
     """Add an item to the print queue."""
-    # Normalize target_model (e.g., "Bambu Lab X1E" / "C13" -> "X1E")
-    target_model_norm = None
-    if data.target_model:
-        target_model_norm = (
-            normalize_printer_model(data.target_model)
-            or normalize_printer_model_id(data.target_model)
-            or data.target_model
-        )
-
-    # Validate that either archive_id or library_file_id is provided
-    if not data.archive_id and not data.library_file_id:
+    # Normalize target_model (e.g., "Bambu Lab X1E" / "C13" -> "X1E").
+    # normalize_model_name resolves internal codes first: the previous
+    # `normalize_printer_model(x) or normalize_printer_model_id(x)` chain never
+    # reached the code map, because the first call returns unknown input
+    # unchanged — so a "C13" target stayed "C13", matched no printer row and
+    # left the item waiting forever. Identical result for every other spelling.
+    target_model_norm = normalize_model_name(data.target_model)
+
+    # Cross-model alternatives (#671): several sliced files, whichever printer
+    # frees up first. The whole candidate set is validated before anything is
+    # written — a half-valid set would produce a job that can only reach some of
+    # the printers the user asked for, with nothing to say which.
+    variant_specs: list[tuple[QueueVariantCreate, LibraryFile, str]] = []
+    if data.variants:
+        if data.printer_id:
+            raise HTTPException(
+                400, "Cannot specify both printer_id and variants — pick a printer or offer alternatives"
+            )
+        if data.archive_id or data.library_file_id:
+            raise HTTPException(
+                400, "Cannot combine variants with archive_id or library_file_id — the variants are the files"
+            )
+        variant_specs = await _resolve_queue_variants(db, data.variants, current_user)
+        # Mirror the first candidate onto the item so the queue listing, the SJF
+        # grouping and the "Any H2S" label have something before a printer is
+        # picked. Resolution overwrites it with whichever candidate actually runs.
+        target_model_norm = variant_specs[0][2]
+
+    # Validate that either archive_id or library_file_id is provided.
+    # A cross-model item deliberately holds neither: its files live on the
+    # variant rows. Pointing library_file_id at one of them would be worse than
+    # useless — that FK is ON DELETE CASCADE, so deleting a single alternative
+    # would take the whole queue item with it.
+    if not data.archive_id and not data.library_file_id and not data.variants:
         raise HTTPException(400, "Either archive_id or library_file_id must be provided")
 
     # Cannot specify both printer_id and target_model
@@ -412,8 +575,10 @@ async def add_to_queue(
         if not result.scalar_one_or_none():
             raise HTTPException(400, "Printer not found")
 
-    # Validate target_model has active printers
-    if target_model_norm:
+    # Validate target_model has active printers. Skipped for cross-model items:
+    # target_model there is just the first candidate, and _resolve_queue_variants
+    # has already required that *some* candidate has a printer.
+    if target_model_norm and not data.variants:
         result = await db.execute(
             select(Printer).where(Printer.model == target_model_norm).where(Printer.is_active == True)  # noqa: E712
         )
@@ -744,6 +909,22 @@ async def add_to_queue(
         db.add(item)
         items.append(item)
 
+    if variant_specs:
+        variant_values = [
+            _variant_values(spec, library_file, model, position)
+            for position, (spec, library_file, model) in enumerate(variant_specs)
+        ]
+        # SJF orders pending items before any printer is known, so the row carries
+        # the shortest candidate's estimate. Resolution replaces it with the one
+        # that actually runs.
+        estimates = [v["print_time_seconds"] for v in variant_values if v["print_time_seconds"]]
+        for item in items:
+            # Each copy in a quantity>1 batch gets its own candidate rows —
+            # attempt counts are per-item, and two copies must be free to land on
+            # different printers.
+            item.variants.extend(PrintQueueVariant(**values) for values in variant_values)
+            item.print_time_seconds = min(estimates) if estimates else None
+
     await db.commit()
 
     # Refresh the first item for the response
@@ -1158,13 +1339,10 @@ async def update_queue_item(
 
     update_data = data.model_dump(exclude_unset=True)
 
-    # Normalize target_model if being updated
+    # Normalize target_model if being updated (see add_to_queue for why the
+    # code map has to run first).
     if "target_model" in update_data and update_data["target_model"]:
-        update_data["target_model"] = (
-            normalize_printer_model(update_data["target_model"])
-            or normalize_printer_model_id(update_data["target_model"])
-            or update_data["target_model"]
-        )
+        update_data["target_model"] = normalize_model_name(update_data["target_model"])
 
     # Cannot specify both printer_id and target_model
     new_printer_id = update_data.get("printer_id", item.printer_id)

backend/app/core/database.py

@@ -3961,6 +3961,9 @@ async def run_migrations(conn):
         "REFERENCES file_variant_groups(id) ON DELETE SET NULL",
     )
     await _safe_execute(conn, "ALTER TABLE library_files ADD COLUMN variant_position INTEGER DEFAULT 0")
+    # User-declared target model for a file whose 3MF does not say (#671).
+    # VARCHAR(50) is spelled identically on SQLite and Postgres.
+    await _safe_execute(conn, "ALTER TABLE library_files ADD COLUMN variant_target_model VARCHAR(50)")
     # The model declares index=True, so fresh installs get this from create_all();
     # migrated databases need it spelled out. Resolution looks members up by group
     # on every scheduler pass that touches a grouped item.

backend/app/main.py

@@ -39,6 +39,7 @@ from backend.app.api.routes import (
     library,
     library_tags,
     library_trash,
+    library_variants,
     local_backup,
     local_presets,
     maintenance,
@@ -7900,6 +7901,7 @@ app.include_router(projects.router, prefix=app_settings.api_prefix)
 app.include_router(library.router, prefix=app_settings.api_prefix)
 app.include_router(library_tags.router, prefix=app_settings.api_prefix)
 app.include_router(library_trash.router, prefix=app_settings.api_prefix)
+app.include_router(library_variants.router, prefix=app_settings.api_prefix)
 app.include_router(slice_jobs.router, prefix=app_settings.api_prefix)
 app.include_router(slicer_pipelines.router, prefix=app_settings.api_prefix)
 app.include_router(pipeline_runs.pipeline_run_create_router, prefix=app_settings.api_prefix)

backend/app/models/library.py

@@ -143,6 +143,12 @@ class LibraryFile(Base):
         ForeignKey("file_variant_groups.id", ondelete="SET NULL"), nullable=True, index=True
     )
     variant_position: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+    # User's answer to "which printer is this for", for a file that does not say.
+    # Files imported before Bambuddy parsed ``sliced_for_model`` — and raw .gcode —
+    # declare nothing, and without this they could never be grouped. Deliberately
+    # NOT written into ``file_metadata``: that holds what was parsed out of the
+    # file, and a user's assertion must not become indistinguishable from it.
+    variant_target_model: Mapped[str | None] = mapped_column(String(50), nullable=True)
 
     # User tracking (Issue #206)
     created_by_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)

backend/app/schemas/library.py

@@ -397,3 +397,58 @@ class BatchThumbnailResponse(BaseModel):
     succeeded: int
     failed: int
     results: list[BatchThumbnailResult]
+
+
+# ============ Variant Group Schemas (#671 / #2570) ============
+
+
+class VariantGroupMemberRequest(BaseModel):
+    """One file joining a variant group.
+
+    ``target_model`` is optional and normally omitted — it is read from the
+    file's own ``sliced_for_model``. Supply it only for a legacy 3MF that
+    declares no model, where there is nothing else to go on.
+    """
+
+    library_file_id: int
+    target_model: str | None = Field(None, max_length=50)
+
+
+class VariantGroupCreate(BaseModel):
+    """Declare that these files are the same job sliced for different printers.
+
+    Order is significant: it is the priority used when more than one printer is
+    idle at the same moment. Two members minimum — a group of one expresses no
+    choice.
+    """
+
+    members: list[VariantGroupMemberRequest] = Field(..., min_length=2)
+    name: str | None = Field(None, max_length=255)
+
+
+class VariantGroupUpdate(BaseModel):
+    """Rename a group and/or re-order its members.
+
+    ``member_file_ids`` must list exactly the group's current members; a partial
+    list is rejected rather than guessing where the omitted ones belong.
+    """
+
+    name: str | None = Field(None, max_length=255)
+    member_file_ids: list[int] | None = None
+
+
+class VariantGroupMemberResponse(BaseModel):
+    """A file within a group, with the model it will be dispatched to."""
+
+    library_file_id: int
+    filename: str
+    target_model: str
+    position: int
+
+
+class VariantGroupResponse(BaseModel):
+    """A variant group and its members, in priority order."""
+
+    id: int
+    name: str
+    members: list[VariantGroupMemberResponse]

backend/app/schemas/print_queue.py

@@ -44,6 +44,25 @@ def _coerce_tristate(v: object) -> object:
 TriState = Annotated[Literal["off", "on", "auto"], BeforeValidator(_coerce_tristate)]
 
 
+class QueueVariantCreate(BaseModel):
+    """One candidate file for a cross-model queue item (#671).
+
+    Per-file rather than per-item because the settings genuinely differ between
+    candidates: an H2C slice is dual-nozzle and will not share slot count, AMS
+    mapping or nozzle mapping with the H2S slice of the same model.
+
+    ``target_model`` is normally omitted and read from the file's own
+    ``sliced_for_model``; supply it only for a legacy 3MF that declares none.
+    """
+
+    library_file_id: int
+    target_model: str | None = None
+    plate_id: int | None = None
+    ams_mapping: list[int] | None = None
+    nozzle_mapping: list[int] | None = None
+    filament_overrides: list[dict] | None = None
+
+
 class PrintQueueItemCreate(BaseModel):
     printer_id: int | None = None  # None = unassigned, user assigns later
     target_model: str | None = None  # Target printer model (mutually exclusive with printer_id)
@@ -98,6 +117,12 @@ class PrintQueueItemCreate(BaseModel):
     # Direct printer-card uploads are temporary library files. The scheduler
     # deletes them after creating the durable archive copy.
     cleanup_library_after_dispatch: bool = False
+    # Cross-model alternatives (#671): several sliced files, one job, whichever
+    # printer frees up first. Mutually exclusive with printer_id (a specific
+    # printer defeats the purpose) and with archive_id/library_file_id (the
+    # candidates ARE the files). The scheduler resolves one onto the row at
+    # dispatch, after which the item is an ordinary single-file job.
+    variants: list[QueueVariantCreate] | None = None
 
 
 class PrintQueueItemUpdate(BaseModel):

backend/app/services/library_trash.py

@@ -27,6 +27,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.app.core.config import settings as app_settings
 from backend.app.core.database import async_session
 from backend.app.models.library import LibraryFile
+from backend.app.models.print_queue import PrintQueueVariant
 from backend.app.models.settings import Settings
 
 logger = logging.getLogger(__name__)
@@ -351,6 +352,7 @@ class LibraryTrashService:
         for row in rows:
             self._unlink_on_disk(row)
             deleted += 1
+        await delete_dependent_variants(db, [r.id for r in rows])
         # Single DELETE is faster than N await db.delete() round-trips; we
         # still need the Python loop above to unlink bytes on disk.
         await db.execute(delete(LibraryFile).where(LibraryFile.id.in_([r.id for r in rows])))
@@ -383,8 +385,25 @@ class LibraryTrashService:
     async def hard_delete_now(self, db: AsyncSession, file: LibraryFile) -> None:
         """Bypass retention and delete this trashed file + its bytes immediately."""
         self._unlink_on_disk(file)
+        await delete_dependent_variants(db, [file.id])
         await db.delete(file)
         await db.commit()
 
 
+async def delete_dependent_variants(db: AsyncSession, file_ids: list[int]) -> None:
+    """Drop cross-model queue candidates that pointed at these files (#671).
+
+    SQLite ships with ``PRAGMA foreign_keys`` off — verified, not assumed — so
+    the ON DELETE CASCADE on ``print_queue_variants.library_file_id`` never fires
+    on the default deployment and the rows would outlive the file.
+
+    The scheduler already refuses to dispatch a candidate whose file is missing
+    or trashed, so nothing prints wrongly without this. It is here so the table
+    does not fill with rows referencing files that no longer exist.
+    """
+    if not file_ids:
+        return
+    await db.execute(delete(PrintQueueVariant).where(PrintQueueVariant.library_file_id.in_(file_ids)))
+
+
 library_trash_service = LibraryTrashService()

backend/app/services/print_scheduler.py

@@ -205,6 +205,11 @@ def _candidates_for(item: PrintQueueItem) -> list[_ModelCandidate]:
     still gives up, it just does not give up without trying them.
     """
     if not item.variants:
+        if not item.archive_id and not item.library_file_id:
+            # Nothing to print at all. Dispatching would fail deep in the upload
+            # on "No archive_id or library_file_id"; the caller holds the item
+            # with an explanation instead.
+            return []
         return [
             _ModelCandidate(
                 target_model=item.target_model,
@@ -214,7 +219,14 @@ def _candidates_for(item: PrintQueueItem) -> list[_ModelCandidate]:
             )
         ]
 
-    ordered = sorted(item.variants, key=lambda v: (v.attempt_count or 0, v.position, v.id))
+    # Drop candidates whose file is gone or in the trash. Both are reachable and
+    # neither is covered by the schema: library deletes are soft (the row lives
+    # on with ``deleted_at`` set, which no foreign key can express), and SQLite
+    # ships with ``PRAGMA foreign_keys`` off, so the ON DELETE CASCADE never
+    # fires there and a hard delete leaves the variant row pointing at nothing.
+    usable = [v for v in item.variants if v.library_file is not None and v.library_file.deleted_at is None]
+
+    ordered = sorted(usable, key=lambda v: (v.attempt_count or 0, v.position, v.id))
     return [
         _ModelCandidate(
             target_model=v.target_model,
@@ -822,12 +834,10 @@ class PrintScheduler:
                     chosen: _ModelCandidate | None = None
                     per_model_reasons: list[tuple[str | None, str]] = []
 
-                    if not item.variants and not item.archive_id and not item.library_file_id:
-                        # Every candidate file was deleted out from under this item
-                        # (variant rows go with their library file). Dispatching would
-                        # fail deep in the upload with "No archive_id or library_file_id";
-                        # hold it here with something the user can act on instead.
-                        candidates = []
+                    if not candidates:
+                        # Every candidate file has been deleted or trashed out from
+                        # under this item. Hold it with something the user can act
+                        # on rather than letting it look dispatchable forever.
                         per_model_reasons.append(
                             (
                                 item.target_model,

backend/tests/integration/test_library_variants_api.py

@@ -0,0 +1,246 @@
+"""Integration tests for variant groups (#671 / #2570).
+
+A variant group is the user declaring that several sliced files are the same
+job for different printers. The endpoints exist to enforce what that statement
+has to mean before the scheduler acts on it without a human in the loop.
+"""
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.fixture
+async def sliced_file_factory(db_session):
+    """Create a sliced library file declaring the model it was sliced for."""
+    _counter = [0]
+
+    async def _create(model: str | None = "H2S", **kwargs):
+        from backend.app.models.library import LibraryFile
+
+        _counter[0] += 1
+        defaults = {
+            "filename": f"job_{_counter[0]}.gcode.3mf",
+            "file_path": f"/test/job_{_counter[0]}.gcode.3mf",
+            "file_size": 100,
+            "file_type": "gcode.3mf",
+            "file_metadata": {"sliced_for_model": model} if model else {},
+        }
+        defaults.update(kwargs)
+        f = LibraryFile(**defaults)
+        db_session.add(f)
+        await db_session.commit()
+        await db_session.refresh(f)
+        return f
+
+    return _create
+
+
+async def _create_group(client: AsyncClient, *file_ids: int, name: str | None = None):
+    payload = {"members": [{"library_file_id": fid} for fid in file_ids]}
+    if name:
+        payload["name"] = name
+    return await client.post("/api/v1/library/variant-groups", json=payload)
+
+
+class TestCreateVariantGroup:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_groups_two_slices_in_priority_order(self, async_client, sliced_file_factory):
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _create_group(async_client, h2s.id, h2c.id, name="bracket")
+        assert r.status_code == 201
+        body = r.json()
+        assert body["name"] == "bracket"
+        assert [m["target_model"] for m in body["members"]] == ["H2S", "H2C"]
+        assert [m["position"] for m in body["members"]] == [0, 1]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_model_is_read_from_the_file_not_the_caller(self, async_client, sliced_file_factory):
+        """The group never carries its own model data, so it cannot disagree with
+        the 3MFs. "Bambu Lab H2S" normalizes to the same H2S the scheduler matches."""
+        a = await sliced_file_factory("Bambu Lab H2S")
+        b = await sliced_file_factory("O1C")  # internal code for H2C
+
+        body = (await _create_group(async_client, a.id, b.id)).json()
+        assert [m["target_model"] for m in body["members"]] == ["H2S", "H2C"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_two_slices_for_the_same_printer_are_rejected(self, async_client, sliced_file_factory):
+        """Not alternatives — the resolver would have no basis to prefer one, and
+        the arbitrary pick would look like a bug the first time it chose wrong."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2S")
+
+        r = await _create_group(async_client, a.id, b.id)
+        assert r.status_code == 400
+        assert "different printers" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_normalization_catches_the_same_printer_spelled_differently(self, async_client, sliced_file_factory):
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("Bambu Lab H2S")
+
+        r = await _create_group(async_client, a.id, b.id)
+        assert r.status_code == 400
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_unsliced_file_cannot_be_a_variant(self, async_client, sliced_file_factory):
+        """A source .3mf has no G-code — it can never be dispatched to anything."""
+        sliced = await sliced_file_factory("H2S")
+        source = await sliced_file_factory(None, filename="model.3mf", file_type="3mf")
+
+        r = await _create_group(async_client, sliced.id, source.id)
+        assert r.status_code == 400
+        assert "not a sliced file" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_file_without_a_model_must_name_one(self, async_client, sliced_file_factory):
+        """Legacy 3MFs declare no model. Rather than guess, make the user say."""
+        known = await sliced_file_factory("H2S")
+        legacy = await sliced_file_factory(None)
+
+        r = await _create_group(async_client, known.id, legacy.id)
+        assert r.status_code == 400
+        assert "does not say which printer" in r.json()["detail"]
+
+        r = await async_client.post(
+            "/api/v1/library/variant-groups",
+            json={
+                "members": [
+                    {"library_file_id": known.id},
+                    {"library_file_id": legacy.id, "target_model": "H2C"},
+                ]
+            },
+        )
+        assert r.status_code == 201
+        assert [m["target_model"] for m in r.json()["members"]] == ["H2S", "H2C"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_a_file_belongs_to_one_group_only(self, async_client, sliced_file_factory):
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        c = await sliced_file_factory("H2D")
+        assert (await _create_group(async_client, a.id, b.id)).status_code == 201
+
+        r = await _create_group(async_client, a.id, c.id)
+        assert r.status_code == 409
+        assert "already belongs" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_single_member_is_rejected_by_the_schema(self, async_client, sliced_file_factory):
+        only = await sliced_file_factory("H2S")
+        r = await _create_group(async_client, only.id)
+        assert r.status_code == 422, "a group of one expresses no choice"
+
+
+class TestVariantGroupMembership:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_add_version_to_an_existing_group(self, async_client, sliced_file_factory):
+        """The common real case: the H2S version was queued last week, the H2C
+        version was sliced today."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        c = await sliced_file_factory("H2D")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.post(f"/api/v1/library/variant-groups/{gid}/members", json={"library_file_id": c.id})
+        assert r.status_code == 200
+        assert [m["target_model"] for m in r.json()["members"]] == ["H2S", "H2C", "H2D"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_added_member_cannot_duplicate_a_model(self, async_client, sliced_file_factory):
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        dupe = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.post(f"/api/v1/library/variant-groups/{gid}/members", json={"library_file_id": dupe.id})
+        assert r.status_code == 400
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_removing_down_to_one_dissolves_the_group(self, async_client, sliced_file_factory):
+        """A leftover one-member group would look like a choice and behave like an
+        ordinary job — worse than no group at all."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.delete(f"/api/v1/library/variant-groups/{gid}/members/{b.id}")
+        assert r.status_code == 204
+        assert (await async_client.get(f"/api/v1/library/variant-groups/{gid}")).status_code == 404
+        # ...and the survivor is still a perfectly good file.
+        assert (await async_client.get(f"/api/v1/library/variant-groups/by-file/{a.id}")).status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_removing_from_a_three_member_group_keeps_it(self, async_client, sliced_file_factory):
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        c = await sliced_file_factory("H2D")
+        gid = (await _create_group(async_client, a.id, b.id, c.id)).json()["id"]
+
+        assert (await async_client.delete(f"/api/v1/library/variant-groups/{gid}/members/{c.id}")).status_code == 204
+        body = (await async_client.get(f"/api/v1/library/variant-groups/{gid}")).json()
+        assert [m["library_file_id"] for m in body["members"]] == [a.id, b.id]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_deleting_a_group_keeps_the_files(self, async_client, sliced_file_factory):
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        assert (await async_client.delete(f"/api/v1/library/variant-groups/{gid}")).status_code == 204
+        for f in (a, b):
+            assert (await async_client.get(f"/api/v1/library/files/{f.id}")).status_code == 200
+
+
+class TestVariantGroupOrdering:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reorder_changes_priority(self, async_client, sliced_file_factory):
+        """Order is the user saying which printer they would rather have when both
+        are free, so it has to be editable."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.patch(f"/api/v1/library/variant-groups/{gid}", json={"member_file_ids": [b.id, a.id]})
+        assert r.status_code == 200
+        assert [m["target_model"] for m in r.json()["members"]] == ["H2C", "H2S"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_partial_reorder_is_rejected(self, async_client, sliced_file_factory):
+        """Listing a subset would leave the rest in an order nobody chose."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.patch(f"/api/v1/library/variant-groups/{gid}", json={"member_file_ids": [a.id]})
+        assert r.status_code == 400
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_lookup_by_file(self, async_client, sliced_file_factory):
+        """Both consumers start from a file: the print modal knows what was
+        clicked, the queue flow knows what was selected."""
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2C")
+        gid = (await _create_group(async_client, a.id, b.id)).json()["id"]
+
+        r = await async_client.get(f"/api/v1/library/variant-groups/by-file/{b.id}")
+        assert r.status_code == 200
+        assert r.json()["id"] == gid

backend/tests/integration/test_queue_variants_api.py

@@ -0,0 +1,245 @@
+"""Queueing a job with cross-model alternatives (#671).
+
+One queue item, several sliced files, whichever printer frees up first. The
+create endpoint's job is to refuse candidate sets that cannot mean what the user
+intends, because after this point the scheduler dispatches to hardware with no
+human in the loop.
+"""
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+
+@pytest.fixture
+async def sliced_file_factory(db_session):
+    _counter = [0]
+
+    async def _create(model: str | None = "H2S", **kwargs):
+        from backend.app.models.library import LibraryFile
+
+        _counter[0] += 1
+        defaults = {
+            "filename": f"job_{_counter[0]}.gcode.3mf",
+            "file_path": f"/test/job_{_counter[0]}.gcode.3mf",
+            "file_size": 100,
+            "file_type": "gcode.3mf",
+            "file_metadata": {"sliced_for_model": model} if model else {},
+        }
+        defaults.update(kwargs)
+        f = LibraryFile(**defaults)
+        db_session.add(f)
+        await db_session.commit()
+        await db_session.refresh(f)
+        return f
+
+    return _create
+
+
+async def _queue_variants(client: AsyncClient, *file_ids: int, **extra):
+    payload = {"variants": [{"library_file_id": fid} for fid in file_ids]}
+    payload.update(extra)
+    return await client.post("/api/v1/queue/", json=payload)
+
+
+async def _variants_of(db_session, item_id: int):
+    from backend.app.models.print_queue import PrintQueueVariant
+
+    rows = (
+        (
+            await db_session.execute(
+                select(PrintQueueVariant)
+                .where(PrintQueueVariant.queue_item_id == item_id)
+                .order_by(PrintQueueVariant.position)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    return rows
+
+
+class TestQueueWithVariants:
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_creates_one_item_with_a_candidate_per_file(
+        self, async_client, db_session, sliced_file_factory, printer_factory
+    ):
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id)
+        assert r.status_code == 200
+        item_id = r.json()["id"]
+
+        variants = await _variants_of(db_session, item_id)
+        assert [v.target_model for v in variants] == ["H2S", "H2C"]
+        assert [v.position for v in variants] == [0, 1]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_the_item_holds_no_file_of_its_own(
+        self, async_client, db_session, sliced_file_factory, printer_factory
+    ):
+        """library_file_id is ON DELETE CASCADE. Pointing it at one candidate
+        would mean deleting that single alternative destroys the whole job."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        item_id = (await _queue_variants(async_client, h2s.id, h2c.id)).json()["id"]
+        item = (await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))).scalar_one()
+        assert item.library_file_id is None
+        assert item.archive_id is None
+        assert item.target_model == "H2S", "mirrors the first candidate so the card has a label"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_deleting_one_candidate_leaves_the_job_and_its_sibling(
+        self, async_client, db_session, sliced_file_factory, printer_factory
+    ):
+        from backend.app.models.print_queue import PrintQueueItem
+
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+        item_id = (await _queue_variants(async_client, h2s.id, h2c.id)).json()["id"]
+
+        # Trash, then permanently delete — the only path that actually removes
+        # the row. SQLite has PRAGMA foreign_keys off, so nothing cleans the
+        # candidate up on its own.
+        assert (await async_client.delete(f"/api/v1/library/files/{h2s.id}")).status_code == 200
+        assert (await async_client.delete(f"/api/v1/library/trash/{h2s.id}")).status_code == 200
+
+        item = (
+            await db_session.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
+        ).scalar_one_or_none()
+        assert item is not None, "the job survives losing one alternative"
+        remaining = await _variants_of(db_session, item_id)
+        assert [v.target_model for v in remaining] == ["H2C"], "no row left pointing at a deleted file"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_rejects_a_specific_printer(self, async_client, sliced_file_factory, printer_factory):
+        """Naming a printer defeats the entire purpose of offering alternatives."""
+        printer = await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id, printer_id=printer.id)
+        assert r.status_code == 400
+        assert "printer_id" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_rejects_a_file_alongside_the_variants(self, async_client, sliced_file_factory, printer_factory):
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+        other = await sliced_file_factory("H2D")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id, library_file_id=other.id)
+        assert r.status_code == 400
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_rejects_two_candidates_for_the_same_printer(
+        self, async_client, sliced_file_factory, printer_factory
+    ):
+        await printer_factory(model="H2S")
+        a = await sliced_file_factory("H2S")
+        b = await sliced_file_factory("H2S")
+
+        r = await _queue_variants(async_client, a.id, b.id)
+        assert r.status_code == 400
+        assert "different printers" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_rejects_the_same_file_twice(self, async_client, sliced_file_factory, printer_factory):
+        await printer_factory(model="H2S")
+        f = await sliced_file_factory("H2S")
+
+        r = await _queue_variants(async_client, f.id, f.id)
+        assert r.status_code == 400
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cross_model_gate_applies_to_every_candidate(
+        self, async_client, sliced_file_factory, printer_factory
+    ):
+        """A set is only as safe as its worst member."""
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        good = await sliced_file_factory("H2S")
+        # Declares X1C but is offered as an H2C candidate.
+        bad = await sliced_file_factory("X1C")
+
+        r = await async_client.post(
+            "/api/v1/queue/",
+            json={
+                "variants": [
+                    {"library_file_id": good.id},
+                    {"library_file_id": bad.id, "target_model": "H2C"},
+                ]
+            },
+        )
+        assert r.status_code == 400
+        assert "sliced for X1C" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_one_candidate_without_a_printer_is_allowed(
+        self, async_client, db_session, sliced_file_factory, printer_factory
+    ):
+        """Slicing for the H2C before the H2C arrives is reasonable. Refusing the
+        whole queue action over it would be worse than that candidate simply
+        never matching."""
+        await printer_factory(model="H2S")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id)
+        assert r.status_code == 200
+        assert len(await _variants_of(db_session, r.json()["id"])) == 2
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_rejected_when_no_candidate_has_a_printer(self, async_client, sliced_file_factory):
+        """Nothing in the set can ever run — that is a job that waits forever."""
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id)
+        assert r.status_code == 400
+        assert "No active printers" in r.json()["detail"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_quantity_gives_each_copy_its_own_candidates(
+        self, async_client, db_session, sliced_file_factory, printer_factory
+    ):
+        """Attempt counts are per-item, and two copies must be free to land on
+        different printers."""
+        from backend.app.models.print_queue import PrintQueueItem, PrintQueueVariant
+
+        await printer_factory(model="H2S")
+        await printer_factory(model="H2C")
+        h2s = await sliced_file_factory("H2S")
+        h2c = await sliced_file_factory("H2C")
+
+        r = await _queue_variants(async_client, h2s.id, h2c.id, quantity=3)
+        assert r.status_code == 200
+
+        item_ids = (await db_session.execute(select(PrintQueueItem.id))).scalars().all()
+        assert len(item_ids) == 3
+        total = (await db_session.execute(select(PrintQueueVariant))).scalars().all()
+        assert len(total) == 6

backend/tests/unit/test_scheduler_cross_model_variants.py

@@ -37,13 +37,18 @@ from backend.app.services.print_scheduler import (
 # ---------------------------------------------------------------------------
 
 
-def _fake_variant(*, vid, position, model, attempts=0):
+def _fake_variant(*, vid, position, model, attempts=0, trashed=False, file_missing=False):
     return SimpleNamespace(
         id=vid,
         position=position,
         target_model=model,
         attempt_count=attempts,
-        library_file=SimpleNamespace(file_metadata={"sliced_for_model": model}),
+        library_file=None
+        if file_missing
+        else SimpleNamespace(
+            file_metadata={"sliced_for_model": model},
+            deleted_at="2026-01-01" if trashed else None,
+        ),
         required_filament_types=None,
         filament_overrides=None,
     )
@@ -54,7 +59,9 @@ def _fake_item(variants):
         variants=variants,
         target_model=None,
         archive=None,
+        archive_id=None,
         library_file=None,
+        library_file_id=None,
         required_filament_types=None,
         filament_overrides=None,
     )
@@ -67,6 +74,8 @@ def test_no_variants_yields_the_items_own_columns():
         variants=[],
         target_model="H2D",
         archive=None,
+        archive_id=None,
+        library_file_id=7,
         library_file=SimpleNamespace(file_metadata={"sliced_for_model": "H2D"}),
         required_filament_types='["PLA"]',
         filament_overrides=None,
@@ -101,6 +110,36 @@ def test_least_attempted_candidate_is_tried_first():
     assert [c.target_model for c in _candidates_for(item)] == ["H2C", "H2S"]
 
 
+def test_trashed_candidate_is_skipped():
+    """Library deletes are soft: the row survives with deleted_at set, which no
+    foreign key can express. Dispatching a file the user put in the bin would be
+    a genuine surprise."""
+    item = _fake_item(
+        [
+            _fake_variant(vid=1, position=0, model="H2S", trashed=True),
+            _fake_variant(vid=2, position=1, model="H2C"),
+        ]
+    )
+    assert [c.target_model for c in _candidates_for(item)] == ["H2C"]
+
+
+def test_orphaned_candidate_is_skipped():
+    """SQLite runs with PRAGMA foreign_keys off, so a hard delete can leave a
+    candidate row pointing at nothing."""
+    item = _fake_item(
+        [
+            _fake_variant(vid=1, position=0, model="H2S", file_missing=True),
+            _fake_variant(vid=2, position=1, model="H2C"),
+        ]
+    )
+    assert [c.target_model for c in _candidates_for(item)] == ["H2C"]
+
+
+def test_item_with_no_usable_candidates_yields_none():
+    item = _fake_item([_fake_variant(vid=1, position=0, model="H2S", trashed=True)])
+    assert _candidates_for(item) == []
+
+
 def test_equal_attempts_fall_back_to_priority():
     """Once every candidate has failed equally often they cycle in the user's
     order, so the item still reaches its DISPATCH_MAX_ATTEMPTS ceiling."""