feat(printing): add authenticated realtime device transport

Replace native job/control polling with client-initiated WebSockets, durable
state and receipt reconciliation, independent verified document streams and
native administrator settings for transfer, connection and admission limits.
Preserve QQQ current-user authorization and device binding on every operation.

Validation: Maven verify passes 117 tests with zero Checkstyle violations;
OpenAPI generation/check and diff checks pass. Consumer CI pins protocol
67be986cb76d6b5d9a9f72cad054193b0d11b0d5. Installation is separate.

Co-Authored-By: Codex (GPT-6) <noreply@openai.com>
Signed-off-by: James Maes <james@kof22.com>

src/test/java/com/kingsrook/qbits/printing/PrintingAdmissionTest.java

@@ -160,13 +160,13 @@ class PrintingAdmissionTest
 
 
    /***************************************************************************
-    ** Legacy upload intents with unknown size reserve the maximum allowed bytes.
+    ** Legacy unknown sizes block new admission even when current capacity exceeds their historical estimate.
     ***************************************************************************/
    @Test
    void legacyUnknownUploadChargesTenantCapacityUntilExactCleanup() throws Exception
    {
       long unknownBytes = 20L * 1024 * 1024;
-      Fixture fixture = fixture(limits(10, 20, 100_000_000, 200_000_000, 100_000_000, unknownBytes, 100, 200, 10, 20));
+      Fixture fixture = fixture(limits(10, 20, 100_000_000, 200_000_000, 100_000_000, 200_000_000, 100, 200, 10, 20));
       String orphan = UUID.randomUUID().toString();
       String reference = fixture.documents.prefix() + orphan + ".zpl";
       fixture.documents.objects.put(reference, DOCUMENT.clone());
@@ -572,7 +572,7 @@ class PrintingAdmissionTest
        ** Read immutable synthetic bytes for the retry path.
        ***********************************************************************/
       @Override
-      public byte[] get(String table, String reference) throws SQLException
+      public byte[] get(String table, String reference, int maximum) throws SQLException
       {
          gets.incrementAndGet();
          byte[] content = objects.get(reference);

src/test/java/com/kingsrook/qbits/printing/PrintingDocumentStorageTest.java

@@ -255,6 +255,40 @@ class PrintingDocumentStorageTest
    }
 
 
+   /** Native StorageAction content larger than the old cap is consumed incrementally and verified. */
+   @Test
+   void streamsLargePdfAndRejectsChangedLengthWithoutBufferingInTheAdapter() throws Exception
+   {
+      Fixture fixture = fixture();
+      fixture.service.inventory(fixture.device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic")
+         .put("formats", new JSONArray(List.of("PDF")))));
+      byte[] document = new byte[21 * 1024 * 1024];
+      System.arraycopy("%PDF-1.7\n".getBytes(java.nio.charset.StandardCharsets.US_ASCII), 0, document, 0, 9);
+      JSONObject submitted = fixture.service.submit(new JSONObject().put("deviceId", fixture.device).put("printerId", "queue").put("title", "Large synthetic PDF")
+         .put("format", "PDF").put("contentBase64", Base64.getEncoder().encodeToString(document)).put("idempotencyKey", "large-pdf"));
+      fixture.service.claim(fixture.device);
+      java.security.MessageDigest digest = java.security.MessageDigest.getInstance("SHA-256");
+      long count = 0;
+      try(var stream = fixture.service.openContent(fixture.device, submitted.getString("jobId")))
+      {
+         byte[] buffer = new byte[65536];
+         int read;
+         while((read = stream.read(buffer)) != -1)
+         {
+            count += read;
+            digest.update(buffer, 0, read);
+         }
+      }
+      assertEquals(document.length, count);
+      assertEquals(submitted.getString("contentSha256"), java.util.HexFormat.of().formatHex(digest.digest()));
+      fixture.storage.put("printing_documents", fixture.storage.lastReference, "PDF", new byte[] {1, 2, 3});
+      try(var stream = fixture.service.openContent(fixture.device, submitted.getString("jobId")))
+      {
+         assertThrows(java.io.IOException.class, stream::readAllBytes);
+      }
+   }
+
+
 
    /***************************************************************************
     ** Synthetic bytes never leave the test process.

src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeProcessesTest.java

@@ -0,0 +1,201 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.printing;
+
+
+import java.math.BigDecimal;
+import java.util.UUID;
+import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.session.QUser;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+/*******************************************************************************
+ ** Native settings forms preserve exact values and enforce current authority.
+ *******************************************************************************/
+class PrintingRealtimeProcessesTest
+{
+   /***************************************************************************
+    ** Opening settings cannot save them until the operator passes native review.
+    ***************************************************************************/
+   @Test
+   void nativeReviewIsReadOnly() throws Exception
+   {
+      PrintingService service = fixture();
+      var preview = new RunProcessAction().execute(new RunProcessInput().withProcessName(PrintingRealtimeProcesses.SETTINGS));
+      assertFalse(preview.getException().isPresent(), () -> preview.getException().toString());
+      assertEquals("review", preview.getProcessState().getNextStepName().orElseThrow());
+      assertEquals(0, service.getRealtimeSettings().getLong("revision"));
+   }
+
+
+
+   /***************************************************************************
+    ** Large PDFs and fractional human units round-trip without a fixed 20 MiB cap.
+    ***************************************************************************/
+   @Test
+   void exactPolicyPersistsAndRejectsPartialOrInconsistentEdits() throws Exception
+   {
+      PrintingService service = fixture();
+      RunBackendStepInput save = prepared(PrintingRealtimeProcesses.SETTINGS);
+      save.addValue("maxPdfMiB", new BigDecimal("250.5"));
+      save.addValue("maxConcurrentDownloads", 3);
+      new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput());
+      assertEquals(262668288L, service.getRealtimeSettings().getLong("maxPdfBytes"));
+      assertEquals(3, service.getRealtimeSettings().getInt("maxConcurrentDownloads"));
+      assertEquals(1, service.getRealtimeSettings().getLong("revision"));
+      RunBackendStepInput fractionalByte = prepared(PrintingRealtimeProcesses.SETTINGS).addValue("maxPdfMiB", new BigDecimal("0.1"));
+      assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(fractionalByte, new RunBackendStepOutput()));
+      RunBackendStepInput tooLittleDisk = prepared(PrintingRealtimeProcesses.SETTINGS).addValue("maxStagedMiB", 1);
+      assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(tooLittleDisk, new RunBackendStepOutput()));
+      assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(input(PrintingRealtimeProcesses.SETTINGS, "execute"), new RunBackendStepOutput()));
+      assertEquals(1, service.getRealtimeSettings().getLong("revision"));
+   }
+
+
+
+   /***************************************************************************
+    ** Backend invocation cannot bypass either tenant scope or administrator role.
+    ***************************************************************************/
+   @Test
+   void currentUserAndTenantAreRequired() throws Exception
+   {
+      fixture();
+      QInstance instance = QContext.getQInstance();
+      RunBackendStepInput save = prepared(PrintingRealtimeProcesses.SETTINGS);
+      QContext.init(instance, session("foreign", true));
+      assertThrows(QPermissionDeniedException.class, () -> new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput()));
+      QContext.init(instance, session("tenant-a", false));
+      assertThrows(QPermissionDeniedException.class, () -> new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput()));
+   }
+
+
+
+   /***************************************************************************
+    ** Queue limits use the same typed native forms and survive new service objects.
+    ***************************************************************************/
+   @Test
+   void admissionSettingsPersistAndRejectInvalidCapacity() throws Exception
+   {
+      PrintingService service = fixture();
+      RunBackendStepInput save = prepared(PrintingRealtimeProcesses.ADMISSION).addValue("deviceJobsPerHour", 15000).addValue("deviceOutstandingMiB", new BigDecimal("10240.5"));
+      new PrintingRealtimeProcesses().run(save, new RunBackendStepOutput());
+      assertEquals(15000, service.getAdmissionSettings().getInt("deviceJobsPerHour"));
+      assertEquals(10737942528L, service.getAdmissionSettings().getLong("deviceOutstandingBytes"));
+      RunBackendStepInput zero = prepared(PrintingRealtimeProcesses.ADMISSION).addValue("devicePrinters", 0);
+      assertThrows(QException.class, () -> new PrintingRealtimeProcesses().run(zero, new RunBackendStepOutput()));
+      assertTrue(service.getAdmissionSettings().getInt("devicePrinters") > 0);
+   }
+
+
+
+   /***************************************************************************
+    ** Prepared form values exercise the same complete fields used by the UI.
+    ***************************************************************************/
+   private static RunBackendStepInput prepared(String process) throws Exception
+   {
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      new PrintingRealtimeProcesses().run(input(process, "prepare"), output);
+      RunBackendStepInput save = input(process, "execute");
+      output.getValues().forEach(save::addValue);
+      return (save);
+   }
+
+
+
+   /***************************************************************************
+    ** Each test owns a database and normal authenticated tenant session.
+    ***************************************************************************/
+   private static PrintingService fixture() throws Exception
+   {
+      DriverDataSource source = new DriverDataSource("jdbc:h2:mem:realtime-ui-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", "");
+      PrintingService service = new PrintingService(source, "tenant-a");
+      service.initialize();
+      QInstance instance = new QInstance();
+      instance.addBackend(new QBackendMetaData().withName("memory").withBackendType(com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule.class));
+      instance.addTable(new QTableMetaData().withName("fixture").withBackendName("memory").withPrimaryKeyField("id").withField(new QFieldMetaData("id", QFieldType.STRING)));
+      instance.addQBit(new QBitMetaData().withGroupId("com.kingsrook.qbits").withArtifactId("qbit-printing").withVersion("test").withNamespace("printing").withConfig(new PrintingQBitConfig().withService(service)));
+      QAppMetaData app = new QAppMetaData().withName("printing").withLabel("Printing");
+      PrintingRealtimeProcesses.register(instance, app);
+      instance.addApp(app);
+      instance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withName("mock").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.MOCK));
+      QContext.init(instance, session("tenant-a", true));
+      return (service);
+   }
+
+
+
+   /***************************************************************************
+    ** Negative authorization cases retain a real user to test role and tenant.
+    ***************************************************************************/
+   private static QSession session(String tenant, boolean administrator)
+   {
+      QSession session = new QSession().withUser(new QUser().withIdReference("operator")).withSecurityKeyValue("printing_tenant", tenant);
+      if(administrator)
+      {
+         session.withPermissions("printing_admin");
+      }
+      return (session);
+   }
+
+
+
+   /***************************************************************************
+    ** Backend entrypoints use explicit process and step names.
+    ***************************************************************************/
+   private static RunBackendStepInput input(String process, String step)
+   {
+      RunBackendStepInput input = new RunBackendStepInput().withProcessName(process);
+      input.setStepName(step);
+      return (input);
+   }
+
+
+
+   /***************************************************************************
+    ** Never leak current tenant authority into another test.
+    ***************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+}

src/test/java/com/kingsrook/qbits/printing/PrintingRealtimeTest.java

@@ -0,0 +1,291 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.printing;
+
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import com.kingsrook.qqq.middleware.javalin.QApplicationLauncher;
+import com.kingsrook.qqq.middleware.javalin.QApplicationLauncherConfig;
+import io.javalin.Javalin;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+/** Real native authentication and WebSocket lifecycle with durable synthetic jobs only. */
+public class PrintingRealtimeTest
+{
+   @TempDir
+   Path directory;
+
+
+   /** Commit wakeups, exact receipts, replacement and revocation all use actual QQQ sessions. */
+   @Test
+   void authenticatedPushReconcilesDurableWorkWithoutHttpControlPolling() throws Exception
+   {
+      String jdbc = "jdbc:h2:file:" + directory.resolve("realtime").toString().replace('\\', '/');
+      PrintingLocalHost host = new PrintingLocalHost(new PrintingLocalSettings(jdbc, "sa", "", "demo", 0, "synthetic-management-token-long-enough", "operator", "synthetic-realtime-password-1234"));
+      AtomicReference<Javalin> app = new AtomicReference<>();
+      QApplicationLauncher launcher = QApplicationLauncher.run(host, new QApplicationLauncherConfig().withRegisterShutdownHook(false).withServerCustomizer(server ->
+      {
+         host.configure(server, false);
+         server.withJavalinConfigurationCustomizer(app::set);
+      }));
+      PrintingService service = launcher.getQInstance().getQBits().values().stream().map(value -> value.getConfig()).filter(PrintingQBitConfig.class::isInstance)
+         .map(PrintingQBitConfig.class::cast).findFirst().orElseThrow().getService();
+      String base = "http://127.0.0.1:" + app.get().port();
+      JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Synthetic realtime").put("deviceName", "Synthetic realtime").put("agentVersion", "test"), "operator");
+      String device = enrolled.getString("deviceId");
+      String proof = enrolled.getString("deviceToken");
+      service.inventory(device, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL"))));
+      service.enablePrinter(device, "queue", true);
+      try(HttpClient http = HttpClient.newHttpClient())
+      {
+         String cookie = login(http, base);
+         assertThrows(Exception.class, () -> PrintingSocketFixture.connect(http, base, "sessionUUID=invalid", device, proof));
+         assertThrows(Exception.class, () -> PrintingSocketFixture.connect(http, base, cookie, device, "wrong-proof"));
+         try(PrintingSocketFixture first = PrintingSocketFixture.connect(http, base, cookie, device, proof))
+         {
+            first.take("JOB_AVAILABLE");
+            first.send("INVENTORY", "idle-inventory", new JSONObject().put("printers", new JSONArray().put(new JSONObject().put("printerId", "queue")
+               .put("name", "Synthetic queue").put("status", "READY").put("formats", new JSONArray().put("ZPL")))));
+            assertEquals("idle-inventory", first.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId"));
+            org.junit.jupiter.api.Assertions.assertNull(first.received.poll(300, TimeUnit.MILLISECONDS), "Idle inventory must not create a self-sustaining job wakeup");
+            sharedBusRoundTrip(launcher, service, jdbc, http, cookie, device, first);
+            first.received.clear();
+            String job = service.submit(request(device, "one")).getString("jobId");
+            first.take("JOB_AVAILABLE");
+            assertEquals(job, service.claim(device).getJSONObject("job").getString("jobId"));
+            JSONObject event = new JSONObject().put("eventId", "stable-event").put("status", "RECEIVED");
+            first.send("JOB_EVENT", "event-stable-event", new JSONObject().put("jobId", job).put("event", event));
+            assertEquals("event-stable-event", first.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId"));
+            assertEquals("RECEIVED", service.getJob(job).getString("status"));
+            JSONObject policy = service.getRealtimeSettings();
+            policy.remove("revision");
+            service.saveRealtimeSettings(policy.put("maxCopies", 99));
+            assertEquals(1, first.take("SETTINGS").getJSONObject("payload").getLong("revision"));
+            first.apply(0, 0);
+            assertEquals(409, assertThrows(PrintingException.class, () -> service.requireRealtimeReady(device)).getStatus());
+            first.apply(0, 1);
+            service.requireRealtimeReady(device);
+            JSONObject pause = service.deviceAction(device, action("PAUSE_DISPATCH"), "operator");
+            JSONObject desired = first.take("DESIRED_STATE").getJSONObject("payload");
+            while(!desired.getBoolean("dispatchPaused"))
+            {
+               desired = first.take("DESIRED_STATE").getJSONObject("payload");
+            }
+            first.apply(desired.getLong("revision"), 1);
+            assertTrue(service.claim(device).isNull("job"));
+            assertFalse(pause.isEmpty());
+            JSONObject command = service.deviceAction(device, action("REFRESH_INVENTORY"), "operator");
+            JSONObject pushed = first.take("COMMAND").getJSONObject("payload");
+            assertEquals(command.getString("commandId"), pushed.getString("commandId"));
+            JSONObject receipt = new JSONObject().put("receiptId", "stable-receipt").put("outcome", "INVENTORY_REFRESHED").put("completedAt", Instant.now().toString());
+            JSONObject delivery = new JSONObject().put("commandId", pushed.getString("commandId")).put("receipt", receipt);
+            first.send("COMMAND_RECEIPT", "receipt-stable-receipt", delivery);
+            first.take("ACK");
+            first.send("COMMAND_RECEIPT", "receipt-pruned", new JSONObject().put("commandId", UUID.randomUUID().toString()).put("receipt", receipt));
+            assertEquals("NOT_FOUND", first.take("ERROR").getJSONObject("payload").getString("code"));
+            first.send("PING", "after-pruned-receipt", new JSONObject());
+            first.take("PONG");
+            try(PrintingSocketFixture second = PrintingSocketFixture.connect(http, base, cookie, device, proof))
+            {
+               assertEquals(1008, first.closed.get(10, TimeUnit.SECONDS));
+               second.send("COMMAND_RECEIPT", "receipt-stable-receipt", delivery);
+               assertEquals("receipt-stable-receipt", second.take("ACK").getJSONObject("payload").getString("acknowledgedMessageId"));
+               second.send("JOB_EVENT", "event-stable-event", new JSONObject().put("jobId", job).put("event", event));
+               second.take("ACK");
+               assertEquals("RECEIVED", service.getJob(job).getString("status"));
+               service.deviceAction(device, action("REVOKE"), "operator");
+               assertEquals(1008, second.closed.get(10, TimeUnit.SECONDS));
+            }
+         }
+         authenticationChangesAreImmediate(launcher, service, http, base, cookie);
+      }
+      finally
+      {
+         launcher.stop();
+      }
+   }
+
+
+   /** Cookie variants, changed provider roles, channel policy and logout are checked after upgrade. */
+   private static void authenticationChangesAreImmediate(QApplicationLauncher launcher, PrintingService service, HttpClient http, String base, String cookie) throws Exception
+   {
+      JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Authentication fixture").put("deviceName", "Authentication fixture").put("agentVersion", "test"), "operator");
+      String device = enrolled.getString("deviceId");
+      String proof = enrolled.getString("deviceToken");
+      var table = launcher.getQInstance().getTable(PrintingTables.DEVICE);
+      var originalRules = table.getPermissionRules();
+      try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie.replace("sessionUUID=", "sessionId="), device, proof))
+      {
+         table.withPermissionRules(PrintingProcesses.permissionRules().withDeniedClients(java.util.Set.of("api")));
+         channel.send("PING", "channel-denied", new JSONObject());
+         assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code"));
+         assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS));
+      }
+      finally
+      {
+         table.withPermissionRules(originalRules);
+      }
+      var auth = launcher.getQInstance().getAuthentication();
+      var originalModule = auth.getModuleCodeReference();
+      try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie, device, proof))
+      {
+         auth.withModuleCodeReference(new com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference(RemovedRoleAuthentication.class));
+         service.getRealtimeBus().publish("demo", device);
+         assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code"));
+         assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS));
+      }
+      finally
+      {
+         auth.withModuleCodeReference(originalModule);
+      }
+      try(PrintingSocketFixture channel = PrintingSocketFixture.connect(http, base, cookie, device, proof))
+      {
+         HttpResponse<String> logout = http.send(HttpRequest.newBuilder(URI.create(base + "/qqq/v1/logout")).header("Origin", base).header("Cookie", cookie)
+            .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString("{}")).build(), HttpResponse.BodyHandlers.ofString());
+         assertEquals(200, logout.statusCode());
+         channel.send("PING", "logged-out", new JSONObject());
+         assertEquals("AUTHORIZATION_REQUIRED", channel.take("ERROR").getJSONObject("payload").getString("code"));
+         assertEquals(1008, channel.closed.get(10, TimeUnit.SECONDS));
+      }
+   }
+
+
+   /** Native authentication still verifies the actual password/session; only current role membership changes. */
+   public static class RemovedRoleAuthentication extends PrintingLocalAuthentication
+   {
+      @Override
+      public com.kingsrook.qqq.backend.core.model.session.QSession createSession(com.kingsrook.qqq.backend.core.model.metadata.QInstance instance, java.util.Map<String, String> context)
+            throws com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException
+      {
+         var session = super.createSession(instance, context);
+         session.setPermissions(java.util.Set.of());
+         return (session);
+      }
+   }
+
+
+   /** Two independent route providers receive only their scoped commit wakeups through an explicitly shared bus. */
+   private static void sharedBusRoundTrip(QApplicationLauncher launcher, PrintingService service, String jdbc, HttpClient http, String cookie, String firstDevice, PrintingSocketFixture first) throws Exception
+   {
+      PrintingService sibling = new PrintingService(new DriverDataSource(jdbc, "sa", ""), "demo").withRealtimeBus(service.getRealtimeBus());
+      PrintingRealtimeRoutes routes = new PrintingRealtimeRoutes(sibling);
+      routes.setQInstance(launcher.getQInstance());
+      Javalin other = Javalin.create(config ->
+      {
+         config.jetty.host = "127.0.0.1";
+         routes.acceptJavalinConfig(config);
+      }).start(0);
+      JSONObject enrolled = service.enrollForUser(service.createEnrollment("default", "Second synthetic device").put("deviceName", "Second synthetic device").put("agentVersion", "test"), "operator");
+      String secondDevice = enrolled.getString("deviceId");
+      service.inventory(secondDevice, new JSONArray().put(new JSONObject().put("printerId", "queue").put("name", "Synthetic queue").put("formats", new JSONArray().put("ZPL"))));
+      service.enablePrinter(secondDevice, "queue", true);
+      try(PrintingSocketFixture second = PrintingSocketFixture.open(http, "http://127.0.0.1:" + other.port(), cookie, secondDevice, enrolled.getString("deviceToken")))
+      {
+         String early = service.submit(request(secondDevice, "commit-between-upgrade-and-hello")).getString("jobId");
+         second.initialize();
+         second.take("JOB_AVAILABLE");
+         assertEquals("QUEUED", sibling.getJob(early).getString("status"));
+         sibling.cancel(secondDevice, early);
+         first.received.clear();
+         second.received.clear();
+         String firstJob = sibling.submit(request(firstDevice, "other-service-to-first-hub")).getString("jobId");
+         first.take("JOB_AVAILABLE");
+         String secondJob = service.submit(request(secondDevice, "first-service-to-other-hub")).getString("jobId");
+         second.take("JOB_AVAILABLE");
+         assertEquals("QUEUED", sibling.getJob(firstJob).getString("status"));
+         assertEquals("QUEUED", service.getJob(secondJob).getString("status"));
+         service.cancel(firstDevice, firstJob);
+         sibling.cancel(secondDevice, secondJob);
+      }
+      finally
+      {
+         other.stop();
+      }
+   }
+
+
+   /** A real browser login supplies only its own temporary test session. */
+   private static String login(HttpClient client, String base) throws Exception
+   {
+      HttpRequest request = HttpRequest.newBuilder(URI.create(base + "/qqq/v1/manageSession")).header("Origin", base)
+         .header("Authorization", "Basic " + Base64.getEncoder().encodeToString("operator:synthetic-realtime-password-1234".getBytes(StandardCharsets.UTF_8)))
+         .header("Content-Type", "application/json").POST(HttpRequest.BodyPublishers.ofString("{}")).build();
+      HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
+      assertEquals(200, response.statusCode());
+      return (response.headers().allValues("set-cookie").stream().filter(value -> value.startsWith("sessionUUID=")).findFirst().orElseThrow().split(";", 2)[0]);
+   }
+
+
+   /** Stable synthetic submission identity; no physical printer exists in this fixture. */
+   private static JSONObject request(String device, String key)
+   {
+      return (new JSONObject().put("deviceId", device).put("printerId", "queue").put("title", "Synthetic realtime")
+         .put("format", "ZPL").put("contentBase64", Base64.getEncoder().encodeToString("^XA^XZ".getBytes(StandardCharsets.US_ASCII))).put("idempotencyKey", key));
+   }
+
+
+   /** Native actions retain their existing temporal replay boundary. */
+   private static JSONObject action(String name)
+   {
+      return (new JSONObject().put("action", name).put("requestId", UUID.randomUUID().toString()).put("issuedAt", Instant.now().toString()));
+   }
+
+
+   /** Policy updates preserve validation, tenant isolation and durable revisions. */
+   @Test
+   void policyRejectsInconsistentLimitsAndPersistsRevisions() throws Exception
+   {
+      DriverDataSource source = new DriverDataSource("jdbc:h2:mem:policy-" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1", "sa", "");
+      PrintingService first = new PrintingService(source, "one");
+      first.initialize();
+      PrintingService other = new PrintingService(source, "two");
+      JSONObject values = first.getRealtimeSettings();
+      assertEquals(104857600, values.getInt("maxPdfBytes"));
+      values.remove("revision");
+      JSONObject invalid = new JSONObject(values.toString()).put("heartbeatTimeoutSeconds", 10);
+      assertThrows(PrintingException.class, () -> first.saveRealtimeSettings(invalid));
+      assertThrows(PrintingException.class, () -> first.saveRealtimeSettings(new JSONObject(values.toString()).put("maxCopies", 1.5)));
+      values.put("maxPdfBytes", 30 * 1024 * 1024);
+      assertEquals(1, first.saveRealtimeSettings(values).getLong("revision"));
+      assertEquals(1, new PrintingService(source, "one").getRealtimeSettings().getLong("revision"));
+      assertNotEquals(first.getRealtimeSettings().getLong("maxPdfBytes"), other.getRealtimeSettings().getLong("maxPdfBytes"));
+      assertThrows(PrintingException.class, () -> PrintingRealtimePolicy.defaults().requireClientLimits(new JSONObject().put("maxPdfBytes", 1)));
+   }
+}

