stream bluetooth responses with notifications

tools/galaxy_bluetooth/README.md

@@ -2,7 +2,7 @@
 
 A working prototype source project that bundles StarPilot's existing Galaxy mobile interface inside a native SwiftUI / WKWebView iPhone app. Requests to the comma travel through Core Bluetooth instead of Wi-Fi or a Galaxy relay.
 
-**Hardware testing is in progress.** A signed app has launched on a physical iPhone, connected to the comma 4 bridge, and loaded the dashboard with the phone's Wi-Fi and cellular disabled. The first Toggles test exposed an oversized internal parameter in StarPilot's settings response; the bridge now excludes that unused learning-history field. The corrected Toggles transfer and a real setting change still need hardware verification. Automated protocol/proxy tests pass. Run the included BlueZ probe before testing on another device.
+**Hardware testing is in progress.** A signed app has launched on a physical iPhone, connected to the comma 4 bridge, and loaded the dashboard with the phone's Wi-Fi and cellular disabled. The bridge excludes unused learning history that initially exceeded its body limit. Subsequent device logs confirmed compact responses but approximately 61 seconds spent transferring Toggles data through sequential Bluetooth reads. The app and bridge now support push notifications in bounded batches; that transfer mode and a real setting change still need hardware verification. Automated protocol/proxy tests pass. Run the included BlueZ probe before testing on another device.
 
 ## What is included
 
@@ -10,7 +10,7 @@ A working prototype source project that bundles StarPilot's existing Galaxy mobi
 - Galaxy mobile and classic assets, pinned to StarPilot commit `2a12dbd0ad94b46f8a7b6099d32d7216b7676f79`.
 - A separate Python service on the comma using its existing BlueZ daemon. It does not patch StarPilot, the driving code or the Bluetooth kernel.
 - AES-256-GCM encryption using a private pairing key, per-session challenges and monotonic request counters, direction-bound authentication, and compressed messages.
-- MTU-aware fragmentation, compact HTTP response compression, and a negotiated read stream with one acknowledgement after the complete response. Older apps/bridges retain the original per-fragment acknowledgement mode. No automatic retry of a settings change.
+- MTU-aware fragmentation, compact HTTP response compression, and negotiated push notifications with eight fragments per acknowledgement window. Older apps/bridges retain the read-stream or original per-fragment acknowledgement mode. No automatic retry of a settings change.
 - Shared concurrent settings reads and a five-minute in-memory catalog/defaults cache, cleared after any write or new connection. Current toggle values are never cached.
 - Read-only hardware diagnostics, automated bridge tests and Swift/Python interoperability checks.
 
@@ -27,7 +27,11 @@ Galaxy interface bundled on iPhone
 
 The two HTTP hops stay inside their respective devices. The iPhone-to-comma connection uses Bluetooth. Galaxy's screens and server-side settings behavior are retained. The phone's loopback server rejects external origins and requires a private app header for API requests.
 
-For faster loading, the authenticated health response advertises `readStream`. The phone opts in only after verifying it. The bridge compresses raw HTTP bytes before base64 encoding, permits response packets up to 512 bytes within the negotiated ATT MTU, and advances one fragment per successful read. The phone verifies the complete authenticated response before sending the final ACK. A read error closes the session; it never retries a fragment or a mutating request. Existing characteristic UUIDs and flags remain unchanged. The old per-fragment mode remains available for compatibility.
+For faster loading, the authenticated health response advertises `notificationStream`. After verifying it, the phone subscribes to a separate notification characteristic (`bd490005-6dc1-4de7-a7d0-6cdb441f7650`) before opening Galaxy. The bridge compresses raw HTTP bytes before base64 encoding and pushes up to eight MTU-bounded fragments without requiring reads. Each window has a tagged ACK that releases the next batch. The final ACK follows successful decryption and verification of the entire response. Shared notifications carry a session/request tag; publishers are serialized, and the phone ignores frames for other clients or requests. Subscriptions alone cannot execute an API request: the request still must pass authentication, session and replay checks.
+
+Bursts stop after eight packets without an ACK. Missing credit times out after 15 seconds without retransmitting fragments or the HTTP request. The phone also detects a 15-second stall after a transfer begins and retains its overall request timeout. An interrupted transfer requires reconnection and checking the setting's outcome. Small requests use the larger negotiated write payload too. The existing RX/TX/INFO characteristics retain their UUIDs and flags, and older bridges/apps use read-stream or per-fragment ACK mode.
+
+When updating from the earlier read-only prototype, disconnect the app and forget the comma/Galaxy entry in iPhone **Settings > Bluetooth** once so iOS discovers the new characteristic. The native app handles service invalidation and reports a stale service cache explicitly rather than silently staying in the slow read mode. Keep the app's pairing key; forgetting the system Bluetooth entry does not replace it.
 
 Startup reads only `LanguageSetting`, rather than the entire parameter snapshot. Toggles reuses its catalog/defaults for up to five minutes; opening it again still fetches current values. Any non-GET request invalidates cached metadata and concurrent reads before and after forwarding the write. The cache does not persist across connections.
 
@@ -144,7 +148,7 @@ node tests/settings_load.mjs
 
 The tests check actual local HTTP forwarding, binary response preservation, finite SSE, body limits, redirect/path restrictions, encryption tamper rejection, GATT packet/ACK behavior, and replay/session rejection. The interoperability check compiles the production Swift wire/parser files on macOS and exchanges messages with Python in both directions. It requires Xcode's command line tools.
 
-To compare response sizes without modifying settings, run `python3 scripts/benchmark_settings.py --galaxy-url http://YOUR_COMMA:8082`. Add `--packet-size 512` only to model a connection whose negotiated MTU permits that payload. The default comparison uses 180-byte packets. This prints byte/operation counts, never parameter values or keys, and does not measure actual Bluetooth loading time. Both the bridge and iPhone app must be updated for the full optimization.
+To compare response sizes without modifying settings, run `python3 scripts/benchmark_settings.py --galaxy-url http://YOUR_COMMA:8082`. Add `--packet-size 512` only to model a connection whose negotiated MTU permits that payload. The default comparison uses 180-byte packets. This prints byte/operation counts for legacy reads, read streams and windowed notifications, never parameter values or keys, and does not measure actual Bluetooth loading time. Both the bridge and iPhone app must be updated for the full optimization.
 
 To regenerate the included Xcode project after adding Swift files, run `python3 scripts/create_xcode_project.py`.
 

tools/galaxy_bluetooth/VALIDATION.md

@@ -5,9 +5,9 @@
 | iPhone compilation and linking | Passed, using Apple Swift 6.4 in Swift 5 mode, iPhoneOS 27.0 SDK, ARM64 target, iOS 17 minimum |
 | iPhone compiler warnings | None in the final compile/link run |
 | Xcode project and Info.plist parsing | Passed |
