fix(queue): ownership gates + TOCTOU lock + /reorder validator (#1625-followup)

Three issues from the post-merge audit of the unified-dispatch PR, all
pre-existed on dev but became more impactful once every print routes
through the queue:

1. Start/Stop ownership gates. /queue/{id}/stop required QUEUE_UPDATE_ALL
   (admin-only) -- operators saw the Stop button in the queue UI but got
   403 on click. /queue/{id}/start required QUEUE_UPDATE_OWN with no
   ownership check -- _OWN holders could start anyone's queue items via
   direct API. Both routes now use require_ownership_permission, mirroring
   /cancel. Stop is strict (rejects unowned items for _OWN); start preserves
   #1670's VP-import flow where _OWN can start NULL-owner items and claim
   ownership at click-time. Frontend QueuePage Start/Stop buttons flip
   from printers:control to canModify('queue', 'update', created_by_id).

2. TOCTOU race on insert_position. Concurrent ASAP inserts to the same
   scope both computed MAX(position) from before the other committed; in
   an empty scope, both inserted at position=1 (duplicate). Wraps the
   read+update in a transaction-scoped Postgres pg_advisory_xact_lock
   keyed on the printer_id. Different printers don't contend. SQLite
   serializes writes implicitly so the path is no-op there. Dialect is
   checked against the live session binding, not the is_sqlite() helper,
   because the test fixture overrides get_db to SQLite while
   settings.database_url still points at Postgres.

3. /reorder duplicate-position validator. POST /queue/reorder set position
   from the payload in a loop with no uniqueness validation -- a buggy
   drag-drop client could leave the queue with ambiguous ordering (the
   scheduler's ORDER BY (printer_id, position) ties break by row order).
   New model_validator on PrintQueueReorder rejects duplicates at the
   schema layer with 422 + "Duplicate positions in reorder request: [N, ...]".

backend/app/api/routes/print_queue.py

@@ -541,6 +541,30 @@ async def add_to_queue(
             PrintQueueItem.status == "pending",
         )
 
+    # Serialize concurrent queue inserts to the same scope (#1625-followup).
+    # The race: two concurrent ASAP inserts both compute MAX(position) before
+    # either commits; in an empty scope, both INSERT at position 1 (duplicate).
+    # In a non-empty scope, Postgres's row-level locks on the UPDATE shift
+    # serialize naturally, but the empty-scope path has no rows to lock.
+    # A transaction-scoped advisory lock keyed on the printer_id closes that
+    # window; the lock is released automatically at commit/rollback. Different
+    # printers don't contend. SQLite serializes writes implicitly so this is a
+    # no-op there.
+    #
+    # Dialect is checked against the actual session binding, NOT the
+    # `is_sqlite()` helper, because the test fixture overrides `get_db` with a
+    # SQLite engine while `settings.database_url` still points at Postgres
+    # (the helper reads settings). Inspecting the connection directly is the
+    # right shape for any code that mutates SQL based on the live dialect.
+    from sqlalchemy import text
+
+    bind = db.get_bind()
+    if bind.dialect.name == "postgresql":
+        scope_key = data.printer_id if data.printer_id is not None else 0
+        # 1625 namespaces the lock so it can't collide with other advisory
+        # locks elsewhere in the codebase.
+        await db.execute(text("SELECT pg_advisory_xact_lock(1625, :k)"), {"k": scope_key})
+
     insert_position = max(1, data.insert_position or 1)
     if data.insert_at_top or data.insert_position is not None:
         result = await db.execute(select(func.max(PrintQueueItem.position)).where(*queue_scope))
@@ -1229,19 +1253,39 @@ async def cancel_queue_item(
 async def stop_queue_item(
     item_id: int,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_ALL),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_UPDATE_ALL,
+            Permission.QUEUE_UPDATE_OWN,
+        )
+    ),
 ):
-    """Stop an actively printing queue item."""
+    """Stop an actively printing queue item.
+
+    Ownership-scoped (#1625-followup): callers with QUEUE_UPDATE_OWN can stop
+    their own items; callers with QUEUE_UPDATE_ALL can stop any item. Mirrors
+    the /cancel shape. Pre-fix this required QUEUE_UPDATE_ALL — Operators
+    holding only _OWN saw the Stop button in the queue UI but got 403 on click.
+    """
 
     from backend.app.models.smart_plug import SmartPlug
     from backend.app.services.printer_manager import printer_manager
     from backend.app.services.tasmota import tasmota_service
 
+    user, can_modify_all = auth_result
+
     result = await db.execute(select(PrintQueueItem).where(PrintQueueItem.id == item_id))
     item = result.scalar_one_or_none()
     if not item:
         raise HTTPException(404, "Queue item not found")
 
