Three fixes on top of the billing branch, all found by running the suite against

both dialects rather than one.

Postgres upgrades never got as far as the finance schema.

  database.py added on_billing_charge_failed with BOOLEAN DEFAULT 1. The 1 is a
  SQLite-ism; Postgres answers DatatypeMismatchError, and _safe_execute
  deliberately re-raises anything that is not an idempotency error, so
  run_migrations died there and rolled the whole transaction back. No finance
  tables, no columns, and the app does not start. Six lines above, the same
  change gets is_voided right with an is_sqlite() branch, so this was an
  oversight rather than a decision. Now branched the same way.

  This also explains the four test_security.py::TestBackupKeyFiles failures
  reporting "column print_archives.cost_center_id does not exist". That column's
  migration exists and works -- it simply never ran, because every startup
  aborted before committing. Reproduced against Postgres 16 by building a
  pre-billing schema from dev and upgrading over it: fails without this,
  completes with it, and re-running the migrations or starting from an empty
  database are both clean.

test_billing_run_id_migration.py failed on any Postgres-configured checkout.

  It builds its own SQLite engine, but run_migrations branches on the global
  dialect rather than the connection in hand, so on a box whose DATABASE_URL
  points at Postgres it emitted md5(random()::text) and btrim() into SQLite.
  Given the same fixture test_ldap_migration.py already carries for exactly this
  reason. The suite now agrees across dialects -- 9190 passed either way, where
  it used to be 9184 on one and 9183 on the other.

The kill switch could not tell a print Bambuddy started from one it merely
watched.

  Authorization fell back to a print_archives row in status="printing" matched on
  subtask_id. But 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 -- the code says as much where it notes "a print Bambuddy didn't
  dispatch". So a foreign print became authorized the moment its 3MF finished
  downloading, and _active_prints was rehydrated from it, making that permanent.
  The switch fired only inside the download race, and never afterwards. Neither
  test caught it: one stubs the authorization call to False, the other stubs the
  query to return an archive, so the real lookup was never exercised against a
  foreign print.

  Authorization now requires a marker Bambuddy writes itself: billing_run_id,
  minted per dispatch in the scheduler, or created_by_id carried over from the
  queue item. Failing that, it looks for a queue row in status="printing" on that
  printer -- committed before the MQTT send, and the only durable trace a
  library-file dispatch leaves, since those have no archive at send time and the
  row created for them moments later carries neither marker. That row cannot be
  tied to a subtask_id, so it defers rather than authorizes.

  Deferring also closes a false positive the previous version shared: a restart
  in the window between the send and the download left no archive at all, and a
  Bambuddy print was stopped as unauthorized. Stopping a print is irreversible
  and declining to act costs a log line, so ambiguity resolves that way.

  Tests cover an unmarked archive not being authorization and not entering
  _active_prints, either marker alone authorizing and rehydrating the fast path
  without touching the queue, an unmarked archive with a live dispatch deferring,
  a dispatch not yet archived deferring, and nothing at all being unauthorized.

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

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(

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'}")

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