-| Python protocol/proxy/GATT session tests | 17 tests passed, including compact binary/JSON bodies, malformed metadata/body bounds, capability-gated larger packets, retained MTU, one final stream ACK, learning-history filtering, replay rejection and toggle-value preservation |
+| Python protocol/proxy/GATT session tests | 23 tests passed, including notification framing/MTU, eight-packet credits, wrong/stale ACK rejection, final verification, subscription/authentication gating, multiple client isolation, timeout/cancellation without repeating a mutation, and existing codec/proxy/replay checks |
 | Installer with unavailable ensurepip | Passed with uv and pip 26.2.1 using real isolated environments and local dependency wheels; recovered a partial environment and preserved the pairing key on reinstall |
-| Swift → Python and Python → Swift | Passed for encrypted, compressed messages using production wire code |
+| Swift → Python and Python → Swift | Passed for encrypted/compact messages, notification stream tags/fragments, unrelated-frame filtering, truncated-frame rejection and tagged window ACKs using production wire code |
 | Swift HTTP parser | Passed for partial bodies and malformed/duplicate/chunked header rejection |
 | Production loopback HTTP server on macOS | Passed for serving Galaxy HTML/modules, API forwarding to a fixture, request key enforcement, origin rejection and disconnected responses |
 | Production Swift settings read cache | Passed for concurrent read sharing, uncached current values, write/reconnect/TTL invalidation, header variants, failed responses, cancellation and reads spanning writes |
@@ -16,7 +16,7 @@
 | Signed Xcode build for a physical iPhone | Optimized app passed in the user's Xcode with automatic signing and their personal team; latest report shows Build succeeded, no issues |
 | Xcode simulator build | Passed in the user's Xcode; a simulator launch was not used for Bluetooth testing |
 | Physical iPhone launch and Bluetooth exchange | App launched after turning off debugger attachment; user reported connection and dashboard loading with iPhone Wi-Fi/cellular disabled |
-| Toggles over Bluetooth | The first hardware test failed at the 1 MiB limit; after updating the bridge, the user reports very slow loading. Optimized bridge/app timing and a real setting change still need hardware verification |
+| Toggles over Bluetooth | Compact read-stream responses delivered on hardware but took approximately 61 seconds of transfer time. Notification-stream timing and a real setting change remain pending |
 | comma 4 BlueZ prerequisites | Passed in the user's device probe: powered hci0, central/peripheral roles, GATT server and advertising interfaces, five supported advertising instances and zero active before bridge startup; Galaxy HTTP status 200 |
 | comma 4 GATT and advertisement registration | Passed; foreground bridge reported ready on hci0 and forwarding to localhost:8082 |
 | Bluetooth coexistence with other accessories | Pending |
@@ -33,4 +33,6 @@ The bridge is installed separately from the user's existing StarPilot Dom instal
 
 Read-only diagnosis of the user's Galaxy API found `/api/params/all` was approximately 1.55 MB, including roughly 1.46 MB of `LiveTorqueParameters`. This key is absent from the settings catalog and unused by the bundled frontend. Running the corrected production proxy against that API returned HTTP 200 with approximately 92 KB, preserved all 765 other values exactly, and produced an encrypted/compressed frame of approximately 27 KB. This check used the Mac's network connection for diagnosis; the iPhone's corrected Bluetooth transfer and a real toggle write remain pending. The settings catalog was approximately 190 KB and defaults approximately 20 KB, both below the existing BLE body limit.
 
-The performance update was measured using read-only HTTP responses from the same device, with a public test key and representative envelope metadata. Compact responses were approximately 19 KB for current parameters (previously 27 KB), 25 KB for the catalog (41 KB), and 6 KB for defaults (9 KB). At the same 180-byte packet size, the first Settings load needs approximately 287 response packets instead of 447; one final ACK per response reduces modeled response read/ACK operations from 894 to 290. Subsequent visits use about 110 parameter packets plus one ACK while metadata remains cached. Request writes, wait polling, radio latency and phone rendering are excluded. Startup also avoids a separate full-parameter transfer just to read language. These are transfer counts, not a measured speedup on the physical Bluetooth connection. The new app and bridge need to be installed together and timed on hardware.
+The first performance update was measured using read-only HTTP responses from the same device, with a public test key and representative envelope metadata. Compact responses were approximately 19 KB for current parameters (previously 27 KB), 25 KB for the catalog (41 KB), and 6 KB for defaults (9 KB). At the same 180-byte packet size, the first Settings load needed approximately 287 response packets instead of 447; one final ACK per response reduced modeled response read/ACK operations from 894 to 290. Startup also avoids a separate full-parameter transfer just to read language. These counts excluded request writes, wait polling, radio latency and rendering.
+
+The user's subsequent hardware trace confirmed the compact read stream was active with 512-byte packets. Catalog: 24,773 bytes / 49 packets / 28.07 seconds. Current parameters: 19,157 bytes / 38 packets / 24.74 seconds. Defaults: 6,074 bytes / 12 packets / 8.19 seconds. Combined preparation time was 0.38 seconds; transfer time was 61.00 seconds. This established sequential read latency as the remaining bottleneck despite the smaller payloads. Those three frames would use 102 notification packets and 14 window ACKs at the same payload size, avoiding the 99 sequential read requests. Notification throughput is not yet measured on the hardware; both app and bridge need the new version, with the phone's remembered Bluetooth service cache refreshed.

tools/galaxy_bluetooth/bridge/protocol.py

@@ -1,6 +1,7 @@
 """Galaxy BLE v1: bounded framing and authenticated, direction-bound messages."""
 import json
 import base64
+import hashlib
 import os
 import struct
 import zlib
@@ -11,6 +12,8 @@ SERVICE_UUID = "bd490001-6dc1-4de7-a7d0-6cdb441f7650"
 RX_UUID = "bd490002-6dc1-4de7-a7d0-6cdb441f7650"
 TX_UUID = "bd490003-6dc1-4de7-a7d0-6cdb441f7650"
 INFO_UUID = "bd490004-6dc1-4de7-a7d0-6cdb441f7650"
+NOTIFY_UUID = "bd490005-6dc1-4de7-a7d0-6cdb441f7650"
+NOTIFICATION_WINDOW = 8
 MAX_FRAME = 2 * 1024 * 1024
 MAX_BODY = 1024 * 1024
 AAD_PREFIX = b"galaxy-ble-v1/"
@@ -125,3 +128,14 @@ def packets(frame, size=180):
         raise ValueError("Invalid ATT payload size")
     for sequence, start in enumerate(range(0, len(frame), size - 5)):
         yield b"\x01" + struct.pack(">I", sequence) + frame[start:start + size - 5]
