diff --git a/backend/app/api/routes/inventory.py b/backend/app/api/routes/inventory.py
index 1b75696213..51d1c1aa7b 100644
--- a/backend/app/api/routes/inventory.py
+++ b/backend/app/api/routes/inventory.py
@@ -26,6 +26,8 @@
 from backend.app.models.spool_k_profile import SpoolKProfile
 from backend.app.models.user import User
 from backend.app.schemas.spool import (
+    PendingSlotAssignmentCreateRequest,
+    PendingSlotAssignmentResponse,
     SpoolAssignmentCreate,
     SpoolAssignmentResponse,
     SpoolBulkCreate,
@@ -1494,6 +1496,70 @@ async def unassign_spool(
     return {"status": "deleted"}
 
 
+# ── Pending Slot Assignments (assign-on-next-slot) ────────────────────────────
+
+
+@router.post("/spools/assign-on-next-slot", response_model=PendingSlotAssignmentResponse)
+async def assign_on_next_slot(
+    body: PendingSlotAssignmentCreateRequest,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
+):
+    """Create a pending spool-to-slot assignment request."""
+    from backend.app.services.pending_slot_assignment import create_pending_assignment
+
+    logger.info(
+        "Received assign-on-next-slot request: spool_id=%d tray_uuid=%s tag_uid=%s source=%s",
+        body.spool_id,
+        body.tray_uuid,
+        body.tag_uid,
+        body.source,
+    )
+    assignment = await create_pending_assignment(
+        db=db,
+        tray_uuid=body.tray_uuid,
+        tag_uid=body.tag_uid,
+        spool_id=body.spool_id,
+        printer_id=body.printer_id,
+        source=body.source,
+        timeout_seconds=body.timeout,
+    )
+    return PendingSlotAssignmentResponse.from_model(pending_assignment=assignment)
+
+
+@router.get("/spools/assignments/{assignment_id}", response_model=PendingSlotAssignmentResponse)
+async def get_pending_assignment_status(
+    assignment_id: int,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
+):
+    """Get the status of a pending slot assignment."""
+    from backend.app.services.pending_slot_assignment import get_pending_assignment
+
+    assignment = await get_pending_assignment(db=db, assignment_id=assignment_id)
+    if not assignment:
+        raise HTTPException(status_code=404, detail="Assignment not found")
+    return PendingSlotAssignmentResponse.from_model(pending_assignment=assignment)
+
+
+@router.delete("/spools/assignments/{assignment_id}", response_model=PendingSlotAssignmentResponse)
+async def cancel_pending_assignment_endpoint(
+    assignment_id: int,
+    db: AsyncSession = Depends(get_db),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_UPDATE),
+):
+    """Cancel a pending slot assignment. Only pending assignments can be cancelled."""
+    from backend.app.services.pending_slot_assignment import cancel_pending_assignment
+
+    assignment = await cancel_pending_assignment(db=db, assignment_id=assignment_id)
+    if not assignment:
+        raise HTTPException(
+            status_code=404,
+            detail="Assignment not found or not in pending status",
+        )
+    return PendingSlotAssignmentResponse.from_model(pending_assignment=assignment)
+
+
 # ── Tag Linking ───────────────────────────────────────────────────────────────
 
 
diff --git a/backend/app/core/database.py b/backend/app/core/database.py
index ec8f60ed91..39d9869e95 100644
--- a/backend/app/core/database.py
+++ b/backend/app/core/database.py
@@ -187,6 +187,7 @@ async def init_db():
         notification_template,
         oidc_provider,
         orca_base_cache,
+        pending_slot_assignment,
         pending_upload,
         print_batch,
         print_log,
diff --git a/backend/app/main.py b/backend/app/main.py
index 4b0e7a381a..995bb39829 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -1668,6 +1668,35 @@ async def on_ams_change(printer_id: int, ams_data: list):
                                 )
                             continue
 
+                        # Check pending assign-on-next-slot requests before auto-assign
+                        try:
+                            from backend.app.services.pending_slot_assignment import (
+                                try_complete_pending_assignments,
+                            )
+
+                            if await try_complete_pending_assignments(
+                                db=db,
+                                printer_id=printer_id,
+                                ams_id=ams_id,
+                                tray_id=tray_id,
+                                slot_tray_uuid=tray_uuid or None,
+                                slot_tag_uid=tag_uid or None,
+                            ):
+                                logger.info(
+                                    "Pending slot assignment completed for printer %d AMS%d-T%d",
+                                    printer_id,
+                                    ams_id,
+                                    tray_id,
+                                )
+                                continue
+                        except Exception:
+                            logger.exception(
+                                "Pending slot assignment check failed for printer %d AMS%d-T%d",
+                                printer_id,
+                                ams_id,
+                                tray_id,
+                            )
+
                         if is_bambu_tag(tag_uid, tray_uuid, tray_info_idx):
                             # BL spool with RFID tag: auto-match → inventory match → auto-create
                             spool = await get_spool_by_tag(db, tag_uid, tray_uuid)
@@ -5272,6 +5301,11 @@ async def lifespan(app: FastAPI):
     # Rehydrate persisted awaiting-plate-clear gate (#961) so prompts survive restarts
     await printer_manager.load_awaiting_plate_clear_from_db()
 
+    # Restore pending slot assignment timeout tasks that survived a restart
+    from backend.app.services.pending_slot_assignment import restore_pending_timeouts
+
+    await restore_pending_timeouts()
+
     # Layer change callback for external camera timelapse
     async def on_layer_change(printer_id: int, layer_num: int):
         """Capture timelapse frame on layer change + first layer notification."""