src/test/java/com/kingsrook/qbits/printing/PrintingRetentionProcessesTest.java

@@ -105,7 +105,7 @@ class PrintingRetentionProcessesTest
    {
       PrintingService service = fixture();
       PrintingRetentionProcesses step = new PrintingRetentionProcesses();
-      RunBackendStepInput save = input(PrintingRetentionProcesses.SETTINGS, "execute").addValue("days", 730).addValue("enabled", false);
+      RunBackendStepInput save = input(PrintingRetentionProcesses.SETTINGS, "execute").addValue("days", 730).addValue("enabled", false).addValue("batchSize", 100);
       QContext.getQSession().setSecurityKeyValues(Map.of("printing_tenant", List.of("foreign")));
       assertThrows(QPermissionDeniedException.class, () -> step.run(save, new RunBackendStepOutput()));
       QContext.init(QContext.getQInstance(), new QSession().withSecurityKeyValue("printing_tenant", "tenant-a"));

src/test/java/com/kingsrook/qbits/printing/PrintingSocketFixture.java

@@ -0,0 +1,157 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.printing;
+
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.WebSocket;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.json.JSONArray;
+import org.json.JSONObject;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+
+/** Real native WebSocket client shared by the HTTP and realtime integration fixtures. */
+final class PrintingSocketFixture implements WebSocket.Listener, AutoCloseable
+{
+   final BlockingQueue<JSONObject> received = new LinkedBlockingQueue<>();
+   final CompletableFuture<Integer> closed = new CompletableFuture<>();
+   private final StringBuilder text = new StringBuilder();
+   private WebSocket socket;
+
+
+   /** Authenticate with the normal session cookie and secondary proof headers. */
+   static PrintingSocketFixture connect(HttpClient client, String base, String cookie, String device, String proof) throws Exception
+   {
+      PrintingSocketFixture fixture = open(client, base, cookie, device, proof);
+      fixture.initialize();
+      return (fixture);
+   }
+
+
+   /** Separate upgrade from HELLO so a commit in that interval can be tested deterministically. */
+   static PrintingSocketFixture open(HttpClient client, String base, String cookie, String device, String proof) throws Exception
+   {
+      PrintingSocketFixture fixture = new PrintingSocketFixture();
+      fixture.socket = client.newWebSocketBuilder().subprotocols(PrintingRealtimeRoutes.SUBPROTOCOL).header("Origin", base)
+         .header("X-QQQ-Client", "api/qqq-print-agent")
+         .header("Cookie", cookie).header(PrintingAgentAuthorization.DEVICE_HEADER, device).header(PrintingAgentAuthorization.TOKEN_HEADER, proof)
+         .buildAsync(URI.create(base.replace("http:", "ws:") + PrintingRealtimeRoutes.PATH), fixture).get(10, TimeUnit.SECONDS);
+      return (fixture);
+   }
+
+
+   /** Reconcile after the connection's subscription is already registered. */
+   void initialize() throws Exception
+   {
+      send("HELLO", "hello", new JSONObject().put("agentVersion", "synthetic-test").put("capabilities", new JSONArray().put("realtime-v1").put("REFRESH_INVENTORY").put("CANCEL_SPOOL"))
+         .put("lastAppliedRevision", 0).put("lastSettingsRevision", 0).put("limits", PrintingRealtimePolicy.defaults().toJson(0)));
+      JSONObject welcome = take("WELCOME").getJSONObject("payload");
+      apply(welcome.getJSONObject("desiredState").getLong("revision"), welcome.getJSONObject("settings").getLong("revision"));
+   }
+
+
+   /** Observe application ACK before allowing a dependent HTTP claim. */
+   void apply(long state, long settings) throws Exception
+   {
+      send("STATE_APPLIED", "applied", new JSONObject().put("lastAppliedRevision", state).put("lastSettingsRevision", settings));
+      take("ACK");
+   }
+
+
+   /** Send one explicit protocol operation, never a timed query. */
+   void send(String type, String id, JSONObject payload)
+   {
+      socket.sendText(new JSONObject().put("type", type).put("messageId", id).put("payload", payload).toString(), true).join();
+   }
+
+
+   /** Bounded waiting diagnoses absent pushes without introducing a polling loop. */
+   JSONObject take(String type) throws Exception
+   {
+      long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+      while(System.nanoTime() < end)
+      {
+         JSONObject next = received.poll(Math.max(1, end - System.nanoTime()), TimeUnit.NANOSECONDS);
+         assertNotNull(next, "Expected realtime frame " + type);
+         if("ERROR".equals(next.getString("type")) && !"ERROR".equals(type))
+         {
+            throw new AssertionError("Realtime rejected synthetic test message: " + next.getJSONObject("payload"));
+         }
+         if(type.equals(next.getString("type")))
+         {
+            return (next);
+         }
+      }
+      throw new AssertionError("No frame " + type);
+   }
+
+
+   @Override
+   public void onOpen(WebSocket webSocket)
+   {
+      webSocket.request(1);
+   }
+
+
+   @Override
+   public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last)
+   {
+      text.append(data);
+      if(last)
+      {
+         received.add(new JSONObject(text.toString()));
+         text.setLength(0);
+      }
+      webSocket.request(1);
+      return (null);
+   }
+
+
+   @Override
+   public CompletionStage<?> onClose(WebSocket webSocket, int statusCode, String reason)
+   {
+      closed.complete(statusCode);
+      return (null);
+   }
+
+
+   @Override
+   public void onError(WebSocket webSocket, Throwable error)
+   {
+      closed.completeExceptionally(error);
+   }
+
+
+   @Override
+   public void close()
+   {
+      if(socket != null)
+      {
+         socket.abort();
+      }
+   }
+}