+
+
+def stream_tag(session, counter):
+    return hashlib.sha256(bytes.fromhex(session) + struct.pack(">Q", counter)).digest()[:8]
+
+
+def notification_packets(frame, tag, size):
+    if len(tag) != 8 or not 20 <= size <= 512:
+        raise ValueError("Invalid notification framing")
+    for sequence, start in enumerate(range(0, len(frame), size - 13)):
+        yield b"\x03" + tag + struct.pack(">I", sequence) + frame[start:start + size - 13]

tools/galaxy_bluetooth/bridge/proxy.py

@@ -58,7 +58,7 @@ class GalaxyProxy:
         try:
             if request.get("path") == "/_bridge/health" and request.get("method") == "GET":
                 return {"id": request_id, "status": 200, "headers": {"content-type": "application/json"},
-                        "body": base64.b64encode(b'{"protocol":1,"transport":"bluetooth","readStream":true}').decode()}
+                        "body": base64.b64encode(b'{"protocol":1,"transport":"bluetooth","readStream":true,"notificationStream":true}').decode()}
             target = validate_target(request.get("path"))
             method = request.get("method", "GET")
             if method not in METHODS:

tools/galaxy_bluetooth/bridge/server.py

@@ -17,7 +17,8 @@ from dbus_next.aio import MessageBus
 from dbus_next.constants import PropertyAccess
 from dbus_next.service import ServiceInterface, dbus_property, method
 
-from protocol import Assembler, INFO_UUID, RX_UUID, SERVICE_UUID, TX_UUID, open_message, packets, seal
+from protocol import (Assembler, INFO_UUID, RX_UUID, SERVICE_UUID, TX_UUID, NOTIFY_UUID,
+                      NOTIFICATION_WINDOW, open_message, packets, seal, stream_tag, notification_packets)
 from proxy import GalaxyProxy, error_response
 
 ROOT = "/link/galaxy/ble"
@@ -25,6 +26,7 @@ SERVICE_PATH = ROOT + "/service0"
 GATT_SERVICE = "org.bluez.GattService1"
 GATT_CHAR = "org.bluez.GattCharacteristic1"
 LOG = logging.getLogger("galaxy-ble")
+NOTIFICATION_ACK_TIMEOUT = 15
 
 
 class Session:
@@ -40,10 +42,19 @@ class Session:
         self.packet_size = 20
         self.read_stream = False
         self.response_started = 0
+        self.notification_stream = False
+        self.notification_tag = b""
+        self.notification_ack = asyncio.Event()
+        self.notification_expected_ack = None
+        self.notification_task = None
+        self.closed = False
 
     def close(self):
+        self.closed = True
         if self.task:
             self.task.cancel()
+        if self.notification_task:
+            self.notification_task.cancel()
 
 
 class Gateway:
@@ -51,6 +62,8 @@ class Gateway:
         self.key = key
         self.proxy = proxy
         self.sessions = {}
+        self.notifier = None
+        self.notification_lock = asyncio.Lock()
 
     def session(self, options, fresh=False):
         device = options.get("device")
@@ -78,9 +91,22 @@ class Gateway:
         if options.get("offset", Variant("q", 0)).value:
             raise DBusError("org.bluez.Error.InvalidOffset", "Long writes are not used")
         try:
+            if len(value) == 13 and value[0] == 4:
+                sequence = struct.unpack(">I", value[9:])[0]
+                if (not state.notification_stream or not state.response or bytes(value[1:9]) != state.notification_tag
+                        or sequence != state.notification_expected_ack or state.index != sequence + 1
+                        or state.notification_ack.is_set()):
+                    raise ValueError("Unexpected notification acknowledgement")
+                state.notification_ack.set()
+                state.notification_expected_ack = None
+                if state.index == len(state.response):
+                    self.acknowledged(state)
+                return
             if len(value) == 5 and value[0] == 2:
                 sequence = struct.unpack(">I", value[1:])[0]
-                if state.read_stream and state.response:
+                if state.notification_stream:
+                    raise ValueError("Notification responses require a tagged acknowledgement")
+                elif state.read_stream and state.response:
                     if state.index != len(state.response) or sequence != len(state.response) - 1:
                         raise ValueError("Unexpected stream acknowledgement")
                     self.acknowledged(state)
@@ -103,7 +129,8 @@ class Gateway:
 
     def acknowledged(self, state):
         LOG.info("Response delivered: %d packets in %.2fs (%s)", len(state.response),
-                 time.monotonic() - state.response_started, "read stream" if state.read_stream else "per-packet ACK")
+                 time.monotonic() - state.response_started,
+                 "notification stream" if state.notification_stream else "read stream" if state.read_stream else "per-packet ACK")
         state.response = []
         state.index = 0
         state.busy = False
@@ -112,6 +139,8 @@ class Gateway:
         state = self.session(options)
         if options.get("offset", Variant("q", 0)).value:
             raise DBusError("org.bluez.Error.InvalidOffset", "Packets fit inside the negotiated MTU")
+        if state.notification_stream:
+            return b"\x00"
         if not state.response or state.index >= len(state.response):
             return b"\x00"
         packet = state.response[state.index]
@@ -131,19 +160,29 @@ class Gateway:
                 raise ValueError("Invalid session or replayed request")
             if not isinstance(request.get("id"), str) or len(request["id"]) > 64:
                 raise ValueError("Invalid request identifier")
+            if counter > 0xffffffffffffffff:
+                raise ValueError("Invalid counter")
+            fast = request.get("responseCodec") == 2
+            notify = fast and request.get("responseFlow") == "notify-window8"
+            if notify and (self.notifier is None or not self.notifier.notifying):
+                raise ValueError("Subscribe to notifications before requesting that response flow")
             state.counter = counter  # Consume before invoking any mutating endpoint.
             response = await asyncio.to_thread(self.proxy.handle, request)
             response["session"] = state.challenge.hex()
             response["counter"] = counter
-            fast = request.get("responseCodec") == 2
+            state.notification_stream = notify
+            state.notification_tag = stream_tag(state.challenge.hex(), counter) if notify else b""
             state.read_stream = fast and request.get("responseFlow") == "read-stream"
             frame = seal(response, self.key, "response", compact_body=fast)
             size = state.packet_size if fast else min(180, state.packet_size)
-            state.response = list(packets(frame, size))
+            state.response = list(notification_packets(frame, state.notification_tag, size) if notify else packets(frame, size))
             state.index = 0
             state.response_started = time.monotonic()
-            LOG.info("Response prepared: %d bytes, %d packets, ATT payload %d, %.2fs",
-                     len(frame), len(state.response), size, time.monotonic() - started)
+            LOG.info("Response prepared: %d bytes, %d packets, ATT payload %d, %.2fs (%s)",
+                     len(frame), len(state.response), size, time.monotonic() - started,
+                     "notification stream" if notify else "read stream" if state.read_stream else "per-packet ACK")
+            if notify:
+                state.notification_task = asyncio.create_task(self.push_response(state))
         except asyncio.CancelledError:
             raise
         except Exception:
