fix(queue): cancelled prints no longer block require_previous_success chain (#1667)

Two bugs in PrintScheduler._check_previous_success:
  - Lookback excluded 'cancelled' so user cancellations were walked past
  - Lookback included 'skipped', so one skip cascaded indefinitely

  Swap to ['completed', 'failed', 'cancelled', 'aborted'] and accept
  both 'completed' and 'cancelled' as predecessor success. Real
  'failed' / 'aborted' still gate.

  One-shot migration in run_migrations resets only the skipped items
  whose true predecessor was cancelled — surgical reversal of the exact
  bug fingerprint, leaves genuine failure-gated skips alone. Portable
  across SQLite and Postgres, idempotent on re-run.

CHANGELOG.md

@@ -5,6 +5,8 @@ All notable changes to Bambuddy will be documented in this file.
 ## [0.2.5b1] - Unreleased
 
 ### Fixed
+- **Print queue `require_previous_success` no longer cascades indefinitely after a user-cancelled print (#1667, fully root-caused by @599w6c26tv-droid)** — Reporter on an A1 saw a single user-cancelled print block 18 downstream queue items over 3 days, all marked `skipped` with `Previous print failed or was aborted`. They captured the override log line proving Bambuddy correctly detects the cancellation (`Overriding status 'failed' -> 'cancelled' for printer 1 (print was stopped from queue by user)`) but the scheduler's gate ignored the override; they dumped the affected DB rows confirming the cascade pattern; and they reproduced from clean state in one cycle. Two distinct bugs in one function (`PrintScheduler._check_previous_success` in `services/print_scheduler.py`): **(a)** The lookback query `.in_(["completed", "failed", "skipped", "aborted"])` excluded `cancelled`, so a user cancellation was never found as the most-recent predecessor — the query walked past it to whatever real outcome existed before. **(b)** The same lookback INCLUDED `skipped`, so once one item got skipped (under any reason — bug-cascaded or genuinely failure-gated) it became the next item's "failed predecessor" and the cascade compounded. **Fix:** swap the lookback list to `["completed", "failed", "cancelled", "aborted"]` and broaden the success check to `prev_item.status in ("completed", "cancelled")`. A user cancellation is a deliberate action — treating it as neutral matches the user's intent ("I'm done with that one, move on"); `skipped` is excluded so the query always walks back to the most recent REAL print attempt and `failed` / `aborted` still gate as before. **Conservative recovery migration**: a one-shot pass in `core/database.py::run_migrations` resets only the skipped items whose immediate real predecessor (by `completed_at` desc, excluding the skipped-cascade itself) was `cancelled` — same fingerprint as the bug, narrow enough not to disturb skipped items whose true predecessor was a real `failed` / `aborted` print. Items match on `status='skipped' AND error_message='Previous print failed or was aborted'` and the predecessor check via correlated subquery; logged per-row at INFO so operators can audit the count after upgrade. Portable across SQLite and Postgres. Idempotent (post-reset rows no longer match). **Tests**: 10 new behaviour tests in `test_check_previous_success.py` pin every status/cascade combination — bug A (cancelled → True), bug B (skipped walked past), the reporter's exact failed→cancelled→skipped→skipped→pending cascade, regression guards on real failed / aborted still gating, edge cases (no-predecessor, only-skipped history, completed-then-failed). 7 new tests in `test_cancellation_cascade_recovery_migration.py` pin the migration — skipped-after-cancelled resets, skipped-after-failed stays, skipped-after-aborted stays, different-error-message untouched, reporter's multi-item cascade resets all, idempotent on re-run, per-printer isolation. All green; full scheduler + migration test suite stays green.
+
 - **Firmware-update check no longer 403s against Bambu Lab's Cloudflare-gated download page (#1666, reported by @arekm, with the working bypass demonstrated)** — Reporter on a fresh install hit `Could not reach Bambu Lab's firmware download page...` when checking firmware for an A1 Mini, and surfaced the diagnostic: `curl -H 'User-Agent: Bambuddy/1.0' https://bambulab.com/en/support/firmware-download/all` returns `HTTP 403 cf-mitigated=challenge` — Cloudflare upped the bot-protection on `bambulab.com` to a JA3 / TLS-fingerprint challenge. Plain Python TLS handshakes (httpx, requests, urllib) don't match Chrome's ClientHello bytes, so CF rejects before the request reaches the app layer. The `Accept` / `Accept-Language` header workaround we shipped for #1350 was below-HTTP and no longer enough. Existing users with a `build_id.json` on disk from a previous successful fetch kept working until Bambu rebuilt the page (every few weeks); fresh installs and wiped data dirs hit the wall immediately — exactly the reporter's path. **Fix: use `curl_cffi` for the two `bambulab.com` fetches only.** New dependency added to `requirements.txt`; `firmware_check.py` lazy-initialises a `curl_cffi.requests.AsyncSession(impersonate="chrome", ...)` for the `bambulab.com` calls (the index page that carries the Next.js `buildId`, and the per-model `_next/data/{buildId}/.../{api_key}.json` endpoint). Smoke-tested end-to-end against the live page: returns 200 OK + valid `buildId`, vs the reporter's 403. **Compliance framing matters here**: per the Bambu-compliance email from 2026-05-12, Bambuddy committed to "no falsified client identity." `curl_cffi`'s Chrome impersonation only governs TLS handshake bytes — the **HTTP-layer User-Agent is overridden back to `Bambuddy/1.0 (+https://github.com/maziggy/bambuddy)`** via the session's `headers=` parameter. Defensible read: TLS fingerprint matches Chrome (necessary because Python's TLS is the signal CF gates on), but every application-layer identity remains honestly Bambuddy. A new test (`test_bambulab_curl_cffi_session_keeps_honest_user_agent`) pins this — a future refactor that drops the `headers=` override would silently revert to curl_cffi's Chrome-default UA and break the compliance commitment; the test fails on any non-Bambuddy UA in the session. **Soft dependency**: if `curl_cffi` fails to import (rare platforms, alpine without wheels, etc.), the service logs a one-time warning at startup and falls back to httpx; wiki-based version detection continues to work for the badge, only the in-app firmware download URL stops resolving. New test `test_bambulab_get_falls_back_to_httpx_when_curl_cffi_missing` pins the fallback path. The wiki path (`wiki.bambulab.com`) and the CDN download path (`public-cdn.bblmw.com`) stay on httpx — neither sits behind the same JA3 gate. Three existing tests (`test_build_id_is_persisted_to_disk`, `test_build_id_falls_back_to_disk_on_403`, `test_download_page_unreachable_flag_set_on_403_json`, `test_download_page_retries_once_when_buildid_stale`) updated to mock `_bambulab_get` instead of the raw httpx client — a tighter mock target that's stable across the curl_cffi / httpx switch.
 
 ### Changed

backend/app/core/database.py

@@ -1873,6 +1873,70 @@ async def run_migrations(conn):
         )
     )
 