+    # Ownership check — mirrors /cancel. Ownerless items (created_by_id IS NULL)
+    # require _ALL: stop is destructive and an _OWN holder can't claim "they
+    # own it" the way /start does (#1670).
+    if not can_modify_all and user is not None:
+        if item.created_by_id is None or item.created_by_id != user.id:
+            raise HTTPException(403, "You can only stop your own queue items")
+
     if item.status != "printing":
         raise HTTPException(400, f"Can only stop items that are printing, current status: '{item.status}'")
 
@@ -1311,10 +1355,21 @@ async def start_queue_item(
     item_id: int,
     skip_filament_check: bool = Query(default=False),
     db: AsyncSession = Depends(get_db),
-    user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
+    auth_result: tuple[User | None, bool] = Depends(
+        require_ownership_permission(
+            Permission.QUEUE_UPDATE_ALL,
+            Permission.QUEUE_UPDATE_OWN,
+        )
+    ),
 ):
     """Manually start a staged (manual_start) queue item.
 
+    Ownership-scoped (#1625-followup): callers with QUEUE_UPDATE_OWN can
+    start their own items + claim ownership of NULL-owner items (VP-uploaded
+    items arrive unattributed per #1670). Callers with QUEUE_UPDATE_ALL can
+    start any item. Pre-fix this required QUEUE_UPDATE_OWN with no ownership
+    check, so _OWN holders could start anyone's queue items via direct API.
+
     Clears the manual_start flag so the scheduler picks it up. When
     ``skip_filament_check`` is false (the default) the live filament
     deficit (#1496) is checked first — if the assigned spool can't satisfy
@@ -1322,6 +1377,8 @@ async def start_queue_item(
     payload so the caller can show a confirm dialog and retry with
     ``skip_filament_check=true``.
     """
