feat(finance): add cost center management and wallet transactions in the backend

- Introduced CostCenter and related models for managing print costs and budgets.
- Updated PrintArchive and PrintQueueItem models to include cost_center_id and estimated_cost.
- Implemented budget reservation logic in finance services to validate and manage print costs.
- Enhanced ArchiveService and BackgroundDispatchService to handle cost center information during print jobs.
- Added wallet transaction handling for print charges, including partial charges based on filament usage.
- Created finance billing and budget services to manage user wallets and budget reservations.
- Ensured user finance defaults are created upon user registration, including wallets and private cost centers.
- Updated print scheduler to validate budget before processing print jobs.

backend/app/api/routes/archives.py

@@ -29,6 +29,7 @@ from backend.app.schemas.archive import ArchiveResponse, ArchiveSlim, ArchiveSta
 from backend.app.schemas.print_log import PrintLogResponse
 from backend.app.schemas.slicer import SliceRequest
 from backend.app.services.archive import ArchiveService
+from backend.app.services.finance_budget import validate_print_budget
 from backend.app.utils.http import build_content_disposition
 from backend.app.utils.safe_path import safe_join_under
 from backend.app.utils.threemf_tools import (
@@ -4078,6 +4079,13 @@ async def reprint_archive(
     if not file_path.is_file():
         raise HTTPException(404, "Archive file not found")
 
+    await validate_print_budget(
+        db,
+        cost_center_id=body.cost_center_id,
+        estimated_cost=body.estimated_cost,
+        current_user=user,
+    )
+
     plate_name = body.plate_name
     if not plate_name and body.plate_id is not None:
         plate_name = f"Plate {body.plate_id}"

backend/app/api/routes/finance.py

@@ -0,0 +1,1085 @@
+import calendar
+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.ext.asyncio import AsyncSession
+from sqlalchemy.orm import selectinload
+
+from backend.app.core.auth import RequirePermissionIfAuthEnabled, require_auth_if_enabled
+from backend.app.core.database import get_db
+from backend.app.core.permissions import Permission
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import (
+    BudgetReservation,
+    CostCenter,
+    CostCenterMember,
+    TransactionType,
+    UserWallet,
+    WalletTransaction,
+    normalize_transaction_type,
+)
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.schemas.finance import (
+    CostCenterBudgetUpdateRequest,
+    CostCenterCreateRequest,
+    CostCenterDetailResponse,
+    CostCenterMemberRequest,
+    CostCenterMemberResponse,
+    CostCenterSummaryResponse,
+    CostCenterUpdateRequest,
+    ManualPrintRequest,
+    TransactionEditRequest,
+    WalletAdjustmentRequest,
+    WalletAdjustmentResponse,
+    WalletBalanceResponse,
+    WalletTransactionListResponse,
+    WalletTransactionResponse,
+)
+
+router = APIRouter(prefix="/finance", tags=["finance"])
+
+
+def _serialize_wallet_transaction(tx: WalletTransaction) -> WalletTransactionResponse:
+    transaction_type = (
+        tx.transaction_type.value if isinstance(tx.transaction_type, TransactionType) else tx.transaction_type
+    )
+    return WalletTransactionResponse.model_construct(
+        id=tx.id,
+        user_id=tx.user_id,
+        cost_center_id=tx.cost_center_id,
+        transaction_type=transaction_type,
+        amount=tx.amount,
+        balance_after=tx.balance_after,
+        description=tx.description,
+        created_by_user_id=tx.created_by_user_id,
+        print_run_id=tx.print_run_id,
+        print_archive_id=tx.print_archive_id,
+        print_queue_id=tx.print_queue_id,
+        created_at=tx.created_at,
+    )
+
+
+def _clamp_day(year: int, month: int, desired_day: int) -> int:
+    return min(max(1, desired_day), calendar.monthrange(year, month)[1])
+
+
+async def _get_budget_window_start_utc(db: AsyncSession) -> datetime:
+    """Resolve monthly budget window start in UTC using configurable reset day/timezone.
+
+    Defaults preserve current behavior: day=1, timezone=UTC.
+    """
+    desired_day = 1
+    tz_name = "UTC"
+
+    result = await db.execute(
+        select(Settings).where(Settings.key.in_(["finance_budget_reset_day", "finance_budget_reset_timezone"]))
+    )
+    for setting in result.scalars().all():
+        if setting.key == "finance_budget_reset_day":
+            try:
+                parsed = int(setting.value)
+                if 1 <= parsed <= 31:
+                    desired_day = parsed
+            except (TypeError, ValueError):
+                pass
+        elif setting.key == "finance_budget_reset_timezone":
+            value = (setting.value or "").strip()
+            if value:
+                tz_name = value
+
+    try:
+        tz = ZoneInfo(tz_name)
+    except ZoneInfoNotFoundError:
+        tz = timezone.utc
+
+    now_local = datetime.now(tz)
+    current_month_reset_day = _clamp_day(now_local.year, now_local.month, desired_day)
+
+    if now_local.day >= current_month_reset_day:
+        start_local = datetime(now_local.year, now_local.month, current_month_reset_day, tzinfo=tz)
+    else:
+        prev_year = now_local.year
+        prev_month = now_local.month - 1
+        if prev_month == 0:
+            prev_month = 12
+            prev_year -= 1
+        prev_month_reset_day = _clamp_day(prev_year, prev_month, desired_day)
+        start_local = datetime(prev_year, prev_month, prev_month_reset_day, tzinfo=tz)
+
+    return start_local.astimezone(timezone.utc)
+
+
+async def _get_cost_center_usage_maps(
+    db: AsyncSession,
+    cost_center_ids: list[int],
+) -> tuple[dict[int, float], dict[int, float]]:
+    if not cost_center_ids:
+        return {}, {}
+
+    spend_expr = case((WalletTransaction.amount < 0, -WalletTransaction.amount), else_=0.0)
+
+    total_rows = await db.execute(
+        select(WalletTransaction.cost_center_id, func.coalesce(func.sum(spend_expr), 0.0))
+        .where(
+            WalletTransaction.cost_center_id.in_(cost_center_ids),
+            WalletTransaction.cost_center_id.is_not(None),
+        )
+        .group_by(WalletTransaction.cost_center_id)
+    )
+
+    budget_window_start_utc = await _get_budget_window_start_utc(db)
+
+    month_rows = await db.execute(
+        select(WalletTransaction.cost_center_id, func.coalesce(func.sum(spend_expr), 0.0))
+        .where(
+            WalletTransaction.cost_center_id.in_(cost_center_ids),
+            WalletTransaction.cost_center_id.is_not(None),
+            WalletTransaction.created_at >= budget_window_start_utc,
+        )
+        .group_by(WalletTransaction.cost_center_id)
+    )
+
+    total_map = {int(center_id): float(value) for center_id, value in total_rows.all() if center_id is not None}
+    month_map = {int(center_id): float(value) for center_id, value in month_rows.all() if center_id is not None}
+    return total_map, month_map
+
+
+async def _get_cost_center_balance_map(
+    db: AsyncSession,
+    cost_center_ids: list[int],
+) -> dict[int, float]:
+    if not cost_center_ids:
+        return {}
+
+    rows = await db.execute(
+        select(WalletTransaction.cost_center_id, func.coalesce(func.sum(WalletTransaction.amount), 0.0))
+        .where(
+            WalletTransaction.cost_center_id.in_(cost_center_ids),
+            WalletTransaction.cost_center_id.is_not(None),
+        )
+        .group_by(WalletTransaction.cost_center_id)
+    )
+    return {int(center_id): float(value) for center_id, value in rows.all() if center_id is not None}
+
+
+async def _get_cost_center_reserved_map(
+    db: AsyncSession,
+    cost_center_ids: list[int],
+) -> dict[int, float]:
+    if not cost_center_ids:
+        return {}
+
+    budget_rows = await db.execute(
+        select(BudgetReservation.cost_center_id, func.coalesce(func.sum(BudgetReservation.amount), 0.0))
+        .where(
+            BudgetReservation.cost_center_id.in_(cost_center_ids),
+            BudgetReservation.status == "active",
+        )
+        .group_by(BudgetReservation.cost_center_id)
+    )
+    reserved_map = {int(center_id): float(value) for center_id, value in budget_rows.all() if center_id is not None}
+
+    queue_rows = await db.execute(
+        select(PrintQueueItem.cost_center_id, func.coalesce(func.sum(PrintQueueItem.estimated_cost), 0.0))
+        .where(
+            PrintQueueItem.cost_center_id.in_(cost_center_ids),
+            PrintQueueItem.status.in_(("pending", "printing")),
+        )
+        .group_by(PrintQueueItem.cost_center_id)
+    )
+    for center_id, value in queue_rows.all():
+        if center_id is not None:
+            reserved_map[int(center_id)] = reserved_map.get(int(center_id), 0.0) + float(value or 0.0)
+    return reserved_map
+
+
+def _budget_mode_and_limit(center: CostCenter) -> tuple[str, float | None]:
+    # Monthly takes precedence if legacy data still has both set.
+    if center.monthly_budget is not None:
+        return "monthly", float(center.monthly_budget)
+    if center.total_budget is not None:
+        return "total", float(center.total_budget)
+    return "none", None
+
+
+def _to_cost_center_summary(
+    center: CostCenter,
+    *,
+    can_print: bool,
+    total_usage: float,
+    month_usage: float,
+    total_balance: float,
+    reserved: float = 0.0,
+) -> CostCenterSummaryResponse:
+    budget_mode, budget_limit = _budget_mode_and_limit(center)
+    budget_used = month_usage if budget_mode == "monthly" else total_usage if budget_mode == "total" else None
+    budget_available = (
+        max(0.0, budget_limit - budget_used - reserved)
+        if budget_limit is not None and budget_used is not None
+        else None
+    )
+
+    return CostCenterSummaryResponse(
+        id=center.id,
+        name=center.name,
+        is_private=center.is_private,
+        owner_user_id=center.owner_user_id,
+        is_active=center.is_active,
+        total_balance=total_balance,
+        total_budget=center.total_budget,
+        monthly_budget=center.monthly_budget,
+        budget_mode=budget_mode,
+        budget_limit=budget_limit,
+        budget_used=budget_used,
+        budget_available=budget_available,
+        can_print=can_print,
+    )
+
+
+async def _require_authenticated_user(current_user: User | None) -> User:
+    if current_user is None:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return current_user
+
+
+def _has_cost_center_admin_access(user: User) -> bool:
+    return user.has_any_permission(
+        Permission.COST_CENTERS_READ_ALL.value,
+        Permission.COST_CENTERS_MODIFY.value,
+        Permission.COST_CENTERS_CREATE.value,
+    )
+
+
+async def _require_cost_center_admin_access(current_user: User | None) -> User:
+    user = await _require_authenticated_user(current_user)
+    if not _has_cost_center_admin_access(user):
+        raise HTTPException(status_code=403, detail="Missing required permissions for cost center administration")
+    return user
+
+
+async def _get_or_create_wallet(db: AsyncSession, user_id: int) -> UserWallet:
+    result = await db.execute(select(UserWallet).where(UserWallet.user_id == user_id))
+    wallet = result.scalar_one_or_none()
+    if wallet:
+        return wallet
+
+    wallet = UserWallet(user_id=user_id, balance=0.0, currency="EUR")
+    db.add(wallet)
+    await db.flush()
+    await db.refresh(wallet)
+    return wallet
+
+
+async def _get_user_or_404(db: AsyncSession, user_id: int) -> User:
+    result = await db.execute(select(User).where(User.id == user_id))
+    user = result.scalar_one_or_none()
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+    return user
+
+
+async def _get_cost_center_or_404(db: AsyncSession, cost_center_id: int) -> CostCenter:
+    result = await db.execute(
+        select(CostCenter).options(selectinload(CostCenter.members)).where(CostCenter.id == cost_center_id)
+    )
+    center = result.scalar_one_or_none()
+    if center is None:
+        raise HTTPException(status_code=404, detail="Cost center not found")
+    return center
+
+
+def _to_balance_response(wallet: UserWallet) -> WalletBalanceResponse:
+    return WalletBalanceResponse(
+        user_id=wallet.user_id,
+        balance=wallet.balance,
+        currency=wallet.currency,
+        updated_at=wallet.updated_at,
+    )
+
+
+async def _build_personal_balance_map(db: AsyncSession, user_id: int) -> dict[int, float]:
+    result = await db.execute(
+        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)
+        .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
+
+    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,
+    *,
+    target_user_id: int,
+    actor_user_id: int,
+    amount: float,
+    transaction_type: str,
+    description: str | None,
+    cost_center_id: int | None,
+) -> WalletAdjustmentResponse:
+    transaction_type = normalize_transaction_type(transaction_type)
+
+    if cost_center_id is not None:
+        await _get_cost_center_or_404(db, cost_center_id)
+
+    wallet = await _get_or_create_wallet(db, target_user_id)
+
+    # Calculate balance_after for this specific transaction context
+    if cost_center_id is None:
+        # Personal transaction: validate and update user wallet
+        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)
+        result = await db.execute(
+            select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)).where(
+                WalletTransaction.cost_center_id == cost_center_id,
+            )
+        )
+        current_cc_balance = float(result.scalar() or 0.0)
+        new_cc_balance = current_cc_balance + amount
+        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
+
+    tx = WalletTransaction(
+        user_id=target_user_id,
+        cost_center_id=cost_center_id,
+        transaction_type=transaction_type,
+        amount=amount,
+        balance_after=balance_after,
+        description=description,
+        created_by_user_id=actor_user_id,
+    )
+    db.add(tx)
+    await db.flush()
+    await db.commit()
+    await db.refresh(wallet)
+    await db.refresh(tx)
+
+    # Return appropriate balance based on transaction type
+    if cost_center_id is None:
+        # Personal transaction: return user wallet balance
+        response_balance = _to_balance_response(wallet)
+    else:
+        # Cost-center transaction: return cost-center balance as if it were a wallet
+        response_balance = WalletBalanceResponse(
+            user_id=target_user_id,
+            balance=balance_after,
+            currency=wallet.currency,
+            updated_at=tx.created_at,
+        )
+
+    return WalletAdjustmentResponse(
+        transaction=_serialize_wallet_transaction(tx),
+        balance=response_balance,
+    )
+
+
+@router.get("/me/balance", response_model=WalletBalanceResponse)
+async def get_my_balance(
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_OWN),
+):
+    """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,
+    )
+
+
+@router.get("/me/transactions", response_model=WalletTransactionListResponse)
+async def get_my_transactions(
+    limit: int = Query(50, ge=1, le=500),
+    offset: int = Query(0, ge=0),
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_OWN),
+):
+    """Return wallet ledger entries for the current user."""
+    user = await _require_authenticated_user(current_user)
+
+    total_result = await db.execute(
+        select(func.count(WalletTransaction.id)).where(WalletTransaction.user_id == user.id)
+    )
+    total = int(total_result.scalar_one() or 0)
+
+    result = await db.execute(
+        select(WalletTransaction)
+        .where(WalletTransaction.user_id == user.id)
+        .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    transactions = result.scalars().all()
+    personal_balance_map = await _build_personal_balance_map(db, user.id)
+    return WalletTransactionListResponse(
+        items=[
+            _serialize_wallet_transaction(tx).model_copy(
+                update={"balance_after": personal_balance_map.get(tx.id, tx.balance_after)}
+            )
+            for tx in transactions
+        ],
+        total=total,
+        limit=limit,
+        offset=offset,
+    )
+
+
+@router.get("/transactions", response_model=WalletTransactionListResponse)
+async def get_all_transactions(
+    limit: int = Query(50, ge=1, le=500),
+    offset: int = Query(0, ge=0),
+    user_id: int | None = Query(None, description="Optional filter by user id"),
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Return wallet ledger entries across users for admin finance view."""
+    await _require_authenticated_user(current_user)
+
+    conditions = []
+    if user_id is not None:
+        await _get_user_or_404(db, user_id)
+        conditions.append(WalletTransaction.user_id == user_id)
+
+    total_result = await db.execute(select(func.count(WalletTransaction.id)).where(*conditions))
+    total = int(total_result.scalar_one() or 0)
+
+    result = await db.execute(
+        select(WalletTransaction)
+        .where(*conditions)
+        .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    transactions = result.scalars().all()
+    return WalletTransactionListResponse(
+        items=[_serialize_wallet_transaction(tx) for tx in transactions],
+        total=total,
+        limit=limit,
+        offset=offset,
+    )
+
+
+async def _rebuild_wallet_ledger_for_user(db: AsyncSession, user_id: int) -> None:
+    """Recompute `balance_after` for all wallet transactions of a user.
+
+    - 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()
+
+
+@router.delete("/transactions/{transaction_id}")
+async def delete_transaction(
+    transaction_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Delete a wallet transaction and rebuild the user's ledger to keep balances consistent."""
+    await _require_authenticated_user(current_user)
+
+    result = await db.execute(select(WalletTransaction).where(WalletTransaction.id == transaction_id))
+    tx = result.scalar_one_or_none()
+    if tx is None:
+        raise HTTPException(status_code=404, detail="Transaction not found")
+
+    user_id = tx.user_id
+
+    if tx.transaction_type == "print_charge" and tx.print_archive_id is not None:
+        archive_result = await db.execute(select(PrintArchive).where(PrintArchive.id == tx.print_archive_id))
+        archive = archive_result.scalar_one_or_none()
+        if archive is not None:
+            archive.wallet_charge_skipped = True
+            db.add(archive)
+
+    await db.delete(tx)
+    await db.flush()
+
+    await _rebuild_wallet_ledger_for_user(db, user_id)
+
+    return {"status": "success"}
+
+
+@router.patch("/transactions/{transaction_id}", response_model=WalletTransactionResponse)
+async def edit_transaction(
+    transaction_id: int,
+    request: TransactionEditRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Edit a wallet transaction (user_id, cost_center_id, amount, description) and rebuild ledger."""
+    await _require_authenticated_user(current_user)
+
+    result = await db.execute(select(WalletTransaction).where(WalletTransaction.id == transaction_id))
+    tx = result.scalar_one_or_none()
+    if tx is None:
+        raise HTTPException(status_code=404, detail="Transaction not found")
+
+    # Apply edits
+    if request.user_id is not None:
+        tx.user_id = request.user_id
+
+    if request.cost_center_id is not None:
+        tx.cost_center_id = request.cost_center_id
+
+    if request.amount is not None:
+        tx.amount = request.amount
+
+    if request.description is not None:
+        # Append "(Admin edit)" marker if not already present
+        new_desc = request.description
+        if not new_desc.endswith("(Admin edit)"):
+            new_desc = f"{new_desc} (Admin edit)"
+        tx.description = new_desc
+
+    db.add(tx)
+    await db.flush()
+
+    # Rebuild full ledger using the current session
+    from backend.app.core.database import repair_wallet_ledger_internal
+
+    await repair_wallet_ledger_internal(db)
+
+    return tx
+
+
+@router.post("/transactions/manual", response_model=WalletTransactionResponse)
+async def create_manual_print(
+    request: ManualPrintRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Create a manual print charge transaction (for admin purposes)."""
+    await _require_authenticated_user(current_user)
+
+    from datetime import timezone
+
+    # Use provided created_at or current time
+    created_at = request.created_at or datetime.now(timezone.utc)
+
+    # Ensure manual print charges are negative amounts (charges reduce wallet)
+    amount = request.amount
+    if amount > 0:
+        amount = -abs(amount)
+
+    # Create transaction
+    tx = WalletTransaction(
+        user_id=request.user_id,
+        cost_center_id=request.cost_center_id,
+        transaction_type=TransactionType.MANUAL_ADJUSTMENT.value,
+        amount=amount,
+        balance_after=None,  # Will be set by repair_wallet_ledger_internal
+        description=request.description or "Manual print charge",
+        created_by_user_id=current_user.id if current_user else None,
+        created_at=created_at,
+    )
+    db.add(tx)
+    await db.flush()
+
+    # Rebuild full ledger using the current session
+    from backend.app.core.database import repair_wallet_ledger_internal
+
+    await repair_wallet_ledger_internal(db)
+
+    return tx
+
+
+@router.get("/cost-centers/mine", response_model=list[CostCenterSummaryResponse])
+async def get_my_cost_centers(
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = Depends(require_auth_if_enabled),
+):
+    """Return private and assigned cost centers for the current user."""
+    user = await _require_authenticated_user(current_user)
+
+    result = await db.execute(
+        select(CostCenter, CostCenterMember.can_print)
+        .outerjoin(
+            CostCenterMember,
+            (CostCenterMember.cost_center_id == CostCenter.id) & (CostCenterMember.user_id == user.id),
+        )
+        .where(
+            CostCenter.is_active.is_(True),
+            or_(
+                (CostCenter.is_private.is_(True) & (CostCenter.owner_user_id == user.id)),
+                (CostCenterMember.user_id == user.id),
+            ),
+        )
+        .order_by(CostCenter.is_private.desc(), CostCenter.name.asc())
+    )
+
+    rows = result.all()
+    centers_only = [center for center, _ in rows]
+    center_ids = [center.id for center in centers_only]
+    total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, center_ids)
+    total_balance_map = await _get_cost_center_balance_map(db, center_ids)
+    reserved_map = await _get_cost_center_reserved_map(db, center_ids)
+
+    centers: list[CostCenterSummaryResponse] = []
+    for center, can_print in rows:
+        centers.append(
+            _to_cost_center_summary(
+                center,
+                can_print=True if center.is_private and center.owner_user_id == user.id else bool(can_print),
+                total_usage=total_usage_map.get(center.id, 0.0),
+                month_usage=month_usage_map.get(center.id, 0.0),
+                total_balance=total_balance_map.get(center.id, 0.0),
+                reserved=reserved_map.get(center.id, 0.0),
+            )
+        )
+
+    return centers
+
+
+@router.get("/users/{user_id}/balance", response_model=WalletBalanceResponse)
+async def get_user_balance(
+    user_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Return a specific user's wallet balance."""
+    await _require_authenticated_user(current_user)
+    user = await _get_user_or_404(db, user_id)
+    wallet = await _get_or_create_wallet(db, user.id)
+    return _to_balance_response(wallet)
+
+
+@router.get("/users/{user_id}/transactions", response_model=list[WalletTransactionResponse])
+async def get_user_transactions(
+    user_id: int,
+    limit: int = Query(50, ge=1, le=500),
+    offset: int = Query(0, ge=0),
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_READ_ALL),
+):
+    """Return wallet ledger entries for a specific user."""
+    await _require_authenticated_user(current_user)
+    await _get_user_or_404(db, user_id)
+
+    result = await db.execute(
+        select(WalletTransaction)
+        .where(WalletTransaction.user_id == user_id)
+        .order_by(WalletTransaction.created_at.desc(), WalletTransaction.id.desc())
+        .limit(limit)
+        .offset(offset)
+    )
+    return [_serialize_wallet_transaction(tx) for tx in result.scalars().all()]
+
+
+@router.post("/users/{user_id}/deposit", response_model=WalletAdjustmentResponse)
+async def deposit_user_balance(
+    user_id: int,
+    body: WalletAdjustmentRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Add funds to a user's wallet."""
+    actor = await _require_authenticated_user(current_user)
+    await _get_user_or_404(db, user_id)
+    return await _create_wallet_adjustment(
+        db,
+        target_user_id=user_id,
+        actor_user_id=actor.id,
+        amount=body.amount,
+        transaction_type=TransactionType.DEPOSIT.value,
+        description=body.description,
+        cost_center_id=body.cost_center_id,
+    )
+
+
+@router.post("/users/{user_id}/withdraw", response_model=WalletAdjustmentResponse)
+async def withdraw_user_balance(
+    user_id: int,
+    body: WalletAdjustmentRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Withdraw funds from a user's wallet."""
+    actor = await _require_authenticated_user(current_user)
+    await _get_user_or_404(db, user_id)
+    return await _create_wallet_adjustment(
+        db,
+        target_user_id=user_id,
+        actor_user_id=actor.id,
+        amount=-body.amount,
+        transaction_type=TransactionType.WITHDRAW.value,
+        description=body.description,
+        cost_center_id=body.cost_center_id,
+    )
+
+
+@router.post("/rebuild-balance-ledger")
+async def rebuild_balance_ledger(
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Recompute balance_after for all wallet transactions.
+
+    This rebuilds the running balance for all users and cost centers.
+    - Personal transactions: per-user running balance
+    - 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()
+
+    # 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()
+
+    return {
+        "status": "success",
+        "transactions_rebuilt": len(all_transactions),
+        "message": f"Rebuilt balance_after for {len(all_transactions)} transactions",
+    }
+
+
+@router.get("/cost-centers", response_model=list[CostCenterSummaryResponse])
+async def list_cost_centers(
+    include_inactive: bool = Query(False),
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = Depends(require_auth_if_enabled),
+):
+    """List all cost centers.
+
+    Requires admin-level finance permissions.
+    """
+    await _require_cost_center_admin_access(current_user)
+    query = select(CostCenter).order_by(CostCenter.is_private.desc(), CostCenter.name.asc())
+    if not include_inactive:
+        query = query.where(CostCenter.is_active.is_(True))
+
+    result = await db.execute(query)
+    centers = result.scalars().all()
+    center_ids = [center.id for center in centers]
+    total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, center_ids)
+    total_balance_map = await _get_cost_center_balance_map(db, center_ids)
+    reserved_map = await _get_cost_center_reserved_map(db, center_ids)
+
+    return [
+        _to_cost_center_summary(
+            center,
+            can_print=True,
+            total_usage=total_usage_map.get(center.id, 0.0),
+            month_usage=month_usage_map.get(center.id, 0.0),
+            total_balance=total_balance_map.get(center.id, 0.0),
+            reserved=reserved_map.get(center.id, 0.0),
+        )
+        for center in centers
+    ]
+
+
+@router.post("/cost-centers", response_model=CostCenterSummaryResponse)
+async def create_cost_center(
+    body: CostCenterCreateRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_CREATE),
+):
+    """Create a shared cost center."""
+    await _require_authenticated_user(current_user)
+    total_budget = body.total_budget
+    monthly_budget = body.monthly_budget
+    if monthly_budget is not None:
+        total_budget = None
+    elif total_budget is not None:
+        monthly_budget = None
+
+    center = CostCenter(
+        name=body.name.strip(),
+        is_active=body.is_active,
+        is_private=False,
+        owner_user_id=None,
+        total_budget=total_budget,
+        monthly_budget=monthly_budget,
+    )
+    db.add(center)
+    await db.flush()
+    await db.commit()
+    await db.refresh(center)
+
+    return _to_cost_center_summary(center, can_print=True, total_usage=0.0, month_usage=0.0, total_balance=0.0)
+
+
+@router.get("/cost-centers/{cost_center_id}", response_model=CostCenterDetailResponse)
+async def get_cost_center(
+    cost_center_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = Depends(require_auth_if_enabled),
+):
+    """Get one cost center with its memberships."""
+    await _require_cost_center_admin_access(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+    total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id])
+    total_balance_map = await _get_cost_center_balance_map(db, [center.id])
+    reserved_map = await _get_cost_center_reserved_map(db, [center.id])
+    summary = _to_cost_center_summary(
+        center,
+        can_print=True,
+        total_usage=total_usage_map.get(center.id, 0.0),
+        month_usage=month_usage_map.get(center.id, 0.0),
+        total_balance=total_balance_map.get(center.id, 0.0),
+        reserved=reserved_map.get(center.id, 0.0),
+    )
+    return CostCenterDetailResponse(
+        **summary.model_dump(),
+        members=[CostCenterMemberResponse.model_validate(m) for m in center.members],
+    )
+
+
+@router.patch("/cost-centers/{cost_center_id}", response_model=CostCenterSummaryResponse)
+async def update_cost_center(
+    cost_center_id: int,
+    body: CostCenterUpdateRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Update name or active-state of a cost center."""
+    await _require_authenticated_user(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+
+    if body.name is not None:
+        center.name = body.name.strip()
+    if body.is_active is not None:
+        center.is_active = body.is_active
+
+    await db.flush()
+
+    total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id])
+    total_balance_map = await _get_cost_center_balance_map(db, [center.id])
+    reserved_map = await _get_cost_center_reserved_map(db, [center.id])
+    return _to_cost_center_summary(
+        center,
+        can_print=True,
+        total_usage=total_usage_map.get(center.id, 0.0),
+        month_usage=month_usage_map.get(center.id, 0.0),
+        total_balance=total_balance_map.get(center.id, 0.0),
+        reserved=reserved_map.get(center.id, 0.0),
+    )
+
+
+@router.patch("/cost-centers/{cost_center_id}/budgets", response_model=CostCenterSummaryResponse)
+async def update_cost_center_budgets(
+    cost_center_id: int,
+    body: CostCenterBudgetUpdateRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Update budget values of a cost center."""
+    await _require_authenticated_user(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+
+    if body.monthly_budget is not None:
+        center.monthly_budget = body.monthly_budget
+        center.total_budget = None
+    elif body.total_budget is not None:
+        center.total_budget = body.total_budget
+        center.monthly_budget = None
+    else:
+        center.total_budget = None
+        center.monthly_budget = None
+    await db.flush()
+
+    total_usage_map, month_usage_map = await _get_cost_center_usage_maps(db, [center.id])
+    total_balance_map = await _get_cost_center_balance_map(db, [center.id])
+    reserved_map = await _get_cost_center_reserved_map(db, [center.id])
+    return _to_cost_center_summary(
+        center,
+        can_print=True,
+        total_usage=total_usage_map.get(center.id, 0.0),
+        month_usage=month_usage_map.get(center.id, 0.0),
+        total_balance=total_balance_map.get(center.id, 0.0),
+        reserved=reserved_map.get(center.id, 0.0),
+    )
+
+
+@router.post("/cost-centers/{cost_center_id}/members", response_model=CostCenterMemberResponse)
+async def upsert_cost_center_member(
+    cost_center_id: int,
+    body: CostCenterMemberRequest,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Assign or update a user's membership on a cost center."""
+    await _require_authenticated_user(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+
+    if center.is_private:
+        raise HTTPException(status_code=400, detail="Private cost center memberships cannot be modified")
+
+    await _get_user_or_404(db, body.user_id)
+
+    existing = await db.execute(
+        select(CostCenterMember).where(
+            CostCenterMember.cost_center_id == cost_center_id,
+            CostCenterMember.user_id == body.user_id,
+        )
+    )
+    member = existing.scalar_one_or_none()
+    if member is None:
+        member = CostCenterMember(cost_center_id=cost_center_id, user_id=body.user_id, can_print=body.can_print)
+        db.add(member)
+    else:
+        member.can_print = body.can_print
+
+    await db.flush()
+    await db.commit()
+    return CostCenterMemberResponse.model_validate(member)
+
+
+@router.delete("/cost-centers/{cost_center_id}")
+async def delete_cost_center(
+    cost_center_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Delete a shared cost center."""
+    await _require_authenticated_user(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+
+    if center.is_private:
+        raise HTTPException(status_code=400, detail="Private cost centers cannot be deleted")
+
+    balance_map = await _get_cost_center_balance_map(db, [center.id])
+    total_balance = balance_map.get(center.id, 0.0)
+    if abs(total_balance) > 1e-9:
+        raise HTTPException(status_code=400, detail="Cost center can only be deleted when balance is 0")
+
+    await db.delete(center)
+    await db.commit()
+    return {"status": "success"}
+
+
+@router.delete("/cost-centers/{cost_center_id}/members/{user_id}")
+async def remove_cost_center_member(
+    cost_center_id: int,
+    user_id: int,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.COST_CENTERS_MODIFY),
+):
+    """Remove a user from a shared cost center."""
+    await _require_authenticated_user(current_user)
+    center = await _get_cost_center_or_404(db, cost_center_id)
+    if center.is_private:
+        raise HTTPException(status_code=400, detail="Private cost center memberships cannot be modified")
+
+    result = await db.execute(
+        select(CostCenterMember).where(
+            CostCenterMember.cost_center_id == cost_center_id,
+            CostCenterMember.user_id == user_id,
+        )
+    )
+    member = result.scalar_one_or_none()
+    if member is None:
+        raise HTTPException(status_code=404, detail="Membership not found")
+
+    await db.delete(member)
+    await db.commit()
+    return {"status": "success"}

backend/app/api/routes/library.py

@@ -65,6 +65,7 @@ from backend.app.schemas.library import (
 )
 from backend.app.schemas.slicer import SliceRequest, SliceResponse
 from backend.app.services.archive import ThreeMFParser
+from backend.app.services.finance_budget import validate_print_budget
 from backend.app.services.plate_thumbnail import inject_plate_thumbnails_if_missing
 from backend.app.services.stl_thumbnail import MIN_USABLE_STL_BYTES, generate_stl_thumbnail
 from backend.app.utils.filename import InvalidFilenameError, validate_print_filename
@@ -4181,6 +4182,13 @@ async def print_library_file(
         if not project_result.scalar_one_or_none():
             raise HTTPException(status_code=404, detail="Project not found")
 
+    await validate_print_budget(
+        db,
+        cost_center_id=body.cost_center_id,
+        estimated_cost=body.estimated_cost,
+        current_user=current_user,
+    )
+
     plate_name = body.plate_name
     if not plate_name and body.plate_id is not None:
         plate_name = f"Plate {body.plate_id}"

backend/app/api/routes/print_queue.py

@@ -36,6 +36,7 @@ from backend.app.schemas.print_queue import (
     PrintQueueReorder,
 )
 from backend.app.services.filament_deficit import compute_deficit_for_queue_item
+from backend.app.services.finance_budget import validate_print_budget
 from backend.app.services.notification_service import notification_service
 from backend.app.utils.printer_models import normalize_printer_model, normalize_printer_model_id
 from backend.app.utils.threemf_tools import (
@@ -164,6 +165,8 @@ def _enrich_response(item: PrintQueueItem) -> PrintQueueItemResponse:
         "waiting_reason": item.waiting_reason,
         "archive_id": item.archive_id,
         "library_file_id": item.library_file_id,
+        "cost_center_id": item.cost_center_id,
+        "estimated_cost": item.estimated_cost,
         "position": item.position,
         "scheduled_time": item.scheduled_time,
         "require_previous_success": item.require_previous_success,
@@ -545,6 +548,14 @@ async def add_to_queue(
         if not project_result.scalar_one_or_none():
             raise HTTPException(status_code=404, detail="Project not found")
 
+    await validate_print_budget(
+        db,
+        cost_center_id=data.cost_center_id,
+        estimated_cost=data.estimated_cost,
+        current_user=current_user,
+        quantity=quantity,
+    )
+
     ams_mapping_json = json.dumps(data.ams_mapping) if data.ams_mapping else None
     items = []
     for i in range(quantity):
@@ -556,6 +567,8 @@ async def add_to_queue(
             filament_overrides=filament_overrides_json,
             archive_id=data.archive_id,
             library_file_id=data.library_file_id,
+            cost_center_id=data.cost_center_id,
+            estimated_cost=data.estimated_cost,
             scheduled_time=data.scheduled_time,
             require_previous_success=data.require_previous_success,
             auto_off_after=data.auto_off_after,
@@ -672,6 +685,7 @@ async def bulk_update_queue_items(
 
     updated_count = 0
     skipped_count = 0
+    validates_billing_fields = "cost_center_id" in update_data or "estimated_cost" in update_data
 
     for item in items:
         if item.status != "pending":
@@ -683,6 +697,15 @@ async def bulk_update_queue_items(
             skipped_count += 1
             continue
 
+        if validates_billing_fields:
+            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),
+                current_user=user,
+                exclude_queue_item_id=item.id,
+            )
+
         for field, value in update_data.items():
             setattr(item, field, value)
         updated_count += 1
@@ -1022,12 +1045,13 @@ async def update_queue_item(
             json.dumps(update_data["filament_overrides"]) if update_data["filament_overrides"] else None
         )
 
-    # Serialize H2C rack-swap nozzle pick (#1780) to JSON for TEXT column
-    # storage; same Text-as-opaque-blob convention as ams_mapping above.
-    if "nozzle_mapping" in update_data:
-        update_data["nozzle_mapping"] = (
-            json.dumps(update_data["nozzle_mapping"]) if update_data["nozzle_mapping"] else None
-        )
+    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),
+        current_user=user,
+        exclude_queue_item_id=item.id,
+    )
 
     for field, value in update_data.items():
         setattr(item, field, value)
@@ -1212,7 +1236,7 @@ async def start_queue_item(
     item_id: int,
     skip_filament_check: bool = Query(default=False),
     db: AsyncSession = Depends(get_db),
-    user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.QUEUE_UPDATE_OWN),
 ):
     """Manually start a staged (manual_start) queue item.
 
@@ -1240,6 +1264,14 @@ async def start_queue_item(
     if item.status != "pending":
         raise HTTPException(400, f"Can only start pending items, current status: '{item.status}'")
 
+    await validate_print_budget(
+        db,
+        cost_center_id=item.cost_center_id,
+        estimated_cost=item.estimated_cost,
+        current_user=current_user,
+        exclude_queue_item_id=item.id,
+    )
+
     # Live deficit check — re-evaluated against current spool state, so a
     # spool swap between scheduler flagging and the user clicking ▶ clears
     # the block automatically.
@@ -1269,8 +1301,8 @@ async def start_queue_item(
     # (#1670). An item that already has a creator (UI-added queue items)
     # keeps that attribution; the dispatcher is not promoted over the
     # original uploader.
-    if user is not None and item.created_by_id is None:
-        item.created_by_id = user.id
+    if current_user is not None and item.created_by_id is None:
+        item.created_by_id = current_user.id
     await db.commit()
     await db.refresh(item, ["archive", "printer", "library_file", "created_by", "batch"])
 

backend/app/api/routes/settings.py

@@ -135,6 +135,8 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) -
             "default_vibration_cali",
             "default_layer_inspect",
             "default_timelapse",
+            "billing_enabled",
+            "printer_kill_switch_enabled",
             "default_nozzle_offset_cali",
             "ldap_enabled",
             "ldap_auto_provision",
@@ -161,6 +163,7 @@ async def _build_settings_response(db: AsyncSession, is_api_key: bool = False) -
             "stagger_group_size",
             "stagger_interval_minutes",
             "forecast_global_lead_time_days",
+            "finance_budget_reset_day",
             "session_max_hours",
         ]:
             settings_dict[setting.key] = int(setting.value)

backend/app/api/routes/users.py

@@ -41,6 +41,7 @@ from backend.app.services.email_service import (
     get_smtp_settings,
     send_email,
 )
+from backend.app.services.finance_defaults import ensure_user_finance_defaults
 
 router = APIRouter(prefix="/users", tags=["users"])
 
@@ -164,6 +165,8 @@ async def create_user(
         new_user.groups = list(groups)
 
     db.add(new_user)
+    await db.flush()
+    await ensure_user_finance_defaults(db, new_user)
     await db.commit()
     await db.refresh(new_user)
 
@@ -307,6 +310,8 @@ async def update_user(
             )
         user.groups = list(groups)
 
+    await ensure_user_finance_defaults(db, user)
+
     await db.commit()
     result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
     user = result.scalar_one()

backend/app/main.py

@@ -30,6 +30,7 @@ from backend.app.api.routes import (
     discovery,
     external_links,
     filaments,
+    finance,
     firmware,
     github_backup,
     groups,
@@ -372,6 +373,9 @@ _expected_prints: dict[tuple[int, str], int] = {}
 # Used by usage tracker to map 3MF slots to physical AMS trays
 _print_ams_mappings: dict[int, list[int]] = {}
 
+# Track cost center selection for the current print run: {archive_id: cost_center_id}
+_print_cost_center_ids: dict[int, int] = {}
+
 # Track plate_id for prints from multi-plate 3MFs: {archive_id: plate_id}
 # Used by usage tracker to scope 3MF parsing to the dispatched plate (#1697).
 # Populated by direct-Print and queue dispatch paths; queue prints also have a
@@ -386,6 +390,9 @@ _last_progress_milestone: dict[int, int] = {}
 # Track whether first layer complete notification has been sent for current print
 _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()
+
 # 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]] = {}
@@ -611,6 +618,54 @@ _expected_print_registered_at: dict[tuple[int, str], float] = {}
 _EXPECTED_PRINT_CLEANUP_INTERVAL: int = 15 * 60  # 15 minutes
 _expected_prints_cleanup_task: asyncio.Task | None = None
 
+_ACTIVE_PRINT_STATES: set[str] = {"RUNNING", "PRINTING", "PAUSE"}
+
+
+def _build_status_print_keys(printer_id: int, state: PrinterState) -> list[tuple[int, str]]:
+    """Build filename keys for matching a printer status update to Bambuddy-owned jobs."""
+
+    possible_keys: list[tuple[int, str]] = []
+    filename = (state.gcode_file or state.current_print or "").strip()
+    subtask_name = (state.subtask_name or "").strip()
+
+    if subtask_name:
+        possible_keys.append((printer_id, subtask_name))
+        possible_keys.append((printer_id, f"{subtask_name}.3mf"))
+        possible_keys.append((printer_id, f"{subtask_name}.gcode.3mf"))
+
+    if filename:
+        base_name = filename.rsplit("/", 1)[-1]
+        if base_name.endswith(".gcode.3mf"):
+            root_name = base_name[: -len(".gcode.3mf")]
+            possible_keys.append((printer_id, root_name))
+            possible_keys.append((printer_id, base_name))
+            possible_keys.append((printer_id, f"{root_name}.gcode"))
+            possible_keys.append((printer_id, f"{root_name}.3mf"))
+        elif base_name.endswith(".3mf"):
+            root_name = base_name[: -len(".3mf")]
+            possible_keys.append((printer_id, root_name))
+            possible_keys.append((printer_id, base_name))
+        elif base_name.endswith(".gcode"):
+            root_name = base_name[: -len(".gcode")]
+            possible_keys.append((printer_id, root_name))
+            possible_keys.append((printer_id, f"{root_name}.3mf"))
+            possible_keys.append((printer_id, base_name))
+        else:
+            possible_keys.append((printer_id, base_name))
+            possible_keys.append((printer_id, f"{base_name}.3mf"))
+
+    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."""
+
+    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)
+
 
 async def _get_plug_energy(plug, db) -> dict | None:
     """Get energy from plug regardless of type (Tasmota, Home Assistant, MQTT, or REST).
@@ -683,6 +738,7 @@ def register_expected_print(
     archive_id: int,
     ams_mapping: list[int] | None = None,
     created_by_id: int | None = None,
+    cost_center_id: int | None = None,
     plate_id: int | None = None,
 ):
     """Register an expected print from reprint/scheduled so we don't create duplicate archives."""
@@ -696,6 +752,8 @@ def register_expected_print(
     # Store AMS mapping for usage tracking at print completion
     if ams_mapping is not None:
         _print_ams_mappings[archive_id] = ams_mapping
+    if cost_center_id is not None:
+        _print_cost_center_ids[archive_id] = cost_center_id
     # Store plate_id for usage tracking when this is a single-plate dispatch from
     # a multi-plate 3MF — without this, the direct-Print path attributes the whole
     # file's filament total to the spool instead of just the printed plate (#1697).
@@ -1184,6 +1242,45 @@ async def on_printer_status_change(printer_id: int, state: PrinterState):
         f"{ams_dry_key}:{ams_tray_key}:{state.door_open}:{state.ams_filament_backup}"
     )
 
+    is_active_print = state.state in _ACTIVE_PRINT_STATES
+    if not is_active_print:
+        _unauthorized_print_kill_sent.discard(printer_id)
+    else:
+        kill_switch_enabled = False
+        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)
+        except Exception as e:
+            status_logger.warning("[KILL SWITCH] Failed to read kill-switch setting for printer %s: %s", printer_id, e)
+
+        if not kill_switch_enabled or _is_bambuddy_authorized_print(printer_id, state):
+            _unauthorized_print_kill_sent.discard(printer_id)
+        elif printer_id in _unauthorized_print_kill_sent:
+            pass
+        else:
+            try:
+                stopped = printer_manager.stop_print(printer_id)
+                if stopped:
+                    _unauthorized_print_kill_sent.add(printer_id)
+                    status_logger.warning(
+                        "[KILL SWITCH] Stopped unauthorized print on printer %s (state=%s)",
+                        printer_id,
+                        state.state,
+                    )
+                else:
+                    status_logger.warning(
+                        "[KILL SWITCH] Could not stop unauthorized print on printer %s (state=%s)",
+                        printer_id,
+                        state.state,
+                    )
+            except Exception as e:
+                status_logger.warning(
+                    "[KILL SWITCH] Failed to stop unauthorized print on printer %s: %s", printer_id, e
+                )
+
     # MQTT relay - publish status (before dedup check - always publish to MQTT)
     try:
         printer_info = printer_manager.get_printer(printer_id)
@@ -4603,6 +4700,29 @@ async def on_print_complete(printer_id: int, data: dict):
 
     log_timing("Archive status update")
 
+    # Apply finance wallet charge or release reservations once
+    try:
+        if data.get("status") in ("completed", "failed", "aborted", "cancelled"):
+            async with async_session() as db:
+                from backend.app.models.archive import PrintArchive
+                from backend.app.services.finance_billing import apply_print_charge_for_archive
+
+                archive = await db.get(PrintArchive, archive_id)
+                cost_center_id = _print_cost_center_ids.pop(archive_id, None)
+                charged = await apply_print_charge_for_archive(
+                    db,
+                    archive_id,
+                    cost_center_id=cost_center_id,
+                    print_run_id=archive.subtask_id if archive else None,
+                )
+                await db.commit()
+                if charged:
+                    logger.info("[FINANCE] Applied print charge for archive %s", archive_id)
+    except Exception as e:
+        logger.warning("[FINANCE] Failed to apply print charge for archive %s: %s", archive_id, e)
+
+    log_timing("Finance charge update")
+
     # Write independent print log entry (separate table, never touches archives)
     try:
         async with async_session() as db:
@@ -5830,6 +5950,7 @@ def _evict_stale_expected_prints() -> None:
     for archive_id in evicted_archive_ids:
         if archive_id not in live_archive_ids:
             _print_ams_mappings.pop(archive_id, None)
+            _print_cost_center_ids.pop(archive_id, None)
             _print_plate_ids.pop(archive_id, None)
 
     logging.getLogger(__name__).info(
@@ -6711,6 +6832,7 @@ app.include_router(groups.router, prefix=app_settings.api_prefix)
 app.include_router(printers.router, prefix=app_settings.api_prefix)
 app.include_router(archives.router, prefix=app_settings.api_prefix)
 app.include_router(filaments.router, prefix=app_settings.api_prefix)
+app.include_router(finance.router, prefix=app_settings.api_prefix)
 app.include_router(inventory.router, prefix=app_settings.api_prefix)
 app.include_router(labels.router, prefix=app_settings.api_prefix)
 app.include_router(settings_routes.router, prefix=app_settings.api_prefix)

backend/app/models/archive.py

@@ -12,6 +12,9 @@ class PrintArchive(Base):
     id: Mapped[int] = mapped_column(primary_key=True)
     printer_id: Mapped[int | None] = mapped_column(ForeignKey("printers.id"), nullable=True)
     project_id: Mapped[int | None] = mapped_column(ForeignKey("projects.id", ondelete="SET NULL"), nullable=True)
+    cost_center_id: Mapped[int | None] = mapped_column(
+        ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True
+    )
 
     # File info
     filename: Mapped[str] = mapped_column(String(255))
@@ -68,6 +71,7 @@ class PrintArchive(Base):
 
     # User additions
     is_favorite: Mapped[bool] = mapped_column(Boolean, default=False)
+    wallet_charge_skipped: Mapped[bool] = mapped_column(Boolean, default=False)
     tags: Mapped[str | None] = mapped_column(Text)
     notes: Mapped[str | None] = mapped_column(Text)
     cost: Mapped[float | None] = mapped_column(Float)
@@ -98,9 +102,11 @@ class PrintArchive(Base):
     # Relationships
     printer: Mapped["Printer | None"] = relationship(back_populates="archives")
     project: Mapped["Project | None"] = relationship(back_populates="archives")
+    cost_center: Mapped["CostCenter | None"] = relationship()
     created_by: Mapped["User | None"] = relationship()
 
 
+from backend.app.models.finance import CostCenter  # noqa: E402, F811
 from backend.app.models.printer import Printer  # noqa: E402, F811
 from backend.app.models.project import Project  # noqa: E402, F811
 from backend.app.models.user import User  # noqa: E402, F811

backend/app/models/finance.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+from enum import Enum as PyEnum
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Boolean, CheckConstraint, DateTime, Float, ForeignKey, String, Text, UniqueConstraint, func
+from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
+
+from backend.app.core.database import Base
+
+if TYPE_CHECKING:
+    from backend.app.models.archive import PrintArchive
+    from backend.app.models.print_queue import PrintQueueItem
+    from backend.app.models.user import User
+
+
+class TransactionType(str, PyEnum):
+    PRINT_CHARGE = "print_charge"
+    DEPOSIT = "deposit"
+    WITHDRAW = "withdraw"
+    MANUAL_ADJUSTMENT = "manual_adjustment"
+
+
+VALID_TRANSACTION_TYPES = {item.value for item in TransactionType}
+
+
+def normalize_transaction_type(value: str | TransactionType) -> str:
+    if isinstance(value, TransactionType):
+        return value.value
+    if value not in VALID_TRANSACTION_TYPES:
+        raise ValueError(f"Invalid transaction type: {value}")
+    return value
+
+
+class UserWallet(Base):
+    """Per-user wallet balance.
+
+    Balance updates are driven by wallet transactions.
+    """
+
+    __tablename__ = "user_wallets"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), unique=True, index=True)
+    balance: Mapped[float] = mapped_column(Float, default=0.0)
+    currency: Mapped[str] = mapped_column(String(3), default="EUR")
+    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
+
+    user: Mapped[User] = relationship()
+
+
+class CostCenter(Base):
+    """Cost center for assigning print costs and budgets."""
+
+    __tablename__ = "cost_centers"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    code: Mapped[str] = mapped_column(String(32), unique=True, index=True, default=lambda: uuid.uuid4().hex[:12])
+    name: Mapped[str] = mapped_column(String(150), index=True)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    is_private: Mapped[bool] = mapped_column(Boolean, default=False)
+    owner_user_id: Mapped[int | None] = mapped_column(ForeignKey("users.id", ondelete="SET NULL"), nullable=True)
+
+    total_budget: Mapped[float | None] = mapped_column(Float, nullable=True)
+    monthly_budget: Mapped[float | None] = mapped_column(Float, nullable=True)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
+
+    owner: Mapped[User | None] = relationship()
+    members: Mapped[list[CostCenterMember]] = relationship(
+        "CostCenterMember",
+        back_populates="cost_center",
+        cascade="all, delete-orphan",
+        lazy="selectin",
+    )
+
+
+class CostCenterMember(Base):
+    """User-to-cost-center assignment with print permission."""
+
+    __tablename__ = "cost_center_members"
+    __table_args__ = (UniqueConstraint("cost_center_id", "user_id", name="uq_cost_center_members_cc_user"),)
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    can_print: Mapped[bool] = mapped_column(Boolean, default=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+
+    cost_center: Mapped[CostCenter] = relationship("CostCenter", back_populates="members")
+    user: Mapped[User] = relationship()
+
+
+class BudgetReservation(Base):
+    """Persisted budget hold for accepted print work that has not been charged yet."""
+
+    __tablename__ = "budget_reservations"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    cost_center_id: Mapped[int] = mapped_column(ForeignKey("cost_centers.id", ondelete="CASCADE"), index=True)
+    amount: Mapped[float] = mapped_column(Float)
+    status: Mapped[str] = mapped_column(String(20), default="active", index=True)
+    source_type: Mapped[str] = mapped_column(String(50), index=True)
+    source_id: Mapped[int | None] = mapped_column(index=True)
+    print_archive_id: Mapped[int | None] = mapped_column(
+        ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+    released_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
+
+    cost_center: Mapped[CostCenter] = relationship()
+    print_archive: Mapped[PrintArchive | None] = relationship()
+
+
+class WalletTransaction(Base):
+    """Immutable wallet ledger entry."""
+
+    __tablename__ = "wallet_transactions"
+    __table_args__ = (
+        CheckConstraint(
+            "transaction_type IN ('print_charge', 'deposit', 'withdraw', 'manual_adjustment')",
+            name="ck_wallet_transactions_transaction_type",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), index=True)
+    cost_center_id: Mapped[int | None] = mapped_column(
+        ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+
+    transaction_type: Mapped[str] = mapped_column(String(40), index=True)
+    amount: Mapped[float] = mapped_column(Float)
+    balance_after: Mapped[float | None] = mapped_column(Float, nullable=True)
+    description: Mapped[str | None] = mapped_column(Text, nullable=True)
+
+    created_by_user_id: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    print_run_id: Mapped[str | None] = mapped_column(String(100), nullable=True, index=True)
+    print_archive_id: Mapped[int | None] = mapped_column(
+        ForeignKey("print_archives.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+    print_queue_id: Mapped[int | None] = mapped_column(
+        ForeignKey("print_queue.id", ondelete="SET NULL"), nullable=True, index=True
+    )
+
+    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), index=True)
+
+    user: Mapped[User] = relationship(foreign_keys=[user_id])
+    cost_center: Mapped[CostCenter | None] = relationship()
+    created_by: Mapped[User | None] = relationship(foreign_keys=[created_by_user_id])
+    print_archive: Mapped[PrintArchive | None] = relationship()
+    print_queue: Mapped[PrintQueueItem | None] = relationship()
+
+    @validates("transaction_type")
+    def _validate_transaction_type(self, key: str, value: str | TransactionType) -> str:
+        return normalize_transaction_type(value)

backend/app/models/print_queue.py

@@ -1,6 +1,6 @@
 from datetime import datetime
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from backend.app.core.database import Base
@@ -32,6 +32,10 @@ class PrintQueueItem(Base):
     library_file_id: Mapped[int | None] = mapped_column(
         ForeignKey("library_files.id", ondelete="CASCADE"), nullable=True
     )
+    cost_center_id: Mapped[int | None] = mapped_column(
+        ForeignKey("cost_centers.id", ondelete="SET NULL"), nullable=True
+    )
+    estimated_cost: Mapped[float | None] = mapped_column(Float, nullable=True)
     project_id: Mapped[int | None] = mapped_column(ForeignKey("projects.id", ondelete="SET NULL"), nullable=True)
     batch_id: Mapped[int | None] = mapped_column(ForeignKey("print_batches.id", ondelete="SET NULL"), nullable=True)
 
@@ -120,12 +124,14 @@ class PrintQueueItem(Base):
     printer: Mapped["Printer"] = relationship()
     archive: Mapped["PrintArchive | None"] = relationship()
     library_file: Mapped["LibraryFile | None"] = relationship()
+    cost_center: Mapped["CostCenter | None"] = relationship()
     project: Mapped["Project | None"] = relationship(back_populates="queue_items")
     batch: Mapped["PrintBatch | None"] = relationship(back_populates="queue_items")
     created_by: Mapped["User | None"] = relationship()
 
 
 from backend.app.models.archive import PrintArchive  # noqa: E402
+from backend.app.models.finance import CostCenter  # noqa: E402
 from backend.app.models.library import LibraryFile  # noqa: E402
 from backend.app.models.print_batch import PrintBatch  # noqa: E402
 from backend.app.models.printer import Printer  # noqa: E402

backend/app/services/archive.py

@@ -1127,6 +1127,7 @@ class ArchiveService:
         created_by_id: int | None = None,
         original_filename: str | None = None,
         project_id: int | None = None,
+        cost_center_id: int | None = None,
         subtask_id: str | None = None,
         prefer_filename_for_name: bool = False,
     ) -> PrintArchive | None:
@@ -1300,6 +1301,7 @@ class ArchiveService:
             extra_data=metadata,
             created_by_id=created_by_id,
             project_id=project_id,
+            cost_center_id=cost_center_id,
             subtask_id=subtask_id,
         )
 

backend/app/services/background_dispatch.py

@@ -22,8 +22,10 @@ from backend.app.core.config import settings
 from backend.app.core.database import async_session
 from backend.app.core.tasks import spawn_background_task
 from backend.app.core.websocket import ws_manager
+from backend.app.models.finance import BudgetReservation, CostCenter
 from backend.app.models.library import LibraryFile
 from backend.app.models.printer import Printer
+from backend.app.models.user import User
 from backend.app.services.archive import ArchiveService
 from backend.app.services.bambu_ftp import (
     cache_3mf_download,
@@ -32,6 +34,7 @@ from backend.app.services.bambu_ftp import (
     upload_file_async,
     with_ftp_retry,
 )
+from backend.app.services.finance_budget import create_budget_reservation, release_budget_reservation
 from backend.app.services.printer_manager import printer_manager
 from backend.app.utils.filename import derive_remote_filename
 
@@ -116,12 +119,16 @@ class BackgroundDispatchService:
             dispatcher = self._dispatcher_task
             self._dispatcher_task = None
             running_tasks = list(self._running_tasks.values())
+            jobs_to_release = [*self._queued_jobs, *(state.job for state in self._active_jobs.values())]
             self._running_tasks.clear()
             self._active_jobs.clear()
             self._queued_jobs.clear()
             self._cancel_requested_job_ids.clear()
             self._job_event.set()
 
+        for job in jobs_to_release:
+            await self._release_budget_reservation(job, status="released")
+
         if dispatcher:
             dispatcher.cancel()
         for task in running_tasks:
@@ -290,8 +297,22 @@ class BackgroundDispatchService:
                 raise DispatchEnqueueRejected(f"Printer {printer_name} is currently busy printing")
 
             dispatch_position = len(self._queued_jobs) + len(self._active_jobs) + 1
+            job_id = self._next_job_id
+            async with async_session() as db:
+                requested_by = await db.get(User, requested_by_user_id) if requested_by_user_id is not None else None
+                await create_budget_reservation(
+                    db,
+                    cost_center_id=options.get("cost_center_id"),
+                    estimated_cost=options.get("estimated_cost"),
+                    current_user=requested_by,
+                    source_type="background_dispatch",
+                    source_id=job_id,
+                    print_archive_id=source_id if kind == "reprint_archive" else None,
+                )
+                await db.commit()
+
             job = PrintDispatchJob(
-                id=self._next_job_id,
+                id=job_id,
                 kind=kind,
                 source_id=source_id,
                 source_name=source_name,
@@ -431,6 +452,9 @@ class BackgroundDispatchService:
         await ws_manager.broadcast({"type": "background_dispatch", "data": payload})
 
     async def _mark_job_finished(self, job: PrintDispatchJob, *, failed: bool, message: str):
+        if failed:
+            await self._release_budget_reservation(job, status="released")
+
         async with self._lock:
             if failed:
                 self._batch_failed += 1
@@ -463,6 +487,8 @@ class BackgroundDispatchService:
                     self._batch_failed = 0
 
     async def _mark_job_cancelled(self, job: PrintDispatchJob):
+        await self._release_budget_reservation(job, status="released")
+
         async with self._lock:
             self._active_jobs.pop(job.id, None)
             self._running_tasks.pop(job.id, None)
@@ -493,6 +519,19 @@ class BackgroundDispatchService:
         if self._is_cancel_requested(job.id):
             raise DispatchJobCancelled(f"Dispatch job {job.id} cancelled")
 
+    @staticmethod
+    async def _release_budget_reservation(job: PrintDispatchJob, *, status: str):
+        if job.options.get("cost_center_id") is None:
+            return
+        async with async_session() as db:
+            await release_budget_reservation(
+                db,
+                source_type="background_dispatch",
+                source_id=job.id,
+                status=status,
+            )
+            await db.commit()
+
     def _build_state_payload_unlocked(self, recent_event: dict[str, Any] | None = None) -> dict[str, Any]:
         processing = len(self._active_jobs)
         dispatched = len(self._queued_jobs)
@@ -565,6 +604,12 @@ class BackgroundDispatchService:
             if not archive:
                 raise RuntimeError("Archive not found")
 
+            cost_center_id = job.options.get("cost_center_id")
+            if cost_center_id is not None:
+                cost_center = await db.scalar(select(CostCenter).where(CostCenter.id == cost_center_id))
+                if not cost_center:
+                    raise RuntimeError("Cost center not found")
+
             printer = await db.scalar(select(Printer).where(Printer.id == job.printer_id))
             if not printer:
                 raise RuntimeError("Printer not found")
@@ -669,6 +714,7 @@ class BackgroundDispatchService:
                     remote_filename,
                     job.source_id,
                     ams_mapping=job.options.get("ams_mapping"),
+                    cost_center_id=job.options.get("cost_center_id"),
                     plate_id=plate_id,
                 )
 
@@ -783,11 +829,21 @@ class BackgroundDispatchService:
                 original_filename=lib_file.filename,
                 project_id=job.project_id,
                 created_by_id=job.requested_by_user_id,
+                cost_center_id=job.options.get("cost_center_id"),
             )
             if not archive:
                 raise RuntimeError("Failed to create archive")
-
-            await db.flush()
+            if job.options.get("cost_center_id") is not None:
+                reservation = await db.scalar(
+                    select(BudgetReservation).where(
+                        BudgetReservation.source_type == "background_dispatch",
+                        BudgetReservation.source_id == job.id,
+                        BudgetReservation.status == "active",
+                    )
+                )
+                if reservation:
+                    reservation.print_archive_id = archive.id
+                    await db.flush()
 
             remote_filename = derive_remote_filename(lib_file.filename)
             remote_path = f"/{remote_filename}"
@@ -876,6 +932,7 @@ class BackgroundDispatchService:
                     remote_filename,
                     archive.id,
                     ams_mapping=job.options.get("ams_mapping"),
+                    cost_center_id=job.options.get("cost_center_id"),
                     plate_id=plate_id,
                 )
 

backend/app/services/finance_billing.py

@@ -0,0 +1,213 @@
+import logging
+
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError, SQLAlchemyError
+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_budget import is_billing_enabled, release_budget_reservation
+
+logger = logging.getLogger(__name__)
+
+
+async def _get_balance_after_for_transaction(
+    db: AsyncSession,
+    user_id: int,
+    cost_center_id: int | None,
+    amount: float,
+) -> float:
+    """Calculate balance_after for a transaction.
+
+    For cost-center transactions: sum of ALL transactions for that cost center (global).
+    For personal transactions (cost_center_id=None): user's wallet balance (personal).
+
+    Args:
+        user_id: The user making the transaction
+        cost_center_id: The cost center (None for personal)
+        amount: The transaction amount (positive/negative)
+
+    Returns:
+        The balance after this transaction would be applied
+    """
+    try:
+        if cost_center_id is None:
+            # Personal transaction: use user wallet balance
+            wallet = (await db.execute(select(UserWallet).where(UserWallet.user_id == user_id))).scalar_one_or_none()
+            if wallet is None:
+                return float(amount)
+            return float(wallet.balance) + amount
+        else:
+            # Cost-center transaction: sum of ALL transactions for this cost center (global, not per-user)
+            result = await db.execute(
+                select(func.coalesce(func.sum(WalletTransaction.amount), 0.0)).where(
+                    WalletTransaction.cost_center_id == cost_center_id,
+                )
+            )
+            current_balance = float(result.scalar() or 0.0)
+            return current_balance + amount
+    except SQLAlchemyError as e:
+        logger.error(f"Database error in _get_balance_after_for_transaction: {e}", exc_info=True)
+        raise
+
+
+def _calculate_partial_charge(
+    archive: PrintArchive,
+    base_cost: float,
+) -> tuple[float, str]:
+    """Calculate proportional charge for partial prints based on filament usage.
+
+    Returns (charge_amount, description_suffix) where:
+    - charge_amount: absolute cost to charge (0 if insufficient data)
+    - description_suffix: reason/details for transaction description
+    """
+    try:
+        # Only apply proportional calculation for non-completed prints
+        if archive.status == "completed":
+            return round(float(base_cost), 2), ""
+
+        filament_used = float(archive.filament_used_grams or 0.0)
+        filament_planned = None
+
+        if archive.extra_data and isinstance(archive.extra_data, dict):
+            filament_planned = archive.extra_data.get("filament_grams_total")
+            if filament_planned is not None:
+                filament_planned = float(filament_planned)
+
+        # If we don't have reliable planned filament data, do not guess a partial charge.
+        # Charging a failed/aborted print without an estimated baseline can overcharge users.
+        if filament_planned is None or filament_planned <= 0:
+            return 0.0, f"[{archive.status}: insufficient filament data]"
+
+        # Calculate proportional cost
+        filament_ratio = min(1.0, max(0.0, filament_used / filament_planned))  # Clamp to [0, 1]
+        charge = float(base_cost) * filament_ratio
+
+        # Round charges to 2 decimals for consistent persistence
+        charge = round(charge, 2)
+
+        suffix = f"[{archive.status}: {filament_ratio:.1%} filament ({filament_used:.1f}g/{filament_planned:.1f}g)]"
+        return charge, suffix
+    except ValueError as e:
+        logger.error(f"Value error in _calculate_partial_charge: {e}", exc_info=True)
+        raise
+
+
+async def apply_print_charge_for_archive(
+    db: AsyncSession,
+    archive_id: int,
+    *,
+    cost_center_id: int | None = None,
+    print_run_id: str | None = None,
+) -> bool:
+    """Apply an idempotent wallet charge for a print archive.
+
+    Charges completed prints at full cost, and partial/failed prints proportionally
+    based on actual filament used vs. planned filament.
+
+    Returns True when a new wallet transaction was created.
+    """
+    try:
+        if not await is_billing_enabled(db):
+            await release_budget_reservation(db, print_archive_id=archive_id, status="released")
+            logger.info("Billing is disabled; skipping print charge for archive ID %s.", archive_id)
+            return False
+
+        archive = (
+            await db.execute(select(PrintArchive).where(PrintArchive.id == archive_id).with_for_update())
+        ).scalar_one_or_none()
+        if archive is None:
+            logger.warning(f"Archive with ID {archive_id} not found.")
+            return False
+
+        if archive.wallet_charge_skipped:
+            logger.info(f"Wallet charge skipped for archive ID {archive_id}.")
+            return False
+
+        # Accept completed, aborted, cancelled, and failed prints
+        if archive.status not in ("completed", "aborted", "cancelled", "failed"):
+            logger.info(f"Archive ID {archive_id} has status {archive.status}, which is not chargeable.")
+            return False
+
+        if archive.created_by_id is None:
+            logger.warning(f"Archive ID {archive_id} has no creator ID.")
+            return False
+
+        base_cost = float(archive.cost or 0.0)
+        if base_cost <= 0:
+            logger.info(f"Base cost for archive ID {archive_id} is zero or negative.")
+            return False
+
+        tx_conditions = [WalletTransaction.transaction_type == TransactionType.PRINT_CHARGE.value]
+        if print_run_id:
+            tx_conditions.append(WalletTransaction.print_run_id == print_run_id)
+        else:
+            tx_conditions.append(WalletTransaction.print_archive_id == archive.id)
+
+        existing_tx = (await db.execute(select(WalletTransaction).where(*tx_conditions))).scalar_one_or_none()
+        if existing_tx is not None:
+            logger.info(f"Transaction already exists for archive ID {archive_id}.")
+            return False
+
+        # Calculate charge (full for completed, partial for others)
+        charge, reason_suffix = _calculate_partial_charge(archive, base_cost)
+        if charge <= 0:
+            await release_budget_reservation(db, print_archive_id=archive.id, status="released")
+            logger.info(f"Calculated charge for archive ID {archive_id} is zero or negative.")
+            return False
+
+        actual_cost_center_id = cost_center_id if cost_center_id is not None else archive.cost_center_id
+
+        wallet = (
+            await db.execute(select(UserWallet).where(UserWallet.user_id == archive.created_by_id))
+        ).scalar_one_or_none()
+        if wallet is None:
+            wallet = UserWallet(user_id=archive.created_by_id, balance=0.0, currency="EUR")
+            db.add(wallet)
+            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 ''}"
+
+        balance_after = await _get_balance_after_for_transaction(
+            db, archive.created_by_id, actual_cost_center_id, -charge
+        )
+        if balance_after is not None:
+            balance_after = round(float(balance_after), 2)
+
+        tx = WalletTransaction(
+            user_id=archive.created_by_id,
+            cost_center_id=actual_cost_center_id,
+            transaction_type=TransactionType.PRINT_CHARGE.value,
+            amount=-charge,
+            balance_after=balance_after,
+            description=description,
+            created_by_user_id=None,
+            print_run_id=print_run_id or archive.subtask_id,
+            print_archive_id=archive.id,
+        )
+        db.add(tx)
+        # Ensure the transaction is flushed to detect unique/index constraint violations
+        try:
+            await db.flush()
+        except IntegrityError as e:
+            # Another concurrent worker likely created the same transaction
+            logger.info("Transaction already exists for archive ID %s (concurrent), skipping: %s", archive_id, e)
+            await db.rollback()
+            return False
+
+        # 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}.")
+        return True
+    except SQLAlchemyError as e:
+        logger.error(f"Database error in apply_print_charge_for_archive: {e}", exc_info=True)
+        return False
+    except ValueError as e:
+        logger.error(f"Value error in apply_print_charge_for_archive: {e}", exc_info=True)
+        return False

backend/app/services/finance_budget.py

@@ -0,0 +1,234 @@
+"""Budget validation helpers for finance-aware print dispatch."""
+
+import calendar
+from datetime import datetime, timezone
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from fastapi import HTTPException
+from sqlalchemy import case, func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.finance import BudgetReservation, CostCenter, CostCenterMember, WalletTransaction
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+
+
+async def is_billing_enabled(db: AsyncSession) -> bool:
+    # Consider any 'billing_enabled' setting with a true-ish value as enabling billing.
+    result = await db.execute(
+        select(func.count())
+        .select_from(Settings)
+        .where(Settings.key == "billing_enabled", func.lower(func.coalesce(Settings.value, "")) == "true")
+    )
+    count = int(result.scalar_one() or 0)
+    return count > 0
+
+
+async def is_printer_kill_switch_enabled(db: AsyncSession) -> bool:
+    """Return True when billing and the printer kill-switch are both enabled."""
+
+    result = await db.execute(
+        select(Settings.key, Settings.value).where(Settings.key.in_(("billing_enabled", "printer_kill_switch_enabled")))
+    )
+    values = {key: (value or "").strip().lower() for key, value in result.all()}
+    return values.get("billing_enabled") == "true" and values.get("printer_kill_switch_enabled") == "true"
+
+
+async def _get_budget_window_start_utc(db: AsyncSession) -> datetime:
+    result = await db.execute(
+        select(Settings).where(Settings.key.in_(["finance_budget_reset_day", "finance_budget_reset_timezone"]))
+    )
+    values = {setting.key: setting.value for setting in result.scalars().all()}
+
+    desired_day = 1
+    try:
+        parsed = int(values.get("finance_budget_reset_day") or 1)
+        if 1 <= parsed <= 31:
+            desired_day = parsed
+    except (TypeError, ValueError):
+        pass
+
+    timezone_name = values.get("finance_budget_reset_timezone") or "UTC"
+    try:
+        tz = ZoneInfo(timezone_name)
+    except ZoneInfoNotFoundError:
+        tz = ZoneInfo("UTC")
+
+    now = datetime.now(tz)
+    current_month_reset_day = min(desired_day, calendar.monthrange(now.year, now.month)[1])
+    if now.day < current_month_reset_day:
+        month = now.month - 1
+        year = now.year
+        if month == 0:
+            month = 12
+            year -= 1
+    else:
+        month = now.month
+        year = now.year
+
+    reset_day = min(desired_day, calendar.monthrange(year, month)[1])
+    return datetime(year, month, reset_day, tzinfo=tz).astimezone(timezone.utc)
+
+
+async def _cost_center_spend(db: AsyncSession, cost_center_id: int, *, monthly: bool) -> float:
+    spend_expr = case((WalletTransaction.amount < 0, -WalletTransaction.amount), else_=0.0)
+    conditions = [
+        WalletTransaction.cost_center_id == cost_center_id,
+        WalletTransaction.cost_center_id.is_not(None),
+    ]
+    if monthly:
+        conditions.append(WalletTransaction.created_at >= await _get_budget_window_start_utc(db))
+
+    result = await db.execute(select(func.coalesce(func.sum(spend_expr), 0.0)).where(*conditions))
+    return float(result.scalar() or 0.0)
+
+
+async def _cost_center_open_queue_reservations(
+    db: AsyncSession,
+    cost_center_id: int,
+    *,
+    exclude_queue_item_id: int | None = None,
+) -> float:
+    conditions = [
+        PrintQueueItem.cost_center_id == cost_center_id,
+        PrintQueueItem.status.in_(("pending", "printing")),
+    ]
+    if exclude_queue_item_id is not None:
+        conditions.append(PrintQueueItem.id != exclude_queue_item_id)
+
+    result = await db.execute(select(func.coalesce(func.sum(PrintQueueItem.estimated_cost), 0.0)).where(*conditions))
+    return float(result.scalar() or 0.0)
+
+
+async def _cost_center_active_budget_reservations(db: AsyncSession, cost_center_id: int) -> float:
+    result = await db.execute(
+        select(func.coalesce(func.sum(BudgetReservation.amount), 0.0)).where(
+            BudgetReservation.cost_center_id == cost_center_id,
+            BudgetReservation.status == "active",
+        )
+    )
+    return float(result.scalar() or 0.0)
+
+
+async def validate_print_budget(
+    db: AsyncSession,
+    *,
+    cost_center_id: int | None,
+    estimated_cost: float | None,
+    current_user: User | None,
+    quantity: int = 1,
+    exclude_queue_item_id: int | None = None,
+) -> None:
+    """Validate that a print can be assigned to a cost center budget."""
+    if not await is_billing_enabled(db):
+        return
+
+    if cost_center_id is None:
+        raise HTTPException(status_code=400, detail="Cost center is required when billing is enabled")
+
+    if estimated_cost is None or estimated_cost <= 0:
+        raise HTTPException(status_code=400, detail="Estimated cost is required for cost center prints")
+
+    center = await db.scalar(select(CostCenter).where(CostCenter.id == cost_center_id).with_for_update())
+    if not center:
+        raise HTTPException(status_code=404, detail="Cost center not found")
+    if not center.is_active:
+        raise HTTPException(status_code=400, detail="Cost center is inactive")
+
+    if current_user is not None and not current_user.is_admin:
+        if center.is_private:
+            if center.owner_user_id != current_user.id:
+                raise HTTPException(status_code=403, detail="You cannot print with this private cost center")
+        else:
+            member = await db.scalar(
+                select(CostCenterMember).where(
+                    CostCenterMember.cost_center_id == cost_center_id,
+                    CostCenterMember.user_id == current_user.id,
+                )
+            )
+            if not member or not member.can_print:
+                raise HTTPException(status_code=403, detail="You cannot print with this cost center")
+
+    budget_limit = center.monthly_budget if center.monthly_budget is not None else center.total_budget
+    if budget_limit is None:
+        return
+
+    used = await _cost_center_spend(db, cost_center_id, monthly=center.monthly_budget is not None)
+    reserved = await _cost_center_open_queue_reservations(
+        db,
+        cost_center_id,
+        exclude_queue_item_id=exclude_queue_item_id,
+    )
+    reserved += await _cost_center_active_budget_reservations(db, cost_center_id)
+    requested = estimated_cost * max(1, quantity)
+    available = float(budget_limit) - used - reserved
+    if requested > available:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Estimated print cost exceeds available cost center budget ({requested:.2f} > {available:.2f})",
+        )
+
+
+async def create_budget_reservation(
+    db: AsyncSession,
+    *,
+    cost_center_id: int | None,
+    estimated_cost: float | None,
+    current_user: User | None,
+    source_type: str,
+    source_id: int | None,
+    print_archive_id: int | None = None,
+) -> BudgetReservation | None:
+    if not await is_billing_enabled(db):
+        return None
+
+    if cost_center_id is None:
+        raise HTTPException(status_code=400, detail="Cost center is required when billing is enabled")
+
+    await validate_print_budget(
+        db,
+        cost_center_id=cost_center_id,
+        estimated_cost=estimated_cost,
+        current_user=current_user,
+    )
+    reservation = BudgetReservation(
+        cost_center_id=cost_center_id,
+        amount=float(estimated_cost or 0.0),
+        status="active",
+        source_type=source_type,
+        source_id=source_id,
+        print_archive_id=print_archive_id,
+    )
+    db.add(reservation)
+    await db.flush()
+    return reservation
+
+
+async def release_budget_reservation(
+    db: AsyncSession,
+    *,
+    source_type: str | None = None,
+    source_id: int | None = None,
+    print_archive_id: int | None = None,
+    status: str = "released",
+) -> int:
+    conditions = [BudgetReservation.status == "active"]
+    if print_archive_id is not None:
+        conditions.append(BudgetReservation.print_archive_id == print_archive_id)
+    else:
+        conditions.extend(
+            [
+                BudgetReservation.source_type == source_type,
+                BudgetReservation.source_id == source_id,
+            ]
+        )
+
+    result = await db.execute(select(BudgetReservation).where(*conditions))
+    reservations = result.scalars().all()
+    for reservation in reservations:
+        reservation.status = status
+        reservation.released_at = datetime.now(timezone.utc)
+    if reservations:
+        await db.flush()
+    return len(reservations)

backend/app/services/finance_defaults.py

@@ -0,0 +1,68 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.finance import CostCenter, CostCenterMember, UserWallet
+from backend.app.models.settings import Settings as AppSettingModel
+from backend.app.models.user import User
+from backend.app.schemas.settings import AppSettings as AppSettingsSchema
+
+
+async def ensure_user_finance_defaults(db: AsyncSession, user: User) -> bool:
+    """Ensure wallet and private cost center defaults exist for a user.
+
+    Returns True when database objects were created or changed.
+    """
+    changed = False
+
+    wallet = (await db.execute(select(UserWallet).where(UserWallet.user_id == user.id))).scalar_one_or_none()
+    if wallet is None:
+        # Respect admin-configured currency if present, otherwise fall back to app default
+        default_currency = AppSettingsSchema().currency
+        result = await db.execute(select(AppSettingModel).where(AppSettingModel.key == "currency"))
+        setting = result.scalar_one_or_none()
+        currency = setting.value if setting and setting.value else default_currency
+        db.add(UserWallet(user_id=user.id, balance=0.0, currency=currency))
+        changed = True
+
+    private_center = (
+        (
+            await db.execute(
+                select(CostCenter)
+                .where(
+                    CostCenter.is_private.is_(True),
+                    CostCenter.owner_user_id == user.id,
+                )
+                .order_by(CostCenter.id.asc())
+            )
+        )
+        .scalars()
+        .first()
+    )
+
+    if private_center is None:
+        private_center = CostCenter(
+            name=user.username,
+            is_active=True,
+            is_private=True,
+            owner_user_id=user.id,
+        )
+        db.add(private_center)
+        await db.flush()
+        changed = True
+    elif private_center.name != user.username:
+        private_center.name = user.username
+        changed = True
+
+    membership = (
+        await db.execute(
+            select(CostCenterMember).where(
+                CostCenterMember.cost_center_id == private_center.id,
+                CostCenterMember.user_id == user.id,
+            )
+        )
+    ).scalar_one_or_none()
+    if membership is None:
+        db.add(CostCenterMember(cost_center_id=private_center.id, user_id=user.id, can_print=True))
+        changed = True
+
+    return changed

backend/app/services/print_scheduler.py

@@ -7,6 +7,7 @@ import time
 from datetime import datetime, timezone
 from pathlib import Path
 
+from fastapi import HTTPException
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
@@ -15,6 +16,7 @@ from backend.app.core.config import settings
 from backend.app.core.database import async_session, run_with_retry
 from backend.app.core.tasks import spawn_background_task
 from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, CostCenterMember
 from backend.app.models.library import LibraryFile
 from backend.app.models.print_queue import PrintQueueItem
 from backend.app.models.printer import Printer
@@ -22,6 +24,7 @@ from backend.app.models.settings import Settings
 from backend.app.models.smart_plug import SmartPlug
 from backend.app.models.spool_assignment import SpoolAssignment
 from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
+from backend.app.models.user import User
 from backend.app.services.bambu_ftp import (
     cache_3mf_download,
     delete_file_async,
@@ -30,6 +33,7 @@ from backend.app.services.bambu_ftp import (
     with_ftp_retry,
 )
 from backend.app.services.filament_deficit import compute_deficit_for_queue_item
+from backend.app.services.finance_budget import validate_print_budget
 from backend.app.services.notification_service import notification_service
 from backend.app.services.printer_manager import printer_manager, supports_drying
 from backend.app.services.smart_plug_manager import smart_plug_manager
@@ -2054,6 +2058,44 @@ class PrintScheduler:
         """
         logger.info("Starting queue item %s", item.id)
 
+        try:
+            queue_user = await db.get(User, item.created_by_id) if item.created_by_id is not None else None
+            if queue_user is not None and item.cost_center_id is not None and not queue_user.is_admin:
+                center = await db.scalar(
+                    select(CostCenter).where(CostCenter.id == item.cost_center_id).with_for_update()
+                )
+                if not center:
+                    raise HTTPException(status_code=404, detail="Cost center not found")
+                if not center.is_active:
+                    raise HTTPException(status_code=400, detail="Cost center is inactive")
+                if center.is_private:
+                    if center.owner_user_id != queue_user.id:
+                        raise HTTPException(status_code=403, detail="You cannot print with this private cost center")
+                else:
+                    member = await db.scalar(
+                        select(CostCenterMember).where(
+                            CostCenterMember.cost_center_id == item.cost_center_id,
+                            CostCenterMember.user_id == queue_user.id,
+                        )
+                    )
+                    if not member or not member.can_print:
+                        raise HTTPException(status_code=403, detail="You cannot print with this cost center")
+            await validate_print_budget(
+                db,
+                cost_center_id=item.cost_center_id,
+                estimated_cost=item.estimated_cost,
+                current_user=queue_user,
+                exclude_queue_item_id=item.id,
+            )
+        except HTTPException as exc:
+            item.status = "failed"
+            item.error_message = getattr(exc, "detail", str(exc))
+            item.completed_at = datetime.now(timezone.utc)
+            await db.commit()
+            logger.error("Queue item %s: Budget check failed: %s", item.id, item.error_message)
+            await self._power_off_if_needed(db, item)
+            return
+
         # Get printer first (needed for both paths)
         result = await db.execute(select(Printer).where(Printer.id == item.printer_id))
         printer = result.scalar_one_or_none()
@@ -2126,6 +2168,7 @@ class PrintScheduler:
                     original_filename=filename,
                     created_by_id=item.created_by_id,
                     project_id=item.project_id,
+                    cost_center_id=item.cost_center_id,
                 )
                 if archive:
                     item.archive_id = archive.id
@@ -2288,6 +2331,7 @@ class PrintScheduler:
                 archive.id,
                 ams_mapping=ams_mapping,
                 created_by_id=item.created_by_id,
+                cost_center_id=item.cost_center_id,
                 plate_id=item.plate_id,
             )