+    # Migration: Recover queue items that got stuck in `skipped` because of
+    # the cancellation-cascade bug (#1667). Pre-fix, the scheduler's
+    # `_check_previous_success` lookback excluded `cancelled` but included
+    # `skipped`, so a single user-cancelled print poisoned every downstream
+    # item with `require_previous_success=True` indefinitely. The reporter saw
+    # 18 items blocked over 3 days from one cancellation.
+    #
+    # Conservative reversal: ONLY reset rows whose immediate predecessor on
+    # the same printer (by completed_at desc, excluding the skipped-bug
+    # cascade) was `cancelled`. Skipped items whose true predecessor was a
+    # real `failed` or `aborted` print stay skipped — those were legitimate.
+    # Genuine failure-skips share the same status + error_message + completed_at
+    # fingerprint as bug-skips, so the predecessor check is what distinguishes
+    # them. Idempotent (post-reset rows no longer match the WHERE clause).
+    #
+    # Correlated subquery is portable across SQLite and Postgres. The
+    # `error_message` literal matches the exact string the buggy scheduler
+    # wrote — narrowing further on intent.
+    stuck_skipped_result = await conn.execute(
+        text(
+            "SELECT pq.id, pq.printer_id "
+            "FROM print_queue pq "
+            "WHERE pq.status = 'skipped' "
+            "  AND pq.error_message = 'Previous print failed or was aborted' "
+            "  AND pq.completed_at IS NOT NULL "
+            "  AND ("
+            "    SELECT prev.status FROM print_queue prev "
+            "    WHERE prev.printer_id = pq.printer_id "
+            "      AND prev.id != pq.id "
+            "      AND prev.status IN ('completed', 'failed', 'cancelled', 'aborted') "
+            "      AND prev.completed_at IS NOT NULL "
+            "      AND prev.completed_at < pq.completed_at "
+            "    ORDER BY prev.completed_at DESC LIMIT 1"
+            "  ) = 'cancelled'"
+        )
+    )
+    stuck_ids = [row.id for row in stuck_skipped_result.fetchall()]
+    if stuck_ids:
+        logger.info(
+            "Queue cancellation-cascade migration (#1667): resetting %d skipped item(s) to pending",
+            len(stuck_ids),
+        )
+        await conn.execute(
+            text(
+                "UPDATE print_queue "
+                "SET status = 'pending', error_message = NULL, completed_at = NULL "
+                "WHERE id IN ("
+                "  SELECT pq.id FROM print_queue pq "
+                "  WHERE pq.status = 'skipped' "
+                "    AND pq.error_message = 'Previous print failed or was aborted' "
+                "    AND pq.completed_at IS NOT NULL "
+                "    AND ("
+                "      SELECT prev.status FROM print_queue prev "
+                "      WHERE prev.printer_id = pq.printer_id "
+                "        AND prev.id != pq.id "
+                "        AND prev.status IN ('completed', 'failed', 'cancelled', 'aborted') "
+                "        AND prev.completed_at IS NOT NULL "
+                "        AND prev.completed_at < pq.completed_at "
+                "      ORDER BY prev.completed_at DESC LIMIT 1"
+                "    ) = 'cancelled'"
+                ")"
+            )
+        )
+
     # Migration: Unify `LibraryFile.file_type` across ingest paths (#1600).
     # Pre-#1600, only the external-folder scan path stored `gcode.3mf` for
     # sliced outputs — the upload, ZIP-extract, and in-process paths all