src/test/java/com/kingsrook/qbits/printing/PrintingStorageBindingTest.java

@@ -405,7 +405,7 @@ class PrintingStorageBindingTest
        ** Unexpected reads are observable independently of returned bytes.
        ************************************************************************/
       @Override
-      public byte[] get(String table, String reference)
+      public byte[] get(String table, String reference, int maximum)
       {
          calls.add("GET " + reference);
          return (objects.get(reference).clone());

src/test/java/com/kingsrook/qbits/printing/PrintingUiApiContractTest.java

@@ -61,7 +61,7 @@ class PrintingUiApiContractTest
       var model = version.generateExtensionOpenAPIModel("qqq", extension);
       assertEquals(17, model.getPaths().size());
       assertTrue(model.getPaths().keySet().stream().allMatch(path -> path.startsWith("/qqq/v1/qbit/printing/")));
-      assertEquals("1.0.0", extension.getSemanticVersion());
+      assertEquals("1.1.0", extension.getSemanticVersion());
       assertTrue(generated.contains("X-QQQ-Device-Token"));
       assertTrue(generated.contains("application/octet-stream"));
       assertTrue(generated.contains("principalId"));

src/test/java/com/kingsrook/qbits/printing/PrintingUiApiTest.java

@@ -136,6 +136,8 @@ class PrintingUiApiTest
          JSONObject others = service.enrollForUser(service.createEnrollment("default", "Other owner").put("deviceName", "Other owner").put("agentVersion", "test"), "another-principal");
          assertEquals(403, send(client, base, prefix + "/device/heartbeat", "POST", new JSONObject().put("agentVersion", "test"), cookie, others.getString("deviceId"), others.getString("deviceToken")).statusCode());
          assertEquals(device, json(send(client, base, prefix + "/session", "GET", null, cookie, device, proof), 200).getString("deviceId"));
+         assertEquals(409, send(client, base, prefix + "/device/jobs/claim", "POST", new JSONObject(), cookie, device, proof).statusCode());
+         PrintingSocketFixture channel = PrintingSocketFixture.connect(client, base, cookie, device, proof);
          assertEquals(job, json(send(client, base, prefix + "/device/jobs/claim", "POST", new JSONObject(), cookie, device, proof), 200).getJSONObject("job").getString("jobId"));
          HttpResponse<byte[]> downloaded = send(client, base, prefix + "/device/jobs/" + job + "/content", "GET", null, cookie, device, proof);
          assertEquals(200, downloaded.statusCode());
@@ -146,6 +148,7 @@ class PrintingUiApiTest
          JSONObject action = action("REVOKE");
          json(send(client, base, prefix + "/admin/devices/" + device + "/actions", "POST", action, cookie, null, null), 200);
          assertEquals(403, send(client, base, prefix + "/device/heartbeat", "POST", new JSONObject().put("agentVersion", "test"), cookie, device, proof).statusCode());
+         channel.close();
          JSONObject nativeEnrollment = service.createEnrollment("default", "Native enrollment").put("deviceName", "Native enrollment").put("agentVersion", "test");
          JSONObject fresh = json(send(client, base, prefix + "/devices/enroll", "POST", nativeEnrollment, cookie, null, null), 200);
          service.authorizeAgent(fresh.getString("deviceId"), fresh.getString("deviceToken"), "operator", false);