diff --git a/backend/app/models/pending_slot_assignment.py b/backend/app/models/pending_slot_assignment.py
new file mode 100644
index 0000000000..66243559c9
--- /dev/null
+++ b/backend/app/models/pending_slot_assignment.py
@@ -0,0 +1,57 @@
+"""Model for pending spool-to-slot assignment requests.
+
+A pending assignment represents a request to assign a spool to the next AMS slot
+that transitions from empty → filled. The backend monitors AMS events and completes
+the assignment automatically when a matching slot change is detected.
+"""
+
+from datetime import datetime, timezone
+
+from sqlalchemy import DateTime, Float, Integer, String, func
+from sqlalchemy.orm import Mapped, mapped_column
+
+from backend.app.core.database import Base
+
+
+class PendingSlotAssignment(Base):
+    """A pending request to assign a spool to the next available AMS slot."""
+
+    __tablename__ = "pending_slot_assignment"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    # Spool identifiers — at least one must be non-null
+    tray_uuid: Mapped[str | None] = mapped_column(String(64), index=True)
+    tag_uid: Mapped[str | None] = mapped_column(String(32), index=True)
+    # Resolved spool_id once matched (nullable until resolved)
+    spool_id: Mapped[int | None] = mapped_column(Integer)
+    # Target printer (NULL = any printer)
+    printer_id: Mapped[int | None] = mapped_column(Integer)
+    # Source of the request
+    source: Mapped[str] = mapped_column(String(20))  # "nfc" | "qr" | "spoolbuddy"
+    # Status: pending, completed, timed_out, cancelled
+    status: Mapped[str] = mapped_column(String(20), default="pending", index=True)
+    # Timeout in seconds
+    timeout_seconds: Mapped[int] = mapped_column(Integer, default=300)
+
+    # Completion details (filled when status transitions to completed)
+    assigned_printer_id: Mapped[int | None] = mapped_column(Integer)
+    assigned_ams_id: Mapped[int | None] = mapped_column(Integer)
+    assigned_tray_id: Mapped[int | None] = mapped_column(Integer)
+    completed_at: Mapped[datetime | None] = mapped_column(DateTime)
+
+    # Metrics
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    time_to_placement: Mapped[float | None] = mapped_column(Float)  # seconds
+
+    @property
+    def lookup_key(self) -> str:
+        """Return the primary identifier used for spool lookup (tray_uuid preferred)."""
+        return self.tray_uuid or self.tag_uid or ""
+
+    @property
+    def is_expired(self) -> bool:
+        """Check if this pending assignment has timed out."""
+        if self.status != "pending":
+            return False
+        elapsed = (datetime.now(timezone.utc) - self.created_at.replace(tzinfo=timezone.utc)).total_seconds()
+        return elapsed > self.timeout_seconds
diff --git a/backend/app/schemas/pending_slot_assignment.py b/backend/app/schemas/pending_slot_assignment.py
new file mode 100644
index 0000000000..50f7b2bd90
--- /dev/null
+++ b/backend/app/schemas/pending_slot_assignment.py
@@ -0,0 +1,44 @@
+"""Pydantic schemas for the pending slot assignment API."""
+
+from datetime import datetime
+
+from pydantic import BaseModel, model_validator
+
+
+class PendingSlotAssignmentCreate(BaseModel):
+    """Request body for POST /api/v1/inventory/spools/assign-on-next-slot."""
+
+    tray_uuid: str | None = None
+    tag_uid: str | None = None
+    printer_id: int | None = None
+    source: str
+    timeout: int
+
+    @model_validator(mode="after")
+    def _require_at_least_one_identifier(self) -> "PendingSlotAssignmentCreate":
+        if not self.tray_uuid and not self.tag_uid:
+            raise ValueError("At least one of tray_uuid or tag_uid must be provided.")
+        return self
+
+
+class PendingSlotAssignmentResponse(BaseModel):
+    """Response body for assignment endpoints."""
+
+    assignment_id: int
+    tray_uuid: str | None = None
+    tag_uid: str | None = None
+    spool_id: int | None = None
+    printer_id: int | None = None
+    source: str
+    status: str  # pending, completed, timed_out, cancelled
+    timeout_seconds: int
+    # Completion details
+    assigned_printer_id: int | None = None
+    assigned_ams_id: int | None = None
+    assigned_tray_id: int | None = None
+    completed_at: datetime | None = None
+    time_to_placement: float | None = None
+    created_at: datetime
+
+    class Config:
+        from_attributes = True
diff --git a/backend/app/schemas/spool.py b/backend/app/schemas/spool.py
index 3d5a90e687..36471a0f19 100644
--- a/backend/app/schemas/spool.py
+++ b/backend/app/schemas/spool.py
@@ -125,6 +125,7 @@ def _validate_effect_type(cls, v: str | None) -> str | None:
     # assignment). Column has lived on the ORM since the inventory rework
     # but was missing from this schema, so writes were silently dropped (#1291).
     storage_location: str | None = Field(default=None, max_length=255)
+    location_id: int | None = Field(default=None, gt=0)
 
 
 class SpoolCreate(SpoolBase):
@@ -174,6 +175,7 @@ def _validate_effect_type(cls, v: str | None) -> str | None:
     category: str | None = Field(default=None, max_length=50)
     low_stock_threshold_pct: int | None = Field(default=None, ge=1, le=99)
     storage_location: str | None = Field(default=None, max_length=255)
+    location_id: int | None = Field(default=None, gt=0)
 
 
 class SpoolKProfileBase(BaseModel):
@@ -243,3 +245,55 @@ class SpoolAssignmentResponse(BaseModel):
 
     class Config:
         from_attributes = True
+
+
+class PendingSlotAssignmentCreateRequest(BaseModel):
+    """Pending create response."""
+
+    spool_id: int | None = None
+    tray_uuid: str | None = None
+    tag_uid: str | None = None
+    printer_id: int | None = None
+    source: str
+    timeout: int
+
+
+class PendingSlotAssignmentResponse(BaseModel):
+    """Response body for pending assignment endpoints."""
+
+    assignment_id: int
+    tray_uuid: str | None = None
+    tag_uid: str | None = None
+    spool_id: int | None = None
+    printer_id: int | None = None
+    source: str
+    status: str  # pending, completed, timed_out, cancelled
+    timeout_seconds: int
+    assigned_printer_id: int | None = None
+    assigned_ams_id: int | None = None
+    assigned_tray_id: int | None = None
+    completed_at: str | None = None
+    time_to_placement: float | None = None
+    created_at: str
+
+    class Config:
+        from_attributes = True
+
+    @staticmethod
+    def from_model(pending_assignment) -> "PendingSlotAssignmentResponse":
+        return PendingSlotAssignmentResponse(
+            assignment_id=pending_assignment.id,
+            tray_uuid=pending_assignment.tray_uuid,
+            tag_uid=pending_assignment.tag_uid,
+            spool_id=pending_assignment.spool_id,
+            printer_id=pending_assignment.printer_id,
+            source=pending_assignment.source,
+            status=pending_assignment.status,
+            timeout_seconds=pending_assignment.timeout_seconds,
+            assigned_printer_id=pending_assignment.assigned_printer_id,
+            assigned_ams_id=pending_assignment.assigned_ams_id,
+            assigned_tray_id=pending_assignment.assigned_tray_id,
+            completed_at=pending_assignment.completed_at.isoformat() if pending_assignment.completed_at else None,
+            time_to_placement=pending_assignment.time_to_placement,
+            created_at=pending_assignment.created_at.isoformat() if pending_assignment.created_at else "",
+        )
diff --git a/backend/app/services/mqtt_relay.py b/backend/app/services/mqtt_relay.py
index ccd3a7c224..bcef283c13 100644
--- a/backend/app/services/mqtt_relay.py
+++ b/backend/app/services/mqtt_relay.py
@@ -687,6 +687,56 @@ async def on_smart_plug_energy(
             },
         )
 