backend/app/services/print_scheduler.py

@@ -1763,13 +1763,21 @@ class PrintScheduler:
         return False
 
     async def _check_previous_success(self, db: AsyncSession, item: PrintQueueItem) -> bool:
-        """Check if the previous print on this printer succeeded."""
-        # Find the most recent completed queue item for this printer
+        """Check if the previous print on this printer succeeded.
+
+        A user-cancelled predecessor is treated as neutral — `cancelled` is a
+        deliberate action, not a failure, so subsequent items should still
+        dispatch (#1667). `skipped` is excluded from the lookback entirely:
+        a skip isn't an actual print attempt, so it must not gate downstream
+        items — counting it as a failed predecessor was the cascade bug that
+        let a single cancellation block 18 items over 3 days for the reporter.
+        Only `failed` and `aborted` — real print-attempt failures — block.
+        """
         result = await db.execute(
             select(PrintQueueItem)
             .where(PrintQueueItem.printer_id == item.printer_id)
             .where(PrintQueueItem.id != item.id)
-            .where(PrintQueueItem.status.in_(["completed", "failed", "skipped", "aborted"]))
+            .where(PrintQueueItem.status.in_(["completed", "failed", "cancelled", "aborted"]))
             .order_by(PrintQueueItem.completed_at.desc())
             .limit(1)
         )
@@ -1779,7 +1787,7 @@ class PrintScheduler:
         if not prev_item:
             return True
 
-        return prev_item.status == "completed"
+        return prev_item.status in ("completed", "cancelled")
 
     async def _power_off_if_needed(self, db: AsyncSession, item: PrintQueueItem):
         """Power off printer if auto_off_after is enabled (waits for cooldown)."""

backend/tests/unit/test_cancellation_cascade_recovery_migration.py

