diff --git a/backend/app/core/database.py b/backend/app/core/database.py
index fff2cb882..ff3dcb8e6 100644
--- a/backend/app/core/database.py
+++ b/backend/app/core/database.py
@@ -2978,10 +2978,16 @@ async def run_migrations(conn):
         conn,
         "CREATE INDEX IF NOT EXISTS ix_wallet_transactions_is_voided ON wallet_transactions (is_voided)",
     )
-    await _safe_execute(
-        conn,
-        "ALTER TABLE notification_providers ADD COLUMN on_billing_charge_failed BOOLEAN DEFAULT 1",
-    )
+    if is_sqlite():
+        await _safe_execute(
+            conn,
+            "ALTER TABLE notification_providers ADD COLUMN on_billing_charge_failed BOOLEAN DEFAULT 1",
+        )
+    else:
+        await _safe_execute(
+            conn,
+            "ALTER TABLE notification_providers ADD COLUMN on_billing_charge_failed BOOLEAN DEFAULT TRUE",
+        )
 
     # Reprints reuse their source archive, so archive uniqueness must only be
     # the legacy fallback for rows without a per-run UUID. The globally unique
diff --git a/backend/app/main.py b/backend/app/main.py
index dbe1481ec..1e3c7d117 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -744,9 +744,9 @@ async def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState, db
 
     possible_keys = _build_status_print_keys(printer_id, state)
 
-    # In-memory ownership is lost on every Bambuddy restart. The archive row is
-    # the durable source of truth; subtask_id is minted per print and avoids
-    # authorizing an unrelated job that happens to reuse the same filename.
+    # In-memory ownership is lost on every Bambuddy restart, so fall back to what
+    # is on disk. subtask_id is minted per print and pins the answer to the job
+    # actually running, rather than to an unrelated one that reuses a filename.
     raw_subtask_id = getattr(state, "subtask_id", None)
     subtask_id = str(raw_subtask_id).strip() if raw_subtask_id is not None else ""
     if subtask_id in ("", "0"):
@@ -765,15 +765,44 @@ async def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState, db
         .limit(1)
     )
     archive = result.scalar_one_or_none()
-    if archive is None:
-        return False
 
-    # Rehydrate the fast in-memory path for subsequent status frames. Include
-    # both the archive filename and every normalized key reported by MQTT.
-    _active_prints[(printer_id, archive.filename)] = archive.id
-    for key in possible_keys:
-        _active_prints[key] = archive.id
-    return True
+    # An archive row on its own proves nothing: `on_print_start` archives every
+    # print it observes, including ones started from Bambu Studio or Handy, and
+    # stamps them with the same status and subtask_id. Authorizing on its mere
+    # existence would disable the kill switch the moment the 3MF finishes
+    # downloading. Only a dispatch marker Bambuddy writes itself counts —
+    # `billing_run_id` (minted per dispatch in the scheduler) or `created_by_id`
+    # (carried over from the queue item that started it).
+    if archive is not None and (archive.billing_run_id is not None or archive.created_by_id is not None):
+        # Rehydrate the fast in-memory path for subsequent status frames. Include
+        # both the archive filename and every normalized key reported by MQTT.
+        _active_prints[(printer_id, archive.filename)] = archive.id
+        for key in possible_keys:
+            _active_prints[key] = archive.id
+        return True
+
+    # No dispatch marker. Before calling this someone else's print, check whether
+    # Bambuddy has a job of its own running on this printer: a library-file
+    # dispatch has no archive at send time, and an archive created seconds later
+    # by `on_print_start` carries neither marker. The queue row, which the
+    # scheduler commits to status="printing" before the MQTT send, is the one
+    # durable record every Bambuddy print has. It cannot be tied to this
+    # subtask_id, so it is grounds to defer, never to authorize — stopping a
+    # print is irreversible, and refusing to act costs nothing but a log line.
+    from backend.app.models.print_queue import PrintQueueItem
+
+    dispatched_here = await db.scalar(
+        select(PrintQueueItem.id)
+        .where(
+            PrintQueueItem.printer_id == printer_id,
+            PrintQueueItem.status == "printing",
+        )
+        .limit(1)
+    )
+    if dispatched_here is not None:
+        return None
+
+    return False
 
 
 async def _send_kill_switch_provider_notification(
diff --git a/backend/tests/unit/test_billing_run_id_migration.py b/backend/tests/unit/test_billing_run_id_migration.py
index 8484ea5c1..d2b44dfd0 100644
--- a/backend/tests/unit/test_billing_run_id_migration.py
+++ b/backend/tests/unit/test_billing_run_id_migration.py
@@ -10,6 +10,22 @@ import backend.app.models.print_log  # noqa: F401 - required by a legacy ALTER i
 from backend.app.core.database import Base, run_migrations
 
 
+@pytest.fixture(autouse=True)
+def force_sqlite_dialect(monkeypatch):
+    """The engine below is SQLite, but settings.database_url may point at Postgres in a
+    dev config — and run_migrations branches on the global dialect, not on the
+    connection. Without this the Postgres branch runs against SQLite and the migration
+    fails on Postgres-only syntax. Same fixture as test_ldap_migration.py."""
+    from backend.app.core import db_dialect
+
+    monkeypatch.setattr(db_dialect, "is_sqlite", lambda: True)
+    monkeypatch.setattr(db_dialect, "is_postgres", lambda: False)
+    # database.py imported is_sqlite at module load time — patch there too.
+    from backend.app.core import database as database_module
+
+    monkeypatch.setattr(database_module, "is_sqlite", lambda: True)
+
+
 @pytest.mark.asyncio
 async def test_billing_run_columns_and_legacy_archive_index_are_migrated(tmp_path):
     engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'billing-run.db'}")