@@ -153,6 +192,36 @@ class Gateway:
         finally:
             state.task = None
 
+    async def push_response(self, state):
+        response = state.response
+        try:
+            # BlueZ broadcasts Value notifications. Serialize publishers and
+            # tag frames so other subscribed clients ignore unrelated traffic.
+            async with self.notification_lock:
+                while state.response is response and response and not state.closed:
+                    if self.notifier is None or not self.notifier.notifying:
+                        raise RuntimeError("Notification subscription ended")
+                    window = response[state.index:state.index + NOTIFICATION_WINDOW]
+                    state.notification_ack.clear()
+                    state.notification_expected_ack = state.index + len(window) - 1
+                    for packet in window:
+                        if state.closed or not self.notifier.notifying:
+                            raise RuntimeError("Notification session ended")
+                        state.index += 1
+                        self.notifier.send(packet)
+                        # Bound the burst and yield to D-Bus/ATT. No fragment or
+                        # original HTTP request is retransmitted on a timeout.
+                        await asyncio.sleep(0.004)
+                    await asyncio.wait_for(state.notification_ack.wait(), timeout=NOTIFICATION_ACK_TIMEOUT)
+        except asyncio.CancelledError:
+            raise
+        except Exception as error:
+            LOG.warning("Notification transfer interrupted (%s); reconnect to check request outcome", type(error).__name__)
+            # Keep the session busy until a fresh INFO handshake replaces it.
+        finally:
+            if state.notification_task is asyncio.current_task():
+                state.notification_task = None
+
 
 class GattService(ServiceInterface):
     def __init__(self):
@@ -210,6 +279,47 @@ class Characteristic(ServiceInterface):
         return {"UUID": Variant("s", self.UUID), "Service": Variant("o", self.Service), "Flags": Variant("as", self.Flags)}
 
 
+class NotificationCharacteristic(Characteristic):
+    def __init__(self, gateway):
+        super().__init__(NOTIFY_UUID, ["notify"], gateway, "notify")
+        self.value = b""
+        self.notifying = False
+        gateway.notifier = self
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Value(self) -> 'ay':
+        return self.value
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Notifying(self) -> 'b':
+        return self.notifying
+
+    @method()
+    def StartNotify(self):
+        if not self.notifying:
+            self.notifying = True
+            self.emit_properties_changed({"Notifying": True})
+            LOG.info("Bluetooth push enabled")
+
+    @method()
+    def StopNotify(self):
+        self.notifying = False
+        self.emit_properties_changed({"Notifying": False})
+        LOG.info("Bluetooth push disabled")
+        for state in self.gateway.sessions.values():
+            if state.notification_task:
+                state.notification_task.cancel()
+
+    def send(self, packet):
+        if not self.notifying:
+            raise RuntimeError("Not subscribed")
+        self.value = bytes(packet)
+        self.emit_properties_changed({"Value": self.value})
+
+    def props(self):
+        return {**super().props(), "Value": Variant("ay", self.value), "Notifying": Variant("b", self.notifying)}
+
+
 class Application(ServiceInterface):
     def __init__(self, objects):
         super().__init__("org.freedesktop.DBus.ObjectManager")
@@ -264,6 +374,7 @@ async def serve(args):
         SERVICE_PATH + "/rx": Characteristic(RX_UUID, ["write"], gateway, "rx"),
         SERVICE_PATH + "/tx": Characteristic(TX_UUID, ["read"], gateway, "tx"),
         SERVICE_PATH + "/info": Characteristic(INFO_UUID, ["read"], gateway, "info"),
+        SERVICE_PATH + "/notify": NotificationCharacteristic(gateway),
     }
     bus.export(ROOT, Application(objects))
     for path, interface in objects.items():

tools/galaxy_bluetooth/ios/GalaxyBluetooth/BluetoothTransport.swift

@@ -22,12 +22,20 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
     private var peripheral: CBPeripheral?
     private var rx: CBCharacteristic?
     private var tx: CBCharacteristic?
+    private var notify: CBCharacteristic?
     private var key: SymmetricKey?
     private var keyText = ""
     private var session = ""
     var sessionIdentifier: String { session }
     private var supportsReadStream = false
     private var activeReadStream = false
+    private var supportsNotifications = false
+    private var activeNotifications = false
+    private var pairingVerified = false
+    private var notificationTag = Data()
+    private var notificationCount = 0
+    private var notificationBuffer: [Data] = []
+    private var notificationTimeout: Task<Void, Never>?
     private var counter: UInt64 = 0
     private var queue: [Pending] = []
     private var active: Pending?
@@ -105,10 +113,15 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         peripheral = nil
         rx = nil
         tx = nil
+        notify = nil
         session = ""
         counter = 0
         supportsReadStream = false
         activeReadStream = false