@@ -0,0 +1,282 @@
+"""Regression test for the cancellation-cascade recovery migration (#1667).
+
+Pre-fix: the scheduler's `_check_previous_success` lookback included
+`skipped` and excluded `cancelled`, so a single user-cancelled print
+poisoned every downstream item with `require_previous_success=True`
+indefinitely (reporter saw 18 items blocked over 3 days from one
+cancellation).
+
+This migration reverses the bug surgically: ONLY skipped items whose
+immediate real predecessor (by `completed_at` desc, excluding skipped
+items themselves) was `cancelled` get reset to `pending`. Items whose
+true predecessor was `failed` or `aborted` stay skipped — those were
+legitimate failure-gated skips.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from backend.app.core.database import run_migrations
+
+
+@pytest.fixture(autouse=True)
+def force_sqlite_dialect(monkeypatch):
+    """Force the SQLite branch regardless of test env settings."""
+    from backend.app.core import db_dialect
+
+    monkeypatch.setattr(db_dialect, "is_sqlite", lambda: True)
+    monkeypatch.setattr(db_dialect, "is_postgres", lambda: False)
+    from backend.app.core import database as database_module
+
+    monkeypatch.setattr(database_module, "is_sqlite", lambda: True)
+
+
+def _register_all_models():
+    """run_migrations touches multiple tables; the full schema must exist."""
+    from backend.app.models import (  # noqa: F401
+        ams_history,
+        ams_label,
+        api_key,
+        archive,
+        color_catalog,
+        external_link,
+        filament,
+        group,
+        kprofile_note,
+        maintenance,
+        notification,
+        notification_template,
+        print_log,
+        print_queue,
+        printer,
+        project,
+        project_bom,
+        settings,
+        slot_preset,
+        smart_plug,
+        smart_plug_energy_snapshot,
+        spool,
+        spool_assignment,
+        spool_catalog,
+        spool_k_profile,
+        spool_usage_history,
+        spoolbuddy_device,
+        user,
+        user_email_pref,
+        virtual_printer,
+    )
+
+
+@pytest.fixture
+async def engine():
+    from backend.app.core.database import Base
+
+    _register_all_models()
+
+    eng = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
+    async with eng.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield eng
+    await eng.dispose()
+
+
+BASE_TIME = datetime(2026, 6, 6, 12, 0, 0, tzinfo=timezone.utc)
+
+
+async def _insert_queue_item(
+    engine, *, id: int, printer_id: int, status: str, minutes_offset: int, error_message: str | None = None
+) -> None:
+    """Insert a print_queue row via the ORM so Python-side defaults
+    (manual_start, position, bed_levelling, …) all apply without us having
+    to mirror every NOT NULL column."""
+    from backend.app.models.print_queue import PrintQueueItem
+
+    async with AsyncSession(engine) as session:
+        session.add(
+            PrintQueueItem(
+                id=id,
+                printer_id=printer_id,
+                status=status,
+                error_message=error_message,
+                completed_at=BASE_TIME + timedelta(minutes=minutes_offset),
+                require_previous_success=True,
+                position=id,
+            )
+        )
+        await session.commit()
+
+
+async def _get_status(engine, item_id: int) -> tuple[str, str | None]:
+    async with engine.connect() as conn:
+        row = (
+            await conn.execute(text("SELECT status, error_message FROM print_queue WHERE id = :id"), {"id": item_id})
+        ).first()
+    return row.status, row.error_message
+
+
+@pytest.mark.asyncio
+async def test_skipped_after_cancelled_resets_to_pending(engine):
+    """Bug A + B: cancelled → skipped → migration resets the skipped item."""
+    await _insert_queue_item(engine, id=10, printer_id=1, status="cancelled", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=11,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    status, error_message = await _get_status(engine, 11)
+    assert status == "pending"
+    assert error_message is None
+
+
+@pytest.mark.asyncio
+async def test_skipped_after_failed_stays_skipped(engine):
+    """Genuine failure-gated skip must NOT be reset — the user really did
+    have a failure they need to deal with before downstream items run."""
+    await _insert_queue_item(engine, id=20, printer_id=1, status="failed", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=21,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    status, _ = await _get_status(engine, 21)
+    assert status == "skipped"
+
+
+@pytest.mark.asyncio
+async def test_skipped_after_aborted_stays_skipped(engine):
+    """Printer-detected abort is a real failure too — gate stays in place."""
+    await _insert_queue_item(engine, id=30, printer_id=1, status="aborted", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=31,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    status, _ = await _get_status(engine, 31)
+    assert status == "skipped"
+
+
+@pytest.mark.asyncio
+async def test_skipped_with_other_error_message_untouched(engine):
+    """Migration narrows on the exact buggy error string. A skipped item
+    written by some other code path (different error_message) is left alone."""
+    await _insert_queue_item(engine, id=40, printer_id=1, status="cancelled", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=41,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Some other reason",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    status, error_message = await _get_status(engine, 41)
+    assert status == "skipped"
+    assert error_message == "Some other reason"
+
+
+@pytest.mark.asyncio
+async def test_reporter_exact_cascade_resets_all_three(engine):
+    """The reporter's exact pattern: failed → cancelled → skipped → skipped.
+    Predecessors (by completed_at desc, skipped excluded) are cancelled for
+    both stuck items, so both reset."""
+    await _insert_queue_item(engine, id=50, printer_id=1, status="failed", minutes_offset=1)
+    await _insert_queue_item(engine, id=51, printer_id=1, status="cancelled", minutes_offset=2)
+    await _insert_queue_item(
+        engine,
+        id=52,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=3,
+        error_message="Previous print failed or was aborted",
+    )
+    await _insert_queue_item(
+        engine,
+        id=53,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=4,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    assert (await _get_status(engine, 52))[0] == "pending"
+    assert (await _get_status(engine, 53))[0] == "pending"
+    # The original failed/cancelled items are untouched
+    assert (await _get_status(engine, 50))[0] == "failed"
+    assert (await _get_status(engine, 51))[0] == "cancelled"
+
+
+@pytest.mark.asyncio
+async def test_migration_is_idempotent(engine):
+    """Running the migration twice doesn't re-touch already-reset rows."""
+    await _insert_queue_item(engine, id=60, printer_id=1, status="cancelled", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=61,
+        printer_id=1,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+    async with engine.begin() as conn:
+        await run_migrations(conn)  # second pass should be a no-op
+
+    status, _ = await _get_status(engine, 61)
+    assert status == "pending"
+
+
+@pytest.mark.asyncio
+async def test_per_printer_isolation(engine):
+    """A cancelled item on printer A must not affect a skipped item on
+    printer B (different printer queues are independent)."""
+    await _insert_queue_item(engine, id=70, printer_id=1, status="cancelled", minutes_offset=1)
+    await _insert_queue_item(engine, id=71, printer_id=2, status="failed", minutes_offset=1)
+    await _insert_queue_item(
+        engine,
+        id=72,
+        printer_id=2,
+        status="skipped",
+        minutes_offset=2,
+        error_message="Previous print failed or was aborted",
+    )
+
+    async with engine.begin() as conn:
+        await run_migrations(conn)
+
+    # printer 2's skipped item had a failed predecessor → stays skipped
+    status, _ = await _get_status(engine, 72)
+    assert status == "skipped"

backend/tests/unit/test_check_previous_success.py

@@ -0,0 +1,172 @@
+"""Tests for `PrintScheduler._check_previous_success` (#1667).
+
+Pre-fix behaviour: the lookback `.in_([...])` list excluded `cancelled` and
+included `skipped`, so a single user-cancelled print blocked every downstream
+item with `require_previous_success=True` permanently (the reporter saw 18
+items blocked over 3 days from one cancellation, because each new skip
+became the next skip's "failed predecessor").
+
+Post-fix behaviour:
+- `cancelled` is a neutral outcome → returns True (a deliberate user action
+  is not a print failure)
+- `skipped` is excluded from the lookback → an already-skipped item never
+  counts as a predecessor; the query walks back to the most recent real
+  print attempt
+- `failed` and `aborted` still gate as before
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.services.print_scheduler import PrintScheduler
+
+
+@pytest.fixture
+def scheduler():
+    return PrintScheduler()
+
+
+@pytest.fixture
+def queue_factory(db_session, printer_factory):
+    """Helper to drop completed/failed/cancelled/skipped queue items in order.
+
+    Each call assigns a monotonically increasing `completed_at` so the
+    scheduler's `ORDER BY completed_at DESC` reliably picks the latest as
+    the predecessor. `printer_id` is shared so all items count.
+    """
+    base_time = datetime(2026, 6, 6, 12, 0, 0, tzinfo=timezone.utc)
+    counter = {"n": 0}
+    printer_holder: dict = {}
+
+    async def _make_printer():
+        if "p" not in printer_holder:
+            printer_holder["p"] = await printer_factory()
+        return printer_holder["p"]
+
+    async def _add(status: str, error_message: str | None = None) -> PrintQueueItem:
+        printer = await _make_printer()
+        counter["n"] += 1
+        item = PrintQueueItem(
+            printer_id=printer.id,
+            status=status,
+            error_message=error_message,
+            completed_at=base_time + timedelta(minutes=counter["n"]),
+            require_previous_success=True,
+        )
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+        return item
+
+    async def _add_pending() -> PrintQueueItem:
+        printer = await _make_printer()
+        item = PrintQueueItem(
+            printer_id=printer.id,
+            status="pending",
+            require_previous_success=True,
+        )
+        db_session.add(item)
+        await db_session.commit()
+        await db_session.refresh(item)
+        return item
+
+    return {"add": _add, "add_pending": _add_pending}
+
+
+@pytest.mark.asyncio
+async def test_no_previous_item_returns_true(scheduler, db_session, queue_factory):
+    """First item in the queue has no predecessor → always passes."""
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_previous_completed_returns_true(scheduler, db_session, queue_factory):
+    await queue_factory["add"]("completed")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_previous_failed_returns_false(scheduler, db_session, queue_factory):
+    await queue_factory["add"]("failed")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is False
+
+
+@pytest.mark.asyncio
+async def test_previous_aborted_returns_false(scheduler, db_session, queue_factory):
+    """A printer-detected abort (e.g. clogged nozzle) is a real failure → blocks."""
+    await queue_factory["add"]("aborted")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is False
+
+
+@pytest.mark.asyncio
+async def test_previous_cancelled_returns_true_bug_a(scheduler, db_session, queue_factory):
+    """#1667 bug A: user cancellation is deliberate, not a failure → passes."""
+    await queue_factory["add"]("cancelled")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_skipped_predecessor_is_walked_past_bug_b(scheduler, db_session, queue_factory):
+    """#1667 bug B: a skipped item is not an attempt — query walks back to the
+    most recent real outcome instead of treating skipped as failed."""
+    await queue_factory["add"]("completed")  # real predecessor that should be found
+    await queue_factory["add"]("skipped", "Previous print failed or was aborted")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_only_skipped_history_returns_true(scheduler, db_session, queue_factory):
+    """Edge case: every prior item is skipped → no real predecessor found,
+    returns True (first-in-queue semantics)."""
+    await queue_factory["add"]("skipped", "Previous print failed or was aborted")
+    await queue_factory["add"]("skipped", "Previous print failed or was aborted")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_cascade_reporters_scenario(scheduler, db_session, queue_factory):
+    """The exact #1667 reporter scenario: failed → cancelled → skipped → pending.
+
+    Pre-fix: pending blocked because the buggy lookback walked past the
+    cancelled item (excluded) and the prior skipped item (included), found
+    the failed item, and returned False.
+    Post-fix: cancelled is the predecessor (skipped is excluded; cancelled
+    is included and passes), pending dispatches.
+    """
+    await queue_factory["add"]("failed")
+    await queue_factory["add"]("cancelled")
+    await queue_factory["add"]("skipped", "Previous print failed or was aborted")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_failed_then_cancelled_still_passes(scheduler, db_session, queue_factory):
+    """User cancelled after a failure → most recent action wins. The cancellation
+    is the user explicitly choosing to move on, so dispatching the next item
+    respects their intent."""
+    await queue_factory["add"]("failed")
+    await queue_factory["add"]("cancelled")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is True
+
+
+@pytest.mark.asyncio
+async def test_completed_then_failed_blocks(scheduler, db_session, queue_factory):
+    """Regression guard: a real failure after a previously-successful print
+    still gates downstream items. Only the MOST RECENT outcome matters."""
+    await queue_factory["add"]("completed")
+    await queue_factory["add"]("failed")
+    pending = await queue_factory["add_pending"]()
+    assert await scheduler._check_previous_success(db_session, pending) is False