implemented pr feedback #2

backend/app/api/routes/finance.py

@@ -3,7 +3,7 @@ from datetime import datetime, timezone
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from fastapi import APIRouter, Depends, HTTPException, Query
-from sqlalchemy import and_, case, func, or_, select
+from sqlalchemy import case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
@@ -39,6 +39,11 @@ from backend.app.schemas.finance import (
     WalletTransactionListResponse,
     WalletTransactionResponse,
 )
+from backend.app.services.finance_balance import (
+    is_personal_transaction,
+    personal_balance_condition,
+    sync_personal_wallet_balance,
+)
 
 router = APIRouter(prefix="/finance", tags=["finance"])
 
@@ -306,35 +311,21 @@ async def _build_personal_balance_map(db: AsyncSession, user_id: int) -> dict[in
         select(
             WalletTransaction.id,
             WalletTransaction.amount,
-            WalletTransaction.cost_center_id,
-            CostCenter.is_private,
-            CostCenter.owner_user_id,
-        )
-        .where(
-            WalletTransaction.user_id == user_id,
         )
         .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id)
+        .where(WalletTransaction.user_id == user_id, personal_balance_condition(user_id))
         .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc())
     )
 
     running_balance = 0.0
     balance_map: dict[int, float] = {}
-    for transaction_id, amount, cost_center_id, is_private, owner_user_id in result.all():
-        is_personal_cost_center = bool(cost_center_id is not None and is_private and owner_user_id == user_id)
-        if cost_center_id is None or is_personal_cost_center:
-            running_balance += float(amount)
-            balance_map[int(transaction_id)] = running_balance
+    for transaction_id, amount in result.all():
+        running_balance += float(amount)
+        balance_map[int(transaction_id)] = running_balance
 
     return balance_map
 
 