+    user, can_modify_all = auth_result
+
     result = await db.execute(
         select(PrintQueueItem)
         .options(
@@ -1336,6 +1393,14 @@ async def start_queue_item(
     if not item:
         raise HTTPException(404, "Queue item not found")
 
+    # Ownership check — softer than /cancel because /start is the entry point
+    # for #1670's VP-import flow: an unowned item is claimable by the first
+    # _OWN holder who clicks ▶, and the route below credits them as owner.
+    # An item with a DIFFERENT owner → 403.
+    if not can_modify_all and user is not None:
+        if item.created_by_id is not None and item.created_by_id != user.id:
+            raise HTTPException(403, "You can only start your own queue items")
+
     if item.status != "pending":
         raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'")
 

backend/app/schemas/print_queue.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Annotated, Literal
 
-from pydantic import BaseModel, PlainSerializer
+from pydantic import BaseModel, PlainSerializer, model_validator
 
 
 # Custom serializer to ensure UTC datetimes have Z suffix
@@ -190,6 +190,30 @@ class PrintQueueReorderItem(BaseModel):
 class PrintQueueReorder(BaseModel):
     items: list[PrintQueueReorderItem]
 
+    @model_validator(mode="after")
+    def _validate_positions_unique(self) -> "PrintQueueReorder":
+        """Reject reorder requests with duplicate positions in the payload
+        (#1625-followup).
+
+        The /reorder route is the drag-drop renumber path on the queue UI;
+        a well-behaved client sends a contiguous renumbering of a single
+        printer's pending queue. A buggy client that sends two items at
+        the same position would leave the queue in an inconsistent state
+        (scheduler's ORDER BY (printer_id, position) ties get broken by
+        physical row order). Fail closed at the schema boundary so the
+        bug is caught before any DB mutation.
+
+        Uniqueness is enforced WITHIN THE PAYLOAD only — cross-printer
+        reorders that intentionally share positions across different
+        printer queues are a non-goal of the drag-drop UI, so this is the
+        right scope.
+        """
+        positions = [it.position for it in self.items]
+        if len(positions) != len(set(positions)):
+            duplicates = sorted({p for p in positions if positions.count(p) > 1})
+            raise ValueError(f"Duplicate positions in reorder request: {duplicates}")
+        return self
+
 
 class PrintQueueBulkUpdate(BaseModel):
     """Bulk update multiple queue items with the same values."""

backend/tests/integration/test_ownership_permissions.py

@@ -541,6 +541,139 @@ class TestQueueOwnershipPermissions(TestOwnershipPermissionsSetup):
 
         assert response.status_code == 403
 
+    # ========================================================================
+    # Start / Stop ownership gates (#1625-followup)
+    # ========================================================================
+    # Pre-fix /stop required QUEUE_UPDATE_ALL (admin-only) — operators saw the
+    # Stop button in the queue UI but got 403 on click. /start required
+    # QUEUE_UPDATE_OWN with no ownership check — operators could start anyone's
+    # queue items via direct API. Both now use require_ownership_permission.
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_can_start_own_queue_item(self, async_client: AsyncClient, auth_setup, queue_item_factory):
+        """Operator can start their own staged queue item."""
+        item = await queue_item_factory(
+            created_by_id=auth_setup["operator_user"]["id"],
+            manual_start=True,
+        )
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/start?skip_filament_check=true",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_cannot_start_others_queue_item(
+        self, async_client: AsyncClient, auth_setup, queue_item_factory
+    ):
+        """Operator cannot start another user's queue item."""
+        item = await queue_item_factory(
+            created_by_id=auth_setup["operator2_user"]["id"],
+            manual_start=True,
+        )
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/start?skip_filament_check=true",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 403
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_can_start_unowned_queue_item(
+        self, async_client: AsyncClient, auth_setup, queue_item_factory, db_session
+    ):
+        """Operator can start a NULL-owner queue item (VP-uploaded, #1670)
+        and claims ownership in the process.
+
+        Stop and Cancel reject unowned items for _OWN holders (destructive,
+        no "I own it" claim available), but Start is the entry point for the
+        VP-import flow where attribution happens at click-time.
+        """
+        from backend.app.models.print_queue import PrintQueueItem
+
+        item = await queue_item_factory(created_by_id=None, manual_start=True)
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/start?skip_filament_check=true",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 200
+        # Ownership claimed: operator is now the item's owner.
+        await db_session.refresh(item)
+        refetch = await db_session.get(PrintQueueItem, item.id)
+        assert refetch.created_by_id == auth_setup["operator_user"]["id"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_can_stop_own_queue_item(self, async_client: AsyncClient, auth_setup, queue_item_factory):
+        """Operator can stop their own currently-printing queue item."""
+        item = await queue_item_factory(
+            created_by_id=auth_setup["operator_user"]["id"],
+            status="printing",
+        )
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/stop",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_cannot_stop_others_queue_item(
+        self, async_client: AsyncClient, auth_setup, queue_item_factory
+    ):
+        """Operator cannot stop another user's printing queue item."""
+        item = await queue_item_factory(
+            created_by_id=auth_setup["operator2_user"]["id"],
+            status="printing",
+        )
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/stop",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 403
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_operator_cannot_stop_unowned_queue_item(
+        self, async_client: AsyncClient, auth_setup, queue_item_factory
+    ):
+        """Operator cannot stop a NULL-owner printing queue item — stop mirrors
+        cancel (destructive, no claim semantics). Admins with _ALL can still stop it.
+        """
+        item = await queue_item_factory(created_by_id=None, status="printing")
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/stop",
+            headers={"Authorization": f"Bearer {auth_setup['operator_token']}"},
+        )
+
+        assert response.status_code == 403
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_admin_can_stop_any_queue_item(self, async_client: AsyncClient, auth_setup, queue_item_factory):
+        """Admin with _ALL can stop any printing queue item including unowned."""
+        item = await queue_item_factory(created_by_id=None, status="printing")
+
+        response = await async_client.post(
+            f"/api/v1/queue/{item.id}/stop",
+            headers={"Authorization": f"Bearer {auth_setup['admin_token']}"},
+        )
+
+        assert response.status_code == 200
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_bulk_update_skips_non_owned_items(self, async_client: AsyncClient, auth_setup, queue_item_factory):

backend/tests/integration/test_print_queue_api.py

@@ -2684,3 +2684,126 @@ class TestResumeQueueAfterFailure:
 
         second = await async_client.post(f"/api/v1/queue/printer/{printer.id}/resume")
         assert second.json() == {"acknowledged": 0, "restored": 0}
+
+
+class TestReorderEndpoint:
+    """Tests for the /queue/reorder endpoint (#1625-followup duplicate-position validator)."""
+
+    @pytest.fixture
+    async def printer_factory(self, db_session):
+        async def _create(**kwargs):
+            from backend.app.models.printer import Printer
+
+            defaults = {
+                "name": "Reorder Test Printer",
+                "ip_address": "192.168.1.220",
+                "serial_number": "TESTREORDER001",
+                "access_code": "12345678",
+                "model": "X1C",
+            }
+            defaults.update(kwargs)
+            printer = Printer(**defaults)
+            db_session.add(printer)
+            await db_session.commit()
+            await db_session.refresh(printer)
+            return printer
+
+        return _create
+
+    @pytest.fixture
+    async def archive_factory(self, db_session):
+        _counter = [0]
+
+        async def _create(**kwargs):
+            from backend.app.models.archive import PrintArchive
+
+            _counter[0] += 1
+            defaults = {
+                "filename": f"reorder_{_counter[0]}.3mf",
+                "print_name": f"Reorder {_counter[0]}",
+                "file_path": f"/tmp/reorder_{_counter[0]}.3mf",
+                "file_size": 1024,
+                "content_hash": f"reorderhash{_counter[0]:06d}",
+                "status": "completed",
+            }
+            defaults.update(kwargs)
+            archive = PrintArchive(**defaults)
+            db_session.add(archive)
+            await db_session.commit()
+            await db_session.refresh(archive)
+            return archive
+
+        return _create
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reorder_rejects_duplicate_positions(
+        self, async_client: AsyncClient, db_session, printer_factory, archive_factory
+    ):
+        """Reorder payload with duplicate positions → 422 at schema layer.
+
+        Regression guard: pre-fix, a buggy client sending two items at the
+        same position would leave the queue in an inconsistent state (the
+        scheduler's ORDER BY (printer_id, position) tie would be broken by
+        physical row order — non-deterministic dispatch order).
+        """
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printer = await printer_factory()
+        a1 = await archive_factory()
+        a2 = await archive_factory()
+        item1 = PrintQueueItem(printer_id=printer.id, archive_id=a1.id, status="pending", position=1)
+        item2 = PrintQueueItem(printer_id=printer.id, archive_id=a2.id, status="pending", position=2)
+        db_session.add_all([item1, item2])
+        await db_session.commit()
+        await db_session.refresh(item1)
+        await db_session.refresh(item2)
+
+        response = await async_client.post(
+            "/api/v1/queue/reorder",
+            json={
+                "items": [
+                    {"id": item1.id, "position": 1},
+                    {"id": item2.id, "position": 1},  # duplicate
+                ]
+            },
+        )
+        assert response.status_code == 422
+        body = response.json()
+        # Pydantic v2 wraps custom validator errors; the message must mention "Duplicate"
+        # so the FE can surface the actionable detail.
+        assert any("duplicate" in str(err).lower() for err in body.get("detail", []))
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_reorder_accepts_unique_positions(
+        self, async_client: AsyncClient, db_session, printer_factory, archive_factory
+    ):
+        """Reorder with unique positions succeeds and updates them in DB."""
+        from backend.app.models.print_queue import PrintQueueItem
+
+        printer = await printer_factory()
+        a1 = await archive_factory()
+        a2 = await archive_factory()
+        item1 = PrintQueueItem(printer_id=printer.id, archive_id=a1.id, status="pending", position=1)
+        item2 = PrintQueueItem(printer_id=printer.id, archive_id=a2.id, status="pending", position=2)
+        db_session.add_all([item1, item2])
+        await db_session.commit()
+        await db_session.refresh(item1)
+        await db_session.refresh(item2)
+
+        response = await async_client.post(
+            "/api/v1/queue/reorder",
+            json={
+                "items": [
+                    {"id": item1.id, "position": 2},
+                    {"id": item2.id, "position": 1},
+                ]
+            },
+        )
+        assert response.status_code == 200
+
+        await db_session.refresh(item1)
+        await db_session.refresh(item2)
+        assert item1.position == 2
+        assert item2.position == 1

frontend/src/pages/QueuePage.tsx

@@ -672,8 +672,8 @@ function SortableQueueItem({
                 variant="ghost"
                 size="sm"
                 onClick={onStop}
-                disabled={!hasPermission('printers:control')}
-                title={!hasPermission('printers:control') ? t('queue.permissions.noStopPrint') : t('queue.actions.stopPrint')}
+                disabled={!canModify('queue', 'update', item.created_by_id)}
+                title={!canModify('queue', 'update', item.created_by_id) ? t('queue.permissions.noStopPrint') : t('queue.actions.stopPrint')}
                 className="text-red-400 hover:text-red-300 hover:bg-red-500/10 p-1.5 sm:p-2"
               >
                 <StopCircle className="w-4 h-4" />
@@ -686,8 +686,8 @@ function SortableQueueItem({
                     variant="ghost"
                     size="sm"
                     onClick={onStart}
-                    disabled={!hasPermission('printers:control')}
-                    title={!hasPermission('printers:control') ? t('queue.permissions.noStartPrint') : t('queue.actions.startPrint')}
+                    disabled={!canModify('queue', 'update', item.created_by_id)}
+                    title={!canModify('queue', 'update', item.created_by_id) ? t('queue.permissions.noStartPrint') : t('queue.actions.startPrint')}
                     className="text-bambu-green hover:text-bambu-green-light hover:bg-bambu-green/10 p-1.5 sm:p-2"
                   >
                     <Play className="w-4 h-4" />

static/index.html

@@ -26,7 +26,7 @@
 
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-CkDEALWj.js"></script>
+    <script type="module" crossorigin src="/assets/index-BSFuaWEQ.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-BKwIZ5yr.css">
   </head>
   <body>