+        supportsNotifications = false
+        activeNotifications = false
+        pairingVerified = false
+        notificationBuffer = []
     }
 
     private func failConnection(_ message: String) {
@@ -161,13 +174,17 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         activeCancelled = false
         awaitingAck = false
         writeIndex = 0
+        activeNotifications = supportsNotifications && active.path != "/_bridge/health"
+        notificationCount = 0
+        notificationBuffer = []
         activeReadStream = supportsReadStream && active.path != "/_bridge/health"
         do {
+            notificationTag = try Wire.streamTag(session: session, counter: counter)
             let request = BridgeRequest(id: active.id, session: session, counter: counter, path: active.path,
                                         method: active.method, headers: active.headers, body: active.body.base64EncodedString(),
-                                        responseCodec: 2, responseFlow: activeReadStream ? "read-stream" : nil)
+                                        responseCodec: 2, responseFlow: activeNotifications ? "notify-window8" : activeReadStream ? "read-stream" : nil)
             outgoing = Wire.packets(try Wire.seal(request, key: key, direction: "request"),
-                                    size: min(180, peripheral.maximumWriteValueLength(for: .withoutResponse)))
+                                    size: min(supportsNotifications ? 512 : 180, peripheral.maximumWriteValueLength(for: .withoutResponse)))
             timeoutTask = Task { @MainActor [weak self] in
                 let seconds: UInt64 = active.path == "/_bridge/health" ? 10 : 90
                 try? await Task.sleep(nanoseconds: seconds * 1_000_000_000)
@@ -184,6 +201,8 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         guard let peripheral, let rx, active != nil else { return }
         if writeIndex < outgoing.count {
             peripheral.writeValue(outgoing[writeIndex], for: rx, type: .withResponse)
+        } else if activeNotifications {
+            drainNotifications()
         } else if let tx {
             peripheral.readValue(for: tx)
         }
@@ -191,10 +210,12 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
 
     private func finish(_ result: Result<BridgeResponse, Error>) {
         timeoutTask?.cancel()
+        notificationTimeout?.cancel()
         timeoutTask = nil
         let previous = active
         active = nil
         outgoing = []
+        notificationBuffer = []
         if activeCancelled { previous?.completion.resume(throwing: CancellationError()) }
         else { previous?.completion.resume(with: result) }
         activeCancelled = false
@@ -203,11 +224,13 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
 
     private func failAll(_ error: Error) {
         timeoutTask?.cancel()
+        notificationTimeout?.cancel()
         let pending = queue
         queue = []
         let previous = active
         active = nil
         outgoing = []
+        notificationBuffer = []
         previous?.completion.resume(throwing: error)
         pending.forEach { $0.completion.resume(throwing: error) }
     }
@@ -251,7 +274,7 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         guard error == nil, let service = peripheral.services?.first(where: { $0.uuid == CBUUID(string: Wire.service) }) else {
             failConnection("Galaxy Bluetooth service was not found."); return
         }
-        peripheral.discoverCharacteristics([CBUUID(string: Wire.rx), CBUUID(string: Wire.tx), CBUUID(string: Wire.info)], for: service)
+        peripheral.discoverCharacteristics([CBUUID(string: Wire.rx), CBUUID(string: Wire.tx), CBUUID(string: Wire.info), CBUUID(string: Wire.notify)], for: service)
     }
 
     func peripheral(_ peripheral: CBPeripheral, didDiscoverCharacteristicsFor service: CBService, error: Error?) {
@@ -259,6 +282,7 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         guard error == nil else { failConnection(error!.localizedDescription); return }
         rx = service.characteristics?.first { $0.uuid == CBUUID(string: Wire.rx) }
         tx = service.characteristics?.first { $0.uuid == CBUUID(string: Wire.tx) }
+        notify = service.characteristics?.first { $0.uuid == CBUUID(string: Wire.notify) }
         guard rx != nil, tx != nil, let info = service.characteristics?.first(where: { $0.uuid == CBUUID(string: Wire.info) }) else {
             failConnection("The comma bridge has incompatible Bluetooth characteristics."); return
         }
@@ -273,6 +297,8 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
             if let response = completedResponse {
                 completedResponse = nil
                 finish(.success(response))
+            } else if activeNotifications {
+                drainNotifications()
             } else if let tx {
                 peripheral.readValue(for: tx)
             }
@@ -286,26 +312,40 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
         guard peripheral == self.peripheral else { return }
         if let error { failConnection(error.localizedDescription); return }
         guard let data = characteristic.value else { failConnection("Empty Bluetooth response."); return }
+        if characteristic.uuid == CBUUID(string: Wire.notify) {
+            receiveNotification(data)
+            return
+        }
         if characteristic.uuid == CBUUID(string: Wire.info) {
             guard data.count == 17, data.first == 1 else { failConnection("Unsupported bridge protocol."); return }
             session = Data(data.dropFirst()).hex
+            let handshakeSession = session
             status = "Verifying the pairing key…"
-            Task { @MainActor [weak self] in
-                guard let self else { return }
+            Task { @MainActor [weak self, weak peripheral] in
+                guard let self, let peripheral else { return }
                 do {
                     let response = try await self.request(path: "/_bridge/health", method: "GET", headers: [:], body: Data())
+                    guard peripheral == self.peripheral, self.session == handshakeSession else { return }
                     guard response.status == 200 else { throw BridgeError.message("Bridge verification failed.") }
                     let health = try JSONDecoder().decode(BridgeHealth.self, from: response.bodyData)
                     guard health.protocol == 1, health.transport == "bluetooth" else {
                         throw BridgeError.message("Unsupported bridge protocol.")
                     }
                     self.supportsReadStream = health.readStream == true
-                    try PairingKeyStore.save(self.keyText)
-                    self.connectionTimeout?.cancel()
-                    self.connected = true
-                    self.connecting = false
-                    self.status = "Connected over Bluetooth"
-                } catch { self.failConnection("Pairing failed: \(error.localizedDescription)") }
+                    self.pairingVerified = true
+                    if health.notificationStream == true {
+                        guard let notify = self.notify, notify.properties.contains(.notify) else {
+                            throw BridgeError.message("Bluetooth services are cached. Forget Galaxy in iPhone Bluetooth Settings, then reconnect.")
+                        }
+                        self.status = "Enabling fast Bluetooth transfers…"
+                        peripheral.setNotifyValue(true, for: notify)
+                    } else {
+                        try self.completeConnection()
+                    }
+                } catch {
+                    guard peripheral == self.peripheral, self.session == handshakeSession else { return }
+                    self.failConnection("Pairing failed: \(error.localizedDescription)")
+                }
             }
             return
         }
@@ -337,4 +377,77 @@ final class BluetoothTransport: NSObject, ObservableObject, GalaxyRequestTranspo
             peripheral.writeValue(Data([2]) + data.subdata(in: 1..<5), for: rx, type: .withResponse)
         } catch { failConnection("Could not verify the encrypted response: \(error.localizedDescription)") }
     }
+
+    private func completeConnection() throws {
+        try PairingKeyStore.save(keyText)
+        connectionTimeout?.cancel()
+        connected = true
+        connecting = false
+        status = supportsNotifications ? "Connected with fast Bluetooth transfers" : "Connected over Bluetooth"
+    }
+
+    func peripheral(_ peripheral: CBPeripheral, didUpdateNotificationStateFor characteristic: CBCharacteristic, error: Error?) {
+        guard peripheral == self.peripheral, characteristic.uuid == CBUUID(string: Wire.notify), pairingVerified else { return }
+        guard error == nil, characteristic.isNotifying else {
+            failConnection(error?.localizedDescription ?? "Bluetooth push subscription ended. Reconnect to check any pending setting change.")
+            return
+        }
+        supportsNotifications = true
+        do { try completeConnection() }
+        catch { failConnection(error.localizedDescription) }
+    }
+
+    func peripheral(_ peripheral: CBPeripheral, didModifyServices invalidatedServices: [CBService]) {
+        guard peripheral == self.peripheral, invalidatedServices.contains(where: { $0.uuid == CBUUID(string: Wire.service) }) else { return }
+        failConnection("The bridge's Bluetooth services changed. Scan and reconnect.")
+    }
+
+    private func receiveNotification(_ data: Data) {
+        guard active != nil, activeNotifications else { return }
+        do {
+            // Unrelated client/request frames must not enter this assembler.
+            guard try Wire.notificationPacket(data, tag: notificationTag) != nil else { return }
+            notificationTimeout?.cancel()
+            let requestID = active?.id
+            notificationTimeout = Task { @MainActor [weak self] in
+                try? await Task.sleep(nanoseconds: 15_000_000_000)
+                guard !Task.isCancelled, let self, self.active?.id == requestID else { return }
+                self.failConnection("Bluetooth push stalled. Reconnect to check any pending setting change.")
+            }
+            if awaitingAck || writeIndex < outgoing.count {
+                guard notificationBuffer.count < Wire.notificationWindow else {
+                    throw BridgeError.message("Bluetooth notification window exceeded.")
+                }
+                notificationBuffer.append(data)
+            } else {
+                try consumeNotification(data)
+            }
+        } catch { failConnection("Could not verify Bluetooth push data: \(error.localizedDescription)") }
+    }
+
+    private func drainNotifications() {
+        while active != nil, activeNotifications, !awaitingAck, !notificationBuffer.isEmpty {
+            let data = notificationBuffer.removeFirst()
+            do { try consumeNotification(data) }
+            catch { failConnection("Could not verify Bluetooth push data: \(error.localizedDescription)") }
+        }
+    }
+
+    private func consumeNotification(_ data: Data) throws {
+        guard let packet = try Wire.notificationPacket(data, tag: notificationTag), let peripheral, let rx else { return }
+        if let message = try assembler.append(packet), let key {
+            let response = try Wire.open(message, as: BridgeResponse.self, key: key, direction: "response")
+            guard response.id == active?.id, response.session == session, response.counter == counter,
+                  (100...599).contains(response.status), let body = Data(base64Encoded: response.body), body.count <= Wire.maxBody else {
+                throw BridgeError.message("Bluetooth response identity did not match the request.")
+            }
+            completedResponse = response
+        }
+        notificationCount += 1
+        if completedResponse != nil || notificationCount % Wire.notificationWindow == 0 {
+            awaitingAck = true
+            let sequence = Wire.number(packet.subdata(in: 1..<5))
+            peripheral.writeValue(Wire.notificationAck(tag: notificationTag, sequence: sequence), for: rx, type: .withResponse)
+        }
+    }
 }