diff --git a/backend/tests/unit/test_printer_kill_switch.py b/backend/tests/unit/test_printer_kill_switch.py
index d2a184376..4e750b434 100644
--- a/backend/tests/unit/test_printer_kill_switch.py
+++ b/backend/tests/unit/test_printer_kill_switch.py
@@ -286,7 +286,14 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
 @pytest.mark.asyncio
 @pytest.mark.parametrize("printer_state", ["RUNNING", "PAUSE"])
 async def test_persisted_print_is_authorized_after_restart(monkeypatch, printer_state):
-    archive = SimpleNamespace(id=123, filename="owned_job.gcode.3mf")
+    # billing_run_id is the marker the scheduler stamps on its own dispatches;
+    # an archive without one proves only that Bambuddy watched the print.
+    archive = SimpleNamespace(
+        id=123,
+        filename="owned_job.gcode.3mf",
+        billing_run_id="d7c1f0b2-0000-4000-8000-000000000001",
+        created_by_id=None,
+    )
     query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
     db = SimpleNamespace(execute=AsyncMock(return_value=query_result))
 
@@ -368,3 +375,94 @@ async def test_kill_switch_defers_when_restart_identity_is_not_available(monkeyp
 
     assert authorization is None
     db.execute.assert_not_awaited()
+
+
+def _authorization_db(archive, dispatched_queue_item_id=None):
+    """Fake session answering the two lookups `_is_bambuddy_authorized_print` makes."""
+
+    query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
+    return SimpleNamespace(
+        execute=AsyncMock(return_value=query_result),
+        scalar=AsyncMock(return_value=dispatched_queue_item_id),
+    )
+
+
+def _running_state(subtask_id="external-task-9"):
+    return SimpleNamespace(
+        current_print=None,
+        subtask_name="some_job",
+        subtask_id=subtask_id,
+        gcode_file="some_job.gcode.3mf",
+    )
+
+
+@pytest.mark.asyncio
+async def test_archive_without_a_dispatch_marker_is_not_authorization(monkeypatch):
+    """on_print_start archives prints started from Studio or Handy too.
+
+    Those rows carry the same status and subtask_id as Bambuddy's own, so treating
+    the row's existence as proof would switch the feature off a few seconds into
+    every foreign print — as soon as the 3MF finished downloading.
+    """
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    observed_only = SimpleNamespace(
+        id=55,
+        filename="some_job.gcode.3mf",
+        billing_run_id=None,
+        created_by_id=None,
+    )
+    db = _authorization_db(observed_only, dispatched_queue_item_id=None)
+
+    assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is False
+    assert (9, "some_job.gcode.3mf") not in main_module._active_prints
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "marker",
+    [
+        {"billing_run_id": "9f0c2b6e-0000-4000-8000-00000000abcd", "created_by_id": None},
+        {"billing_run_id": None, "created_by_id": 4},
+    ],
+    ids=["billing_run_id", "created_by_id"],
+)
+async def test_either_dispatch_marker_authorizes_after_a_restart(monkeypatch, marker):
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    archive = SimpleNamespace(id=77, filename="some_job.gcode.3mf", **marker)
+    db = _authorization_db(archive, dispatched_queue_item_id=None)
+
+    assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is True
+    assert main_module._active_prints[(9, "some_job.gcode.3mf")] == 77
+    # The fast path is rehydrated, so the queue is never consulted.
+    db.scalar.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_defers_while_bambuddy_has_a_job_running_on_that_printer(monkeypatch):
+    """A library-file dispatch has no archive at send time, and the row created for
+    it moments later by on_print_start carries neither marker. The queue row is the
+    only durable trace, and it cannot be tied to a subtask_id — so it defers."""
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    unmarked = SimpleNamespace(id=56, filename="some_job.gcode.3mf", billing_run_id=None, created_by_id=None)
+    db = _authorization_db(unmarked, dispatched_queue_item_id=310)
+
+    assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is None
+    # Deferring must not authorize the print for every later frame.
+    assert (9, "some_job.gcode.3mf") not in main_module._active_prints
+
+
+@pytest.mark.asyncio
+async def test_defers_when_the_dispatch_has_not_been_archived_yet(monkeypatch):
+    """Restart during the window between the MQTT send and the 3MF download."""
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    db = _authorization_db(None, dispatched_queue_item_id=311)
+
+    assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is None
+
+
+@pytest.mark.asyncio
+async def test_foreign_print_with_no_archive_and_no_dispatch_is_unauthorized(monkeypatch):
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    db = _authorization_db(None, dispatched_queue_item_id=None)
+
+    assert await main_module._is_bambuddy_authorized_print(9, _running_state(), db) is False