fix(queue): every way of adding a job announces it, applies the default print options and keeps its cost centre

- File Manager, webhook, virtual printer, pipeline runs and order dispatch
  send the "Job added" notification (one summary per printer) and the MQTT
  queue/job_added event (one per job), after the commit, never failing the add.
- File Manager, webhook and pipeline jobs use the default print options; the
  VP reads them from the same helper.
- Order copies keep their cost centre; with billing on, dispatch is refused
  when the dispatcher may not use it, the copies have none, or its budget
  cannot cover every copy.
- A scheduled_time with a UTC offset is converted to UTC (queue create/edit/
  bulk edit, webhook); the Postgres bind hook converts instead of stripping.

.basedpyright/baseline.json

@@ -1083,14 +1083,6 @@
             }
         ],
         "./backend/app/api/routes/print_queue.py": [
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 21,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -1099,14 +1091,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 29,
-                    "endColumn": 34,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -1115,14 +1099,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 21,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportOptionalMemberAccess",
                 "range": {
@@ -1146,14 +1122,6 @@
                     "endColumn": 76,
                     "lineCount": 1
                 }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 21,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
             }
         ],
         "./backend/app/api/routes/printer_locations.py": [

CHANGELOG.md

@@ -164,6 +164,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **Every way of adding a queue job now works like the print dialog** — Jobs added from File Manager's **Add to queue**, the webhook, a virtual printer, a pipeline run or an order's **Dispatch remaining** sent no "Job added" notification and no MQTT `queue/job_added` event; now every path sends them. Several jobs added at once are one notification per printer, such as "3 jobs (bracket, hinge, lid)", and MQTT gets one event per job. File Manager, the webhook and pipeline runs ignored the default print options (timelapse, bed levelling and the rest) and now use them. The copies an order dispatches lost their cost centre, so they were billed to none and never checked against its budget; they now keep it, and with billing on, dispatching is refused when you may not print with that cost centre or its budget cannot cover every copy. A `scheduled_time` sent to the API or webhook with a UTC offset, such as `10:00+02:00`, was stored as 10:00 UTC and the job started two hours late; the offset is now converted.
 - **Long prints keep their layer timelapse** — Stitching a layer timelapse was stopped after 5 minutes and its frames were then deleted, so a print with more than a few hundred layers on a Raspberry Pi lost its timelapse for good. Stitching now gets half a second per layer, from 5 minutes up to an hour. If ffmpeg fails anyway, or Bambuddy stops while it runs, the frames are kept for about 24 hours with an ffmpeg command to make the video by hand; when ffmpeg fails, the log names the folder. Without ffmpeg the frames are still deleted.
 - **An in-app update that hangs no longer blocks further updates** — A stalled `git fetch`, `pip install` or `npm install` left the update showing "downloading" or "installing" forever, and every new attempt was refused with "Update already in progress" until Bambuddy was restarted. Each step now has a time limit (5 minutes for git, 15 for pip and npm); a git or pip step that runs out of time ends the update with an error, and checking for updates again lets you retry. When pip or `git reset` is the one that ran out of time the new code may already be in place, so run the update again before restarting. A frontend build that runs out of time keeps the files shipped with the release.
 - **The plate picker downloads a printer's 3MF once** — Opening a 3MF in the printer's file browser downloaded the whole file again for the plate list and for every plate thumbnail, all at once: a ten-plate 60 MB file meant eleven full downloads, slow on any printer and slower still on an A1. The plate list, plate thumbnails and G-code preview now share one download and keep it for a minute (the 3D view still fetches its own copy).

backend/app/api/routes/library.py

@@ -98,7 +98,9 @@ from backend.app.services.library_folder_access import (
 from backend.app.services.pdf_thumbnail import generate_pdf_thumbnail
 from backend.app.services.plate_thumbnail import inject_plate_thumbnails_if_missing
 from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
+from backend.app.services.print_defaults import default_print_options
 from backend.app.services.process_overrides import apply_process_overrides
+from backend.app.services.queue_announcements import announce_queue_jobs_added
 from backend.app.services.slice_output_check import (
     missing_start_gcode_message,
     start_gcode_is_missing,
@@ -3244,6 +3246,9 @@ async def add_files_to_queue(
     # There is no per-job ask-for-outcome toggle on a bulk add, so the rows take
     # the same default the print dialog seeds itself from (#1898).
     confirm_outcome = await confirm_outcome_for_new_queue_item(db)
+    # No print options on a bulk add either: the install's defaults, as the
+    # print dialog would have offered them.
+    print_options = await default_print_options(db)
 
     for file_id in request.file_ids:
         lib_file = files.get(file_id)
@@ -3345,6 +3350,7 @@ async def add_files_to_queue(
                 created_by_id=actor.id if actor else None,
                 # Waits for someone to start it unless they may print without review (#1620)
                 manual_start=review_required,
+                **print_options,
             )
             db.add(queue_item)
 
@@ -3377,6 +3383,7 @@ async def add_files_to_queue(
         )
 
     await db.commit()
+    await announce_queue_jobs_added(db, [a.queue_item_id for a in added])
 
     return AddToQueueResponse(added=added, errors=errors)
 

backend/app/api/routes/pipeline_runs.py

@@ -70,6 +70,8 @@ from backend.app.services.pipeline_eligibility import (
     check_pipeline_eligibility,
 )
 from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
+from backend.app.services.print_defaults import default_print_options
+from backend.app.services.queue_announcements import announce_queue_jobs_added
 
 logger = logging.getLogger(__name__)
 
@@ -592,6 +594,8 @@ def _make_orchestration_callable(
 
             # No per-job ask-for-outcome toggle on a pipeline run either (#1898).
             confirm_outcome = await confirm_outcome_for_new_queue_item(session)
+            # Nor print options: the install's defaults.
+            print_options = await default_print_options(session)
 
             for job, (printer_id, target_model) in zip(jobs, assignments, strict=False):
                 queue_item = PrintQueueItem(
@@ -615,6 +619,7 @@ def _make_orchestration_callable(
                         and printer_id is None
                         and not printer_scope.is_unrestricted
                     ),
+                    **print_options,
                 )
                 session.add(queue_item)
                 await session.flush()
@@ -658,6 +663,7 @@ def _make_orchestration_callable(
                 job.status = "queued"
             run.status = "dispatching"
             await session.commit()
+            await announce_queue_jobs_added(session, [job.queue_entry_id for job in jobs if job.queue_entry_id])
             await _publish_run_event(session, run)
 
             return slice_response.model_dump()

backend/app/api/routes/print_queue.py

@@ -55,7 +55,6 @@ from backend.app.schemas.print_queue import (
 from backend.app.services.filament_deficit import compute_deficit_for_queue_item
 from backend.app.services.filament_requirements import overrides_for_plate
 from backend.app.services.finance_budget import release_budget_reservation, validate_print_budget
-from backend.app.services.notification_service import notification_service
 from backend.app.services.print_batch import (
     BatchDispatchError,
     dispatch_remaining,
@@ -65,6 +64,7 @@ from backend.app.services.print_batch import (
     refresh_batches_after,
 )
 from backend.app.services.print_cost_estimate import estimate_queue_source_cost
+from backend.app.services.queue_announcements import announce_queue_jobs_added
 from backend.app.services.queue_item_printer import PRINTER_RESOLVED_DEFAULTS, drop_printer_resolved
 from backend.app.services.queue_position import lock_queue_positions, max_queue_position
 from backend.app.utils.printer_models import (
@@ -1258,43 +1258,7 @@ async def add_to_queue(
     qty_desc = f" (×{quantity})" if quantity > 1 else ""
     logger.info("Added %s to queue for %s%s", source_name, target_desc, qty_desc)
 
-    # MQTT relay - publish queue job added
-    try:
-        from backend.app.services.mqtt_relay import mqtt_relay
-
-        await mqtt_relay.on_queue_job_added(
-            job_id=item.id,
-            filename=item.archive.filename if item.archive else "",
-            printer_id=item.printer_id,
-            printer_name=item.printer.name if item.printer else None,
-        )
-    except Exception:
-        pass  # Don't fail queue add if MQTT fails
-
-    # Send notification for job added
-    try:
-        job_name = (
-            item.archive.filename
-            if item.archive
-            else item.library_file.filename
-            if item.library_file
-            else f"Job #{item.id}"
-        )
-        job_name = job_name.replace(".gcode.3mf", "").replace(".3mf", "")
-        if quantity > 1:
-            job_name = f"{job_name} ×{quantity}"
-        target = (
-            item.printer.name if item.printer else (f"Any {item.target_model}" if target_model_norm else "Unassigned")
-        )
-        await notification_service.on_queue_job_added(
-            job_name=job_name,
-            target=target,
-            db=db,
-            printer_id=item.printer_id,
-            printer_name=item.printer.name if item.printer else None,
-        )
-    except Exception:
-        pass  # Don't fail queue add if notification fails
+    await announce_queue_jobs_added(db, [i.id for i in items])
 
     return _enrich_response(item)
 
@@ -1690,6 +1654,35 @@ async def update_batch(
     return await _build_batch_response(db, batch)
 
 
+async def _validate_copies_budget(
+    db: AsyncSession, copies: list[PrintQueueItem], actor: User | ApiKeyActor | None
+) -> None:
+    """An order's copies are billed to their template's cost centre: the
+    dispatcher must be allowed to print with it, and its budget must hold
+    every copy, as POST /queue/ checks for a quantity. Only with billing on;
+    then a copy without a cost centre is refused as POST /queue/ refuses it."""
+    by_center: dict[int | None, list[PrintQueueItem]] = {}
+    for copy in copies:
+        by_center.setdefault(copy.cost_center_id, []).append(copy)
+    for cost_center_id, center_copies in by_center.items():
+        estimates = [
+            await _trusted_item_estimated_cost(
+                db, copy, printer_id=copy.printer_id, plate_id=copy.plate_id, ams_mapping=copy.ams_mapping
+            )
+            for copy in center_copies
+        ]
+        await validate_print_budget(
+            db,
+            cost_center_id=cost_center_id,
+            estimated_cost=None if any(e is None for e in estimates) else sum(e for e in estimates if e is not None),
+            current_user=actor,
+        )
+        # Set after the check: a pending copy's estimate counts as held, and
+        # the copies must not be counted against themselves.
+        for copy, estimate in zip(center_copies, estimates, strict=True):
+            copy.estimated_cost = estimate
+
+
 @router.post("/batches/{batch_id}/dispatch", response_model=PrintBatchResponse)
 async def dispatch_batch(
     batch_id: int,
@@ -1733,6 +1726,7 @@ async def dispatch_batch(
             printer_scope.ensure(item.printer_id)
             if item.printer_id is None:
                 ensure_model_target_allowed(current_user, printer_scope)
+        await _validate_copies_budget(db, created, actor)
     except HTTPException:
         await db.rollback()
         raise
@@ -1743,6 +1737,7 @@ async def dispatch_batch(
             item.manual_start = True
 
     await db.commit()
+    await announce_queue_jobs_added(db, [item.id for item in created])
     await db.refresh(batch)
 
     logger.info("Batch %s dispatched %d item(s)", batch.id, len(created))

backend/app/api/routes/webhook.py

@@ -24,8 +24,11 @@ from backend.app.models.printer import Printer
 from backend.app.schemas.printer import HMSErrorResponse, hms_error_responses
 from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
 from backend.app.services.print_control import stop_print_by_user
+from backend.app.services.print_defaults import default_print_options
 from backend.app.services.printer_manager import printer_manager
+from backend.app.services.queue_announcements import announce_queue_jobs_added
 from backend.app.services.queue_position import next_queue_position
+from backend.app.utils.local_time import to_naive_utc
 
 logger = logging.getLogger(__name__)
 
@@ -140,8 +143,9 @@ async def webhook_add_to_queue(
         from datetime import datetime
 
         try:
-            scheduled_time = datetime.fromisoformat(data.scheduled_time.replace("Z", "+00:00"))
-        except ValueError:
+            # Stored as naive UTC: an offset is converted, not dropped.
+            scheduled_time = to_naive_utc(datetime.fromisoformat(data.scheduled_time.replace("Z", "+00:00")))
+        except (ValueError, OverflowError):
             raise HTTPException(status_code=400, detail="Invalid scheduled_time format")
 
     # Create queue item
@@ -162,10 +166,13 @@ async def webhook_add_to_queue(
         created_by_id=api_key.user_id,
         # Waits for someone to start it unless the owner may print without review (#1620)
         manual_start=await is_auth_enabled(db) and queue_review_required_for(await resolve_apikey_owner(db, api_key)),
+        # No print options in the request: the install's defaults.
+        **await default_print_options(db),
     )
     db.add(queue_item)
-    await db.flush()
+    await db.commit()
     await db.refresh(queue_item)
+    await announce_queue_jobs_added(db, [queue_item.id])
 
     return QueueAddResponse(
         id=queue_item.id,

backend/app/core/database.py

@@ -162,10 +162,11 @@ def _create_engine(url: str | None = None):
     if is_sqlite():
         event.listen(eng.sync_engine, "connect", _set_sqlite_pragmas)
     else:
-        # Strip timezone info from aware datetimes before they reach asyncpg.
+        # Make aware datetimes naive UTC before they reach asyncpg.
         # asyncpg rejects timezone-aware values for TIMESTAMP WITHOUT TIME ZONE columns.
-        # The codebase uses datetime.now(timezone.utc) in many places — this makes
-        # Postgres behave like SQLite which ignores timezone info entirely.
+        # The codebase uses datetime.now(timezone.utc) in many places. A value
+        # with another offset is converted, not stripped; SQLite drops offsets,
+        # so request schemas convert on the way in (NaiveUTCDatetime).
         @event.listens_for(eng.sync_engine, "before_cursor_execute", retval=True)
         def _strip_tz_from_params(conn, cursor, statement, parameters, context, executemany):
             import datetime
@@ -181,7 +182,8 @@ def _create_engine(url: str | None = None):
             # correct answer is "strip datetimes at any depth".
             def _strip(val):
                 if isinstance(val, datetime.datetime) and val.tzinfo is not None:
-                    return val.replace(tzinfo=None)
+                    # Naive columns hold UTC: convert, so an offset is not lost.
+                    return val.astimezone(datetime.timezone.utc).replace(tzinfo=None)
                 if isinstance(val, dict):
                     return {k: _strip(v) for k, v in val.items()}
                 if isinstance(val, list):

backend/app/schemas/print_queue.py

@@ -1,9 +1,10 @@
 from datetime import datetime
 from typing import Annotated, Literal
 
-from pydantic import BaseModel, BeforeValidator, Field, PlainSerializer, model_validator
+from pydantic import AfterValidator, BaseModel, BeforeValidator, Field, PlainSerializer, model_validator
 
 from backend.app.schemas.not_null import IgnoresNullForRequiredColumns
+from backend.app.utils.local_time import to_naive_utc
 from backend.app.utils.printer_models import MAX_CHAMBER_TEMP_C
 
 
@@ -18,6 +19,19 @@ def serialize_utc_datetime(dt: datetime | None) -> str | None:
 UTCDatetime = Annotated[datetime | None, PlainSerializer(serialize_utc_datetime)]
 
 
+def _naive_utc(value: datetime | None) -> datetime | None:
+    try:
+        return to_naive_utc(value)
+    except OverflowError:
+        # At the edge of the date range: a 422, not a 500.
+        raise ValueError("date is out of range") from None
+
+
+# Coerces any client-sent UTC offset to the naive UTC the DB stores. Without
+# it the offset was dropped, not converted: 10:00+02:00 was stored as 10:00 UTC.
+NaiveUTCDatetime = Annotated[datetime | None, AfterValidator(_naive_utc)]
+
+
 def _coerce_tristate(v: object) -> object:
     """Map legacy on/off booleans onto the tri-state calibration options.
 
@@ -77,7 +91,7 @@ class PrintQueueItemCreate(BaseModel):
     # Either archive_id OR library_file_id must be provided
     archive_id: int | None = None
     library_file_id: int | None = None
-    scheduled_time: datetime | None = None  # None = ASAP (next when idle)
+    scheduled_time: NaiveUTCDatetime = None  # None = ASAP (next when idle)
     require_previous_success: bool = False
     auto_off_after: bool = False  # Power off printer after print completes
     manual_start: bool = False  # Requires manual trigger to start (staged)
@@ -145,7 +159,7 @@ class PrintQueueItemUpdate(IgnoresNullForRequiredColumns):
     target_location: str | None = None  # Target location filter (only used with target_model)
     filament_overrides: list[dict] | None = None  # Filament overrides for model-based assignment
     position: int | None = None
-    scheduled_time: datetime | None = None
+    scheduled_time: NaiveUTCDatetime = None
     require_previous_success: bool | None = None
     auto_off_after: bool | None = None
     manual_start: bool | None = None
@@ -338,7 +352,7 @@ class PrintQueueBulkUpdate(IgnoresNullForRequiredColumns):
     item_ids: list[int]
     # Fields to update (all optional - only set fields are applied)
     printer_id: int | None = None
-    scheduled_time: datetime | None = None
+    scheduled_time: NaiveUTCDatetime = None
     require_previous_success: bool | None = None
     auto_off_after: bool | None = None
     manual_start: bool | None = None

backend/app/schemas/scheduled_drying.py

@@ -1,13 +1,6 @@
-from datetime import datetime
-from typing import Annotated
+from pydantic import BaseModel, Field
 
-from pydantic import AfterValidator, BaseModel, Field
-
-from backend.app.schemas.print_queue import UTCDatetime
-from backend.app.utils.local_time import to_naive_utc
-
-# Coerces any client-sent UTC offset to the naive UTC the DB stores.
-NaiveUTCDatetime = Annotated[datetime | None, AfterValidator(to_naive_utc)]
+from backend.app.schemas.print_queue import NaiveUTCDatetime, UTCDatetime
 
 
 class ScheduledDryingCreate(BaseModel):

backend/app/services/finance_budget.py

@@ -2,6 +2,7 @@
 
 import calendar
 from datetime import datetime, timezone
+from typing import TYPE_CHECKING
 from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
 
 from fastapi import HTTPException
@@ -13,6 +14,9 @@ from backend.app.models.print_queue import PrintQueueItem
 from backend.app.models.settings import Settings
 from backend.app.models.user import User
 
+if TYPE_CHECKING:
+    from backend.app.core.auth import ApiKeyActor
+
 
 async def is_billing_enabled(db: AsyncSession) -> bool:
     # Consider any 'billing_enabled' setting with a true-ish value as enabling billing.
@@ -154,7 +158,7 @@ async def validate_print_budget(
     *,
     cost_center_id: int | None,
     estimated_cost: float | None,
-    current_user: User | None,
+    current_user: "User | ApiKeyActor | None",
     quantity: int = 1,
     exclude_queue_item_id: int | None = None,
     exclude_reservation_source_type: str | None = None,

backend/app/services/print_batch.py

@@ -78,6 +78,9 @@ CLONED_SETTING_COLUMNS = (
     "preheat_override",
     "preheat_chamber_target_override",
     "skip_filament_check",
+    # The scheduler bills and budget-checks from it; dispatch_batch checks the
+    # dispatcher may use it.
+    "cost_center_id",
 )
 
 CLONED_VARIANT_COLUMNS = (

backend/app/services/print_defaults.py

@@ -0,0 +1,52 @@
+"""The install's default print options for a queue job nobody set them on.
+
+The print dialog seeds its toggles from these settings. File Manager's "Add to
+queue", the webhook and a pipeline run have no toggles, so they take the
+defaults from here; the virtual printer takes them where the slicer sent
+nothing. Fallbacks match AppSettings in schemas/settings.py.
+"""
+
+from typing import TypedDict
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+
+class DefaultPrintOptions(TypedDict):
+    bed_levelling: str
+    flow_cali: str
+    nozzle_offset_cali: str
+    vibration_cali: bool
+    layer_inspect: bool
+    timelapse: bool
+
+
+def _bool_setting(value: str | None, default: bool) -> bool:
+    return value.lower() == "true" if value is not None else default
+
+
+def _tristate_setting(value: str | None, default: str) -> str:
+    """Tri-state default (off/on/auto), coercing legacy true/false rows."""
+    if value is None:
+        return default
+    low = value.strip().lower()
+    if low in ("on", "off", "auto"):
+        return low
+    if low in ("true", "1"):
+        return "on"
+    if low in ("false", "0"):
+        return "off"
+    return default
+
+
+async def default_print_options(db: AsyncSession) -> DefaultPrintOptions:
+    """The default print options, keyed by PrintQueueItem column."""
+    from backend.app.api.routes.settings import get_setting
+
+    return {
+        "bed_levelling": _tristate_setting(await get_setting(db, "default_bed_levelling"), "auto"),
+        "flow_cali": _tristate_setting(await get_setting(db, "default_flow_cali"), "auto"),
+        "nozzle_offset_cali": _tristate_setting(await get_setting(db, "default_nozzle_offset_cali"), "auto"),
+        "vibration_cali": _bool_setting(await get_setting(db, "default_vibration_cali"), True),
+        "layer_inspect": _bool_setting(await get_setting(db, "default_layer_inspect"), False),
+        "timelapse": _bool_setting(await get_setting(db, "default_timelapse"), False),
+    }

backend/app/services/queue_announcements.py

@@ -0,0 +1,120 @@
+"""Telling the outside world that jobs were added to the queue.
+
+Every path that creates queue jobs calls this once, after its commit: the print
+dialog, File Manager's "Add to queue", the webhook, a virtual printer, a
+pipeline run and an order's "dispatch remaining". The "job added" notification
+is one summary per printer target, so a bulk add is one message, not one per
+job, and a provider scoped to one printer hears only of that printer's jobs.
+The MQTT relay is a machine event and gets one per job.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Sequence
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.models.archive import PrintArchive
+from backend.app.models.library import LibraryFile
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+
+logger = logging.getLogger(__name__)
+
+# Names listed in a summary of different files; the rest is "…".
+_NAMES_LISTED = 3
+
+
+def _job_name(filename: str | None, item_id: int) -> str:
+    if not filename:
+        return f"Job #{item_id}"
+    return filename.replace(".gcode.3mf", "").replace(".3mf", "")
+
+
+def _summary(names: list[str]) -> str:
+    distinct = list(dict.fromkeys(names))
+    if len(distinct) == 1:
+        return f"{distinct[0]} ×{len(names)}" if len(names) > 1 else distinct[0]
+    listed = ", ".join(distinct[:_NAMES_LISTED]) + (", …" if len(distinct) > _NAMES_LISTED else "")
+    return f"{len(names)} jobs ({listed})"
+
+
+def _target(printer_name: str | None, models: list[str]) -> str:
+    if printer_name:
+        return printer_name
+    distinct = list(dict.fromkeys(models))
+    return f"Any {' / '.join(distinct)}" if distinct else "Unassigned"
+
+
+async def announce_queue_jobs_added(db: AsyncSession, item_ids: Sequence[int]) -> None:
+    """Send the "job added" notification and relay events for new queue jobs.
+
+    Never raises: a queue add must not fail because a notification did.
+    """
+    if not item_ids:
+        return
+    try:
+        await _announce(db, item_ids)
+    except Exception as e:
+        logger.warning("Could not announce new queue jobs: %s", e)
+
+
+async def _announce(db: AsyncSession, item_ids: Sequence[int]) -> None:
+    try:
+        rows = (
+            await db.execute(
+                select(
+                    PrintQueueItem.id,
+                    PrintQueueItem.printer_id,
+                    PrintQueueItem.target_model,
+                    Printer.name.label("printer_name"),
+                    PrintArchive.filename.label("archive_filename"),
+                    LibraryFile.filename.label("library_filename"),
+                )
+                .outerjoin(Printer, Printer.id == PrintQueueItem.printer_id)
+                .outerjoin(PrintArchive, PrintArchive.id == PrintQueueItem.archive_id)
+                .outerjoin(LibraryFile, LibraryFile.id == PrintQueueItem.library_file_id)
+                .where(PrintQueueItem.id.in_(list(item_ids)))
+                .order_by(PrintQueueItem.position, PrintQueueItem.id)
+            )
+        ).all()
+    except Exception as e:
+        logger.warning("Could not announce new queue jobs: %s", e)
+        return
+
+    from backend.app.services.mqtt_relay import mqtt_relay
+    from backend.app.services.notification_service import notification_service
+
+    groups: dict[int | None, list] = {}
+    for row in rows:
+        groups.setdefault(row.printer_id, []).append(row)
+        try:
+            await mqtt_relay.on_queue_job_added(
+                job_id=row.id,
+                filename=row.archive_filename or row.library_filename or "",
+                printer_id=row.printer_id,
+                printer_name=row.printer_name,
+            )
+        except Exception:
+            pass  # Don't fail queue add if MQTT fails
+
+    for printer_id, group in groups.items():
+        printer_name = group[0].printer_name
+        try:
+            await notification_service.on_queue_job_added(
+                job_name=_summary([_job_name(r.archive_filename or r.library_filename, r.id) for r in group]),
+                target=_target(printer_name, [r.target_model for r in group if r.target_model]),
+                db=db,
+                printer_id=printer_id,
+                printer_name=printer_name,
+            )
+        except Exception as e:
+            logger.warning("Queue job added notification failed: %s", e)
+            # Its log write commits on this session: a failed one must not leave
+            # the request's session unusable after the job was committed.
+            try:
+                await db.rollback()
+            except Exception:
+                pass

backend/app/services/virtual_printer/manager.py

@@ -918,32 +918,18 @@ class VirtualPrinterInstance:
             from backend.app.services.archive import ArchiveService
             from backend.app.services.filament_requirements import extract_filament_requirements
             from backend.app.services.print_confirmation import confirm_outcome_for_new_queue_item
+            from backend.app.services.print_defaults import default_print_options
 
             async with self._session_factory() as db:
                 name_source = await get_setting(db, "virtual_printer_archive_name_source")
                 prefer_filename = name_source == "filename"
 
-                # Read workflow defaults from settings. Without this the
+                # Workflow defaults from settings. Without this the
                 # PrintQueueItem below would fall back to the column-level
                 # defaults and ignore the user's workflow preferences (#1235).
-                # Fallbacks match AppSettings defaults in schemas/settings.py.
                 # The slicer-side options captured above (if any) take
                 # precedence per-field over these defaults.
-                def _bool_setting(value: str | None, default: bool) -> bool:
-                    return value.lower() == "true" if value is not None else default
-
-                def _tristate_setting(value: str | None, default: str) -> str:
-                    """Tri-state workflow default, coercing legacy true/false rows."""
-                    if value is None:
-                        return default
-                    low = value.strip().lower()
-                    if low in ("on", "off", "auto"):
-                        return low
-                    if low in ("true", "1"):
-                        return "on"
-                    if low in ("false", "0"):
-                        return "off"
-                    return default
+                defaults = await default_print_options(db)
 
                 def _slicer_or(field_mqtt: str, settings_default: bool) -> bool:
                     """Slicer's MQTT value if present, else the settings default.
@@ -970,25 +956,21 @@ class VirtualPrinterInstance:
                 bed_levelling = _slicer_tristate(
                     "bed_leveling",
                     "auto_bed_leveling",
-                    _tristate_setting(await get_setting(db, "default_bed_levelling"), "auto"),
+                    defaults["bed_levelling"],
                 )
                 flow_cali = _slicer_tristate(
                     "flow_cali",
                     "extrude_cali_flag",
-                    _tristate_setting(await get_setting(db, "default_flow_cali"), "auto"),
+                    defaults["flow_cali"],
                 )
                 nozzle_offset_cali = _slicer_tristate(
                     None,
                     "nozzle_offset_cali",
-                    _tristate_setting(await get_setting(db, "default_nozzle_offset_cali"), "auto"),
+                    defaults["nozzle_offset_cali"],
                 )
-                vibration_cali = _slicer_or(
-                    "vibration_cali", _bool_setting(await get_setting(db, "default_vibration_cali"), True)
-                )
-                layer_inspect = _slicer_or(
-                    "layer_inspect", _bool_setting(await get_setting(db, "default_layer_inspect"), False)
-                )
-                timelapse = _slicer_or("timelapse", _bool_setting(await get_setting(db, "default_timelapse"), False))
+                vibration_cali = _slicer_or("vibration_cali", defaults["vibration_cali"])
+                layer_inspect = _slicer_or("layer_inspect", defaults["layer_inspect"])
+                timelapse = _slicer_or("timelapse", defaults["timelapse"])
                 # No setting for it: the column default (on) when the slicer is silent.
                 use_ams = _slicer_or("use_ams", True)
 
@@ -1300,6 +1282,9 @@ class VirtualPrinterInstance:
                             queue_item_ids,
                         )
                     await self._broadcast_archive_created(archive)
+                    from backend.app.services.queue_announcements import announce_queue_jobs_added
+
+                    await announce_queue_jobs_added(db, queue_item_ids)
                 else:
                     logger.error("Failed to archive file: %s", file_path.name)
         except Exception as e:

backend/tests/integration/test_queue_creation_paths_c4.py

@@ -0,0 +1,681 @@
+"""Every path that creates queue jobs behaves like the print dialog (C4).
+
+Only ``POST /queue/`` announced a new job ("job added" notification and the
+MQTT relay event); File Manager's "Add to queue", the webhook, a virtual
+printer, a pipeline run and an order's "dispatch remaining" said nothing. The
+default print options (timelapse, bed levelling, ...) reached only the print
+dialog and the virtual printer. The copies an order dispatches lost their cost
+centre, so they were billed to none and never budget-checked. And a
+``scheduled_time`` sent with a UTC offset lost the offset, so the job started
+hours late.
+"""
+
+from datetime import datetime
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+from uuid import uuid4
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from backend.app.core.auth import generate_api_key, get_password_hash
+from backend.app.core.config import settings as app_settings
+from backend.app.models.api_key import APIKey
+from backend.app.models.archive import PrintArchive
+from backend.app.models.finance import CostCenter, CostCenterMember
+from backend.app.models.group import Group
+from backend.app.models.print_queue import PrintQueueItem
+from backend.app.models.printer import Printer
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+from backend.app.services.mqtt_relay import mqtt_relay
+from backend.app.services.notification_service import notification_service
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
+
+
+async def _set(db_session, key: str, value: str) -> None:
+    row = await db_session.scalar(select(Settings).where(Settings.key == key))
+    if row is None:
+        db_session.add(Settings(key=key, value=value))
+    else:
+        row.value = value
+    await db_session.commit()
+
+
+async def _items(test_engine, ids=None) -> list[PrintQueueItem]:
+    maker = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
+    async with maker() as fresh:
+        query = select(PrintQueueItem).order_by(PrintQueueItem.id)
+        if ids is not None:
+            query = query.where(PrintQueueItem.id.in_(ids))
+        return list((await fresh.execute(query)).scalars().all())
+
+
+@pytest.fixture
+def announced():
+    """The notification and relay calls, recorded."""
+    notify = AsyncMock()
+    relay = AsyncMock()
+    with (
+        patch.object(notification_service, "on_queue_job_added", notify),
+        patch.object(mqtt_relay, "on_queue_job_added", relay),
+    ):
+        yield notify, relay
+
+
+@pytest.fixture
+async def library_files(db_session):
+    from backend.app.models.library import LibraryFile
+
+    made = []
+
+    async def make(stem: str):
+        name = f"{stem}_{uuid4().hex[:8]}.gcode.3mf"
+        rel_path = f"archive/library/files/{name}"
+        abs_path = Path(app_settings.base_dir) / rel_path
+        abs_path.parent.mkdir(parents=True, exist_ok=True)
+        abs_path.write_bytes(b"probe")
+        made.append(abs_path)
+        lib_file = LibraryFile(filename=name, file_path=rel_path, file_size=5, file_type="3mf")
+        db_session.add(lib_file)
+        await db_session.commit()
+        await db_session.refresh(lib_file)
+        return lib_file
+
+    yield make
+    for path in made:
+        path.unlink(missing_ok=True)
+
+
+async def _printer(db_session, name="C4 Printer", model="X1C") -> Printer:
+    printer = Printer(
+        name=name,
+        ip_address=f"192.168.44.{len(name)}",
+        serial_number=f"C4{uuid4().hex[:12].upper()}",
+        access_code="12345678",
+        model=model,
+    )
+    db_session.add(printer)
+    await db_session.commit()
+    await db_session.refresh(printer)
+    return printer
+
+
+async def _archive(db_session, filename="widget.gcode.3mf", **extra) -> PrintArchive:
+    archive = PrintArchive(
+        filename=filename,
+        print_name="widget",
+        file_path=f"/tmp/c4_{uuid4().hex}.3mf",  # nosec B108
+        file_size=1,
+        content_hash=uuid4().hex,
+        status="completed",
+        **extra,
+    )
+    db_session.add(archive)
+    await db_session.commit()
+    await db_session.refresh(archive)
+    return archive
+
+
+async def _webhook_key(db_session) -> dict[str, str]:
+    full_key, key_hash, key_prefix = generate_api_key()
+    db_session.add(
+        APIKey(name="c4", key_hash=key_hash, key_prefix=key_prefix, user_id=None, can_queue=True, can_read_status=True)
+    )
+    await db_session.commit()
+    return {"X-API-Key": full_key}
+
+
+class TestEveryPathAnnouncesItsJobs:
+    async def test_a_file_manager_bulk_add_sends_one_summary(
+        self, async_client: AsyncClient, db_session, library_files, announced
+    ):
+        notify, relay = announced
+        printer = await _printer(db_session)
+        files = [await library_files("bracket"), await library_files("hinge")]
+
+        response = await async_client.post(
+            "/api/v1/library/files/add-to-queue",
+            json={"file_ids": [f.id for f in files], "printer_id": printer.id},
+        )
+
+        assert response.status_code == 200, response.text
+        notify.assert_awaited_once()
+        kwargs = notify.await_args.kwargs
+        assert kwargs["job_name"].startswith("2 jobs (bracket_")
+        assert "hinge_" in kwargs["job_name"]
+        assert kwargs["target"] == printer.name
+        assert kwargs["printer_id"] == printer.id
+        # The relay is a machine event: one per job.
+        assert sorted(c.kwargs["job_id"] for c in relay.await_args_list) == sorted(
+            a["queue_item_id"] for a in response.json()["added"]
+        )
+
+    async def test_the_webhook_announces_its_job(self, async_client: AsyncClient, db_session, announced):
+        notify, relay = announced
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+
+        response = await async_client.post(
+            "/api/v1/webhook/queue/add",
+            json={"printer_id": printer.id, "archive_id": archive.id},
+            headers=await _webhook_key(db_session),
+        )
+
+        assert response.status_code == 200, response.text
+        notify.assert_awaited_once()
+        assert notify.await_args.kwargs["job_name"] == "widget"
+        assert relay.await_args.kwargs["job_id"] == response.json()["id"]
+
+    async def test_an_orders_dispatch_announces_the_copies(self, async_client: AsyncClient, db_session, announced):
+        notify, _ = announced
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+        order = (
+            await async_client.post(
+                "/api/v1/queue/batches",
+                json={"name": "o", "archive_id": archive.id, "plates": [{"plate_id": None, "quantity_target": 3}]},
+            )
+        ).json()
+        first = await async_client.post(
+            "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "batch_id": order["id"]}
+        )
+        assert first.status_code == 200, first.text
+        notify.reset_mock()
+
+        response = await async_client.post(f"/api/v1/queue/batches/{order['id']}/dispatch", json={})
+
+        assert response.status_code == 200, response.text
+        notify.assert_awaited_once()
+        assert notify.await_args.kwargs["job_name"] == "widget ×2"
+
+    async def test_a_failing_notification_does_not_fail_the_add(
+        self, async_client: AsyncClient, db_session, library_files
+    ):
+        lib_file = await library_files("plain")
+        with patch.object(notification_service, "on_queue_job_added", AsyncMock(side_effect=RuntimeError("smtp"))):
+            response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
+
+        assert response.status_code == 200
+        assert len(response.json()["added"]) == 1
+
+    async def test_the_print_dialog_still_announces_once(self, async_client: AsyncClient, db_session, announced):
+        notify, relay = announced
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+
+        response = await async_client.post(
+            "/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id, "quantity": 3}
+        )
+
+        assert response.status_code == 200, response.text
+        notify.assert_awaited_once()
+        assert notify.await_args.kwargs["job_name"] == "widget ×3"
+        assert relay.await_count == 3
+
+
+class TestTheSummary:
+    async def test_jobs_for_different_printers_are_announced_per_printer(self, db_session, test_engine, announced):
+        """A provider scoped to one printer must not hear of another's jobs."""
+        from backend.app.services.queue_announcements import announce_queue_jobs_added
+
+        notify, _ = announced
+        a, b = await _printer(db_session, "Alpha"), await _printer(db_session, "Bravo")
+        archive = await _archive(db_session)
+        items = [
+            PrintQueueItem(printer_id=a.id, archive_id=archive.id, position=1),
+            PrintQueueItem(printer_id=b.id, archive_id=archive.id, position=2),
+            PrintQueueItem(printer_id=None, target_model="P1S", archive_id=archive.id, position=3),
+        ]
+        db_session.add_all(items)
+        await db_session.commit()
+
+        await announce_queue_jobs_added(db_session, [i.id for i in items])
+
+        sent = sorted((c.kwargs["printer_id"] or 0, c.kwargs["target"]) for c in notify.await_args_list)
+        assert sent == sorted([(a.id, "Alpha"), (b.id, "Bravo"), (0, "Any P1S")])
+
+    async def test_many_different_files_are_named_briefly(self, db_session, announced):
+        from backend.app.services.queue_announcements import announce_queue_jobs_added
+
+        notify, _ = announced
+        items = []
+        for n in range(5):
+            archive = await _archive(db_session, filename=f"part{n}.3mf")
+            items.append(PrintQueueItem(printer_id=None, archive_id=archive.id, position=n))
+        db_session.add_all(items)
+        await db_session.commit()
+
+        await announce_queue_jobs_added(db_session, [i.id for i in items])
+
+        assert notify.await_args.kwargs["job_name"] == "5 jobs (part0, part1, part2, …)"
+        assert notify.await_args.kwargs["target"] == "Unassigned"
+
+    async def test_nothing_added_says_nothing(self, db_session, announced):
+        from backend.app.services.queue_announcements import announce_queue_jobs_added
+
+        notify, relay = announced
+        await announce_queue_jobs_added(db_session, [])
+
+        notify.assert_not_awaited()
+        relay.assert_not_awaited()
+
+
+class TestTheDefaultPrintOptions:
+    async def test_unset_they_are_the_settings_defaults(self, db_session):
+        from backend.app.services.print_defaults import default_print_options
+
+        assert await default_print_options(db_session) == {
+            "bed_levelling": "auto",
+            "flow_cali": "auto",
+            "nozzle_offset_cali": "auto",
+            "vibration_cali": True,
+            "layer_inspect": False,
+            "timelapse": False,
+        }
+
+    async def test_set_they_are_read_including_legacy_values(self, db_session):
+        from backend.app.services.print_defaults import default_print_options
+
+        for key, value in {
+            "default_bed_levelling": "false",  # legacy boolean row
+            "default_flow_cali": "on",
+            "default_nozzle_offset_cali": "off",
+            "default_vibration_cali": "false",
+            "default_layer_inspect": "true",
+            "default_timelapse": "true",
+        }.items():
+            await _set(db_session, key, value)
+
+        assert await default_print_options(db_session) == {
+            "bed_levelling": "off",
+            "flow_cali": "on",
+            "nozzle_offset_cali": "off",
+            "vibration_cali": False,
+            "layer_inspect": True,
+            "timelapse": True,
+        }
+
+    async def test_a_file_manager_add_uses_them(
+        self, async_client: AsyncClient, db_session, test_engine, library_files
+    ):
+        await _set(db_session, "default_timelapse", "true")
+        await _set(db_session, "default_bed_levelling", "off")
+        lib_file = await library_files("defaults")
+
+        response = await async_client.post("/api/v1/library/files/add-to-queue", json={"file_ids": [lib_file.id]})
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [response.json()["added"][0]["queue_item_id"]])
+        assert (item.timelapse, item.bed_levelling) == (True, "off")
+
+    async def test_the_webhook_uses_them(self, async_client: AsyncClient, db_session, test_engine):
+        await _set(db_session, "default_timelapse", "true")
+        await _set(db_session, "default_layer_inspect", "true")
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+
+        response = await async_client.post(
+            "/api/v1/webhook/queue/add",
+            json={"printer_id": printer.id, "archive_id": archive.id},
+            headers=await _webhook_key(db_session),
+        )
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [response.json()["id"]])
+        assert (item.timelapse, item.layer_inspect) == (True, True)
+
+
+class TestAPipelineRun:
+    async def _run(self, db_session, test_engine, tmp_path):
+        from backend.app.api.routes.pipeline_runs import _make_orchestration_callable
+        from backend.app.core.printer_scope import ALL_PRINTERS
+        from backend.app.models.library import LibraryFile
+        from backend.app.models.pipeline_run import PipelineJob, PipelineRun
+        from backend.app.models.slicer_pipeline import SlicerPipeline
+        from backend.app.schemas.slicer import SliceResponse
+
+        printer = await _printer(db_session, "Pipe")
+        source = LibraryFile(filename="pipe.3mf", file_path="library/pipe.3mf", file_type="3mf", file_size=1)
+        pipeline = SlicerPipeline(
+            name="p",
+            printer_preset_source="local",
+            printer_preset_id="1",
+            process_preset_source="local",
+            process_preset_id="2",
+            filament_presets_json='[{"source": "local", "id": "3"}]',
+            target_kind="specific_printer",
+            target_printer_id=printer.id,
+        )
+        db_session.add_all([source, pipeline])
+        await db_session.commit()
+        run = PipelineRun(pipeline_id=pipeline.id, source_library_file_id=source.id, copies=2, created_by=None)
+        db_session.add(run)
+        await db_session.commit()
+        db_session.add_all([PipelineJob(pipeline_run_id=run.id, copy_index=i) for i in range(2)])
+        await db_session.commit()
+        model = tmp_path / "pipe.3mf"
+        model.write_bytes(b"model")
+
+        async def sliced(session, **kwargs):
+            out = LibraryFile(filename="pipe.gcode.3mf", file_path="library/pipe.gcode.3mf", file_type="3mf")
+            out.file_size = 1
+            session.add(out)
+            await session.flush()
+            return SliceResponse(
+                library_file_id=out.id, name="pipe", print_time_seconds=1, filament_used_g=1.0, filament_used_mm=1.0
+            )
+
+        orchestrate = _make_orchestration_callable(
+            run_id=run.id,
+            pipeline_id=pipeline.id,
+            src_kind="library_file",
+            src_id=source.id,
+            src_filename="pipe.3mf",
+            src_path=Path(model),
+            creator_user_id=None,
+            copies=2,
+            printer_scope=ALL_PRINTERS,
+        )
+        with (
+            patch(
+                "backend.app.api.routes.pipeline_runs.async_session",
+                async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False),
+            ),
+            patch("backend.app.api.routes.library.slice_and_persist", new=sliced),
+            patch("backend.app.api.routes.pipeline_runs._publish_run_event", new=AsyncMock()),
+        ):
+            await orchestrate(1)
+        return await _items(test_engine)
+
+    async def test_announces_its_copies_and_uses_the_defaults(self, db_session, test_engine, tmp_path, announced):
+        notify, _ = announced
+        await _set(db_session, "default_timelapse", "true")
+
+        items = await self._run(db_session, test_engine, tmp_path)
+
+        assert len(items) == 2
+        assert all(item.timelapse for item in items)
+        notify.assert_awaited_once()
+        assert notify.await_args.kwargs["job_name"] == "pipe ×2"
+
+
+class TestAVirtualPrinter:
+    async def test_announces_what_it_queued(self, tmp_path):
+        from backend.tests.unit.services.test_vp_queue_intake_d2 import _enqueue, _session, _vp
+
+        added: list = []
+        factory, _ = _session(added)
+        inst = _vp(tmp_path, factory, target_printer_id=7)
+        announce = AsyncMock()
+
+        await _enqueue(inst, tmp_path, {"timelapse": True}, announce=announce)
+
+        announce.assert_awaited_once()
+        assert announce.await_args.args[1] == [item.id for item in added]
+
+    async def test_where_the_slicer_says_nothing_it_uses_the_defaults(self, tmp_path):
+        """Read from the same helper as every other path now."""
+        from backend.tests.unit.services.test_vp_queue_intake_d2 import _enqueue, _session, _vp
+
+        added: list = []
+        factory, _ = _session(added)
+        inst = _vp(tmp_path, factory, target_printer_id=7)
+
+        await _enqueue(
+            inst,
+            tmp_path,
+            {"flow_cali": True},
+            {"default_timelapse": "true", "default_layer_inspect": "true", "default_vibration_cali": "false"},
+        )
+
+        assert (added[0].timelapse, added[0].layer_inspect, added[0].vibration_cali) == (True, True, False)
+
+
+class TestAnOrdersCopiesKeepTheirCostCentre:
+    @pytest.fixture
+    async def billed(self, db_session):
+        await _set(db_session, "auth_enabled", "true")
+        await _set(db_session, "advanced_auth_enabled", "false")
+        group = Group(
+            name="c4-dispatchers",
+            description="t",
+            permissions=[
+                "queue:create",
+                "queue:read_all",
+                "queue:update_all",
+                "archives:read_all",
+                "archives:reprint_all",
+            ],
+            is_system=False,
+        )
+        db_session.add(group)
+        await db_session.flush()
+        admin = User(username="c4admin", password_hash=get_password_hash("Pass1word!"), role="admin", is_active=True)
+        clerk = User(username="c4clerk", password_hash=get_password_hash("Pass1word!"), is_active=True, groups=[group])
+        center = CostCenter(name="c4-center", is_active=True, is_private=False)
+        db_session.add_all([admin, clerk, center])
+        await db_session.commit()
+        printer = await _printer(db_session)
+        archive = await _archive(db_session, cost=1.25, filament_used_grams=50.0)
+        await _set(db_session, "billing_enabled", "true")
+        return {"admin": admin, "clerk": clerk, "center": center, "printer": printer, "archive": archive}
+
+    async def _login(self, client, username):
+        response = await client.post("/api/v1/auth/login", json={"username": username, "password": "Pass1word!"})
+        assert response.status_code == 200, response.text
+        return {"Authorization": f"Bearer {response.json()['access_token']}"}
+
+    async def _order(self, client, headers, world):
+        order = await client.post(
+            "/api/v1/queue/batches",
+            json={"name": "o", "archive_id": world["archive"].id, "plates": [{"plate_id": None, "quantity_target": 3}]},
+            headers=headers,
+        )
+        assert order.status_code == 200, order.text
+        first = await client.post(
+            "/api/v1/queue/",
+            json={
+                "printer_id": world["printer"].id,
+                "archive_id": world["archive"].id,
+                "batch_id": order.json()["id"],
+                "cost_center_id": world["center"].id,
+            },
+            headers=headers,
+        )
+        assert first.status_code == 200, first.text
+        return order.json()["id"]
+
+    async def test_the_copies_are_billed_to_it(self, async_client: AsyncClient, test_engine, billed):
+        admin = await self._login(async_client, "c4admin")
+        batch_id = await self._order(async_client, admin, billed)
+
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin)
+
+        assert response.status_code == 200, response.text
+        items = [i for i in await _items(test_engine) if i.batch_id == batch_id]
+        assert len(items) == 3
+        assert {i.cost_center_id for i in items} == {billed["center"].id}
+
+    async def test_a_dispatcher_who_may_not_use_it_is_refused(
+        self, async_client: AsyncClient, db_session, test_engine, billed
+    ):
+        admin = await self._login(async_client, "c4admin")
+        batch_id = await self._order(async_client, admin, billed)
+        clerk = await self._login(async_client, "c4clerk")
+
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=clerk)
+
+        assert response.status_code == 403, response.text
+        assert len([i for i in await _items(test_engine) if i.batch_id == batch_id]) == 1
+
+        db_session.add(CostCenterMember(cost_center_id=billed["center"].id, user_id=billed["clerk"].id, can_print=True))
+        await db_session.commit()
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=clerk)
+        assert response.status_code == 200, response.text
+
+    async def test_the_budget_counts_every_copy(self, async_client: AsyncClient, db_session, test_engine, billed):
+        center = await db_session.get(CostCenter, billed["center"].id)
+        admin = await self._login(async_client, "c4admin")
+        batch_id = await self._order(async_client, admin, billed)
+        (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id]
+        # Room for this one and one more, not for the two copies still owed.
+        center.total_budget = 2 * template.estimated_cost + 0.01
+        await db_session.commit()
+
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin)
+
+        assert response.status_code == 400, response.text
+        assert "budget" in response.json()["detail"]
+
+    async def test_copies_without_one_are_refused_while_billing_is_on(
+        self, async_client: AsyncClient, db_session, test_engine, billed
+    ):
+        """Queued before billing was switched on: refused here, as POST /queue/
+        would, rather than failing later at the start."""
+        admin = await self._login(async_client, "c4admin")
+        batch_id = await self._order(async_client, admin, billed)
+        (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id]
+        row = await db_session.get(PrintQueueItem, template.id)
+        row.cost_center_id = None
+        await db_session.commit()
+
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin)
+
+        assert response.status_code == 400, response.text
+        assert "Cost center is required" in response.json()["detail"]
+
+    async def test_a_budget_that_holds_every_copy_dispatches(
+        self, async_client: AsyncClient, db_session, test_engine, billed
+    ):
+        """The copies are not counted against themselves."""
+        center = await db_session.get(CostCenter, billed["center"].id)
+        admin = await self._login(async_client, "c4admin")
+        batch_id = await self._order(async_client, admin, billed)
+        (template,) = [i for i in await _items(test_engine) if i.batch_id == batch_id]
+        center.total_budget = 3 * template.estimated_cost + 0.01
+        await db_session.commit()
+
+        response = await async_client.post(f"/api/v1/queue/batches/{batch_id}/dispatch", json={}, headers=admin)
+
+        assert response.status_code == 200, response.text
+        copies = [i for i in await _items(test_engine) if i.batch_id == batch_id and i.id != template.id]
+        assert [c.estimated_cost for c in copies] == [template.estimated_cost] * 2
+
+
+class TestAScheduledTimeWithAnOffset:
+    async def test_the_queue_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine):
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+
+        response = await async_client.post(
+            "/api/v1/queue/",
+            json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "2026-10-15T10:00:00+02:00"},
+        )
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [response.json()["id"]])
+        assert item.scheduled_time == datetime(2026, 10, 15, 8, 0)
+        assert response.json()["scheduled_time"] == "2026-10-15T08:00:00Z"
+
+    async def test_an_edit_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine):
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+        created = await async_client.post("/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id})
+
+        response = await async_client.patch(
+            f"/api/v1/queue/{created.json()['id']}", json={"scheduled_time": "2026-10-15T10:00:00-05:00"}
+        )
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [created.json()["id"]])
+        assert item.scheduled_time == datetime(2026, 10, 15, 15, 0)
+
+    async def test_a_bulk_edit_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine):
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+        created = await async_client.post("/api/v1/queue/", json={"printer_id": printer.id, "archive_id": archive.id})
+
+        response = await async_client.patch(
+            "/api/v1/queue/bulk",
+            json={"item_ids": [created.json()["id"]], "scheduled_time": "2026-10-15T10:00:00+02:00"},
+        )
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [created.json()["id"]])
+        assert item.scheduled_time == datetime(2026, 10, 15, 8, 0)
+
+    async def test_the_webhook_stores_it_in_utc(self, async_client: AsyncClient, db_session, test_engine):
+        printer = await _printer(db_session)
+        archive = await _archive(db_session)
+
+        response = await async_client.post(
+            "/api/v1/webhook/queue/add",
+            json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "2026-10-15T10:00:00+02:00"},
+            headers=await _webhook_key(db_session),
+        )
+
+        assert response.status_code == 200, response.text
+        (item,) = await _items(test_engine, [response.json()["id"]])
+        assert item.scheduled_time == datetime(2026, 10, 15, 8, 0)
+
+
+async def test_the_postgres_bind_hook_converts_rather_than_drops_the_offset():
+    """The backstop for any other aware value: naive columns hold UTC."""
+    from datetime import timedelta, timezone
+
+    from backend.app.core import database
+
+    with patch.object(database, "is_sqlite", return_value=False):
+        engine = database._create_engine("postgresql+asyncpg://u:p@127.0.0.1:1/never")
+    try:
+        (hook,) = [
+            fn for fn in engine.sync_engine.dispatch.before_cursor_execute if "strip" in getattr(fn, "__name__", "")
+        ]
+        aware = datetime(2026, 10, 15, 10, 0, tzinfo=timezone(timedelta(hours=2)))
+
+        _, params = hook(None, None, "", {"t": aware, "rows": [(aware,)]}, None, False)
+
+        assert params == {"t": datetime(2026, 10, 15, 8, 0), "rows": [(datetime(2026, 10, 15, 8, 0),)]}
+    finally:
+        await engine.dispose()
+
+
+async def test_a_time_at_the_edge_of_the_calendar_is_refused_not_a_crash(async_client: AsyncClient, db_session):
+    printer = await _printer(db_session)
+    archive = await _archive(db_session)
+
+    queued = await async_client.post(
+        "/api/v1/queue/",
+        json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "0001-01-01T00:00:00+05:00"},
+    )
+    webhook = await async_client.post(
+        "/api/v1/webhook/queue/add",
+        json={"printer_id": printer.id, "archive_id": archive.id, "scheduled_time": "0001-01-01T00:00:00+05:00"},
+        headers=await _webhook_key(db_session),
+    )
+
+    assert (queued.status_code, webhook.status_code) == (422, 400)
+
+
+async def test_a_failed_notification_log_leaves_the_session_usable(db_session):
+    from backend.app.services.queue_announcements import announce_queue_jobs_added
+
+    archive = await _archive(db_session)
+    item = PrintQueueItem(printer_id=None, archive_id=archive.id, position=1)
+    db_session.add(item)
+    await db_session.commit()
+
+    async def log_then_fail(**kwargs):
+        db_session.add(Settings(key="dup", value="1"))
+        db_session.add(Settings(key="dup", value="2"))
+        await kwargs["db"].commit()
+
+    with patch.object(notification_service, "on_queue_job_added", side_effect=log_then_fail):
+        await announce_queue_jobs_added(db_session, [item.id])
+
+    db_session.add(Settings(key="after", value="1"))
+    await db_session.commit()