tools/galaxy_bluetooth/ios/GalaxyBluetooth/Wire.swift

@@ -29,6 +29,7 @@ struct BridgeHealth: Decodable {
     let `protocol`: Int
     let transport: String
     let readStream: Bool?
+    let notificationStream: Bool?
 }
 
 enum BridgeError: LocalizedError {
@@ -44,6 +45,8 @@ enum Wire {
     static let rx = "bd490002-6dc1-4de7-a7d0-6cdb441f7650"
     static let tx = "bd490003-6dc1-4de7-a7d0-6cdb441f7650"
     static let info = "bd490004-6dc1-4de7-a7d0-6cdb441f7650"
+    static let notify = "bd490005-6dc1-4de7-a7d0-6cdb441f7650"
+    static let notificationWindow = 8
     static let maxFrame = 2 * 1024 * 1024
     static let maxBody = 1024 * 1024
 
@@ -115,6 +118,27 @@ enum Wire {
             Data([1]) + uint32(UInt32(sequence)) + frame.subdata(in: start..<min(start + payloadSize, frame.count))
         }
     }
+
+    static func streamTag(session: String, counter: UInt64) throws -> Data {
+        guard let challenge = Data(hex: session), challenge.count == 16 else {
+            throw BridgeError.message("Invalid Bluetooth session.")
+        }
+        var value = counter.bigEndian
+        let number = withUnsafeBytes(of: &value) { Data($0) }
+        return Data(SHA256.hash(data: challenge + number).prefix(8))
+    }
+
+    static func notificationPacket(_ packet: Data, tag: Data) throws -> Data? {
+        guard packet.count >= 9, packet.first == 3 else { throw BridgeError.message("Invalid Bluetooth notification.") }
+        guard packet.subdata(in: 1..<9) == tag else { return nil }
+        guard packet.count >= 14 else { throw BridgeError.message("Truncated Bluetooth notification.") }
+        // Reuse the bounded, ordered assembler used by the read transport.
+        return Data([1]) + packet.dropFirst(9)
+    }
+
+    static func notificationAck(tag: Data, sequence: UInt32) -> Data {
+        Data([4]) + tag + uint32(sequence)
+    }
 }
 
 struct FrameAssembler {

tools/galaxy_bluetooth/scripts/benchmark_settings.py

@@ -8,7 +8,7 @@ import sys
 import urllib.request
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "bridge"))
-from protocol import MAX_BODY, packets, seal
+from protocol import MAX_BODY, packets, seal, stream_tag, notification_packets, NOTIFICATION_WINDOW
 from proxy import INTERNAL_PARAMS, PARAMS_SNAPSHOT_LIMIT
 
 parser = argparse.ArgumentParser(description=__doc__)
@@ -17,6 +17,7 @@ parser.add_argument("--packet-size", type=int, default=180, choices=range(20, 51
 args = parser.parse_args()
 paths = ["/api/params/all", "/assets/components/tools/device_settings_layout.json?v=settings-tier-1", "/api/params/defaults"]
 old_operations = new_operations = 0
+notification_count = notification_acks = 0
 for path in paths:
     with urllib.request.urlopen(args.galaxy_url.rstrip("/") + path, timeout=10) as response:
         content_type = response.headers.get("Content-Type", "application/json")
@@ -41,7 +42,12 @@ for path in paths:
     new_packets = len(list(packets(compact, args.packet_size)))
     old_operations += old_packets * 2
     new_operations += new_packets + 1
+    pushed = len(list(notification_packets(compact, stream_tag("01" * 16, 1), args.packet_size)))
+    credits = (pushed + NOTIFICATION_WINDOW - 1) // NOTIFICATION_WINDOW
+    notification_count += pushed
+    notification_acks += credits
     print(f"{path}: HTTP {len(body)} B; encrypted {len(legacy)} → {len(compact)} B; "
           f"response packets {old_packets} → {new_packets}")
 print(f"First settings load: {old_operations} → {new_operations} response read/ACK operations")
+print(f"Notification mode: {notification_count} pushed packets, {notification_acks} window ACKs, zero response reads")
 print("Counts exclude request writes and wait polling; this is not measured Bluetooth latency.")

tools/galaxy_bluetooth/scripts/check_interop.py

@@ -6,11 +6,12 @@ import json
 import platform
 import subprocess
 import sys
+import struct
 import tempfile
 
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "bridge"))
-from protocol import open_message, seal
+from protocol import open_message, seal, stream_tag, notification_packets, NOTIFICATION_WINDOW
 
 if platform.system() != "Darwin":
     raise SystemExit("The Swift/CryptoKit interoperability check requires macOS and Xcode.")
@@ -35,7 +36,19 @@ with tempfile.TemporaryDirectory(prefix="galaxy-interop-") as folder:
     compact_file.write_bytes(seal({"id": "compact", "session": "01" * 16, "counter": 2,
                                   "status": 200, "headers": {}, "body": base64.b64encode(raw).decode()},
                                  bytes(range(32)), "response", compact_body=True))