-def _personal_balance_condition(user_id: int):
-    return or_(
-        WalletTransaction.cost_center_id.is_(None),
-        and_(CostCenter.is_private.is_(True), CostCenter.owner_user_id == user_id),
-    )
-
-
 async def _create_wallet_adjustment(
     db: AsyncSession,
     *,
@@ -351,6 +342,7 @@ async def _create_wallet_adjustment(
         await _get_cost_center_or_404(db, cost_center_id)
 
     wallet = await _get_or_create_wallet(db, target_user_id)
+    affects_personal_wallet = await is_personal_transaction(db, target_user_id, cost_center_id)
 
     # Calculate balance_after for this specific transaction context
     if cost_center_id is None:
@@ -358,7 +350,6 @@ async def _create_wallet_adjustment(
         new_balance = wallet.balance + amount
         if new_balance < 0:
             raise HTTPException(status_code=400, detail="Insufficient balance for withdrawal")
-        wallet.balance = new_balance
         balance_after = new_balance
     else:
         # Cost-center transaction: validate against cost center balance only (global, not per-user)
@@ -372,7 +363,8 @@ async def _create_wallet_adjustment(
         if new_cc_balance < 0:
             raise HTTPException(status_code=400, detail="Insufficient cost center balance for withdrawal")
         balance_after = new_cc_balance
-        # Do NOT update wallet.balance for cost-center transactions
+        if affects_personal_wallet and wallet.balance + amount < 0:
+            raise HTTPException(status_code=400, detail="Insufficient balance for withdrawal")
 
     tx = WalletTransaction(
         user_id=target_user_id,
@@ -385,12 +377,13 @@ async def _create_wallet_adjustment(
     )
     db.add(tx)
     await db.flush()
+    await sync_personal_wallet_balance(db, wallet)
     await db.commit()
     await db.refresh(wallet)
     await db.refresh(tx)
 
     # Return appropriate balance based on transaction type
-    if cost_center_id is None:
+    if affects_personal_wallet:
         # Personal transaction: return user wallet balance
         response_balance = _to_balance_response(wallet)
     else:
@@ -416,19 +409,7 @@ async def get_my_balance(
     """Return the current user's wallet balance."""
     user = await _require_authenticated_user(current_user)
     wallet = await _get_or_create_wallet(db, user.id)
-    personal_balance_result = await db.execute(
-        select(func.coalesce(func.sum(WalletTransaction.amount), 0.0))
-        .select_from(WalletTransaction)
-        .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id)
-        .where(WalletTransaction.user_id == user.id, _personal_balance_condition(user.id))
-    )
-    personal_balance = float(personal_balance_result.scalar_one() or 0.0)
-    return WalletBalanceResponse(
-        user_id=user.id,
-        balance=personal_balance,
-        currency=wallet.currency,
-        updated_at=wallet.updated_at,
-    )
+    return _to_balance_response(wallet)
 
 
 @router.get("/me/transactions", response_model=WalletTransactionListResponse)
@@ -504,47 +485,10 @@ async def get_all_transactions(
 
 
 async def _rebuild_wallet_ledger_for_user(db: AsyncSession, user_id: int) -> None:
-    """Recompute `balance_after` for all wallet transactions of a user.
+    """Rebuild through the same canonical ledger repair used at startup."""
+    from backend.app.core.database import repair_wallet_ledger_internal
 
-    - Personal transactions (cost_center_id=None): running balance per user
-    - Cost-center transactions: running balance GLOBAL for entire cost center (not per-user)
-    - Also updates the user's wallet balance (sum of personal transactions only)
-    """
-    result = await db.execute(
-        select(WalletTransaction)
-        .where(WalletTransaction.user_id == user_id)
-        .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc())
-    )
-    user_transactions = result.scalars().all()
-
-    # Handle personal transactions (cost_center_id=None)
-    personal_balance = 0.0
-    for tx in user_transactions:
-        if tx.cost_center_id is None:
-            personal_balance += float(tx.amount)
-            tx.balance_after = personal_balance
-            db.add(tx)
-
-    for cc_id in {tx.cost_center_id for tx in user_transactions if tx.cost_center_id is not None}:
-        # Get all transactions for this cost center (all users, all time)
-        result_all_cc = await db.execute(
-            select(WalletTransaction)
-            .where(WalletTransaction.cost_center_id == cc_id)
-            .order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc())
-        )
-        all_cc_transactions = result_all_cc.scalars().all()
-
-        running = 0.0
-        for tx in all_cc_transactions:
-            running += float(tx.amount)
-            tx.balance_after = running
-            db.add(tx)
-
-    # Update user wallet balance (sum of all personal transactions only)
-    wallet = await _get_or_create_wallet(db, user_id)
-    wallet.balance = personal_balance
-    await db.flush()
-    await db.commit()
+    await repair_wallet_ledger_internal(db)
 
 
 @router.delete("/transactions/{transaction_id}")
@@ -794,51 +738,19 @@ async def rebuild_balance_ledger(
     """Recompute balance_after for all wallet transactions.
 
     This rebuilds the running balance for all users and cost centers.
-    - Personal transactions: per-user running balance
+    - Personal transactions: unassigned plus the user's own private cost center
     - Cost-center transactions: global running balance for the entire cost center
     """
     await _require_authenticated_user(current_user)
 
-    # Get ALL transactions sorted by timestamp
-    result = await db.execute(
-        select(WalletTransaction).order_by(WalletTransaction.created_at.asc(), WalletTransaction.id.asc())
-    )
-    all_transactions = result.scalars().all()
+    from backend.app.core.database import repair_wallet_ledger_internal
 
-    # Build running balances per (user, cost_center_id) pair
-    # For each cost center, track its global running balance
-    # For each user's personal balance, track that separately
-    cc_running_balances: dict[int, float] = {}  # cost_center_id -> running balance
-    user_personal_balances: dict[int, float] = {}  # user_id -> personal running balance
-
-    tx_updates: list[tuple[WalletTransaction, float]] = []
-
-    for tx in all_transactions:
-        if tx.cost_center_id is None:
-            # Personal transaction: per-user running balance
-            current = user_personal_balances.get(tx.user_id, 0.0)
-            new_balance = current + float(tx.amount)
-            user_personal_balances[tx.user_id] = new_balance
-            tx_updates.append((tx, new_balance))
-        else:
-            # Cost-center transaction: global running balance for this cost center
-            current = cc_running_balances.get(tx.cost_center_id, 0.0)
-            new_balance = current + float(tx.amount)
-            cc_running_balances[tx.cost_center_id] = new_balance
-            tx_updates.append((tx, new_balance))
-
-    # Update all transactions with the new balance_after values
-    for tx, new_balance in tx_updates:
-        tx.balance_after = new_balance
-        db.add(tx)
-
-    await db.flush()
-    await db.commit()
+    rebuilt = await repair_wallet_ledger_internal(db)
 
     return {
         "status": "success",
-        "transactions_rebuilt": len(all_transactions),
-        "message": f"Rebuilt balance_after for {len(all_transactions)} transactions",
+        "transactions_rebuilt": rebuilt,
+        "message": f"Rebuilt {rebuilt} wallet ledger values",
     }
 
 
@@ -1053,6 +965,12 @@ async def delete_cost_center(
         raise HTTPException(status_code=400, detail="Cost center can only be deleted when balance is 0")
 
     await db.delete(center)
+    await db.flush()
+    # ON DELETE SET NULL reclassifies the former shared-center entries as
+    # personal transactions, so synchronize affected wallets immediately.
+    from backend.app.core.database import repair_wallet_ledger_internal
+
+    await repair_wallet_ledger_internal(db)
     await db.commit()
     return {"status": "success"}
 

backend/app/api/routes/print_queue.py

@@ -51,6 +51,7 @@ from backend.app.services.print_batch import (
     load_progress,
     refresh_batch_status,
 )
+from backend.app.services.print_cost_estimate import estimate_queue_source_cost
 from backend.app.utils.printer_models import (
     is_gcode_compatible,
 )
@@ -279,6 +280,55 @@ async def _resolve_source_path(db: AsyncSession, item: PrintQueueItem) -> Path |
     return None
 
 
+async def _trusted_item_estimated_cost(
+    db: AsyncSession,
+    item: PrintQueueItem,
+    *,
+    printer_id: int | None,
+    plate_id: int | None,
+    ams_mapping: list[int] | str | None,
+) -> float | None:
+    """Recompute an existing queue item's cost from its persisted source."""
+
+    if item.archive_id:
+        archive = await db.scalar(select(PrintArchive).where(PrintArchive.id == item.archive_id))
+        return await estimate_queue_source_cost(
+            db,
+            archive=archive,
+            plate_id=plate_id,
+            ams_mapping=ams_mapping,
+            printer_id=printer_id,
+        )
+    if item.library_file_id:
+        library_file = await db.scalar(LibraryFile.active().where(LibraryFile.id == item.library_file_id))
+        return await estimate_queue_source_cost(
+            db,
+            library_file=library_file,
+            plate_id=plate_id,
+            ams_mapping=ams_mapping,
+            printer_id=printer_id,
+        )
+
+    result = await db.execute(
+        select(PrintQueueVariant, LibraryFile)
+        .join(LibraryFile, LibraryFile.id == PrintQueueVariant.library_file_id)
+        .where(PrintQueueVariant.queue_item_id == item.id)
+    )
+    estimates = [
+        await estimate_queue_source_cost(
+            db,
+            library_file=library_file,
+            plate_id=variant.plate_id,
+            ams_mapping=variant.ams_mapping,
+            printer_id=printer_id,
+        )
+        for variant, library_file in result.all()
+    ]
+    if not estimates or any(cost is None for cost in estimates):
+        return None
+    return max(cost for cost in estimates if cost is not None)
+
+
 def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
     """Add nested archive/printer/library_file info to response."""
     # Parse ams_mapping from JSON string BEFORE model_validate
@@ -945,10 +995,39 @@ async def add_to_queue(
         if not project_result.scalar_one_or_none():
             raise HTTPException(status_code=404, detail="Project not found")
 
+    # Security boundary: the browser's estimated_cost is only a display hint.
+    # Budget enforcement and the persisted reservation value must be derived
+    # from the server-owned archive/library metadata and spool assignments.
+    if variant_specs:
+        variant_costs = [
+            await estimate_queue_source_cost(
+                db,
+                library_file=variant_file,
+                plate_id=spec.plate_id,
+                ams_mapping=spec.ams_mapping,
+                printer_id=data.printer_id,
+            )
+            for spec, variant_file, _model in variant_specs
+        ]
+        trusted_estimated_cost = (
+            max(cost for cost in variant_costs if cost is not None)
+            if variant_costs and all(cost is not None for cost in variant_costs)
+            else None
+        )
+    else:
+        trusted_estimated_cost = await estimate_queue_source_cost(
+            db,
+            archive=archive,
+            library_file=library_file,
+            plate_id=data.plate_id,
+            ams_mapping=data.ams_mapping,
+            printer_id=data.printer_id,
+        )
+
     await validate_print_budget(
         db,
         cost_center_id=data.cost_center_id,
-        estimated_cost=data.estimated_cost,
+        estimated_cost=trusted_estimated_cost,
         current_user=current_user,
         quantity=quantity,
     )
@@ -1010,7 +1089,7 @@ async def add_to_queue(
             archive_id=data.archive_id,
             library_file_id=data.library_file_id,
             cost_center_id=data.cost_center_id,
-            estimated_cost=data.estimated_cost,
+            estimated_cost=trusted_estimated_cost,
             scheduled_time=data.scheduled_time,
             require_previous_success=data.require_previous_success,
             auto_off_after=data.auto_off_after,
@@ -1161,16 +1240,25 @@ async def bulk_update_queue_items(
             skipped_count += 1
             continue
 
+        item_update_data = update_data.copy()
         if validates_billing_fields:
+            trusted_estimated_cost = await _trusted_item_estimated_cost(
+                db,
+                item,
+                printer_id=item_update_data.get("printer_id", item.printer_id),
+                plate_id=item.plate_id,
+                ams_mapping=item.ams_mapping,
+            )
+            item_update_data["estimated_cost"] = trusted_estimated_cost
             await validate_print_budget(
                 db,
-                cost_center_id=update_data.get("cost_center_id", item.cost_center_id),
-                estimated_cost=update_data.get("estimated_cost", item.estimated_cost),
+                cost_center_id=item_update_data.get("cost_center_id", item.cost_center_id),
+                estimated_cost=trusted_estimated_cost,
                 current_user=user,
                 exclude_queue_item_id=item.id,
             )
 
-        for field, value in update_data.items():
+        for field, value in item_update_data.items():
             setattr(item, field, value)
         updated_count += 1
 
@@ -1826,10 +1914,19 @@ async def update_queue_item(
             json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None
         )
 
+    trusted_estimated_cost = await _trusted_item_estimated_cost(
+        db,
+        item,
+        printer_id=update_data.get("printer_id", item.printer_id),
+        plate_id=update_data.get("plate_id", item.plate_id),
+        ams_mapping=update_data.get("ams_mapping", item.ams_mapping),
+    )
+    update_data["estimated_cost"] = trusted_estimated_cost
+
     await validate_print_budget(
         db,
         cost_center_id=update_data.get("cost_center_id", item.cost_center_id),
-        estimated_cost=update_data.get("estimated_cost", item.estimated_cost),
+        estimated_cost=trusted_estimated_cost,
         current_user=user,
         exclude_queue_item_id=item.id,
     )
@@ -2169,6 +2266,14 @@ async def start_queue_item(
     if item.status != "pending":
         raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'")
 
+    item.estimated_cost = await _trusted_item_estimated_cost(
+        db,
+        item,
+        printer_id=item.printer_id,
+        plate_id=item.plate_id,
+        ams_mapping=item.ams_mapping,
+    )
+
     await validate_print_budget(
         db,
         cost_center_id=item.cost_center_id,

backend/app/core/database.py

@@ -1232,6 +1232,14 @@ async def _migrate_create_finance_indexes(conn) -> None:
         await _safe_execute(conn, statement)
 
 
+async def _migrate_add_print_archive_cost_center(conn) -> None:
+    """Add the nullable cost-center link missing from pre-billing archives."""
+    await _safe_execute(
+        conn,
+        "ALTER TABLE print_archives ADD COLUMN cost_center_id INTEGER REFERENCES cost_centers(id) ON DELETE SET NULL",
+    )
+
+
 async def run_migrations(conn):
     """Run all schema migrations and data backfills on startup.
 
@@ -1585,6 +1593,9 @@ async def run_migrations(conn):
     except (OperationalError, ProgrammingError):
         pass  # Already applied
 
+    # Migration: Add cost_center_id column to print_archives for billing metadata
+    await _migrate_add_print_archive_cost_center(conn)
+
     # Migration: Add wiki_url column to maintenance_types for documentation links
     await _safe_execute(conn, "ALTER TABLE maintenance_types ADD COLUMN wiki_url VARCHAR(500)")
 
@@ -4913,21 +4924,11 @@ async def repair_wallet_ledger():
 
     This is called during database initialization to ensure all balance_after
     values are correct after code changes. Fixes old balance_after values to:
-    - Personal transactions: per-user running balance
+    - Personal transactions: unassigned plus the user's own private cost center
     - Cost-center transactions: global running balance for the entire cost center
-    Also updates UserWallet.balance to match final personal transaction balance.
+    Also updates UserWallet.balance to the canonical personal ledger sum.
     """
-    from sqlalchemy import select
-
-    from backend.app.models.finance import WalletTransaction
-
     async with async_session() as session:
-        # Check if there are any transactions first
-        result = await session.execute(select(WalletTransaction).limit(1))
-        if not result.scalar_one_or_none():
-            logger.info("No wallet transactions found, skipping ledger rebuild")
-            return
-
         updated_count = await repair_wallet_ledger_internal(session)
         await session.commit()
 
@@ -4942,7 +4943,8 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
     """
     from sqlalchemy import select
 
-    from backend.app.models.finance import UserWallet, WalletTransaction
+    from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction
+    from backend.app.services.finance_balance import transaction_affects_personal_balance
 
     # Get ALL transactions sorted by timestamp
     result = await session.execute(
@@ -4950,8 +4952,10 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
     )
     all_transactions = result.scalars().all()
 
-    if not all_transactions:
-        return 0
+    center_rows = await session.execute(select(CostCenter.id, CostCenter.is_private, CostCenter.owner_user_id))
+    centers = {
+        int(center_id): (bool(is_private), owner_user_id) for center_id, is_private, owner_user_id in center_rows
+    }
 
     # Build running balances per (user, cost_center_id) pair
     cc_running_balances: dict[int, float] = {}  # cost_center_id -> running balance
@@ -4960,6 +4964,13 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
     tx_updates: list[tuple[WalletTransaction, float]] = []
 
     for tx in all_transactions:
+        center_is_private, center_owner_user_id = centers.get(tx.cost_center_id, (False, None))
+        affects_personal = transaction_affects_personal_balance(
+            tx.user_id,
+            tx.cost_center_id,
+            is_private=center_is_private,
+            owner_user_id=center_owner_user_id,
+        )
         if tx.cost_center_id is None:
             # Personal transaction: per-user running balance
             current = user_personal_balances.get(tx.user_id, 0.0)
@@ -4973,6 +4984,11 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
             cc_running_balances[tx.cost_center_id] = new_balance
             tx_updates.append((tx, new_balance))
 
+            # The owner's private cost center is also part of that user's
+            # personal wallet. Shared centers never enter this sum.
+            if affects_personal:
+                user_personal_balances[tx.user_id] = user_personal_balances.get(tx.user_id, 0.0) + float(tx.amount)
+
     # Update all transactions with the new balance_after values
     updated_count = 0
     for tx, new_balance in tx_updates:
@@ -4981,11 +4997,12 @@ async def repair_wallet_ledger_internal(session: AsyncSession):
             session.add(tx)
             updated_count += 1
 
-    # Update UserWallet balances to match final personal balances
-    for user_id, balance in user_personal_balances.items():
-        wallet_result = await session.execute(select(UserWallet).where(UserWallet.user_id == user_id))
-        wallet = wallet_result.scalar_one_or_none()
-        if wallet and wallet.balance != balance:
+    # Update every wallet, including stale wallets whose canonical balance is
+    # now zero because their last personal transaction was deleted.
+    wallet_result = await session.execute(select(UserWallet))
+    for wallet in wallet_result.scalars().all():
+        balance = round(user_personal_balances.get(wallet.user_id, 0.0), 2)
+        if wallet.balance != balance:
             wallet.balance = balance
             session.add(wallet)
             updated_count += 1

backend/app/main.py

@@ -106,6 +106,7 @@ from backend.app.services.mqtt_relay import mqtt_relay
 from backend.app.services.mqtt_smart_plug import mqtt_smart_plug_service
 from backend.app.services.notification_service import notification_service
 from backend.app.services.obico_detection import obico_detection_service
+from backend.app.services.print_cost_estimate import plate_scoped_run_estimate as _plate_scoped_run_estimate
 from backend.app.services.print_scheduler import scheduler as print_scheduler
 from backend.app.services.printer_manager import (
     init_printer_connections,
@@ -425,6 +426,11 @@ _first_layer_notified: dict[int, bool] = {}
 # Track whether we already sent a kill-switch stop for the current unauthorized print
 _unauthorized_print_kill_sent: set[int] = set()
 
+# Provider notification started when the kill switch stops a print. The later
+# MQTT print-complete callback awaits this task and only sends its regular
+# provider notification when the immediate attempt failed.
+_kill_switch_notification_tasks: dict[int, asyncio.Task[bool]] = {}
+
 # Track HMS errors that have been notified: {printer_id: set of error codes}
 # This prevents sending duplicate notifications for the same error
 _notified_hms_errors: dict[int, set[str]] = {}
@@ -689,14 +695,94 @@ def _build_status_print_keys(printer_id: int, state: PrinterState) -> list[tuple
     return possible_keys
 
 
-def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState) -> bool:
-    """Return True when the current status belongs to a print started by Bambuddy."""
+async def _is_bambuddy_authorized_print(printer_id: int, state: PrinterState, db) -> bool | None:
+    """Resolve whether the current print was started by Bambuddy.
+
+    ``None`` means identity is not yet safe to decide. The kill switch must
+    defer in that case: stopping a print is irreversible, and the first status
+    frames after a restart may arrive before all subtask fields are populated.
+    """
 
     if printer_manager.get_current_print_user(printer_id):
         return True
 
     possible_keys = _build_status_print_keys(printer_id, state)
-    return any(key in _expected_prints or key in _active_prints for key in possible_keys)
+    if any(key in _expected_prints or key in _active_prints for key in possible_keys):
+        return True
+
+    # 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.
+    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"):
+        return None
+
+    from backend.app.models.archive import PrintArchive
+
+    result = await db.execute(
+        select(PrintArchive)
+        .where(
+            PrintArchive.printer_id == printer_id,
+            PrintArchive.status == "printing",
+            PrintArchive.subtask_id == subtask_id,
+        )
+        .order_by(PrintArchive.created_at.desc())
+        .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
+
+
+async def _send_kill_switch_provider_notification(
+    printer_id: int,
+    printer_name: str,
+    data: dict,
+) -> bool:
+    """Send the immediate print-stopped provider notification.
+
+    Returning a success flag lets the normal MQTT completion path retry when
+    this early notification could not be delivered.
+    """
+
+    logger = logging.getLogger(__name__)
+    try:
+        async with async_session() as db:
+            await notification_service.on_print_complete(
+                printer_id,
+                printer_name,
+                "stopped",
+                data,
+                db,
+            )
+        return True
+    except Exception as e:
+        logger.warning(
+            "[KILL SWITCH] Immediate provider notification failed for printer %s: %s",
+            printer_id,
+            e,
+        )
+        return False
+
+
+async def _kill_switch_notification_already_sent(task: asyncio.Task[bool] | None) -> bool:
+    """Wait for an immediate kill-switch notification, if one was scheduled."""
+
+    if task is None:
+        return False
+    try:
+        return await task
+    except Exception as e:
+        logging.getLogger(__name__).warning("[KILL SWITCH] Notification task failed: %s", e)
+        return False
 
 
 async def _get_plug_energy(plug, db) -> dict | None:
@@ -888,41 +974,6 @@ def _compute_run_filament_grams(
     return None
 
 
-def _plate_scoped_run_estimate(archive, full_path) -> tuple[float | None, float | None]:
-    """Per-run (grams, cost) scoped to the plate this run actually printed (#2614).
-
-    ``PrintArchive.filament_used_grams`` / ``.cost`` are the sum over EVERY plate of
-    the source 3MF — correct for the archive card and project rollup, but wrong for a
-    single plate dispatched from a multi-plate file: without scoping, each printed
-    plate of a 22-plate file logs the whole ~12 kg and inflates every statistic. When
-    the archive carries a ``plate_id`` and its 3MF is on disk, return that plate's
-    slicer estimate instead; cost is scaled by the plate's share of the whole so it
-    stays consistent with the scoped grams without re-doing the filament price lookup.
-    Falls back to the archive's whole-file values when there's no plate to scope to.
-    """
-    whole_grams = archive.filament_used_grams
-    if archive.plate_id is None or full_path is None or not full_path.exists():
-        return whole_grams, archive.cost
-    try:
-        from backend.app.utils.threemf_tools import extract_plate_metadata_from_3mf
-
-        plate_grams = extract_plate_metadata_from_3mf(full_path, archive.plate_id).filament_used_grams
-    except Exception as exc:
-        logging.getLogger(__name__).debug(
-            "[#2614] plate-scoped estimate failed for archive %s (plate %s): %s",
-            archive.id,
-            archive.plate_id,
-            exc,
-        )
-        return whole_grams, archive.cost
-    if not plate_grams or plate_grams <= 0:
-        return whole_grams, archive.cost
-    plate_cost = archive.cost
-    if archive.cost and whole_grams and whole_grams > 0:
-        plate_cost = round(archive.cost * (plate_grams / whole_grams), 2)
-    return round(plate_grams, 2), plate_cost
-
-
 def _get_start_ams_mapping(data: dict, archive_id: int | None) -> list[int] | None:
     """Resolve AMS mapping for print start without consuming stored queue/reprint state."""
     stored_ams_mapping = data.get("ams_mapping")
@@ -1380,17 +1431,31 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
         _unauthorized_print_kill_sent.discard(printer_id)
     else:
         kill_switch_enabled = False
+        authorization: bool | None = None
         status_logger = logging.getLogger(__name__)
         try:
             async with async_session() as db:
                 from backend.app.services.finance_budget import is_printer_kill_switch_enabled
 
                 kill_switch_enabled = await is_printer_kill_switch_enabled(db)
+                if kill_switch_enabled:
+                    authorization = await _is_bambuddy_authorized_print(printer_id, state, db)
         except Exception as e:
-            status_logger.warning("[KILL SWITCH] Failed to read kill-switch setting for printer %s: %s", printer_id, e)
+            # Fail safe: a database/reconciliation error must never turn into an
+            # irreversible stop of a print whose ownership is still unknown.
+            authorization = None
+            status_logger.warning(
+                "[KILL SWITCH] Failed to reconcile print authorization for printer %s: %s", printer_id, e
+            )
 
-        if not kill_switch_enabled or _is_bambuddy_authorized_print(printer_id, state):
+        if not kill_switch_enabled or authorization is True:
             _unauthorized_print_kill_sent.discard(printer_id)
+        elif authorization is None:
+            _unauthorized_print_kill_sent.discard(printer_id)
+            status_logger.debug(
+                "[KILL SWITCH] Deferring authorization for printer %s until archive state is reconciled",
+                printer_id,
+            )
         elif printer_id in _unauthorized_print_kill_sent:
             pass
         else:
@@ -1398,11 +1463,43 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
                 stopped = printer_manager.stop_print(printer_id)
                 if stopped:
                     _unauthorized_print_kill_sent.add(printer_id)
+                    printer_info = printer_manager.get_printer(printer_id)
+                    printer_name = printer_info.name if printer_info else f"Printer {printer_id}"
+                    filename = state.subtask_name or state.gcode_file or state.current_print or "Unknown"
+                    notification_data = {
+                        "status": "stopped",
+                        "filename": state.gcode_file or state.current_print or "",
+                        "subtask_name": state.subtask_name or "",
+                        "progress": state.progress,
+                        "reason": "unauthorized_print",
+                    }
                     status_logger.warning(
                         "[KILL SWITCH] Stopped unauthorized print on printer %s (state=%s)",
                         printer_id,
                         state.state,
                     )
+                    try:
+                        await ws_manager.broadcast(
+                            {
+                                "type": "kill_switch_triggered",
+                                "printer_id": printer_id,
+                                "printer_name": printer_name,
+                                "filename": filename,
+                                "reason": "unauthorized_print",
+                            }
+                        )
+                    except Exception as e:
+                        status_logger.warning(
+                            "[KILL SWITCH] WebSocket notification failed for printer %s: %s", printer_id, e
+                        )
+
+                    previous_task = _kill_switch_notification_tasks.pop(printer_id, None)
+                    if previous_task is not None and not previous_task.done():
+                        previous_task.cancel()
+                    _kill_switch_notification_tasks[printer_id] = spawn_background_task(
+                        _send_kill_switch_provider_notification(printer_id, printer_name, notification_data),
+                        name=f"kill-switch-notification-{printer_id}",
+                    )
                 else:
                     status_logger.warning(
                         "[KILL SWITCH] Could not stop unauthorized print on printer %s (state=%s)",
@@ -2583,6 +2680,7 @@ async def on_print_start(printer_id: int, data: dict):
 
     # Clear any stale user-stopped flag from previous print cycles
     _user_stopped_printers.discard(printer_id)
+    _kill_switch_notification_tasks.pop(printer_id, None)
 
     # #1721: drop any leftover pre-captured finish frame from a prior print
     # so a never-consumed cache entry can't bleed into the new print's photo.
@@ -4252,16 +4350,12 @@ async def _upgrade_finish_photo_from_timelapse(archive_id: int, archive_dir: Pat
 
 
 async def on_print_running_observed(printer_id: int, data: dict):
-    """Restart-recovery: capture a fresh timelapse baseline for a print that
-    started before Bambuddy came up.
+    """Restart-recovery for a print that started before Bambuddy came up.
 
     bambu_mqtt.py suppresses ``on_print_start`` on the first RUNNING push
     after Bambuddy startup (#1304 guard, prevents duplicate archive
-    creation). Without that path, ``_capture_timelapse_baseline_at_start``
-    never runs and ``_scan_for_timelapse_with_retries`` falls into its
-    "take baseline now" fallback at completion time — but by then the
-    printer has already uploaded the in-flight MP4, so the baseline
-    includes it and no diff ever matches (#1485 follow-up).
+    creation). This hook restores the persisted archive into ``_active_prints``
+    and captures the timelapse baseline that normally hangs off print start.
 
     Fires once per session, in lieu of on_print_start when restart-recovery
     kicks in. The printer doesn't upload the timelapse until after PRINT
@@ -4270,21 +4364,15 @@ async def on_print_running_observed(printer_id: int, data: dict):
     """
     logger = logging.getLogger(__name__)
 
-    # Avoid double-capture: on_print_start may have run earlier in this
-    # Bambuddy process if the print started AFTER startup and we crashed
-    # later in the same session. (Realistically this can't happen — the
-    # MQTT client object would have been recreated — but the cheap guard
-    # is correct regardless.)
-    if printer_id in _timelapse_baselines:
-        logger.debug(
-            "[TIMELAPSE] on_print_running_observed: baseline already present for printer %s, skipping",
-            printer_id,
-        )
-        return
-
     async with async_session() as db:
         from backend.app.models.printer import Printer
 
+        state = printer_manager.get_status(printer_id)
+        if state is not None:
+            authorization = await _is_bambuddy_authorized_print(printer_id, state, db)
+            if authorization is True:
+                logger.info("[RESTART] Restored active Bambuddy print for printer %s", printer_id)
+
         result = await db.execute(select(Printer).where(Printer.id == printer_id))
         printer = result.scalar_one_or_none()
         if not printer:
@@ -4294,6 +4382,15 @@ async def on_print_running_observed(printer_id: int, data: dict):
             )
             return
 
+    # Avoid double-capture: ownership reconciliation above must still run when
+    # a baseline already exists, but the camera work itself is one-shot.
+    if printer_id in _timelapse_baselines:
+        logger.debug(
+            "[TIMELAPSE] on_print_running_observed: baseline already present for printer %s, skipping",
+            printer_id,
+        )
+        return
+
     await _capture_timelapse_baseline_at_start(printer, printer_id, logger)
 
 
@@ -4872,6 +4969,11 @@ async def on_print_complete(printer_id: int, data: dict):
 
     logger.info("[CALLBACK] on_print_complete started for printer %s", printer_id)
 
+    # A kill-switch stop sends its provider notification immediately. Keep the
+    # task so the later notification path can await it and avoid a duplicate;
+    # if that immediate attempt failed, the regular completion path retries.
+    kill_switch_notification_task = _kill_switch_notification_tasks.pop(printer_id, None)
+
     # Drop the 3MF download cache for this printer (#972). The print is over,
     # nothing else legitimately needs the bytes; keeping them would only risk
     # handing a stale file to the next print if it reuses the same name.
@@ -5452,9 +5554,12 @@ async def on_print_complete(printer_id: int, data: dict):
                     logger.info(
                         "[NOTIFY-BG] Sending notification without archive: printer=%s, status=%s", printer_id, ps
                     )
-                    await notification_service.on_print_complete(
-                        printer_id, p_name, ps, data, db, archive_data=no_archive_data
-                    )
+                    if not await _kill_switch_notification_already_sent(kill_switch_notification_task):
+                        await notification_service.on_print_complete(
+                            printer_id, p_name, ps, data, db, archive_data=no_archive_data
+                        )
+                    else:
+                        logger.info("[NOTIFY-BG] Skipped duplicate kill-switch provider notification")
 
                     # Send user-specific email if we have a created_by_id
                     if no_archive_data and no_archive_data.get("created_by_id"):
@@ -6134,9 +6239,12 @@ async def on_print_complete(printer_id: int, data: dict):
                             except Exception as e:
                                 logger.warning("[NOTIFY-BG] Failed to read finish photo bytes: %s", e)
 
-                await notification_service.on_print_complete(
-                    printer_id, printer_name, print_status, data, db, archive_data=archive_data
-                )
+                if not await _kill_switch_notification_already_sent(kill_switch_notification_task):
+                    await notification_service.on_print_complete(
+                        printer_id, printer_name, print_status, data, db, archive_data=archive_data
+                    )
+                else:
+                    logger.info("[NOTIFY-BG] Skipped duplicate kill-switch provider notification")
 
                 # Send user-specific email notification
                 if archive_data:

backend/app/services/finance_balance.py

@@ -0,0 +1,65 @@
+"""Canonical definition and synchronization of a user's personal balance."""
+
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction
+
+
+def transaction_affects_personal_balance(
+    user_id: int,
+    cost_center_id: int | None,
+    *,
+    is_private: bool = False,
+    owner_user_id: int | None = None,
+) -> bool:
+    """Apply the canonical definition to already-loaded transaction data."""
+
+    return cost_center_id is None or (is_private and owner_user_id == user_id)
+
+
+def personal_balance_condition(user_id: int):
+    """Return the SQL condition for transactions in a personal wallet.
+
+    Unassigned transactions and transactions assigned to the user's own
+    private cost center are personal. Shared cost centers are not.
+    """
+
+    return or_(
+        WalletTransaction.cost_center_id.is_(None),
+        and_(CostCenter.is_private.is_(True), CostCenter.owner_user_id == user_id),
+    )
+
+
+async def calculate_personal_balance(db: AsyncSession, user_id: int) -> float:
+    result = await db.execute(
+        select(func.coalesce(func.sum(WalletTransaction.amount), 0.0))
+        .select_from(WalletTransaction)
+        .outerjoin(CostCenter, WalletTransaction.cost_center_id == CostCenter.id)
+        .where(WalletTransaction.user_id == user_id, personal_balance_condition(user_id))
+    )
+    return round(float(result.scalar_one() or 0.0), 2)
+
+
+async def is_personal_transaction(db: AsyncSession, user_id: int, cost_center_id: int | None) -> bool:
+    if cost_center_id is None:
+        return True
+    result = await db.execute(
+        select(CostCenter.is_private, CostCenter.owner_user_id).where(CostCenter.id == cost_center_id)
+    )
+    center = result.one_or_none()
+    if center is None:
+        return False
+    return transaction_affects_personal_balance(
+        user_id,
+        cost_center_id,
+        is_private=bool(center.is_private),
+        owner_user_id=center.owner_user_id,
+    )
+
+
+async def sync_personal_wallet_balance(db: AsyncSession, wallet: UserWallet) -> float:
+    balance = await calculate_personal_balance(db, wallet.user_id)
+    wallet.balance = balance
+    db.add(wallet)
+    return balance

backend/app/services/finance_billing.py

@@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.models.archive import PrintArchive
 from backend.app.models.finance import TransactionType, UserWallet, WalletTransaction
+from backend.app.services.finance_balance import sync_personal_wallet_balance
 from backend.app.services.finance_budget import is_billing_enabled, release_budget_reservation
 
 logger = logging.getLogger(__name__)
@@ -185,10 +186,6 @@ async def apply_print_charge_for_archive(
             await db.flush()
             logger.info(f"Created new wallet for user ID {archive.created_by_id}.")
 
-        # Persist wallet balances rounded to cents
-        new_wallet_balance = round(float(wallet.balance) - charge, 2)
-        wallet.balance = new_wallet_balance
-
         label = archive.print_name or archive.filename or f"Archive {archive.id}"
         description = f"Print charge: {label}{' ' + reason_suffix if reason_suffix else ''}"
 
@@ -219,6 +216,10 @@ async def apply_print_charge_for_archive(
             await db.rollback()
             return False
 
+        # Rebuild from the canonical personal-ledger definition. A shared cost
+        # center charge must not debit the user's personal wallet.
+        new_wallet_balance = await sync_personal_wallet_balance(db, wallet)
+
         # Consume matching budget reservations after the transaction is persisted
         await release_budget_reservation(db, print_archive_id=archive.id, status="consumed")
         logger.info(f"Applied print charge for archive ID {archive_id}. New balance: {new_wallet_balance}.")

backend/app/services/print_cost_estimate.py

@@ -0,0 +1,155 @@
+"""Trusted server-side cost estimates for queued prints."""
+
+import json
+import logging
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from backend.app.core.config import settings
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.models.spool_assignment import SpoolAssignment
+from backend.app.utils import threemf_tools
+from backend.app.utils.safe_path import safe_join_under
+
+logger = logging.getLogger(__name__)
+
+
+def plate_scoped_run_estimate(
+    archive: PrintArchive,
+    full_path: Path | None,
+    plate_id: int | None = None,
+) -> tuple[float | None, float | None]:
+    """Return trusted ``(grams, cost)`` for one run of an archive plate."""
+
+    whole_grams = archive.filament_used_grams
+    selected_plate = archive.plate_id if plate_id is None else plate_id
+    if selected_plate is None or full_path is None or not full_path.exists():
+        return whole_grams, archive.cost
+    try:
+        plate_grams = threemf_tools.extract_plate_metadata_from_3mf(full_path, selected_plate).filament_used_grams
+    except Exception as exc:
+        logger.debug(
+            "Plate-scoped estimate failed for archive %s (plate %s): %s",
+            archive.id,
+            selected_plate,
+            exc,
+        )
+        return whole_grams, archive.cost
+    if not plate_grams or plate_grams <= 0:
+        return whole_grams, archive.cost
+    plate_cost = archive.cost
+    if archive.cost and whole_grams and whole_grams > 0:
+        plate_cost = round(archive.cost * (plate_grams / whole_grams), 2)
+    return round(plate_grams, 2), plate_cost
+
+
+def _source_path(library_file: LibraryFile) -> Path:
+    path = Path(library_file.file_path)
+    if path.is_absolute():
+        # SEC-PATH-OK: absolute paths are persisted LibraryFile locations for
+        # configured external libraries; this branch performs no path join.
+        return path
+    return safe_join_under(settings.base_dir, library_file.file_path, http=False)
+
+
+def _parse_mapping(mapping: list[int] | str | None) -> list[int] | None:
+    if isinstance(mapping, list):
+        return mapping
+    if isinstance(mapping, str):
+        try:
+            parsed = json.loads(mapping)
+        except (TypeError, json.JSONDecodeError):
+            return None
+        return parsed if isinstance(parsed, list) else None
+    return None
+
+
+def _global_tray_id(assignment: SpoolAssignment) -> int:
+    if assignment.ams_id == 255:
+        return 254 + assignment.tray_id
+    if assignment.ams_id >= 128:
+        return assignment.ams_id
+    return assignment.ams_id * 4 + assignment.tray_id
+
+
+async def _default_cost_per_kg(db: AsyncSession) -> float:
+    from backend.app.api.routes.settings import get_setting
+
+    raw = await get_setting(db, "default_filament_cost")
+    try:
+        return float(raw) if raw is not None else 25.0
+    except (TypeError, ValueError):
+        return 25.0
+
+
+async def estimate_queue_source_cost(
+    db: AsyncSession,
+    *,
+    archive: PrintArchive | None = None,
+    library_file: LibraryFile | None = None,
+    plate_id: int | None = None,
+    ams_mapping: list[int] | str | None = None,
+    printer_id: int | None = None,
+) -> float | None:
+    """Compute a queue cost without trusting the request's display hint."""
+
+    if archive is not None:
+        archive_path = settings.base_dir / archive.file_path
+        _grams, cost = plate_scoped_run_estimate(archive, archive_path, plate_id)
+        return float(cost) if cost is not None and cost > 0 else None
+
+    if library_file is None:
+        return None
+
+    path = _source_path(library_file)
+    usage: list[dict] = []
+    if path.exists():
+        usage = threemf_tools.extract_plate_metadata_from_3mf(path, plate_id).filament_usage
+
+    metadata = library_file.file_metadata or {}
+    if not usage:
+        try:
+            grams = float(metadata.get("filament_used_grams") or 0)
+        except (TypeError, ValueError):
+            grams = 0
+        if grams > 0:
+            usage = [{"slot_id": 1, "used_g": grams}]
+
+    if not usage:
+        return None
+
+    default_cost = await _default_cost_per_kg(db)
+    cost_by_tray: dict[int, float | None] = {}
+    mapping = _parse_mapping(ams_mapping)
+    if printer_id is not None and mapping:
+        assignments = (
+            (
+                await db.execute(
+                    select(SpoolAssignment)
+                    .options(selectinload(SpoolAssignment.spool))
+                    .where(SpoolAssignment.printer_id == printer_id)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        cost_by_tray = {_global_tray_id(a): a.spool.cost_per_kg for a in assignments}
+
+    total = 0.0
+    for filament in usage:
+        try:
+            slot_id = int(filament.get("slot_id") or 0)
+            grams = float(filament.get("used_g") or 0)
+        except (TypeError, ValueError):
+            continue
+        tray_id = mapping[slot_id - 1] if mapping and 0 < slot_id <= len(mapping) else None
+        cost_per_kg = cost_by_tray.get(tray_id) if tray_id is not None else None
+        if cost_per_kg is None or cost_per_kg <= 0:
+            cost_per_kg = default_cost
+        total += (grams / 1000.0) * cost_per_kg
+
+    return round(total, 2) if total > 0 else None

backend/app/services/print_scheduler.py

@@ -43,6 +43,7 @@ from backend.app.services.finance_budget import (
 )
 from backend.app.services.ha_sensor_manager import ha_sensor_manager
 from backend.app.services.notification_service import notification_service
+from backend.app.services.print_cost_estimate import estimate_queue_source_cost
 from backend.app.services.printer_manager import (
     printer_manager,
     supports_airduct,
@@ -3684,6 +3685,20 @@ class PrintScheduler:
             from backend.app.models.user import User
 
             queue_user = await db.get(User, item.created_by_id) if item.created_by_id is not None else None
+            # Recompute at the final authorization boundary as well as enqueue
+            # time. This covers rows created before the server-side estimate
+            # migration and prevents any alternate write path from weakening
+            # the budget reservation.
+            archive = await db.get(PrintArchive, item.archive_id) if item.archive_id is not None else None
+            library_file = await db.get(LibraryFile, item.library_file_id) if item.library_file_id is not None else None
+            item.estimated_cost = await estimate_queue_source_cost(
+                db,
+                archive=archive,
+                library_file=library_file,
+                plate_id=item.plate_id,
+                ams_mapping=item.ams_mapping,
+                printer_id=item.printer_id,
+            )
             await validate_print_budget(
                 db,
                 cost_center_id=item.cost_center_id,

backend/tests/integration/test_finance_api.py

@@ -7,8 +7,10 @@ from sqlalchemy import select
 from backend.app.core.auth import get_password_hash
 from backend.app.models.archive import PrintArchive
 from backend.app.models.finance import CostCenter, UserWallet, WalletTransaction
+from backend.app.models.group import Group
 from backend.app.models.settings import Settings
 from backend.app.models.user import User
+from backend.app.services.finance_billing import apply_print_charge_for_archive
 
 
 class TestFinanceAPI:
@@ -143,34 +145,30 @@ class TestFinanceAPI:
         admin_user,
         db_session,
     ):
-        """Test cost-center and personal transaction handling.
-
-        Cost-center transactions affect cost-center balance only.
-        Personal transactions (no cost_center_id) affect user wallet.
-        """
+        """A user's private cost center and unassigned entries share one wallet."""
         await self._enable_basic_user_creation(db_session)
         created_user = await self._create_user_via_api(async_client, auth_headers, "dave")
 
-        shared_center = await db_session.scalar(
+        private_center = await db_session.scalar(
             select(CostCenter).where(CostCenter.owner_user_id == created_user["id"], CostCenter.is_private.is_(True))
         )
-        assert shared_center is not None
+        assert private_center is not None
 
-        # Deposit to cost center: affects cost-center balance, not user wallet
+        # The owner's private cost center affects both its own ledger and the wallet.
         deposit = await async_client.post(
             f"/api/v1/finance/users/{created_user['id']}/deposit",
-            json={"amount": 25.0, "description": "Initial CC top-up", "cost_center_id": shared_center.id},
+            json={"amount": 25.0, "description": "Initial CC top-up", "cost_center_id": private_center.id},
             headers=auth_headers,
         )
         assert deposit.status_code == 200
-        assert deposit.json()["transaction"]["cost_center_id"] == shared_center.id
+        assert deposit.json()["transaction"]["cost_center_id"] == private_center.id
         assert deposit.json()["transaction"]["balance_after"] == 25.0  # CC balance
         assert deposit.json()["balance"]["balance"] == 25.0  # Response shows CC balance
 
-        # Withdraw from cost center: affects cost-center balance only
+        # A withdrawal updates both views by the same amount.
         withdraw = await async_client.post(
             f"/api/v1/finance/users/{created_user['id']}/withdraw",
-            json={"amount": 5.0, "description": "CC Usage", "cost_center_id": shared_center.id},
+            json={"amount": 5.0, "description": "CC Usage", "cost_center_id": private_center.id},
             headers=auth_headers,
         )
         assert withdraw.status_code == 200
@@ -186,8 +184,8 @@ class TestFinanceAPI:
         )
         assert personal_deposit.status_code == 200
         assert personal_deposit.json()["transaction"]["cost_center_id"] is None
-        assert personal_deposit.json()["transaction"]["balance_after"] == 30.0  # Personal balance
-        assert personal_deposit.json()["balance"]["balance"] == 30.0  # User wallet updated
+        assert personal_deposit.json()["transaction"]["balance_after"] == 50.0
+        assert personal_deposit.json()["balance"]["balance"] == 50.0
 
         transactions_response = await async_client.get(
             f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
@@ -195,7 +193,7 @@ class TestFinanceAPI:
         assert transactions_response.status_code == 200
         transactions = transactions_response.json()
         assert len(transactions) == 3
-        cc_txs = [tx for tx in transactions if tx["cost_center_id"] == shared_center.id]
+        cc_txs = [tx for tx in transactions if tx["cost_center_id"] == private_center.id]
         personal_txs = [tx for tx in transactions if tx["cost_center_id"] is None]
         assert len(cc_txs) == 2
         assert len(personal_txs) == 1
@@ -204,7 +202,7 @@ class TestFinanceAPI:
             f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
         )
         assert balance_response.status_code == 200
-        assert balance_response.json()["balance"] == 30.0  # Only personal balance
+        assert balance_response.json()["balance"] == 50.0
 
         # Delete personal transaction, user wallet should decrease
         personal_tx = next(tx for tx in transactions if tx["cost_center_id"] is None)
@@ -217,7 +215,7 @@ class TestFinanceAPI:
             f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
         )
         assert balance_after_delete.status_code == 200
-        assert balance_after_delete.json()["balance"] == 0.0  # Personal balance back to 0
+        assert balance_after_delete.json()["balance"] == 20.0
 
         remaining = await async_client.get(
             f"/api/v1/finance/users/{created_user['id']}/transactions", headers=auth_headers
@@ -225,6 +223,74 @@ class TestFinanceAPI:
         assert remaining.status_code == 200
         assert len(remaining.json()) == 2  # 2 CC transactions remain
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_admin_and_user_balances_agree_after_charge_adjustment_and_delete(
+        self,
+        async_client: AsyncClient,
+        auth_headers: dict[str, str],
+        db_session,
+    ):
+        created_user = await self._create_user_via_api(async_client, auth_headers, "balance-lifecycle")
+        user = await db_session.get(User, created_user["id"])
+        operators = await db_session.scalar(select(Group).where(Group.name == "Operators"))
+        assert operators is not None
+        user.groups.append(operators)
+        await db_session.commit()
+        user_headers = await self._login_user(async_client, "balance-lifecycle")
+        private_center = await db_session.scalar(
+            select(CostCenter).where(CostCenter.owner_user_id == user.id, CostCenter.is_private.is_(True))
+        )
+        assert private_center is not None
+
+        deposit = await async_client.post(
+            f"/api/v1/finance/users/{user.id}/deposit",
+            json={"amount": 20.0, "cost_center_id": private_center.id},
+            headers=auth_headers,
+        )
+        assert deposit.status_code == 200
+
+        archive = PrintArchive(
+            filename="charged.gcode.3mf",
+            file_path="archives/test/charged.gcode.3mf",
+            file_size=10,
+            status="completed",
+            cost=4.0,
+            created_by_id=user.id,
+            cost_center_id=private_center.id,
+        )
+        db_session.add(archive)
+        await db_session.commit()
+        assert await apply_print_charge_for_archive(db_session, archive.id) is True
+        await db_session.commit()
+
+        adjustment = await async_client.post(
+            f"/api/v1/finance/users/{user.id}/deposit",
+            json={"amount": 5.0, "description": "temporary adjustment"},
+            headers=auth_headers,
+        )
+        assert adjustment.status_code == 200
+
+        async def assert_views_agree(expected: float):
+            admin_view = await async_client.get(
+                f"/api/v1/finance/users/{user.id}/balance",
+                headers=auth_headers,
+            )
+            user_view = await async_client.get("/api/v1/finance/me/balance", headers=user_headers)
+            assert admin_view.status_code == 200
+            assert user_view.status_code == 200
+            assert admin_view.json()["balance"] == expected
+            assert user_view.json()["balance"] == expected
+
+        await assert_views_agree(21.0)
+
+        delete_response = await async_client.delete(
+            f"/api/v1/finance/transactions/{adjustment.json()['transaction']['id']}",
+            headers=auth_headers,
+        )
+        assert delete_response.status_code == 200
+        await assert_views_agree(16.0)
+
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_delete_cost_center_transaction_rebuilds_remaining_ledger(
@@ -283,7 +349,7 @@ class TestFinanceAPI:
             f"/api/v1/finance/users/{created_user['id']}/balance", headers=auth_headers
         )
         assert balance_response.status_code == 200
-        assert balance_response.json()["balance"] == 12.0
+        assert balance_response.json()["balance"] == 7.0
 
     @pytest.mark.asyncio
     @pytest.mark.integration

backend/tests/integration/test_print_queue_api.py

@@ -146,7 +146,7 @@ class TestPrintQueueAPI:
         """Verify item can be added to queue with cost_center_id."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=1.25, filament_used_grams=50.0)
         cost_center = CostCenter(name="Queue CC", is_active=True, is_private=False)
         db_session.add(cost_center)
         await db_session.commit()
@@ -158,7 +158,7 @@ class TestPrintQueueAPI:
                 "printer_id": printer.id,
                 "archive_id": archive.id,
                 "cost_center_id": cost_center.id,
-                "estimated_cost": 1.25,
+                "estimated_cost": 0.01,
             },
         )
         assert response.status_code == 200
@@ -175,13 +175,13 @@ class TestPrintQueueAPI:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_add_to_queue_with_cost_center_requires_estimated_cost(
+    async def test_add_to_queue_derives_cost_without_client_estimate(
         self, async_client: AsyncClient, printer_factory, archive_factory, db_session
     ):
-        """Cost-center queue items require an estimated cost for budget checks."""
+        """The persisted budget estimate comes from the archive, not the request."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=1.25, filament_used_grams=50.0)
         cost_center = CostCenter(name="Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
         db_session.add(cost_center)
         await db_session.commit()
@@ -196,18 +196,18 @@ class TestPrintQueueAPI:
             },
         )
 
-        assert response.status_code == 400
-        assert "Estimated cost is required" in response.json()["detail"]
+        assert response.status_code == 200
+        assert response.json()["estimated_cost"] == 1.25
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_add_to_queue_rejects_when_estimated_cost_exceeds_budget(
+    async def test_add_to_queue_rejects_tampered_client_cost_when_server_cost_exceeds_budget(
         self, async_client: AsyncClient, printer_factory, archive_factory, db_session
     ):
-        """Queue creation is rejected if the estimated cost exceeds remaining budget."""
+        """A forged low client hint cannot bypass the server-derived budget check."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=2.0, filament_used_grams=50.0)
         cost_center = CostCenter(name="Tiny Budget CC", is_active=True, is_private=False, monthly_budget=1.0)
         db_session.add(cost_center)
         await db_session.commit()
@@ -219,7 +219,7 @@ class TestPrintQueueAPI:
                 "printer_id": printer.id,
                 "archive_id": archive.id,
                 "cost_center_id": cost_center.id,
-                "estimated_cost": 2.0,
+                "estimated_cost": 0.01,
             },
         )
 
@@ -234,7 +234,7 @@ class TestPrintQueueAPI:
         """Billing enforcement rejects queue jobs that omit cost center."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=1.0, filament_used_grams=50.0)
 
         response = await async_client.post(
             "/api/v1/queue/",
@@ -255,7 +255,7 @@ class TestPrintQueueAPI:
         """Open queue items reserve budget until they leave pending/printing states."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=3.0, filament_used_grams=50.0)
         cost_center = CostCenter(name="Reserved Budget CC", is_active=True, is_private=False, monthly_budget=10.0)
         db_session.add(cost_center)
         await db_session.commit()
@@ -274,7 +274,7 @@ class TestPrintQueueAPI:
                 "printer_id": printer.id,
                 "archive_id": archive.id,
                 "cost_center_id": cost_center.id,
-                "estimated_cost": 3.0,
+                "estimated_cost": 0.01,
             },
         )
 
@@ -345,7 +345,7 @@ class TestPrintQueueAPI:
         """Verify a pending queue item can be updated with cost_center_id."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=1.25, filament_used_grams=50.0)
         item = await queue_item_factory(printer_id=printer.id, archive_id=archive.id)
         cost_center = CostCenter(name="Update CC", is_active=True, is_private=False)
         db_session.add(cost_center)
@@ -354,7 +354,7 @@ class TestPrintQueueAPI:
 
         response = await async_client.patch(
             f"/api/v1/queue/{item.id}",
-            json={"cost_center_id": cost_center.id, "estimated_cost": 1.25},
+            json={"cost_center_id": cost_center.id, "estimated_cost": 0.01},
         )
 
         assert response.status_code == 200
@@ -822,7 +822,7 @@ class TestPrintQueueAPI:
         """Deleting a pending cost-center queue item releases its reserved budget."""
         await enable_billing(db_session)
         printer = await printer_factory()
-        archive = await archive_factory()
+        archive = await archive_factory(cost=3.0, filament_used_grams=50.0)
         cost_center = CostCenter(
             name="Delete Releases Budget CC", is_active=True, is_private=False, monthly_budget=10.0
         )
@@ -843,7 +843,7 @@ class TestPrintQueueAPI:
                 "printer_id": printer.id,
                 "archive_id": archive.id,
                 "cost_center_id": cost_center.id,
-                "estimated_cost": 3.0,
+                "estimated_cost": 0.01,
             },
         )
         assert blocked.status_code == 400
@@ -857,7 +857,7 @@ class TestPrintQueueAPI:
                 "printer_id": printer.id,
                 "archive_id": archive.id,
                 "cost_center_id": cost_center.id,
-                "estimated_cost": 3.0,
+                "estimated_cost": 0.01,
             },
         )
         assert allowed.status_code == 200
@@ -1730,11 +1730,12 @@ class TestBulkUpdateEndpoint:
     @pytest.mark.asyncio
     @pytest.mark.integration
     async def test_bulk_update_cost_center_counts_pending_reservations(
-        self, async_client: AsyncClient, queue_item_factory, db_session
+        self, async_client: AsyncClient, queue_item_factory, archive_factory, db_session
     ):
-        """Bulk updates cannot reserve more than the available cost-center budget."""
+        """A forged bulk-update hint cannot weaken the server-derived reservation."""
         await enable_billing(db_session)
-        item = await queue_item_factory()
+        archive = await archive_factory(cost=3.0, filament_used_grams=50.0)
+        item = await queue_item_factory(archive_id=archive.id)
         existing = await queue_item_factory()
         cost_center = CostCenter(name="Bulk Reserved CC", is_active=True, is_private=False, monthly_budget=10.0)
         db_session.add(cost_center)
@@ -1747,7 +1748,7 @@ class TestBulkUpdateEndpoint:
 
         response = await async_client.patch(
             "/api/v1/queue/bulk",
-            json={"item_ids": [item.id], "cost_center_id": cost_center.id, "estimated_cost": 3.0},
+            json={"item_ids": [item.id], "cost_center_id": cost_center.id, "estimated_cost": 0.01},
         )
 
         assert response.status_code == 400

backend/tests/unit/services/test_finance_service_billing.py

@@ -60,7 +60,7 @@ class TestFinanceBilling:
 
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
         assert wallet is not None
-        assert wallet.balance == -7.5
+        assert wallet.balance == 0.0
 
         tx = await db_session.scalar(select(WalletTransaction).where(WalletTransaction.print_run_id == "run-1"))
         assert tx is not None
@@ -86,7 +86,7 @@ class TestFinanceBilling:
         assert second_run is True
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
         assert wallet is not None
-        assert wallet.balance == -15.0
+        assert wallet.balance == 0.0
 
         rows = (
             (await db_session.execute(select(WalletTransaction).where(WalletTransaction.user_id == user.id)))
@@ -277,7 +277,7 @@ class TestPartialPrintCharges:
         assert changed is True
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
         assert wallet is not None
-        assert wallet.balance == -3.0
+        assert wallet.balance == 0.0
 
         transaction = await db_session.scalar(
             select(WalletTransaction).where(WalletTransaction.print_archive_id == archive.id)
@@ -374,7 +374,7 @@ class TestPartialPrintCharges:
 
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
         assert wallet is not None
-        assert wallet.balance == -5.0  # 50% of 10.0
+        assert wallet.balance == 0.0
 
         tx = await db_session.scalar(
             select(WalletTransaction)
@@ -476,7 +476,7 @@ class TestPartialPrintCharges:
 
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
         assert wallet is not None
-        assert wallet.balance == pytest.approx(-1.0, abs=0.01)  # 5% of 20.0
+        assert wallet.balance == 0.0
 
     @pytest.mark.asyncio
     async def test_completed_print_still_charges_full_cost(self, db_session):
@@ -512,7 +512,7 @@ class TestPartialPrintCharges:
         assert changed is True
 
         wallet = await db_session.scalar(select(UserWallet).where(UserWallet.user_id == user.id))
-        assert wallet.balance == -15.0  # Full cost, not proportional
+        assert wallet.balance == 0.0
 
     @pytest.mark.asyncio
     async def test_partial_charge_with_cost_center_override(self, db_session):

backend/tests/unit/services/test_print_cost_estimate.py

@@ -0,0 +1,85 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from backend.app.services import print_cost_estimate
+
+
+@pytest.fixture
+def library_file(tmp_path):
+    path = tmp_path / "queued.gcode.3mf"
+    path.write_bytes(b"stub")
+    return SimpleNamespace(file_path=str(path), file_metadata={})
+
+
+@pytest.mark.asyncio
+async def test_library_estimate_uses_server_default_cost(monkeypatch, library_file):
+    monkeypatch.setattr(
+        print_cost_estimate.threemf_tools,
+        "extract_plate_metadata_from_3mf",
+        lambda *_args: SimpleNamespace(
+            filament_usage=[
+                {"slot_id": 1, "used_g": 100.0},
+                {"slot_id": 2, "used_g": 50.0},
+            ]
+        ),
+    )
+    monkeypatch.setattr(print_cost_estimate, "_default_cost_per_kg", AsyncMock(return_value=20.0))
+
+    cost = await print_cost_estimate.estimate_queue_source_cost(
+        SimpleNamespace(),
+        library_file=library_file,
+        plate_id=1,
+    )
+
+    assert cost == 3.0
+
+
+@pytest.mark.asyncio
+async def test_library_estimate_uses_server_spool_assignment_costs(monkeypatch, library_file):
+    monkeypatch.setattr(
+        print_cost_estimate.threemf_tools,
+        "extract_plate_metadata_from_3mf",
+        lambda *_args: SimpleNamespace(
+            filament_usage=[
+                {"slot_id": 1, "used_g": 100.0},
+                {"slot_id": 2, "used_g": 50.0},
+            ]
+        ),
+    )
+    monkeypatch.setattr(print_cost_estimate, "_default_cost_per_kg", AsyncMock(return_value=20.0))
+
+    assignments = [
+        SimpleNamespace(ams_id=0, tray_id=0, spool=SimpleNamespace(cost_per_kg=10.0)),
+        SimpleNamespace(ams_id=0, tray_id=1, spool=SimpleNamespace(cost_per_kg=30.0)),
+    ]
+    result = SimpleNamespace(scalars=lambda: SimpleNamespace(all=lambda: assignments))
+    db = SimpleNamespace(execute=AsyncMock(return_value=result))
+
+    cost = await print_cost_estimate.estimate_queue_source_cost(
+        db,
+        library_file=library_file,
+        plate_id=1,
+        ams_mapping=[0, 1],
+        printer_id=7,
+    )
+
+    assert cost == 2.5
+
+
+@pytest.mark.asyncio
+async def test_missing_server_metadata_does_not_fall_back_to_client_hint(monkeypatch, library_file):
+    monkeypatch.setattr(
+        print_cost_estimate.threemf_tools,
+        "extract_plate_metadata_from_3mf",
+        lambda *_args: SimpleNamespace(filament_usage=[]),
+    )
+
+    cost = await print_cost_estimate.estimate_queue_source_cost(
+        SimpleNamespace(),
+        library_file=library_file,
+        plate_id=1,
+    )
+
+    assert cost is None

backend/tests/unit/test_finance_table_migration.py

@@ -7,7 +7,11 @@ import pytest
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import create_async_engine
 
-from backend.app.core.database import _migrate_create_finance_indexes, _migrate_create_finance_tables
+from backend.app.core.database import (
+    _migrate_add_print_archive_cost_center,
+    _migrate_create_finance_indexes,
+    _migrate_create_finance_tables,
+)
 
 EXPECTED_TABLES = {
     "cost_centers",
@@ -66,6 +70,31 @@ async def test_legacy_cost_center_indexes_are_delayed_until_columns_exist():
         await engine.dispose()
 
 
+@pytest.mark.asyncio
+async def test_print_archive_cost_center_is_added_idempotently_on_sqlite():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+
+    try:
+        async with engine.begin() as conn:
+            await conn.execute(text("PRAGMA foreign_keys = ON"))
+            await conn.execute(text("CREATE TABLE cost_centers (id INTEGER PRIMARY KEY)"))
+            await conn.execute(text("CREATE TABLE print_archives (id INTEGER PRIMARY KEY)"))
+
+            await _migrate_add_print_archive_cost_center(conn)
+            await _migrate_add_print_archive_cost_center(conn)
+
+            columns = await conn.execute(text("PRAGMA table_info(print_archives)"))
+            foreign_keys = await conn.execute(text("PRAGMA foreign_key_list(print_archives)"))
+
+        assert "cost_center_id" in {row[1] for row in columns}
+        assert any(
+            row[2] == "cost_centers" and row[3] == "cost_center_id" and row[6].upper() == "SET NULL"
+            for row in foreign_keys
+        )
+    finally:
+        await engine.dispose()
+
+
 @pytest.mark.asyncio
 async def test_postgres_finance_ddl_uses_postgres_types():
     statements: list[str] = []
@@ -110,6 +139,8 @@ async def test_finance_tables_are_created_idempotently_on_postgres():
             with patch("backend.app.core.database.is_sqlite", return_value=False):
                 await _migrate_create_finance_tables(conn)
                 await _migrate_create_finance_tables(conn)
+                await _migrate_add_print_archive_cost_center(conn)
+                await _migrate_add_print_archive_cost_center(conn)
                 await _migrate_create_finance_indexes(conn)
                 await _migrate_create_finance_indexes(conn)
 
@@ -127,8 +158,25 @@ async def test_finance_tables_are_created_idempotently_on_postgres():
                     "AND table_name = 'cost_centers' AND column_name = 'created_at'"
                 )
             )
+            archive_cost_center = await conn.execute(
+                text(
+                    "SELECT c.data_type, rc.delete_rule "
+                    "FROM information_schema.columns c "
+                    "JOIN information_schema.key_column_usage kcu "
+                    "  ON kcu.table_schema = c.table_schema "
+                    " AND kcu.table_name = c.table_name "
+                    " AND kcu.column_name = c.column_name "
+                    "JOIN information_schema.referential_constraints rc "
+                    "  ON rc.constraint_schema = kcu.constraint_schema "
+                    " AND rc.constraint_name = kcu.constraint_name "
+                    "WHERE c.table_schema = 'public' "
+                    "AND c.table_name = 'print_archives' "
+                    "AND c.column_name = 'cost_center_id'"
+                )
+            )
 
         assert {row[0] for row in rows} == EXPECTED_TABLES
         assert timestamp_type.scalar_one() == "timestamp without time zone"
+        assert archive_cost_center.one() == ("integer", "SET NULL")
     finally:
         await engine.dispose()

backend/tests/unit/test_printer_kill_switch.py

@@ -1,4 +1,5 @@
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -8,14 +9,21 @@ from backend.app import main as main_module
 @pytest.fixture(autouse=True)
 def clear_kill_switch_state():
     main_module._unauthorized_print_kill_sent.clear()
+    main_module._kill_switch_notification_tasks.clear()
     main_module._expected_prints.clear()
     main_module._active_prints.clear()
     main_module._expected_print_registered_at.clear()
+    main_module._printer_reconciled_since_connect.clear()
     yield
+    for task in main_module._kill_switch_notification_tasks.values():
+        if not task.done():
+            task.cancel()
     main_module._unauthorized_print_kill_sent.clear()
+    main_module._kill_switch_notification_tasks.clear()
     main_module._expected_prints.clear()
     main_module._active_prints.clear()
     main_module._expected_print_registered_at.clear()
+    main_module._printer_reconciled_since_connect.clear()
 
 
 def test_gcode_3mf_status_filename_matches_registered_expected_print():
@@ -34,6 +42,8 @@ def test_gcode_3mf_status_filename_matches_registered_expected_print():
 @pytest.mark.asyncio
 async def test_unauthorized_active_print_triggers_stop(monkeypatch):
     stop_calls: list[int] = []
+    broadcast = AsyncMock()
+    provider_notification = AsyncMock(return_value=True)
 
     async def fake_status(*args, **kwargs):
         return None
@@ -41,6 +51,9 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch):
     async def kill_switch_enabled(_db):
         return True
 
+    async def unauthorized(*_args):
+        return False
+
     monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
     monkeypatch.setattr(
         main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
@@ -50,6 +63,9 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch):
     monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
     monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
     monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr(main_module.ws_manager, "broadcast", broadcast)
+    monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized)
+    monkeypatch.setattr(main_module, "_send_kill_switch_provider_notification", provider_notification)
     monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
 
     state = SimpleNamespace(
@@ -71,6 +87,7 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch):
         ams_filament_backup=False,
         current_print=None,
         subtask_name="foreign_job",
+        subtask_id="external-task-1",
         gcode_file="foreign_job.gcode",
     )
 
@@ -78,6 +95,39 @@ async def test_unauthorized_active_print_triggers_stop(monkeypatch):
 
     assert stop_calls == [7]
     assert 7 in main_module._unauthorized_print_kill_sent
+    broadcast.assert_awaited_once_with(
+        {
+            "type": "kill_switch_triggered",
+            "printer_id": 7,
+            "printer_name": "Printer 7",
+            "filename": "foreign_job",
+            "reason": "unauthorized_print",
+        }
+    )
+    notification_task = main_module._kill_switch_notification_tasks[7]
+    assert await notification_task is True
+    provider_notification.assert_awaited_once_with(
+        7,
+        "Printer 7",
+        {
+            "status": "stopped",
+            "filename": "foreign_job.gcode",
+            "subtask_name": "foreign_job",
+            "progress": 0,
+            "reason": "unauthorized_print",
+        },
+    )
+
+
+@pytest.mark.asyncio
+async def test_failed_immediate_notification_allows_completion_retry():
+    task = main_module.spawn_background_task(_return_false(), name="test-kill-switch-notification-failure")
+
+    assert await main_module._kill_switch_notification_already_sent(task) is False
+
+
+async def _return_false():
+    return False
 
 
 @pytest.mark.asyncio
@@ -141,6 +191,9 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
     async def kill_switch_enabled(_db):
         return True
 
+    async def unauthorized(*_args):
+        return False
+
     monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
     monkeypatch.setattr(
         main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
@@ -150,6 +203,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
     monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
     monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
     monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr(main_module, "_is_bambuddy_authorized_print", unauthorized)
     monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
 
     active_state = SimpleNamespace(
@@ -171,6 +225,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
         ams_filament_backup=False,
         current_print=None,
         subtask_name="foreign_job",
+        subtask_id="external-task-1",
         gcode_file="foreign_job.gcode",
     )
 
@@ -193,6 +248,7 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
         ams_filament_backup=False,
         current_print=None,
         subtask_name="",
+        subtask_id=None,
         gcode_file=None,
     )
 
@@ -203,3 +259,90 @@ async def test_unauthorized_print_state_is_cleared_when_print_ends(monkeypatch):
     await main_module.on_printer_status_change(7, idle_state)
 
     assert 7 not in main_module._unauthorized_print_kill_sent
+
+
+@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")
+    query_result = SimpleNamespace(scalar_one_or_none=lambda: archive)
+    db = SimpleNamespace(execute=AsyncMock(return_value=query_result))
+
+    class FakeSessionContext:
+        async def __aenter__(self):
+            return db
+
+        async def __aexit__(self, *_args):
+            return False
+
+    stop_calls: list[int] = []
+
+    async def fake_status(*args, **kwargs):
+        return None
+
+    async def kill_switch_enabled(_db):
+        return True
+
+    def discard_background_task(coro, **_kwargs):
+        coro.close()
+
+    monkeypatch.setattr(main_module, "async_session", FakeSessionContext)
+    monkeypatch.setattr(main_module, "spawn_background_task", discard_background_task)
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+    monkeypatch.setattr(
+        main_module.printer_manager, "stop_print", lambda printer_id: stop_calls.append(printer_id) or True
+    )
+    monkeypatch.setattr(main_module.printer_manager, "get_printer", lambda printer_id: None)
+    monkeypatch.setattr(main_module.printer_manager, "get_model", lambda printer_id: None)
+    monkeypatch.setattr(main_module, "printer_state_to_dict", lambda *args, **kwargs: {})
+    monkeypatch.setattr(main_module.mqtt_relay, "on_printer_status", fake_status)
+    monkeypatch.setattr(main_module.ws_manager, "send_printer_status", fake_status)
+    monkeypatch.setattr("backend.app.services.finance_budget.is_printer_kill_switch_enabled", kill_switch_enabled)
+
+    state = SimpleNamespace(
+        connected=True,
+        state=printer_state,
+        progress=42,
+        remaining_time=600,
+        layer_num=50,
+        temperatures={},
+        raw_data={},
+        stg_cur=0,
+        cooling_fan_speed=None,
+        big_fan1_speed=None,
+        big_fan2_speed=None,
+        chamber_light=False,
+        active_extruder=0,
+        tray_now=255,
+        door_open=False,
+        ams_filament_backup=False,
+        current_print=None,
+        subtask_name="owned_job",
+        subtask_id="bambuddy-task-123",
+        gcode_file="owned_job.gcode.3mf",
+    )
+
+    await main_module.on_printer_status_change(7, state)
+
+    assert stop_calls == []
+    assert (7, "owned_job.gcode.3mf") in main_module._active_prints
+    assert main_module._active_prints[(7, "owned_job.gcode.3mf")] == 123
+    assert 7 not in main_module._unauthorized_print_kill_sent
+
+
+@pytest.mark.asyncio
+async def test_kill_switch_defers_when_restart_identity_is_not_available(monkeypatch):
+    state = SimpleNamespace(
+        current_print=None,
+        subtask_name="owned_job",
+        subtask_id=None,
+        gcode_file="owned_job.gcode.3mf",
+    )
+    db = SimpleNamespace(execute=AsyncMock())
+
+    monkeypatch.setattr(main_module.printer_manager, "get_current_print_user", lambda printer_id: None)
+
+    authorization = await main_module._is_bambuddy_authorized_print(7, state, db)
+
+    assert authorization is None
+    db.execute.assert_not_awaited()

backend/tests/unit/test_timelapse_baseline_restart_recovery.py

@@ -98,7 +98,8 @@ async def test_running_observed_skips_when_baseline_already_present():
     """If on_print_start already ran in this Bambuddy process for the same
     printer (the realistic same-session race), a second capture would
     overwrite the correct pre-print baseline with one taken later — which
-    could include the in-flight MP4. Skip when a baseline exists."""
+    could include the in-flight MP4. The archive lookup still has to run so
+    restart recovery can restore durable print ownership for the kill switch."""
     _timelapse_baselines[1] = {"pre_existing_a.mp4", "pre_existing_b.mp4"}
 
     with (
@@ -118,8 +119,9 @@ async def test_running_observed_skips_when_baseline_already_present():
             },
         )
 
-        # Neither the DB lookup nor the FTP scan should have run.
-        mock_session_maker.assert_not_called()
+        # Ownership reconciliation still consults the DB, but the expensive
+        # timelapse scan remains one-shot.
+        mock_session_maker.assert_called_once()
         mock_list.assert_not_called()
 
     # Original baseline preserved.

frontend/src/__tests__/hooks/useWebSocket.test.ts

@@ -6,7 +6,7 @@
  */
 
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
-import { renderHook, waitFor, act } from '@testing-library/react';
+import { renderHook, waitFor, act, screen } from '@testing-library/react';
 import React from 'react';
 import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
 import { ToastProvider } from '../../contexts/ToastContext';
@@ -23,6 +23,10 @@ vi.mock('react-i18next', () => ({
         const { printer, slots } = options as { printer: string; slots: string };
         return `Missing assignments for ${printer}: ${slots}`;
       }
+      if (key === 'printers.toast.killSwitchTriggered' && options) {
+        const { printer, filename } = options as { printer: string; filename: string };
+        return `Billing kill switch stopped ${filename} on ${printer}`;
+      }
       return key;
     },
     i18n: {},
@@ -490,6 +494,28 @@ describe('useWebSocket hook', () => {
       vi.unstubAllGlobals();
     });
 
+    it('shows an error toast when the billing kill switch stops a print', async () => {
+      const { useWebSocket } = await import('../../hooks/useWebSocket');
+
+      renderHook(() => useWebSocket(), {
+        wrapper: createWrapper(queryClient),
+      });
+
+      const ws = await waitForWs();
+      act(() => {
+        ws.open();
+        ws.simulateMessage({
+          type: 'kill_switch_triggered',
+          printer_id: 7,
+          printer_name: 'Printer B',
+          filename: 'foreign_job.3mf',
+        });
+      });
+
+      const toast = screen.getByText('Billing kill switch stopped foreign_job.3mf on Printer B');
+      expect(toast.parentElement).toHaveClass('bg-red-500/10');
+    });
+
     it('handles spool_assignment_verified messages (success and failure) without error', async () => {
       const { useWebSocket } = await import('../../hooks/useWebSocket');
 

frontend/src/hooks/useWebSocket.ts

@@ -17,6 +17,7 @@ interface WebSocketMessage {
   printer_id?: number;
   data?: Record<string, unknown>;
   printer_name?: string;
+  filename?: string;
   missing_slots?: Array<{ slot?: string }>;
   // Spool-assignment read-back verification (#2582).
   slot?: string;
@@ -338,6 +339,13 @@ export function useWebSocket() {
         debouncedInvalidate('archiveStats');
         break;
 
+      case 'kill_switch_triggered': {
+        const printer = message.printer_name || `Printer ${message.printer_id ?? '?'}`;
+        const filename = message.filename || t('common.unknown');
+        showToast(t('printers.toast.killSwitchTriggered', { printer, filename }), 'error');
+        break;
+      }
+
       case 'archive_created':
         debouncedInvalidate('archives');
         debouncedInvalidate('archiveStats');

frontend/src/i18n/locales/de.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Drucker gelöscht',
       missingSpoolAssignment: 'Druck gestartet auf {{printer}}. Fehlende Spulenzuordnung für: {{slots}}',
+      killSwitchTriggered: 'Der Billing-Kill-Switch hat einen nicht autorisierten Druck auf {{printer}} gestoppt: {{filename}}',
       assignmentVerified: 'Filament in Slot {{slot}} geladen ({{printer}})',
       assignmentVerifiedNoKprofile: 'Slot {{slot}} auf {{printer}} geladen, aber das Fluss-Kalibrierungsprofil (K-Profil) wurde nicht übernommen',
       assignmentNotConfirmed: 'Zuordnung für Slot {{slot}} auf {{printer}} konnte nicht bestätigt werden – bitte den AMS-Slot prüfen',

frontend/src/i18n/locales/en.ts

@@ -450,6 +450,7 @@ export default {
     toast: {
       printerDeleted: 'Printer deleted',
       missingSpoolAssignment: 'Print started on {{printer}}. Missing spool assignment for: {{slots}}',
+      killSwitchTriggered: 'The billing kill switch stopped an unauthorized print on {{printer}}: {{filename}}',
       assignmentVerified: 'Filament loaded on slot {{slot}} ({{printer}})',
       assignmentVerifiedNoKprofile: 'Slot {{slot}} on {{printer}} loaded, but the flow calibration (K-profile) was not applied',
       assignmentNotConfirmed: 'Could not confirm the assignment for slot {{slot}} on {{printer}} — check the AMS slot',

frontend/src/i18n/locales/es.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Impresora eliminada',
       missingSpoolAssignment: 'Impresión iniciada en {{printer}}. Falta la asignación de bobina para: {{slots}}',
+      killSwitchTriggered: 'El interruptor de seguridad de facturación detuvo una impresión no autorizada en {{printer}}: {{filename}}',
       assignmentVerified: 'Filamento cargado en la ranura {{slot}} ({{printer}})',
       assignmentVerifiedNoKprofile: 'Ranura {{slot}} en {{printer}} cargada, pero no se aplicó el perfil de calibración de flujo (perfil K)',
       assignmentNotConfirmed: 'No se pudo confirmar la asignación de la ranura {{slot}} en {{printer}}: revisa la ranura AMS',

frontend/src/i18n/locales/fr.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Imprimante supprimée',
       missingSpoolAssignment: 'Impression démarrée sur {{printer}}. Attribution de bobine manquante pour : {{slots}}',
+      killSwitchTriggered: 'Le coupe-circuit de facturation a arrêté une impression non autorisée sur {{printer}} : {{filename}}',
       assignmentVerified: 'Filament chargé dans l\'emplacement {{slot}} ({{printer}})',
       assignmentVerifiedNoKprofile: 'Emplacement {{slot}} sur {{printer}} chargé, mais le profil de calibration de débit (profil K) n\'a pas été appliqué',
       assignmentNotConfirmed: 'Impossible de confirmer l\'attribution de l\'emplacement {{slot}} sur {{printer}} — vérifiez l\'emplacement AMS',

frontend/src/i18n/locales/it.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Stampante eliminata',
       missingSpoolAssignment: 'Stampa avviata su {{printer}}. Mancano assegnazioni bobina per: {{slots}}',
+      killSwitchTriggered: 'L’interruttore di sicurezza della fatturazione ha arrestato una stampa non autorizzata su {{printer}}: {{filename}}',
       assignmentVerified: 'Filamento caricato nello slot {{slot}} ({{printer}})',
       assignmentVerifiedNoKprofile: 'Slot {{slot}} su {{printer}} caricato, ma il profilo di calibrazione del flusso (profilo K) non è stato applicato',
       assignmentNotConfirmed: 'Impossibile confermare l\'assegnazione dello slot {{slot}} su {{printer}} — controlla lo slot AMS',

frontend/src/i18n/locales/ja.ts

@@ -446,6 +446,7 @@ export default {
     toast: {
       printerDeleted: 'プリンターを削除しました',
       missingSpoolAssignment: '{{printer}}で印刷を開始しました。以下のスプール割り当てがありません: {{slots}}',
+      killSwitchTriggered: '課金キルスイッチが{{printer}}で未承認の印刷を停止しました：{{filename}}',
       assignmentVerified: 'スロット{{slot}}にフィラメントを読み込みました（{{printer}}）',
       assignmentVerifiedNoKprofile: '{{printer}}のスロット{{slot}}を読み込みましたが、フロー校正プロファイル（Kプロファイル）は適用されませんでした',
       assignmentNotConfirmed: '{{printer}}のスロット{{slot}}の割り当てを確認できませんでした。AMSスロットを確認してください',

frontend/src/i18n/locales/ko.ts

@@ -421,6 +421,7 @@ export default {
     toast: {
       printerDeleted: '프린터가 삭제되었습니다',
       missingSpoolAssignment: '{{printer}}에서 인쇄가 시작되었습니다. 슬롯 할당 누락: {{slots}}',
+      killSwitchTriggered: '결제 킬 스위치가 {{printer}}에서 승인되지 않은 인쇄를 중지했습니다: {{filename}}',
       assignmentVerified: '슬롯 {{slot}}에 필라멘트가 로드되었습니다 ({{printer}})',
       assignmentVerifiedNoKprofile: '{{printer}}의 슬롯 {{slot}}이(가) 로드되었지만 유량 보정 프로파일(K 프로파일)이 적용되지 않았습니다',
       assignmentNotConfirmed: '{{printer}}의 슬롯 {{slot}} 할당을 확인할 수 없습니다. AMS 슬롯을 확인하세요',

frontend/src/i18n/locales/pt-BR.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Impressora excluída',
       missingSpoolAssignment: 'Impressão iniciada em {{printer}}. Atribuição de bobina ausente para: {{slots}}',
+      killSwitchTriggered: 'O bloqueio de segurança de cobrança interrompeu uma impressão não autorizada em {{printer}}: {{filename}}',
       assignmentVerified: 'Filamento carregado no compartimento {{slot}} ({{printer}})',
       assignmentVerifiedNoKprofile: 'Compartimento {{slot}} em {{printer}} carregado, mas o perfil de calibração de fluxo (perfil K) não foi aplicado',
       assignmentNotConfirmed: 'Não foi possível confirmar a atribuição do compartimento {{slot}} em {{printer}} — verifique o compartimento AMS',

frontend/src/i18n/locales/ru.ts

@@ -426,6 +426,7 @@ export default {
     toast: {
       printerDeleted: "Принтер удалён",
       missingSpoolAssignment: "На принтере {{printer}} началась печать. Не назначены катушки для слотов: {{slots}}",
+      killSwitchTriggered: 'Аварийный выключатель биллинга остановил несанкционированную печать на {{printer}}: {{filename}}',
       assignmentVerified: "Филамент загружен в слот {{slot}} ({{printer}})",
       assignmentVerifiedNoKprofile: "Слот {{slot}} на {{printer}} загружен, но профиль калибровки потока (K-профиль) не применён",
       assignmentNotConfirmed: "Не удалось подтвердить назначение слота {{slot}} на {{printer}} — проверьте слот AMS",

frontend/src/i18n/locales/tr.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: 'Yazıcı silindi',
       missingSpoolAssignment: '{{printer}} üzerinde baskı başladı. Şunlar için eksik makara ataması: {{slots}}',
+      killSwitchTriggered: 'Faturalandırma durdurma anahtarı {{printer}} üzerindeki yetkisiz baskıyı durdurdu: {{filename}}',
       assignmentVerified: '{{slot}} yuvasına filament yüklendi ({{printer}})',
       assignmentVerifiedNoKprofile: '{{printer}} üzerindeki {{slot}} yuvası yüklendi, ancak akış kalibrasyonu profili (K profili) uygulanmadı',
       assignmentNotConfirmed: '{{printer}} üzerindeki {{slot}} yuvası ataması doğrulanamadı — AMS yuvasını kontrol edin',

frontend/src/i18n/locales/uk.ts

@@ -450,6 +450,7 @@ export default {
     toast: {
       printerDeleted: "Принтер видалено",
       missingSpoolAssignment: "Друк на {{printer}} розпочато. Для слотів {{slots}} не призначено котушки.",
+      killSwitchTriggered: 'Аварійний вимикач білінгу зупинив несанкціонований друк на {{printer}}: {{filename}}',
       assignmentVerified: "Філамент завантажено в слот AMS {{slot}} принтера {{printer}}",
       assignmentVerifiedNoKprofile: "Слот AMS {{slot}} на {{printer}} завантажено, але калібрування потоку (K-профіль) не застосовано",
       assignmentNotConfirmed: "Не вдалося підтвердити призначення для слота {{slot}} на {{printer}} — перевірте слот AMS",

frontend/src/i18n/locales/zh-CN.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: '打印机已删除',
       missingSpoolAssignment: '已在{{printer}}上开始打印。以下料槽未分配耗材: {{slots}}',
+      killSwitchTriggered: '计费终止开关已停止 {{printer}} 上的未授权打印：{{filename}}',
       assignmentVerified: '耗材已加载到料槽{{slot}}（{{printer}}）',
       assignmentVerifiedNoKprofile: '{{printer}}的料槽{{slot}}已加载，但流量校准配置（K配置）未应用',
       assignmentNotConfirmed: '无法确认{{printer}}上料槽{{slot}}的分配，请检查AMS料槽',

frontend/src/i18n/locales/zh-TW.ts

@@ -447,6 +447,7 @@ export default {
     toast: {
       printerDeleted: '印表機已刪除',
       missingSpoolAssignment: '已在{{printer}}上開始列印。以下料槽未分配耗材: {{slots}}',
+      killSwitchTriggered: '計費終止開關已停止 {{printer}} 上的未授權列印：{{filename}}',
       assignmentVerified: '耗材已載入料槽{{slot}}（{{printer}}）',
       assignmentVerifiedNoKprofile: '{{printer}}的料槽{{slot}}已載入，但流量校準設定檔（K設定檔）未套用',
       assignmentNotConfirmed: '無法確認{{printer}}上料槽{{slot}}的分配，請檢查AMS料槽',