backend/tests/unit/services/test_virtual_printer.py

@@ -41,6 +41,12 @@ def _write_3mf_with_filaments(file_path: Path, filaments: list[dict], plate_inde
 class TestVirtualPrinterInstance:
     """Tests for VirtualPrinterInstance class."""
 
+    @pytest.fixture(autouse=True)
+    def _no_queue_announcement(self):
+        """Its own query would take a result the session mocks hold for the test."""
+        with patch("backend.app.services.queue_announcements.announce_queue_jobs_added", AsyncMock()):
+            yield
+
     @pytest.fixture
     def instance(self, tmp_path):
         """Create a VirtualPrinterInstance with test defaults."""

backend/tests/unit/services/test_vp_queue_intake_d2.py

@@ -42,7 +42,7 @@ def _vp(tmp_path, factory, **kwargs):
     )
 
 
-async def _enqueue(inst, tmp_path, slicer_opts: dict | None, settings: dict | None = None):
+async def _enqueue(inst, tmp_path, slicer_opts: dict | None, settings: dict | None = None, announce=None):
     file_path = tmp_path / "test.3mf"
     file_path.write_bytes(b"fake3mf")
     if slicer_opts is not None:
@@ -59,6 +59,8 @@ async def _enqueue(inst, tmp_path, slicer_opts: dict | None, settings: dict | No
             "backend.app.services.archive.ArchiveService.archive_print", new_callable=AsyncMock, return_value=archive
         ),
         patch("backend.app.services.virtual_printer.manager._SLICER_OPTIONS_WAIT_TIMEOUT", 0.05),
+        # Its own query would take a result the session mock holds for the test.
+        patch("backend.app.services.queue_announcements.announce_queue_jobs_added", announce or AsyncMock()),
     ):
         await inst._add_to_print_queue(file_path, "192.168.1.100")
     return file_path