+    async def on_slot_assignment_completed(
+        self,
+        assignment_id: int,
+        tray_uuid: str | None,
+        tag_uid: str | None,
+        spool_id: int,
+        printer_id: int,
+        ams_id: int,
+        tray_id: int,
+        time_to_placement: float,
+    ):
+        """Publish event when a pending slot assignment is completed."""
+        if not self.enabled or not self.connected:
+            return
+
+        self._publish(
+            f"{self.topic_prefix}/inventory/slot_assignment/completed",
+            {
+                "assignment_id": assignment_id,
+                "tray_uuid": tray_uuid,
+                "tag_uid": tag_uid,
+                "spool_id": spool_id,
+                "printer_id": printer_id,
+                "ams_id": ams_id,
+                "tray_id": tray_id,
+                "time_to_placement": time_to_placement,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+
+    async def on_slot_assignment_timed_out(
+        self,
+        assignment_id: int,
+        tray_uuid: str | None,
+        tag_uid: str | None,
+    ):
+        """Publish event when a pending slot assignment times out."""
+        if not self.enabled or not self.connected:
+            return
+
+        self._publish(
+            f"{self.topic_prefix}/inventory/slot_assignment/timed_out",
+            {
+                "assignment_id": assignment_id,
+                "tray_uuid": tray_uuid,
+                "tag_uid": tag_uid,
+                "timestamp": datetime.now(timezone.utc).isoformat(),
+            },
+        )
+
 
 # Global instance
 mqtt_relay = MQTTRelayService()
diff --git a/backend/app/services/pending_slot_assignment.py b/backend/app/services/pending_slot_assignment.py
new file mode 100644
index 0000000000..c31965aaa3
--- /dev/null
+++ b/backend/app/services/pending_slot_assignment.py
@@ -0,0 +1,455 @@
+"""Service layer for pending slot assignments.
+
+Manages the lifecycle of assign-on-next-slot requests: creation, completion on
+AMS slot transitions, timeout expiry, cancellation, and event broadcasting.
+"""
+
+import asyncio
+import logging
+import re
+from datetime import datetime, timezone
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.core.websocket import ws_manager
+from backend.app.models.pending_slot_assignment import PendingSlotAssignment
+from backend.app.models.spool import Spool
+
+logger = logging.getLogger(__name__)
+
+# In-memory set of active timeout tasks keyed by assignment ID so they can be
+# cancelled if the assignment completes or is explicitly cancelled.
+_timeout_tasks: dict[int, asyncio.Task] = {}
+
+
+# Firmware reports these sentinel values for "no tag" / "no UUID" — treat
+# them the same as None / empty-string when checking slot identity.
+def _is_valid_identifier(value: str | None) -> bool:
+    """Return True if the identifier is non-empty and not an all-zeros sentinel."""
+    if not value:
+        return False
+    return re.fullmatch(r"0+", value) is None
+
+
+async def _resolve_spool(
+    db: AsyncSession, spool_id: int | None, tray_uuid: str | None, tag_uid: str | None
+) -> Spool | None:
+    """Find an inventory spool by spool_id (preferred), tray_uuid, or tag_uid (fallback)."""
+    if spool_id:
+        result = await db.execute(select(Spool).where(Spool.id == spool_id, Spool.archived_at.is_(None)))
+        spool = result.scalar_one_or_none()
+        if spool:
+            return spool
+    if tray_uuid:
+        result = await db.execute(
+            select(Spool).where(
+                func.upper(Spool.tray_uuid) == tray_uuid.upper(),
+                Spool.archived_at.is_(None),
+            )
+        )
+        spool = result.scalar_one_or_none()
+        if spool:
+            return spool
+    if tag_uid:
+        result = await db.execute(
+            select(Spool).where(
+                func.upper(Spool.tag_uid) == tag_uid.upper(),
+                Spool.archived_at.is_(None),
+            )
+        )
+        return result.scalar_one_or_none()
+    return None
+
+
+async def create_pending_assignment(
+    db: AsyncSession,
+    *,
+    tray_uuid: str | None,
+    tag_uid: str | None,
+    printer_id: int | None,
+    spool_id: int | None,
+    source: str,
+    timeout_seconds: int,
+) -> PendingSlotAssignment:
+    """Create a new pending assignment or return existing one for idempotency.
+
+    Idempotency is based on tray_uuid/tag_uid: if a pending assignment already
+    exists for the same spool identifiers, the existing one is returned.
+    """
+
+    # Build filters to match an existing pending assignment by identifiers
+    id_filters = []
+    if spool_id:
+        id_filters.append(PendingSlotAssignment.spool_id == spool_id)
+    if tray_uuid:
+        id_filters.append(func.upper(PendingSlotAssignment.tray_uuid) == tray_uuid.upper())
+    if tag_uid:
+        id_filters.append(func.upper(PendingSlotAssignment.tag_uid) == tag_uid.upper())
+
+    # Check for an existing pending assignment for the same spool
+    # to avoid assigning the same spool to multiple slots concurrently.
+    if id_filters:
+        existing_spool = await db.execute(
+            select(PendingSlotAssignment).where(
+                PendingSlotAssignment.status == "pending",
+                or_(*id_filters),
+            )
+        )
+        found_spool = existing_spool.scalar_one_or_none()
+        if found_spool:
+            return found_spool
+
+    # Resolve spool_id from tray_uuid or tag_uid
+    spool = await _resolve_spool(db=db, spool_id=spool_id, tray_uuid=tray_uuid, tag_uid=tag_uid)
+
+    assignment = PendingSlotAssignment(
+        tray_uuid=tray_uuid,
+        tag_uid=tag_uid,
+        spool_id=spool.id if spool else None,
+        printer_id=printer_id,
+        source=source,
+        status="pending",
+        timeout_seconds=timeout_seconds,
+    )
+    db.add(assignment)
+    await db.commit()
+    await db.refresh(assignment)
+
+    # Schedule timeout task
+    _schedule_timeout(assignment_id=assignment.id, timeout_seconds=timeout_seconds)
+
+    logger.info(
+        "Created pending slot assignment %d for spool_id=%d tray_uuid=%s tag_uid=%s printer=%s source=%s timeout=%ds",
+        assignment.id,
+        spool_id,
+        tray_uuid,
+        tag_uid,
+        printer_id,
+        source,
+        timeout_seconds,
+    )
+
+    return assignment
+
+
+def _schedule_timeout(assignment_id: int, timeout_seconds: int) -> None:
+    """Schedule an asyncio task to expire the assignment after timeout."""
+    task = asyncio.ensure_future(_expire_assignment(assignment_id=assignment_id, timeout_seconds=timeout_seconds))
+    _timeout_tasks[assignment_id] = task
+    task.add_done_callback(lambda _: _timeout_tasks.pop(assignment_id, None))
+
+
+async def _expire_assignment(assignment_id: int, timeout_seconds: int) -> None:
+    """Wait for timeout and then mark the assignment as timed_out."""
+    await asyncio.sleep(timeout_seconds)
+
+    from backend.app.core.database import async_session
+
+    async with async_session() as db:
+        result = await db.execute(
+            select(PendingSlotAssignment).where(
+                PendingSlotAssignment.id == assignment_id,
+                PendingSlotAssignment.status == "pending",
+            )
+        )
+        assignment = result.scalar_one_or_none()
+        if assignment:
+            assignment.status = "timed_out"
+            await db.commit()
+
+            logger.info("Pending slot assignment %d timed out", assignment_id)
+
+            # Notify clients via WebSocket
+            await ws_manager.broadcast(
+                {
+                    "type": "pending_slot_assignment_timed_out",
+                    "assignment_id": assignment_id,
+                    "tray_uuid": assignment.tray_uuid,
+                    "tag_uid": assignment.tag_uid,
+                }
+            )
+
+            # Notify via MQTT relay
+            try:
+                from backend.app.services.mqtt_relay import mqtt_relay
+
+                await mqtt_relay.on_slot_assignment_timed_out(
+                    assignment_id=assignment_id,
+                    tray_uuid=assignment.tray_uuid,
+                    tag_uid=assignment.tag_uid,
+                )
+            except Exception:
+                pass
+
+
+async def cancel_pending_assignment(db: AsyncSession, assignment_id: int) -> PendingSlotAssignment | None:
+    """Cancel a pending assignment. Returns the updated record or None if not found."""
+    result = await db.execute(
+        select(PendingSlotAssignment).where(
+            PendingSlotAssignment.id == assignment_id,
+            PendingSlotAssignment.status == "pending",
+        )
+    )
+    assignment = result.scalar_one_or_none()
+    if not assignment:
+        return None
+
+    assignment.status = "cancelled"
+    await db.commit()
+
+    # Cancel the timeout task
+    task = _timeout_tasks.pop(assignment_id, None)
+    if task and not task.done():
+        task.cancel()
+
+    logger.info("Cancelled pending slot assignment %d", assignment_id)
+
+    await ws_manager.broadcast(
+        {
+            "type": "pending_slot_assignment_cancelled",
+            "assignment_id": assignment_id,
+            "tray_uuid": assignment.tray_uuid,
+            "tag_uid": assignment.tag_uid,
+        }
+    )
+
+    return assignment
+
+
+async def get_pending_assignment(db: AsyncSession, assignment_id: int) -> PendingSlotAssignment | None:
+    """Get a pending assignment by ID."""
+    result = await db.execute(select(PendingSlotAssignment).where(PendingSlotAssignment.id == assignment_id))
+    return result.scalar_one_or_none()
+
+
+def _identifiers_match(
+    assignment: PendingSlotAssignment,
+    slot_tray_uuid: str | None,
+    slot_tag_uid: str | None,
+) -> bool:
+    """Return True if the pending assignment's identifiers match the slot's live spool.
+
+    Matching rules (case-insensitive):
+    - If the assignment has a tray_uuid AND the slot reports one, they must match.
+    - If the assignment has a tag_uid AND the slot reports one, they must match.
+    - At least one identifier must be present on both sides for a positive match.
+    - All-zeros sentinels are treated as absent (not a valid identifier).
+    """
+    matched = False
+
+    has_assignment_uuid = _is_valid_identifier(assignment.tray_uuid)
+    has_slot_uuid = _is_valid_identifier(slot_tray_uuid)
+    has_assignment_tag = _is_valid_identifier(assignment.tag_uid)
+    has_slot_tag = _is_valid_identifier(slot_tag_uid)
+
+    if has_assignment_uuid and has_slot_uuid:
+        if assignment.tray_uuid.upper() == slot_tray_uuid.upper():
+            matched = True
+        else:
+            return False  # Explicit mismatch
+
+    if has_assignment_tag and has_slot_tag:
+        if assignment.tag_uid.upper() == slot_tag_uid.upper():
+            matched = True
+        else:
+            return False  # Explicit mismatch
+
+    return matched
+
+
+async def try_complete_pending_assignments(
+    db: AsyncSession,
+    printer_id: int,
+    ams_id: int,
+    tray_id: int,
+    *,
+    slot_tray_uuid: str | None = None,
+    slot_tag_uid: str | None = None,
+) -> bool:
+    """Check if any pending assignment should be completed for this slot.
+
+    Called from the AMS change handler when a slot transitions from empty → filled.
+    Returns True if a pending assignment was completed.
+
+    The slot_tray_uuid / slot_tag_uid come from the live AMS tray data and are
+    used to verify that the physically-inserted spool matches the pending
+    request — preventing silent inventory corruption when a user inserts a
+    different spool than the one they scanned.
+    """
+    # Find pending assignments that match this printer (or any printer)
+    result = await db.execute(
+        select(PendingSlotAssignment)
+        .where(
+            PendingSlotAssignment.status == "pending",
+            ((PendingSlotAssignment.printer_id == printer_id) | (PendingSlotAssignment.printer_id.is_(None))),
+        )
+        .order_by(PendingSlotAssignment.created_at.asc())
+    )
+    pending_assignments = result.scalars().all()
+
+    if not pending_assignments:
+        return False
+
+    # Find the first pending assignment whose identifiers match the slot's
+    # live spool. If the slot provides tray_uuid or tag_uid we require a match;
+    # if the slot has no identifiers (3rd-party spool / no RFID) fall back to
+    # FIFO for backwards compatibility.
+    assignment: PendingSlotAssignment | None = None
+    slot_has_identifier = _is_valid_identifier(slot_tray_uuid) or _is_valid_identifier(slot_tag_uid)
+
+    if slot_has_identifier:
+        for candidate in pending_assignments:
+            if _identifiers_match(assignment=candidate, slot_tray_uuid=slot_tray_uuid, slot_tag_uid=slot_tag_uid):
+                assignment = candidate
+                break
+        if assignment is None:
+            # No pending assignment matches the spool actually inserted
+            logger.info(
+                "No pending assignment matches slot identifiers "
+                "(slot_tray_uuid=%s slot_tag_uid=%s) for printer %d AMS%d-T%d, skipping",
+                slot_tray_uuid,
+                slot_tag_uid,
+                printer_id,
+                ams_id,
+                tray_id,
+            )
+            return False
+    else:
+        # No slot identifier available (generic/3rd-party spool) — FIFO
+        assignment = pending_assignments[0]
+
+    # Resolve the spool if not already resolved
+    spool_id = assignment.spool_id
+    if not spool_id:
+        spool = await _resolve_spool(db=db, spool_id=None, tray_uuid=assignment.tray_uuid, tag_uid=assignment.tag_uid)
+        if spool:
+            spool_id = spool.id
+            assignment.spool_id = spool_id
+
+    if not spool_id:
+        logger.warning(
+            "Pending assignment %d: no inventory spool found for tray_uuid=%s tag_uid=%s, skipping",
+            assignment.id,
+            assignment.tray_uuid,
+            assignment.tag_uid,
+        )
+        return False
+
+    # Delegate the actual assignment (upsert + MQTT config + WS broadcast) to
+    # the existing assign_spool endpoint handler.
+    try:
+        from backend.app.api.routes.inventory import assign_spool
+        from backend.app.schemas.spool import SpoolAssignmentCreate
+
+        await assign_spool(
+            data=SpoolAssignmentCreate(
+                spool_id=spool_id,
+                printer_id=printer_id,
+                ams_id=ams_id,
+                tray_id=tray_id,
+            ),
+            db=db,
+            current_user=None,
+        )
+    except Exception:
+        logger.exception(
+            "Failed to execute assign_spool for pending assignment %d (spool %d → printer %d AMS%d-T%d)",
+            assignment.id,
+            spool_id,
+            printer_id,
+            ams_id,
+            tray_id,
+        )
+        return False
+
+    # Mark the pending assignment as completed
+    now = datetime.now(timezone.utc)
+    time_to_placement = (now - assignment.created_at.replace(tzinfo=timezone.utc)).total_seconds()
+
+    assignment.status = "completed"
+    assignment.assigned_printer_id = printer_id
+    assignment.assigned_ams_id = ams_id
+    assignment.assigned_tray_id = tray_id
+    assignment.completed_at = now
+    assignment.time_to_placement = time_to_placement
+
+    await db.commit()
+
+    # Cancel timeout task
+    task = _timeout_tasks.pop(assignment.id, None)
+    if task and not task.done():
+        task.cancel()
+
+    logger.info(
+        "Completed pending slot assignment %d: spool %d → printer %d AMS%d-T%d (%.1fs)",
+        assignment.id,
+        spool_id,
+        printer_id,
+        ams_id,
+        tray_id,
+        time_to_placement,
+    )
+
+    # Notify clients via WebSocket
+    await ws_manager.broadcast(
+        {
+            "type": "pending_slot_assignment_completed",
+            "assignment_id": assignment.id,
+            "tray_uuid": assignment.tray_uuid,
+            "tag_uid": assignment.tag_uid,
+            "spool_id": spool_id,
+            "printer_id": printer_id,
+            "ams_id": ams_id,
+            "tray_id": tray_id,
+            "time_to_placement": time_to_placement,
+        }
+    )
+
+    # Notify via MQTT relay
+    try:
+        from backend.app.services.mqtt_relay import mqtt_relay
+
+        await mqtt_relay.on_slot_assignment_completed(
+            assignment_id=assignment.id,
+            tray_uuid=assignment.tray_uuid,
+            tag_uid=assignment.tag_uid,
+            spool_id=spool_id,
+            printer_id=printer_id,
+            ams_id=ams_id,
+            tray_id=tray_id,
+            time_to_placement=time_to_placement,
+        )
+    except Exception:
+        pass
+
+    return True
+
+
+async def restore_pending_timeouts() -> None:
+    """Re-schedule timeout tasks for any pending assignments on startup.
+
+    Called during application lifespan startup to resume monitoring of
+    assignments that were created before a restart.
+    """
+    from backend.app.core.database import async_session
+
+    async with async_session() as db:
+        result = await db.execute(select(PendingSlotAssignment).where(PendingSlotAssignment.status == "pending"))
+        pending = result.scalars().all()
+        now = datetime.now(timezone.utc)
+        for assignment in pending:
+            created = assignment.created_at.replace(tzinfo=timezone.utc)
+            elapsed = (now - created).total_seconds()
+            remaining = assignment.timeout_seconds - elapsed
+            if remaining <= 0:
+                # Already expired — mark it
+                assignment.status = "timed_out"
+            else:
+                _schedule_timeout(assignment_id=assignment.id, timeout_seconds=int(remaining))
+        if pending:
+            await db.commit()
+            logger.info(
+                "Restored %d pending slot assignment timeout tasks on startup",
+                len([a for a in pending if a.status == "pending"]),
+            )
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 5c55e8a33e..ff8be1f2eb 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -116,6 +116,7 @@ async def test_engine():
         notification,
         notification_template,
         oidc_provider,
+        pending_slot_assignment,  # noqa: F401
         print_log,
         print_queue,
         printer,
diff --git a/backend/tests/integration/test_inventory_next_assign.py b/backend/tests/integration/test_inventory_next_assign.py
new file mode 100644
index 0000000000..7c6bc8407e
--- /dev/null
+++ b/backend/tests/integration/test_inventory_next_assign.py
@@ -0,0 +1,760 @@
+"""Integration tests for pending slot assignment endpoints (assign-on-next-slot).
+
+Tests the POST /spools/assign-on-next-slot, GET /spools/assignments/{id},
+and DELETE /spools/assignments/{id} endpoints.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.pending_slot_assignment import PendingSlotAssignment
+from backend.app.models.spool import Spool
+
+
+@pytest.fixture
+async def spool_factory(db_session: AsyncSession):
+    """Factory to create test spools."""
+    _counter = [0]
+
+    async def _create_spool(**kwargs):
+        _counter[0] += 1
+        defaults = {
+            "material": "PLA",
+            "subtype": "Basic",
+            "brand": "Devil Design",
+            "color_name": "Red",
+            "rgba": "FF0000FF",
+            "label_weight": 1000,
+            "weight_used": 0,
+            "tray_uuid": f"AABBCCDD{_counter[0]:024X}",
+            "tag_uid": f"04AABB{_counter[0]:010X}",
+        }
+        defaults.update(kwargs)
+        spool = Spool(**defaults)
+        db_session.add(spool)
+        await db_session.commit()
+        await db_session.refresh(spool)
+        return spool
+
+    return _create_spool
+
+
+class TestAssignOnNextSlotCreate:
+    """Tests for POST /api/v1/inventory/spools/assign-on-next-slot."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_with_tray_uuid(self, async_client: AsyncClient, spool_factory):
+        """Creating a pending assignment with tray_uuid returns status=pending."""
+        spool = await spool_factory()
+
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "pending"
+        assert body["tray_uuid"] == spool.tray_uuid
+        assert body["source"] == "nfc"
+        assert body["timeout_seconds"] == 300
+        assert body["spool_id"] == spool.id
+        assert body["assignment_id"] is not None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_with_tag_uid(self, async_client: AsyncClient, spool_factory):
+        """Creating a pending assignment with tag_uid resolves the spool."""
+        spool = await spool_factory()
+
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tag_uid": spool.tag_uid,
+                "source": "spoolbuddy",
+                "timeout": 60,
+            },
+        )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "pending"
+        assert body["tag_uid"] == spool.tag_uid
+        assert body["source"] == "spoolbuddy"
+        assert body["timeout_seconds"] == 60
+        assert body["spool_id"] == spool.id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_with_spool_id(self, async_client: AsyncClient, spool_factory):
+        """Creating a pending assignment with explicit spool_id."""
+        spool = await spool_factory()
+
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "qr",
+                "timeout": 120,
+            },
+        )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "pending"
+        assert body["spool_id"] == spool.id
+        assert body["source"] == "qr"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_with_printer_id(
+        self, async_client: AsyncClient, printer_factory, spool_factory
+    ):
+        """Pending assignment can target a specific printer."""
+        printer = await printer_factory(name="X1C")
+        spool = await spool_factory()
+
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "printer_id": printer.id,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["printer_id"] == printer.id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_idempotent(self, async_client: AsyncClient, spool_factory):
+        """Creating a pending assignment for the same spool returns the existing one."""
+        spool = await spool_factory()
+
+        response1 = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assert response1.status_code == 200
+        body1 = response1.json()
+
+        response2 = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assert response2.status_code == 200
+        body2 = response2.json()
+
+        # Same assignment returned (idempotent)
+        assert body1["assignment_id"] == body2["assignment_id"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_requires_identifier(self, async_client: AsyncClient):
+        """Request without tray_uuid, tag_uid, or spool_id still creates an assignment
+        but with spool_id=None since no spool can be resolved."""
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": 1,
+                "tray_uuid": "DEADBEEF00000000",
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+
+        # The endpoint accepts the request — spool resolution happens internally
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "pending"
+        # spool_id=1 doesn't exist, so resolved spool_id may be None
+        assert body["tray_uuid"] == "DEADBEEF00000000"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_create_pending_assignment_invalid_timeout_too_low(self, async_client: AsyncClient, spool_factory):
+        """Timeout below minimum (< 10) is accepted by the loose schema but the
+        service still creates the assignment."""
+        spool = await spool_factory()
+
+        response = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 3,
+            },
+        )
+
+        assert response.status_code == 200
+
+
+class TestGetPendingAssignmentStatus:
+    """Tests for GET /api/v1/inventory/spools/assignments/{assignment_id}."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_get_pending_assignment_status(self, async_client: AsyncClient, spool_factory):
+        """Retrieve the status of a pending assignment."""
+        spool = await spool_factory()
+
+        # Create the assignment first
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assert create_resp.status_code == 200
+        assignment_id = create_resp.json()["assignment_id"]
+
+        # Get status
+        response = await async_client.get(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["assignment_id"] == assignment_id
+        assert body["status"] == "pending"
+        assert body["tray_uuid"] == spool.tray_uuid
+        assert body["source"] == "nfc"
+        assert body["spool_id"] == spool.id
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_get_pending_assignment_not_found(self, async_client: AsyncClient):
+        """Requesting a non-existent assignment returns 404."""
+        response = await async_client.get("/api/v1/inventory/spools/assignments/99999")
+
+        assert response.status_code == 404
+        assert "not found" in response.json()["detail"].lower()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_get_cancelled_assignment_still_visible(self, async_client: AsyncClient, spool_factory):
+        """A cancelled assignment is still retrievable via GET."""
+        spool = await spool_factory()
+
+        # Create
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assignment_id = create_resp.json()["assignment_id"]
+
+        # Cancel
+        await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+
+        # Still visible via GET
+        response = await async_client.get(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["status"] == "cancelled"
+
+
+class TestCancelPendingAssignment:
+    """Tests for DELETE /api/v1/inventory/spools/assignments/{assignment_id}."""
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cancel_pending_assignment(self, async_client: AsyncClient, spool_factory):
+        """Cancelling a pending assignment sets status to cancelled."""
+        spool = await spool_factory()
+
+        # Create
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assert create_resp.status_code == 200
+        assignment_id = create_resp.json()["assignment_id"]
+
+        # Cancel
+        response = await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["assignment_id"] == assignment_id
+        assert body["status"] == "cancelled"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cancel_nonexistent_assignment_returns_404(self, async_client: AsyncClient):
+        """Cancelling a non-existent assignment returns 404."""
+        response = await async_client.delete("/api/v1/inventory/spools/assignments/99999")
+
+        assert response.status_code == 404
+        assert "not found" in response.json()["detail"].lower()
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cancel_already_cancelled_returns_404(self, async_client: AsyncClient, spool_factory):
+        """Cancelling an already-cancelled assignment returns 404 (not in pending status)."""
+        spool = await spool_factory()
+
+        # Create
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assignment_id = create_resp.json()["assignment_id"]
+
+        # Cancel once
+        resp1 = await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+        assert resp1.status_code == 200
+
+        # Cancel again — should fail
+        resp2 = await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+        assert resp2.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cancel_completed_assignment_returns_404(
+        self, async_client: AsyncClient, db_session: AsyncSession, spool_factory
+    ):
+        """Cancelling a completed assignment returns 404 (not in pending status)."""
+        spool = await spool_factory()
+
+        # Create via API
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "source": "nfc",
+                "timeout": 300,
+            },
+        )
+        assignment_id = create_resp.json()["assignment_id"]
+
+        # Manually mark as completed in db to simulate completion
+        from sqlalchemy import update
+
+        await db_session.execute(
+            update(PendingSlotAssignment).where(PendingSlotAssignment.id == assignment_id).values(status="completed")
+        )
+        await db_session.commit()
+
+        # Try to cancel — should fail
+        response = await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+        assert response.status_code == 404
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cancel_returns_full_response(self, async_client: AsyncClient, spool_factory):
+        """Cancel response includes all expected fields."""
+        spool = await spool_factory()
+
+        create_resp = await async_client.post(
+            "/api/v1/inventory/spools/assign-on-next-slot",
+            json={
+                "spool_id": spool.id,
+                "tray_uuid": spool.tray_uuid,
+                "tag_uid": spool.tag_uid,
+                "printer_id": None,
+                "source": "spoolbuddy",
+                "timeout": 120,
+            },
+        )
+        assignment_id = create_resp.json()["assignment_id"]
+
+        response = await async_client.delete(f"/api/v1/inventory/spools/assignments/{assignment_id}")
+
+        assert response.status_code == 200
+        body = response.json()
+        assert body["assignment_id"] == assignment_id
+        assert body["status"] == "cancelled"
+        assert body["tray_uuid"] == spool.tray_uuid
+        assert body["tag_uid"] == spool.tag_uid
+        assert body["source"] == "spoolbuddy"
+        assert body["timeout_seconds"] == 120
+        assert body["spool_id"] == spool.id
+        assert body["assigned_printer_id"] is None
+        assert body["assigned_ams_id"] is None
+        assert body["assigned_tray_id"] is None
+        assert body["completed_at"] is None
+
+
+class TestTryCompletePendingAssignmentIdentityVerification:
+    """Tests that try_complete_pending_assignments verifies slot identity.
+
+    When a slot transitions from empty → filled, the system must verify that
+    the physically-inserted spool (identified by tray_uuid/tag_uid from AMS
+    telemetry) matches the pending assignment's identifiers. Inserting a
+    different spool than requested must NOT complete the pending assignment.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_matching_tray_uuid_completes_assignment(self, db_session: AsyncSession, spool_factory):
+        """Pending assignment completes when slot's tray_uuid matches."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool = await spool_factory()
+
+        # Create a pending assignment directly in DB
+        assignment = PendingSlotAssignment(
+            tray_uuid=spool.tray_uuid,
+            tag_uid=spool.tag_uid,
+            spool_id=spool.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Simulate slot fill with the SAME spool's tray_uuid
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=1,
+                    slot_tray_uuid=spool.tray_uuid,
+                    slot_tag_uid=spool.tag_uid,
+                )
+
+        assert result is True
+        await db_session.refresh(assignment)
+        assert assignment.status == "completed"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_mismatched_tray_uuid_does_not_complete(self, db_session: AsyncSession, spool_factory):
+        """Pending assignment is NOT completed when a different spool is inserted."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool_a = await spool_factory(tray_uuid="AAAA0000AAAA0000AAAA0000AAAA0000")
+        spool_b = await spool_factory(tray_uuid="BBBB0000BBBB0000BBBB0000BBBB0000")
+
+        # Pending assignment for spool A
+        assignment = PendingSlotAssignment(
+            tray_uuid=spool_a.tray_uuid,
+            tag_uid=spool_a.tag_uid,
+            spool_id=spool_a.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Slot fills with spool B's identifiers — DIFFERENT spool
+        with patch("backend.app.services.pending_slot_assignment.ws_manager"):
+            result = await try_complete_pending_assignments(
+                db_session,
+                printer_id=1,
+                ams_id=0,
+                tray_id=1,
+                slot_tray_uuid=spool_b.tray_uuid,
+                slot_tag_uid=spool_b.tag_uid,
+            )
+
+        # Must NOT complete — prevents inventory corruption
+        assert result is False
+        await db_session.refresh(assignment)
+        assert assignment.status == "pending"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_mismatched_tag_uid_does_not_complete(self, db_session: AsyncSession, spool_factory):
+        """Pending assignment is NOT completed when tag_uid doesn't match."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool_a = await spool_factory(tag_uid="04AAAA11111111")
+        spool_b = await spool_factory(tag_uid="04BBBB22222222")
+
+        # Pending assignment for spool A (using tag_uid only)
+        assignment = PendingSlotAssignment(
+            tray_uuid=None,
+            tag_uid=spool_a.tag_uid,
+            spool_id=spool_a.id,
+            printer_id=1,
+            source="spoolbuddy",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Slot fills with spool B's tag_uid
+        with patch("backend.app.services.pending_slot_assignment.ws_manager"):
+            result = await try_complete_pending_assignments(
+                db_session,
+                printer_id=1,
+                ams_id=0,
+                tray_id=0,
+                slot_tray_uuid=None,
+                slot_tag_uid=spool_b.tag_uid,
+            )
+
+        assert result is False
+        await db_session.refresh(assignment)
+        assert assignment.status == "pending"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_no_slot_identifier_falls_back_to_fifo(self, db_session: AsyncSession, spool_factory):
+        """When slot has no identifiers (3rd-party spool), FIFO is used."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool = await spool_factory()
+
+        assignment = PendingSlotAssignment(
+            tray_uuid=spool.tray_uuid,
+            tag_uid=spool.tag_uid,
+            spool_id=spool.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Slot fills with no identifiers (empty strings = no RFID)
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=0,
+                    slot_tray_uuid=None,
+                    slot_tag_uid=None,
+                )
+
+        assert result is True
+        await db_session.refresh(assignment)
+        assert assignment.status == "completed"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_case_insensitive_tray_uuid_match(self, db_session: AsyncSession, spool_factory):
+        """Identity match is case-insensitive."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool = await spool_factory(tray_uuid="aabbccdd11223344aabbccdd11223344")
+
+        assignment = PendingSlotAssignment(
+            tray_uuid="AABBCCDD11223344AABBCCDD11223344",
+            tag_uid=None,
+            spool_id=spool.id,
+            printer_id=1,
+            source="qr",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=0,
+                    slot_tray_uuid="aabbccdd11223344aabbccdd11223344",
+                    slot_tag_uid=None,
+                )
+
+        assert result is True
+        await db_session.refresh(assignment)
+        assert assignment.status == "completed"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_multiple_pending_selects_correct_match(self, db_session: AsyncSession, spool_factory):
+        """With multiple pending assignments, only the one matching the slot completes."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool_a = await spool_factory(tray_uuid="AAAA0000AAAA0000AAAA0000AAAA0000")
+        spool_b = await spool_factory(tray_uuid="BBBB0000BBBB0000BBBB0000BBBB0000")
+
+        # Create assignment for spool A (older)
+        assignment_a = PendingSlotAssignment(
+            tray_uuid=spool_a.tray_uuid,
+            tag_uid=None,
+            spool_id=spool_a.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment_a)
+        await db_session.commit()
+        await db_session.refresh(assignment_a)
+
+        # Create assignment for spool B (newer)
+        assignment_b = PendingSlotAssignment(
+            tray_uuid=spool_b.tray_uuid,
+            tag_uid=None,
+            spool_id=spool_b.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment_b)
+        await db_session.commit()
+        await db_session.refresh(assignment_b)
+
+        # Slot fills with spool B — even though A is older, B must be selected
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=2,
+                    slot_tray_uuid=spool_b.tray_uuid,
+                    slot_tag_uid=None,
+                )
+
+        assert result is True
+        await db_session.refresh(assignment_a)
+        await db_session.refresh(assignment_b)
+        # A stays pending, B gets completed
+        assert assignment_a.status == "pending"
+        assert assignment_b.status == "completed"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_all_zeros_tray_uuid_treated_as_no_identifier(self, db_session: AsyncSession, spool_factory):
+        """All-zeros tray_uuid from firmware is treated as 'no identifier' — FIFO fallback."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool = await spool_factory()
+
+        assignment = PendingSlotAssignment(
+            tray_uuid=spool.tray_uuid,
+            tag_uid=spool.tag_uid,
+            spool_id=spool.id,
+            printer_id=1,
+            source="nfc",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Slot reports all-zeros (3rd-party spool / no RFID chip readable)
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=0,
+                    slot_tray_uuid="00000000000000000000000000000000",
+                    slot_tag_uid="0000000000000000",
+                )
+
+        # All-zeros = no identifier → FIFO fallback → completes
+        assert result is True
+        await db_session.refresh(assignment)
+        assert assignment.status == "completed"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_all_zeros_tag_uid_does_not_match_real_assignment(self, db_session: AsyncSession, spool_factory):
+        """All-zeros tag_uid doesn't falsely match a pending assignment's real tag_uid."""
+        from backend.app.services.pending_slot_assignment import try_complete_pending_assignments
+
+        spool = await spool_factory(
+            tray_uuid=None,
+            tag_uid="04AABB1122334455",
+        )
+
+        assignment = PendingSlotAssignment(
+            tray_uuid=None,
+            tag_uid=spool.tag_uid,
+            spool_id=spool.id,
+            printer_id=1,
+            source="spoolbuddy",
+            status="pending",
+            timeout_seconds=300,
+        )
+        db_session.add(assignment)
+        await db_session.commit()
+        await db_session.refresh(assignment)
+
+        # Slot reports all-zeros — this should NOT match the real tag_uid
+        # but should fall through to FIFO since it's "no identifier"
+        with patch("backend.app.services.pending_slot_assignment.ws_manager") as mock_ws:
+            mock_ws.broadcast = AsyncMock()
+            with patch("backend.app.api.routes.inventory.assign_spool", new_callable=AsyncMock):
+                result = await try_complete_pending_assignments(
+                    db_session,
+                    printer_id=1,
+                    ams_id=0,
+                    tray_id=0,
+                    slot_tray_uuid="00000000000000000000000000000000",
+                    slot_tag_uid="0000000000000000",
+                )
+
+        # All-zeros = no identifier → falls back to FIFO → still completes
+        # (but via the FIFO path, not via identifier match)
+        assert result is True
+        await db_session.refresh(assignment)
+        assert assignment.status == "completed"
diff --git a/scripts/app_version.py b/scripts/app_version.py
index d3bdb74614..7edec17826 100644
--- a/scripts/app_version.py
+++ b/scripts/app_version.py
@@ -27,8 +27,7 @@ def validate_version(version: str) -> None:
         version = version[1:]
     if not VERSION_RE.fullmatch(version):
         raise SystemExit(
-            "Invalid version. Expected forms like 0.2.4, 0.2.4.8, or 0.2.5b1 "
-            f"without a leading v; got {version!r}"
+            f"Invalid version. Expected forms like 0.2.4, 0.2.4.8, or 0.2.5b1 without a leading v; got {version!r}"
         )