-    subprocess.run([str(executable), str(input_file), str(output_file), str(compact_file), str(body_file)], check=True)
+    tag = stream_tag("01" * 16, 2)
+    notification_file, ack_file = work / "notifications.bin", work / "acks.bin"
+    notifications = list(notification_packets(compact_file.read_bytes(), tag, 512))
+    notification_file.write_bytes(b"".join(struct.pack(">I", len(p)) + p for p in notifications))
+    subprocess.run([str(executable), str(input_file), str(output_file), str(compact_file), str(body_file),
+                    str(notification_file), str(ack_file)], check=True)
+    acks = ack_file.read_bytes()
+    expected = [min(start + NOTIFICATION_WINDOW, len(notifications)) - 1
+                for start in range(0, len(notifications), NOTIFICATION_WINDOW)]
+    assert len(acks) == len(expected) * 13
+    for index, sequence in enumerate(expected):
+        assert acks[index * 13:(index + 1) * 13] == b"\x04" + tag + struct.pack(">I", sequence)
+    print("Swift accepted Python notification packets and produced matching tagged window ACKs")
     response = open_message(output_file.read_bytes()[4:], bytes(range(32)), "response")
     assert response["body"] == request["body"] and response["id"] == request["id"] and response["status"] == 200
     print("Python successfully decrypted and decompressed the Swift response")

tools/galaxy_bluetooth/tests/Interop.swift

@@ -7,7 +7,9 @@ struct Interop {
         let key = SymmetricKey(data: Data((0..<32).map(UInt8.init)))
         let oldHealth = try JSONDecoder().decode(BridgeHealth.self, from: Data("{\"protocol\":1,\"transport\":\"bluetooth\"}".utf8))
         let newHealth = try JSONDecoder().decode(BridgeHealth.self, from: Data("{\"protocol\":1,\"transport\":\"bluetooth\",\"readStream\":true}".utf8))
+        let pushHealth = try JSONDecoder().decode(BridgeHealth.self, from: Data("{\"protocol\":1,\"transport\":\"bluetooth\",\"notificationStream\":true}".utf8))
         precondition(oldHealth.readStream == nil && newHealth.readStream == true)
+        precondition(oldHealth.notificationStream == nil && pushHealth.notificationStream == true)
         let input = try Data(contentsOf: URL(fileURLWithPath: CommandLine.arguments[1]))
         let request = try Wire.open(Data(input.dropFirst(4)), as: BridgeRequest.self, key: key, direction: "request")
         precondition(request.path == "/api/params")
@@ -41,6 +43,35 @@ struct Interop {
             let decoded = try Wire.open(completeCompact!, as: BridgeResponse.self, key: key, direction: "response")
             precondition(decoded.bodyData == rawBody)
         }
+        let notifications = try Data(contentsOf: URL(fileURLWithPath: CommandLine.arguments[5]))
+        let tag = try Wire.streamTag(session: String(repeating: "01", count: 16), counter: 2)
+        let otherTag = try Wire.streamTag(session: String(repeating: "01", count: 16), counter: 3)
+        var offset = 0
+        var count = 0
+        var pushAssembler = FrameAssembler()
+        var acknowledgements = Data()
+        var pushComplete: Data?
+        while offset < notifications.count {
+            let length = Int(Wire.number(notifications.subdata(in: offset..<(offset + 4))))
+            offset += 4
+            let notification = notifications.subdata(in: offset..<(offset + length))
+            offset += length
+            let unrelated = try Wire.notificationPacket(notification, tag: otherTag)
+            precondition(unrelated == nil)
+            let packet = try Wire.notificationPacket(notification, tag: tag)!
+            pushComplete = try pushAssembler.append(packet)
+            count += 1
+            if count % Wire.notificationWindow == 0 || pushComplete != nil {
+                acknowledgements.append(Wire.notificationAck(tag: tag, sequence: Wire.number(packet.subdata(in: 1..<5))))
+            }
+        }
+        let pushed = try Wire.open(pushComplete!, as: BridgeResponse.self, key: key, direction: "response")
+        precondition(pushed.bodyData == rawBody)
+        do {
+            _ = try Wire.notificationPacket(Data([3]) + tag + Data(repeating: 0, count: 4), tag: tag)
+            fatalError("Truncated notification was accepted")
+        } catch {}
+        try acknowledgements.write(to: URL(fileURLWithPath: CommandLine.arguments[6]))
 
         let valid = Data("PUT /api/params HTTP/1.1\r\nHost: 127.0.0.1:9000\r\nContent-Length: 2\r\n\r\n{}".utf8)
         let incomplete = try LocalRequest.parse(Data(valid.dropLast()))

tools/galaxy_bluetooth/tests/test_bridge.py

@@ -8,18 +8,34 @@ import zlib
 import sys
 import threading
 import unittest
+from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "bridge"))
 from dbus_next import Variant, DBusError
 from Crypto.Cipher import AES
-from protocol import Assembler, MAX_BODY, MAX_FRAME, open_message, packets, seal
+from protocol import (Assembler, MAX_BODY, MAX_FRAME, NOTIFICATION_WINDOW, open_message,
+                      packets, seal, stream_tag, notification_packets)
 from proxy import GalaxyProxy, PARAMS_SNAPSHOT_LIMIT, validate_target
-from server import Application, Characteristic, GattService, Gateway, SERVICE_PATH
+from server import Application, Characteristic, NotificationCharacteristic, GattService, Gateway, SERVICE_PATH
 
 KEY = bytes(range(32))
 
 
 class ProtocolTests(unittest.TestCase):
+    def test_notification_packet_tags_binary_and_mtu(self):
+        tag = stream_tag("01" * 16, 1)
+        self.assertNotEqual(tag, stream_tag("01" * 16, 2))
+        self.assertNotEqual(tag, stream_tag("02" * 16, 1))
+        value = {"body": base64.b64encode(bytes(range(256)) * 100).decode(), "status": 200}
+        frame = seal(value, KEY, "response", compact_body=True)
+        for size in (20, 180, 244, 512):
+            assembler = Assembler()
+            for packet in notification_packets(frame, tag, size):
+                self.assertLessEqual(len(packet), size)
+                self.assertEqual(packet[1:9], tag)
+                decoded = assembler.add(b"\x01" + packet[9:])
+            self.assertEqual(open_message(decoded, KEY, "response"), value)
+
     def test_roundtrip_all_att_sizes_and_binary_body(self):
         value = {"body": base64.b64encode(bytes(range(256)) * 32).decode(), "unicode": "Galaxy ✨", "status": 200}
         frame = seal(value, KEY, "response")
@@ -300,6 +316,144 @@ class GatewayTests(unittest.IsolatedAsyncioTestCase):
         self.assertFalse(self.state.response)
         self.assertFalse(self.state.busy)
 
+    def notifications(self):
+        notifier = NotificationCharacteristic(self.gateway)
+        sent = []
+        def changed(values, *args):
+            if "Value" in values:
+                sent.append(values["Value"])
+        notifier.emit_properties_changed = changed
+        NotificationCharacteristic.StartNotify.__wrapped__(notifier)
+        self.assertTrue(notifier.props()["Notifying"].value)
+        self.assertEqual(notifier.props()["Flags"].value, ["notify"])
+        return notifier, sent
+
+    async def wait_for_packets(self, sent, count):
+        async def wait():
+            while len(sent) < count:
+                await asyncio.sleep(0.001)
+        await asyncio.wait_for(wait(), timeout=2)
+
+    def notification_body(self):
+        body = json.dumps([{ "repeated_setting": n, "label": "Galaxy ✨" } for n in range(2000)]).encode()
+        def response(request):
+            self.proxy.calls += 1
+            return {"id": request["id"], "status": 200, "headers": {}, "body": base64.b64encode(body).decode()}
+        self.proxy.handle = response
+        return body
+
+    async def test_notifications_are_bounded_and_verified_before_final_ack(self):
+        notifier, sent = self.notifications()
+        body = self.notification_body()
+        self.options["mtu"] = Variant("q", 517)
+        await self.submit(response_codec=2, response_flow="notify-window8")
+        publisher = self.state.notification_task
+        count = len(self.state.response)
+        self.assertGreater(count, NOTIFICATION_WINDOW)
+        await self.wait_for_packets(sent, NOTIFICATION_WINDOW)
+        await asyncio.sleep(0.03)
+        self.assertEqual(len(sent), NOTIFICATION_WINDOW, "Publisher must wait for window credit")
+        tag = self.state.notification_tag
+        for bad in (b"\x04" + bytes(8) + struct.pack(">I", 7),
+                    b"\x04" + tag + struct.pack(">I", 0), b"\x02" + struct.pack(">I", 7)):
+            with self.assertRaises(DBusError):
+                self.gateway.write(bad, self.options)
+        self.assertEqual(self.gateway.read(self.options), b"\x00")
+        assembler = Assembler()
+        cursor = 0
+        while cursor < count:
+            end = min(cursor + NOTIFICATION_WINDOW, count)
+            await self.wait_for_packets(sent, end)
+            for packet in sent[cursor:end]:
+                self.assertLessEqual(len(packet), 512)
+                data = assembler.add(b"\x01" + packet[9:])
+            self.assertTrue(self.state.busy)
+            if end == count:
+                response = open_message(data, KEY, "response")
+                self.assertEqual(base64.b64decode(response["body"]), body)
+            self.gateway.write(b"\x04" + tag + struct.pack(">I", end - 1), self.options)
+            cursor = end
+        await publisher
+        self.assertFalse(self.state.busy)
+        self.assertEqual(self.proxy.calls, 1)
+        await self.submit(response_codec=2, response_flow="notify-window8")
+        self.assertEqual(self.proxy.calls, 1, "A replay cannot execute the setting change twice")
+        self.assertFalse(self.state.response)
+
+    async def test_no_notifications_or_proxy_call_before_subscription_and_authentication(self):
+        await self.submit(response_codec=2, response_flow="notify-window8")
+        self.assertEqual(self.proxy.calls, 0)
+        notifier, sent = self.notifications()
+        await self.submit(response_codec=2, response_flow="notify-window8", key=bytes(32))
+        await self.submit(response_codec=2, response_flow="notify-window8", session="00" * 16)
+        self.assertFalse(sent)
+        self.assertEqual(self.proxy.calls, 0)
+
+    async def test_notification_subscription_end_and_fresh_session_cancel_without_retry(self):
+        notifier, sent = self.notifications()
+        self.notification_body()
+        await self.submit(response_codec=2, response_flow="notify-window8")
+        publisher = self.state.notification_task
+        await self.wait_for_packets(sent, NOTIFICATION_WINDOW)
+        NotificationCharacteristic.StopNotify.__wrapped__(notifier)
+        with self.assertRaises(asyncio.CancelledError):
+            await publisher
+        self.assertTrue(self.state.busy)
+        self.assertEqual(self.proxy.calls, 1)
+        new_state = self.gateway.session(self.options, fresh=True)
+        self.assertTrue(self.state.closed)
+        self.assertFalse(new_state.busy)
+        self.assertNotEqual(self.state.challenge, new_state.challenge)
+        with self.assertRaises(DBusError):
+            self.gateway.write(b"\x04" + self.state.notification_tag + struct.pack(">I", 7), self.options)
+
+    async def test_missing_notification_ack_stops_at_one_window_without_repeating_mutation(self):
+        notifier, sent = self.notifications()
+        self.notification_body()
+        with patch("server.NOTIFICATION_ACK_TIMEOUT", 0.03):
+            await self.submit(response_codec=2, response_flow="notify-window8")
+            publisher = self.state.notification_task
+            await publisher
+        self.assertEqual(len(sent), NOTIFICATION_WINDOW)
+        self.assertTrue(self.state.busy)
+        self.assertEqual(self.proxy.calls, 1)
+        with self.assertRaises(DBusError):
+            await self.submit(counter=2, response_codec=2, response_flow="notify-window8")
+        self.assertEqual(self.proxy.calls, 1)
+
+    async def test_clients_have_distinct_tags_and_serialized_publishers(self):
+        notifier, sent = self.notifications()
+        self.notification_body()
+        await self.submit(response_codec=2, response_flow="notify-window8")
+        first_task = self.state.notification_task
+        first_count = len(self.state.response)
+        other_options = {"device": Variant("o", "/org/bluez/hci0/dev_11_12_13_14_15_16"), "mtu": Variant("q", 185)}
+        other = self.gateway.session(other_options, fresh=True)
+        request = {"id": "other", "session": other.challenge.hex(), "counter": 1, "method": "PUT", "path": "/api/params",
+                   "body": "", "responseCodec": 2, "responseFlow": "notify-window8"}
+        for packet in packets(seal(request, KEY, "request")):
+            self.gateway.write(packet, other_options)
+        await other.task
+        second_task = other.notification_task
+        self.assertNotEqual(self.state.notification_tag, other.notification_tag)
+        offset = 0
+        for state, options, count in [(self.state, self.options, first_count), (other, other_options, len(other.response))]:
+            cursor = 0
+            while cursor < count:
+                end = min(cursor + NOTIFICATION_WINDOW, count)
+                await self.wait_for_packets(sent, offset + end)
+                window = sent[offset + cursor:offset + end]
+                self.assertTrue(all(p[1:9] == state.notification_tag for p in window))
+                if state is self.state:
+                    with self.assertRaises(DBusError):
+                        self.gateway.write(b"\x04" + state.notification_tag + struct.pack(">I", end - 1), other_options)
+                self.gateway.write(b"\x04" + state.notification_tag + struct.pack(">I", end - 1), options)
+                cursor = end
+            offset += count
+        await first_task
+        await second_task
+        self.assertEqual(self.proxy.calls, 2)
+
 
 if __name__ == "__main__":
     unittest.main()