merge: A-M1c iOS SSE client + reconnect state machine

# Conflicts:
#	ios/README.md

ios/Packages/HapiKit/Package.swift

@@ -21,6 +21,7 @@ let package = Package(
         // Transport layer: APIClient, auth, SSE, stores (lands in M1).
         .target(name: "HapiClient", dependencies: ["HapiProtocol"]),
         .testTarget(name: "HapiProtocolTests", dependencies: ["HapiProtocol"]),
-        .testTarget(name: "HapiClientTests", dependencies: ["HapiClient"]),
+        // HapiProtocol is named directly by SSE tests (SyncEvent assertions).
+        .testTarget(name: "HapiClientTests", dependencies: ["HapiClient", "HapiProtocol"]),
     ]
 )

ios/Packages/HapiKit/Sources/HapiClient/SSE/NetworkPathMonitor.swift

@@ -0,0 +1,59 @@
+import Foundation
+#if canImport(Network)
+import Network
+#endif
+
+/// A snapshot of the device's network path.
+public struct NetworkPathUpdate: Equatable, Sendable {
+    /// Whether the path can carry traffic (`NWPath.Status.satisfied`).
+    public var isSatisfied: Bool
+    /// Cellular / personal-hotspot style paths (informational).
+    public var isExpensive: Bool
+
+    public init(isSatisfied: Bool, isExpensive: Bool = false) {
+        self.isSatisfied = isSatisfied
+        self.isExpensive = isExpensive
+    }
+}
+
+/// Source of network-path change notifications for `SSEClient`.
+///
+/// The stream's FIRST element is the baseline path reported on subscription
+/// (NWPathMonitor always fires once immediately); every subsequent element is
+/// an actual change. `SSEClient` skips the baseline and treats any later
+/// update while connected as a transport error — the old socket is almost
+/// certainly routed over a path that no longer exists, and reconnecting
+/// immediately beats waiting for the 90 s staleness watchdog.
+public protocol NetworkPathObserving: Sendable {
+    func pathUpdates() -> AsyncStream<NetworkPathUpdate>
+}
+
+#if canImport(Network)
+/// Production observer backed by `NWPathMonitor`.
+public struct NWPathObserver: NetworkPathObserving {
+    /// `NWPathMonitor` is not Sendable; it is confined to its own dispatch
+    /// queue and only ever touched from the update handler / termination
+    /// callback, so boxing it is safe.
+    private final class MonitorBox: @unchecked Sendable {
+        let monitor = NWPathMonitor()
+    }
+
+    public init() {}
+
+    public func pathUpdates() -> AsyncStream<NetworkPathUpdate> {
+        AsyncStream { continuation in
+            let box = MonitorBox()
+            box.monitor.pathUpdateHandler = { path in
+                continuation.yield(NetworkPathUpdate(
+                    isSatisfied: path.status == .satisfied,
+                    isExpensive: path.isExpensive
+                ))
+            }
+            box.monitor.start(queue: DispatchQueue(label: "run.hapi.sse.path-monitor"))
+            continuation.onTermination = { _ in
+                box.monitor.cancel()
+            }
+        }
+    }
+}
+#endif

ios/Packages/HapiKit/Sources/HapiClient/SSE/ReconnectPolicy.swift

@@ -0,0 +1,76 @@
+import Foundation
+
+/// Normative SSE timing constants, in milliseconds.
+///
+/// Values come from `docs/api/client-contract/sse.md` ("Reconnect policy")
+/// and the web reference client (`web/src/hooks/useSSE.ts`).
+public enum SSETimings {
+    /// The hub emits a heartbeat frame every 30 s (informational).
+    public static let heartbeatIntervalMs = 30_000
+    /// No frames of any kind for 90 s ⇒ tear down and reconnect.
+    public static let stalenessThresholdMs = 90_000
+    /// On app-foreground, reconnect immediately when the last frame is older
+    /// than this. One missed heartbeat interval after an OS suspend is
+    /// already enough to distrust the socket (no FIN/RST ever surfaces).
+    public static let foregroundResumeStalenessMs = 45_000
+    /// Staleness check interval. Checks are skipped while suspended.
+    public static let watchdogTickMs = 10_000
+    /// An attempt that has not reached OPEN within 10 s is likely hung on a
+    /// dead pooled socket — abandon it and retry on a fresh connection.
+    public static let connectTimeoutMs = 10_000
+}
+
+/// Pure backoff schedule for SSE reconnects.
+///
+/// Delay for attempt `n` (0-based, reset to 0 on every successful open):
+///
+/// - attempt 0 → 0 ms (the first retry is immediate; backoff is for
+///   *repeated* failures, not the initial recovery),
+/// - attempt n ≥ 1 → `min(cap, base × 2^(n-1))`,
+/// - cap is 30 s, widening to 300 s once `n ≥ 8` (a hub that stays
+///   unreachable is usually down for hours, and every retry through a relay
+///   costs a TLS handshake),
+/// - plus uniform jitter of 0–500 ms on every delay, immediate ones included.
+///
+/// Jitter is injectable: the generic `delayMs(forAttempt:using:)` draws from
+/// any `RandomNumberGenerator` so tests can seed it, and callers that manage
+/// randomness themselves can add `exponentialDelayMs(forAttempt:)` + their
+/// own draw from `jitterRangeMs` (this is what `SSEClient` does).
+public struct ReconnectPolicy: Equatable, Sendable {
+    public var baseDelayMs: Int
+    public var maxDelayMs: Int
+    public var slowMaxDelayMs: Int
+    public var slowAfterAttempts: Int
+    public var jitterRangeMs: ClosedRange<Int>
+
+    public init(
+        baseDelayMs: Int = 1_000,
+        maxDelayMs: Int = 30_000,
+        slowMaxDelayMs: Int = 300_000,
+        slowAfterAttempts: Int = 8,
+        jitterRangeMs: ClosedRange<Int> = 0...500
+    ) {
+        self.baseDelayMs = baseDelayMs
+        self.maxDelayMs = maxDelayMs
+        self.slowMaxDelayMs = slowMaxDelayMs
+        self.slowAfterAttempts = slowAfterAttempts
+        self.jitterRangeMs = jitterRangeMs
+    }
+
+    /// The deterministic part of the schedule, without jitter.
+    public func exponentialDelayMs(forAttempt attempt: Int) -> Int {
+        guard attempt >= 1 else {
+            return 0
+        }
+        let cap = attempt >= slowAfterAttempts ? slowMaxDelayMs : maxDelayMs
+        // Clamp the exponent so the shift can never overflow; 2^30 × base is
+        // already far past every cap.
+        let exponent = min(attempt - 1, 30)
+        return min(cap, baseDelayMs << exponent)
+    }
+
+    /// Full delay including a jitter draw from the supplied generator.
+    public func delayMs<R: RandomNumberGenerator>(forAttempt attempt: Int, using rng: inout R) -> Int {
+        exponentialDelayMs(forAttempt: attempt) + Int.random(in: jitterRangeMs, using: &rng)
+    }
+}

ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClient.swift

@@ -0,0 +1,528 @@
+import Foundation
+import HapiProtocol
+
+/// Which events this subscription receives (sse.md "Dual-subscription
+/// model"). The reference client holds one `.global` client for the whole
+/// app session plus one `.session` client per open chat.
+public enum SSEScope: Equatable, Hashable, Sendable {
+    /// `all=true` — every event in the token's namespace.
+    case global
+    /// `sessionId=<id>` — events for one session.
+    case session(String)
+}
+
+public struct SSEClientConfiguration: Sendable {
+    /// Hub origin, e.g. `https://hub.example`. The path is always replaced
+    /// with `/api/events` (mirrors `new URL(path, baseUrl)` on the web).
+    public var baseUrl: URL
+    /// Called before every connection attempt; expected to return a fresh
+    /// JWT (refreshing when its cached one is stale) or `nil` when the
+    /// client cannot authenticate right now (the attempt fails into
+    /// backoff). Re-asked once, without counting a backoff attempt, when a
+    /// connect fails with 401.
+    public var tokenProvider: @Sendable () async -> String?
+    public var scope: SSEScope
+    /// Initial `?visibility=` value: `"visible"` or `"hidden"` (anything
+    /// else reads as hidden hub-side). Later changes go through
+    /// `POST /api/visibility` with the handshake's `subscriptionId`.
+    public var visibility: String
+    /// When true, sends `Accept-Encoding: identity` so the hub skips gzip.
+    ///
+    /// TODO(gzip): sse.md requires verifying on-device that URLSession
+    /// surfaces gzip-decoded bytes incrementally (one visible frame per
+    /// server flush) rather than buffering until EOF. Until that M1
+    /// verification runs against a real hub, this flag is the escape hatch:
+    /// flip it to `true` if frames only arrive in bulk on disconnect.
+    public var acceptEncodingIdentity: Bool
+    public var reconnectPolicy: ReconnectPolicy
+    /// Jitter draw added to every backoff delay. Injectable so tests are
+    /// deterministic; defaults to uniform over the policy's jitter range.
+    public var jitterMs: @Sendable () -> Int
+
+    public init(
+        baseUrl: URL,
+        tokenProvider: @escaping @Sendable () async -> String?,
+        scope: SSEScope,
+        visibility: String = "visible",
+        acceptEncodingIdentity: Bool = false,
+        reconnectPolicy: ReconnectPolicy = ReconnectPolicy(),
+        jitterMs: (@Sendable () -> Int)? = nil
+    ) {
+        self.baseUrl = baseUrl
+        self.tokenProvider = tokenProvider
+        self.scope = scope
+        self.visibility = visibility
+        self.acceptEncodingIdentity = acceptEncodingIdentity
+        self.reconnectPolicy = reconnectPolicy
+        let range = reconnectPolicy.jitterRangeMs
+        self.jitterMs = jitterMs ?? { Int.random(in: range) }
+    }
+}
+
+public enum SSEConnectionState: Equatable, Sendable {
+    case idle
+    case connecting
+    /// Entered only on the first decoded
+    /// `connection-changed {status: "connected"}` frame — a 200 response
+    /// alone is not "up" (the hub's handshake is the source of truth).
+    case connected
+    /// Waiting `reconnectPolicy` delay before attempt `attempt` retries
+    /// (0-based; attempt 0 retries immediately, jitter only).
+    case backoff(attempt: Int)
+    /// Parked by `suspend()` with no live connection; retries are deferred
+    /// until `resume(...)`.
+    case suspended
+}
+
+public enum SSEClientEvent: Equatable, Sendable {
+    /// The hub's subscribe handshake. `resume == .ok` ⇒ the replay that
+    /// follows contains every missed event, skip the REST resync; `.gap`
+    /// (including an absent/unknown wire verdict) ⇒ full refetch.
+    /// `subscriptionId` is needed for `POST /api/visibility` and is new on
+    /// every reconnect.
+    case handshake(resume: ResumeVerdict, subscriptionId: String?)
+    /// A decoded broadcast frame, in arrival order. Heartbeats are consumed
+    /// internally (watchdog) and never surface; unknown event types surface
+    /// as `SyncEvent.unknown` and never kill the stream.
+    case event(SyncEvent)
+    case stateChanged(SSEConnectionState)
+}
+
+/// SSE subscription client: one actor instance per subscription
+/// (scope + hub + namespace). Owns the whole reconnect state machine:
+///
+///     idle → connecting → connected
+///                ↑            │ transport error / staleness / path change
+///                └─ backoff(n) ┘        (suspended parks between attempts)
+///
+/// Behavior follows `docs/api/client-contract/sse.md` and the web reference
+/// `web/src/hooks/useSSE.ts`:
+///
+/// - Handshake-gated: `.connected` only after the first decoded
+///   `connection-changed {status: "connected"}`; its `resume` verdict is
+///   surfaced as `.handshake`.
+/// - Sticky cursor: only non-empty frame `id`s move `lastEventId`; id-less
+///   heartbeats can never blank it. The cursor is scoped to this instance —
+///   a client for a different subscription must be a new instance, so a
+///   cursor can never replay against the wrong filter set.
+/// - At-least-once: the cursor advances only after the event was yielded to
+///   the consumer stream. (`AsyncStream` buffers, so "yielded" means handed
+///   to the consumer's buffer — consumers needing durable handling must
+///   drain promptly and treat redelivery as idempotent.)
+/// - Watchdog: 10 s ticks; connect attempts that have not opened within
+///   10 s are abandoned; 90 s without any bytes tears the connection down.
+///   Checks are skipped while suspended.
+/// - Backoff per `ReconnectPolicy`; the attempt counter resets on every
+///   successful open, so the first retry after a healthy connection is
+///   immediate.
+/// - `suspend()`/`resume(staleThresholdCheck:)` for app lifecycle: retries
+///   are deferred while suspended; on resume a surviving connection that
+///   has been silent ≥ 45 s is torn down and rebuilt immediately.
+/// - 401 connect failures re-ask the token provider once per connect cycle
+///   without counting a backoff attempt (the provider refreshes); a second
+///   consecutive 401 backs off normally.
+///
+/// Lifecycle: `start()` once, consume the returned stream, `stop()` when
+/// done. A stopped client is finished — build a new instance to resubscribe.
+public actor SSEClient {
+    private enum AttemptExit: Equatable, Sendable {
+        case unauthorized
+        case failed
+        case stopped
+    }
+
+    public let configuration: SSEClientConfiguration
+    private let transport: any SSETransport
+    private let clock: any SSEClock
+    private let pathObserver: (any NetworkPathObserving)?
+    private let decoder = JSONDecoder()
+
+    public private(set) var state: SSEConnectionState = .idle
+    /// Sticky resume cursor for THIS subscription (nil until the first
+    /// id-carrying frame). Callers may persist it keyed by
+    /// (hub, namespace, filter set) and seed a future instance via
+    /// `seedCursor(_:)` before `start()`.
+    public private(set) var lastEventId: String?
+    /// From the latest handshake; new on every reconnect. Needed for
+    /// `POST /api/visibility`.
+    public private(set) var subscriptionId: String?
+
+    private var continuation: AsyncStream<SSEClientEvent>.Continuation?
+    private var consumerGeneration = 0
+    private var runTask: Task<Void, Never>?
+    private var pathTask: Task<Void, Never>?
+    private var currentAttemptTask: Task<AttemptExit, Never>?
+    private var sleepTask: Task<Void, Never>?
+    private var resumeWaiters: [CheckedContinuation<Void, Never>] = []
+
+    private var stopped = false
+    private var suspended = false
+    private var connectAttempt = 0
+    private var authRetryUsed = false
+
+    // Per-attempt bookkeeping, read by the watchdog and lifecycle checks.
+    // Internal (not private) so @testable tests can synchronize on them.
+    private(set) var transportOpen = false
+    private(set) var handshakeSeen = false
+    private(set) var connectStartedAtMs = 0
+    private(set) var lastActivityAtMs = 0
+
+    public init(
+        configuration: SSEClientConfiguration,
+        transport: any SSETransport = URLSessionSSETransport(),
+        clock: any SSEClock = SystemSSEClock(),
+        pathObserver: (any NetworkPathObserving)? = nil
+    ) {
+        self.configuration = configuration
+        self.transport = transport
+        self.clock = clock
+        self.pathObserver = pathObserver
+    }
+
+    // MARK: - Lifecycle
+
+    /// Starts the connection loop (first call) and returns the event stream.
+    /// Calling again replaces the consumer stream; calling after `stop()`
+    /// returns an immediately finished stream.
+    public func start() -> AsyncStream<SSEClientEvent> {
+        let (stream, continuation) = AsyncStream.makeStream(of: SSEClientEvent.self)
+        if stopped {
+            continuation.finish()
+            return stream
+        }
+        self.continuation?.finish()
+        self.continuation = continuation
+        consumerGeneration += 1
+        let generation = consumerGeneration
+        continuation.onTermination = { [weak self] _ in
+            // Nobody is listening anymore: shut the connection down (guarded
+            // by generation so replacing the consumer does not stop us).
+            Task { await self?.consumerFinished(generation: generation) }
+        }
+        if runTask == nil {
+            runTask = Task { await self.run() }
+            if let pathObserver {
+                pathTask = Task { await self.observePath(pathObserver) }
+            }
+        }
+        return stream
+    }
+
+    /// Permanently stops the client: tears down the connection, cancels any
+    /// pending retry, and finishes the event stream. Idempotent.
+    public func stop() {
+        stopped = true
+        suspended = false
+        drainResumeWaiters()
+        sleepTask?.cancel()
+        currentAttemptTask?.cancel()
+        pathTask?.cancel()
+        pathTask = nil
+        runTask?.cancel()
+        setState(.idle)
+        finishStream()
+    }
+
+    /// App went to background: defer retries. A live connection is kept
+    /// (the OS may or may not preserve it); watchdog checks pause so a
+    /// frozen app cannot mis-detect staleness on wake.
+    public func suspend() {
+        guard !stopped, !suspended else { return }
+        suspended = true
+        // A pending backoff sleep is abandoned; the run loop parks at the
+        // top until resume() and then retries immediately.
+        sleepTask?.cancel()
+    }
+
+    /// App returned to foreground. A parked retry runs immediately. When a
+    /// connection survived suspension, `staleThresholdCheck` distrusts it if
+    /// it has been silent for ≥ 45 s (`SSETimings.foregroundResumeStalenessMs`)
+    /// — an OS suspend can kill the socket without any error surfacing — and
+    /// reconnects immediately.
+    public func resume(staleThresholdCheck: Bool = true) {
+        guard !stopped, suspended else { return }
+        suspended = false
+        drainResumeWaiters()
+        if staleThresholdCheck, transportOpen,
+           clock.nowMs() - lastActivityAtMs >= SSETimings.foregroundResumeStalenessMs {
+            currentAttemptTask?.cancel()
+        }
+    }
+
+    /// Seeds the resume cursor before `start()` (e.g. one persisted from a
+    /// previous process, keyed to this same subscription filter set — never
+    /// seed a cursor recorded under a different scope/hub/namespace).
+    public func seedCursor(_ id: String?) {
+        guard runTask == nil, !stopped else { return }
+        lastEventId = id
+    }
+
+    private func consumerFinished(generation: Int) {
+        guard generation == consumerGeneration else { return }
+        stop()
+    }
+
+    private func finishStream() {
+        continuation?.finish()
+        continuation = nil
+    }
+
+    // MARK: - Run loop
+
+    private func run() async {
+        while !stopped {
+            if suspended {
+                await parkUntilResumed()
+                continue
+            }
+            setState(.connecting)
+            let token = await configuration.tokenProvider()
+            if stopped { break }
+            if suspended { continue }
+            guard let token, let request = makeRequest(token: token) else {
+                // Cannot authenticate right now; the provider may recover
+                // (refresh in flight, credentials being restored) — retry on
+                // the normal schedule.
+                await scheduleBackoff()
+                continue
+            }
+            let exit = await performAttempt(request: request)
+            if stopped { break }
+            if exit == .stopped { continue } // loop condition exits
+            if exit == .unauthorized, !authRetryUsed {
+                // The token was rejected: ask the provider again (it
+                // refreshes) and retry immediately. Not a backoff attempt,
+                // and capped at once per connect cycle so a hub that keeps
+                // 401-ing still backs off.
+                authRetryUsed = true
+                continue
+            }
+            if suspended { continue } // parks at loop top; retry deferred
+            await scheduleBackoff()
+        }
+        setState(.idle)
+        finishStream()
+    }
+
+    private func parkUntilResumed() async {
+        setState(.suspended)
+        while suspended, !stopped {
+            await withCheckedContinuation { (waiter: CheckedContinuation<Void, Never>) in
+                if !suspended || stopped {
+                    waiter.resume()
+                } else {
+                    resumeWaiters.append(waiter)
+                }
+            }
+        }
+    }
+
+    private func drainResumeWaiters() {
+        let waiters = resumeWaiters
+        resumeWaiters.removeAll()
+        for waiter in waiters {
+            waiter.resume()
+        }
+    }
+
+    private func scheduleBackoff() async {
+        let attempt = connectAttempt
+        connectAttempt += 1
+        setState(.backoff(attempt: attempt))
+        let delay = configuration.reconnectPolicy.exponentialDelayMs(forAttempt: attempt)
+            + max(0, configuration.jitterMs())
+        await backoffSleep(ms: delay)
+    }
+
+    private func backoffSleep(ms: Int) async {
+        guard ms > 0, !suspended, !stopped else { return }
+        let clock = self.clock
+        let task = Task { try? await clock.sleep(ms: ms) }
+        sleepTask = task
+        await task.value
+        sleepTask = nil
+    }
+
+    // MARK: - One connection attempt
+
+    private func performAttempt(request: URLRequest) async -> AttemptExit {
+        connectStartedAtMs = clock.nowMs()
+        lastActivityAtMs = connectStartedAtMs
+        transportOpen = false
+        handshakeSeen = false
+        let attemptTask = Task { await self.readAttempt(request: request) }
+        currentAttemptTask = attemptTask
+        let watchdogTask = Task { await self.runWatchdog(killing: attemptTask) }
+        let exit = await attemptTask.value
+        watchdogTask.cancel()
+        currentAttemptTask = nil
+        transportOpen = false
+        handshakeSeen = false
+        return exit
+    }
+
+    private func readAttempt(request: URLRequest) async -> AttemptExit {
+        let stream = transport.connect(request)
+        var parser = SSELineParser()
+        do {
+            for try await transportEvent in stream {
+                if Task.isCancelled { return cancellationExit() }
+                switch transportEvent {
+                case .connected(let response):
+                    let status = response.statusCode
+                    if status == 401 { return .unauthorized }
+                    guard (200...299).contains(status) else { return .failed }
+                    transportOpen = true
+                    // Mirrors the web client's onopen: the backoff counter
+                    // resets on every successful open, and the one free
+                    // auth retry re-arms for the next cycle.
+                    connectAttempt = 0
+                    authRetryUsed = false
+                    lastActivityAtMs = clock.nowMs()
+                case .bytes(let data):
+                    // Any received bytes — heartbeats, comments, partial
+                    // frames — count as activity for the staleness clock.
+                    lastActivityAtMs = clock.nowMs()
+                    for frame in parser.consume(data) {
+                        handleFrame(frame)
+                    }
+                }
+            }
+            // Server closed the stream (or iteration ended on cancellation).
+            return Task.isCancelled ? cancellationExit() : .failed
+        } catch {
+            if Task.isCancelled || error is CancellationError {
+                return cancellationExit()
+            }
+            return .failed
+        }
+    }
+
+    private func cancellationExit() -> AttemptExit {
+        stopped ? .stopped : .failed
+    }
+
+    private func handleFrame(_ frame: SSEFrame) {
+        let decoded: SyncEvent
+        do {
+            decoded = try decoder.decode(SyncEvent.self, from: Data(frame.data.utf8))
+        } catch {
+            // Known event type with a malformed payload (unknown types decode
+            // to .unknown and never throw). Mirror the web client: drop the
+            // frame, keep the stream alive, and leave the cursor BEHIND it —
+            // never advance past something that was not handled.
+            return
+        }
+
+        switch decoded {
+        case .heartbeat:
+            // Pure liveness; activity was already stamped per-chunk. Carries
+            // no id, so the sticky cursor is untouched by construction.
+            return
+        case .connectionChanged(_, let payload):
+            if !handshakeSeen, payload?.status == "connected" {
+                handshakeSeen = true
+                subscriptionId = payload?.subscriptionId
+                setState(.connected)
+                // Absent verdict = older hub = cannot prove continuity: gap.
+                continuation?.yield(.handshake(
+                    resume: payload?.resume ?? .gap,
+                    subscriptionId: payload?.subscriptionId
+                ))
+            } else {
+                continuation?.yield(.event(decoded))
+            }
+        default:
+            continuation?.yield(.event(decoded))
+        }
+
+        // At-least-once: the cursor moves only after the yield above, and
+        // only for frames that actually carried an id (sticky otherwise).
+        if let id = frame.id, !id.isEmpty {
+            lastEventId = id
+        }
+    }
+
+    private func runWatchdog(killing attemptTask: Task<AttemptExit, Never>) async {
+        while !Task.isCancelled {
+            try? await clock.sleep(ms: SSETimings.watchdogTickMs)
+            if Task.isCancelled { return }
+            if suspended { continue } // checks are skipped while backgrounded
+            let now = clock.nowMs()
+            if !transportOpen {
+                // Likely hung on a dead pooled socket; abandon the attempt
+                // and retry on a fresh connection.
+                if now - connectStartedAtMs >= SSETimings.connectTimeoutMs {
+                    attemptTask.cancel()
+                    return
+                }
+            } else if now - lastActivityAtMs >= SSETimings.stalenessThresholdMs {
+                attemptTask.cancel()
+                return
+            }
+        }
+    }
+
+    private func observePath(_ observer: any NetworkPathObserving) async {
+        // The first element is the baseline emitted on subscription, not a
+        // change — see NetworkPathObserving.
+        var isBaseline = true
+        for await _ in observer.pathUpdates() {
+            if Task.isCancelled { return }
+            if isBaseline {
+                isBaseline = false
+                continue
+            }
+            // A path change while connected: the socket is almost certainly
+            // bound to a route that no longer exists. Treat it as a transport
+            // error; the attempt counter was reset on open, so the reconnect
+            // is immediate.
+            if !stopped, !suspended, transportOpen {
+                currentAttemptTask?.cancel()
+            }
+        }
+    }
+
+    // MARK: - Request building
+
+    private func makeRequest(token: String) -> URLRequest? {
+        guard var components = URLComponents(url: configuration.baseUrl, resolvingAgainstBaseURL: false) else {
+            return nil
+        }
+        components.path = "/api/events"
+        // Query-param auth, exactly like the web reference (buildEventsUrl):
+        // token, visibility, all|sessionId, lastEventId.
+        var items = [
+            URLQueryItem(name: "token", value: token),
+            URLQueryItem(name: "visibility", value: configuration.visibility),
+        ]
+        switch configuration.scope {
+        case .global:
+            items.append(URLQueryItem(name: "all", value: "true"))
+        case .session(let sessionId):
+            items.append(URLQueryItem(name: "sessionId", value: sessionId))
+        }
+        if let lastEventId, !lastEventId.isEmpty {
+            items.append(URLQueryItem(name: "lastEventId", value: lastEventId))
+        }
+        components.queryItems = items
+        guard let url = components.url else { return nil }
+        var request = URLRequest(url: url)
+        request.httpMethod = "GET"
+        request.setValue("text/event-stream", forHTTPHeaderField: "Accept")
+        request.setValue("no-cache", forHTTPHeaderField: "Cache-Control")
+        if configuration.acceptEncodingIdentity {
+            // See SSEClientConfiguration.acceptEncodingIdentity (gzip TODO).
+            request.setValue("identity", forHTTPHeaderField: "Accept-Encoding")
+        }
+        return request
+    }
+
+    // MARK: - State
+
+    private func setState(_ newState: SSEConnectionState) {
+        guard state != newState else { return }
+        state = newState
+        continuation?.yield(.stateChanged(newState))
+    }
+}

ios/Packages/HapiKit/Sources/HapiClient/SSE/SSEClock.swift

@@ -0,0 +1,33 @@
+import Dispatch
+import Foundation
+
+/// Minimal time source used by `SSEClient` for the connect timeout, the
+/// staleness watchdog, and backoff sleeps.
+///
+/// Deliberately NOT the stdlib `Clock` protocol: tests need a manually
+/// advanced clock whose `sleep` parks on a continuation until test code
+/// moves time forward, and the stdlib protocol's associated types and
+/// tolerance machinery buy nothing here.
+public protocol SSEClock: Sendable {
+    /// Current time in milliseconds on a monotonically increasing scale.
+    /// Only ever used for differences; the origin is arbitrary.
+    func nowMs() -> Int
+    /// Suspends for `ms`. Must throw `CancellationError` promptly when the
+    /// surrounding task is cancelled — the client relies on it to abort
+    /// watchdog ticks and pending backoff sleeps.
+    func sleep(ms: Int) async throws
+}
+
+/// Production clock: monotonic uptime + `Task.sleep`.
+public struct SystemSSEClock: SSEClock {
+    public init() {}
+
+    public func nowMs() -> Int {
+        Int(DispatchTime.now().uptimeNanoseconds / 1_000_000)
+    }
+
+    public func sleep(ms: Int) async throws {
+        guard ms > 0 else { return }
+        try await Task.sleep(nanoseconds: UInt64(ms) * 1_000_000)
+    }
+}

ios/Packages/HapiKit/Sources/HapiClient/SSE/SSELineParser.swift

@@ -0,0 +1,115 @@
+import Foundation
+
+/// One dispatched SSE message.
+///
+/// `id` is the value of the `id:` field seen in THIS event block only — `nil`
+/// when the block carried no `id:` line (heartbeat / connection-changed /
+/// toast frames). Cursor stickiness is deliberately NOT the parser's job: the
+/// caller keeps its last-seen id and lets only non-nil (and non-empty) frame
+/// ids overwrite it, so an id-less heartbeat can never reset the cursor.
+public struct SSEFrame: Equatable, Sendable {
+    public var id: String?
+    /// Value of the `event:` field, `nil` for default `message` frames.
+    /// The hub only ever sends unnamed frames; kept for completeness.
+    public var event: String?
+    /// All `data:` lines of the block joined with `\n`.
+    public var data: String
+
+    public init(id: String? = nil, event: String? = nil, data: String) {
+        self.id = id
+        self.event = event
+        self.data = data
+    }
+}
+
+/// Incremental parser for a `text/event-stream` byte stream.
+///
+/// Feed it raw chunks as they arrive (`consume`); it buffers partial lines
+/// across chunk boundaries, understands LF and CRLF line endings, and emits
+/// an `SSEFrame` for every blank-line dispatch that accumulated at least one
+/// `data:` line. Field handling follows the WHATWG EventSource algorithm:
+///
+/// - `data:` — multi-line, joined with `\n` at dispatch.
+/// - `id:` — sets the block id (ignored when it contains U+0000, per spec).
+///   Like EventSource's last-event-id buffer, an id from a block that never
+///   dispatched (no data lines) carries over into the next block; a real
+///   dispatch consumes it.
+/// - `event:` — sets the frame's event name.
+/// - `retry:` — ignored; reconnect policy is entirely client-owned (sse.md).
+/// - `: comment` lines and unknown fields — ignored.
+///
+/// Not handled (the hub never produces them): lone-CR line endings and a
+/// leading UTF-8 BOM. Create a fresh parser per connection; state must never
+/// bleed across reconnects.
+public struct SSELineParser: Sendable {
+    private var pendingLine = Data()
+    private var dataLines: [String] = []
+    private var eventType: String?
+    private var eventId: String?
+
+    public init() {}
+
+    /// Consumes one chunk and returns every frame completed by it, in order.
+    public mutating func consume(_ chunk: Data) -> [SSEFrame] {
+        var frames: [SSEFrame] = []
+        for byte in chunk {
+            if byte == 0x0A { // LF terminates a line; strip an optional CR.
+                if pendingLine.last == 0x0D {
+                    pendingLine.removeLast()
+                }
+                process(line: pendingLine, into: &frames)
+                pendingLine.removeAll(keepingCapacity: true)
+            } else {
+                pendingLine.append(byte)
+            }
+        }
+        return frames
+    }
+
+    private mutating func process(line: Data, into frames: inout [SSEFrame]) {
+        if line.isEmpty {
+            // Blank line: dispatch. Per spec, no data lines means no event —
+            // the event type buffer resets, but the id buffer survives so a
+            // (hypothetical) id-only block still tags the next dispatch.
+            if !dataLines.isEmpty {
+                frames.append(SSEFrame(id: eventId, event: eventType, data: dataLines.joined(separator: "\n")))
+                dataLines.removeAll(keepingCapacity: true)
+                eventId = nil
+            }
+            eventType = nil
+            return
+        }
+        if line.first == 0x3A { // ':' — comment line.
+            return
+        }
+
+        let field: String
+        let value: String
+        if let colonIndex = line.firstIndex(of: 0x3A) {
+            field = String(decoding: line[line.startIndex..<colonIndex], as: UTF8.self)
+            var valueBytes = line[line.index(after: colonIndex)...]
+            if valueBytes.first == 0x20 { // A single leading space is stripped.
+                valueBytes = valueBytes.dropFirst()
+            }
+            value = String(decoding: valueBytes, as: UTF8.self)
+        } else {
+            field = String(decoding: line, as: UTF8.self)
+            value = ""
+        }
+
+        switch field {
+        case "data":
+            dataLines.append(value)
+        case "event":
+            eventType = value
+        case "id":
+            if !value.contains("\u{0000}") {
+                eventId = value
+            }
+        case "retry":
+            break // Server-suggested delays are ignored; policy is ours.
+        default:
+            break // Unknown fields are ignored per spec.
+        }
+    }
+}

ios/Packages/HapiKit/Sources/HapiClient/SSE/SSETransport.swift

@@ -0,0 +1,93 @@
+import Foundation
+
+/// What a transport reports while a single SSE connection attempt is alive:
+/// exactly one `.connected` first (the HTTP response headers), then raw body
+/// chunks. The stream finishes when the server closes the connection and
+/// throws on transport errors.
+public enum SSETransportEvent: Sendable {
+    case connected(HTTPURLResponse)
+    case bytes(Data)
+}
+
+public enum SSETransportError: Error, Sendable {
+    /// The response was not HTTP at all (never happens against a hub).
+    case notHTTP
+}
+
+/// One-shot connection factory. `SSEClient` calls `connect` once per attempt
+/// and drives every retry itself; a transport must never auto-reconnect.
+/// Cancelling the consuming task (or dropping the iterator) must tear the
+/// underlying connection down.
+public protocol SSETransport: Sendable {
+    func connect(_ request: URLRequest) -> AsyncThrowingStream<SSETransportEvent, Error>
+}
+
+/// Production transport on `URLSession.bytes(for:)`.
+///
+/// Uses a dedicated session configuration because the defaults are wrong for
+/// a long-lived stream that manages its own retries:
+///
+/// - `timeoutIntervalForRequest` 300 s — this is URLSession's *idle* timeout
+///   (time between chunks). It must exceed the 30 s heartbeat interval by a
+///   wide margin (≥ 120 s required; the 90 s staleness watchdog in
+///   `SSEClient` fires long before it anyway).
+/// - `timeoutIntervalForResource` 7 days — the stream is expected to live
+///   "forever"; the default 7-minute-per-resource cap would kill it.
+/// - `waitsForConnectivity` false — a dead link must surface as an error
+///   immediately so the reconnect state machine owns the schedule.
+/// - No caching, ephemeral storage — an event stream must never be served
+///   from or written to a cache.
+public struct URLSessionSSETransport: SSETransport {
+    private let session: URLSession
+
+    public init(session: URLSession = URLSessionSSETransport.makeSession()) {
+        self.session = session
+    }
+
+    public static func makeSession() -> URLSession {
+        let configuration = URLSessionConfiguration.ephemeral
+        configuration.timeoutIntervalForRequest = 300
+        configuration.timeoutIntervalForResource = 7 * 24 * 60 * 60
+        configuration.waitsForConnectivity = false
+        configuration.requestCachePolicy = .reloadIgnoringLocalCacheData
+        configuration.urlCache = nil
+        configuration.httpCookieStorage = nil
+        return URLSession(configuration: configuration)
+    }
+
+    public func connect(_ request: URLRequest) -> AsyncThrowingStream<SSETransportEvent, Error> {
+        let session = self.session
+        return AsyncThrowingStream { continuation in
+            let task = Task {
+                do {
+                    let (bytes, response) = try await session.bytes(for: request)
+                    guard let http = response as? HTTPURLResponse else {
+                        throw SSETransportError.notHTTP
+                    }
+                    continuation.yield(.connected(http))
+                    // AsyncBytes iterates single bytes (buffered internally);
+                    // coalesce to one chunk per line so the parser is called
+                    // O(lines), not O(bytes).
+                    var chunk = Data()
+                    chunk.reserveCapacity(1024)
+                    for try await byte in bytes {
+                        chunk.append(byte)
+                        if byte == 0x0A {
+                            continuation.yield(.bytes(chunk))
+                            chunk.removeAll(keepingCapacity: true)
+                        }
+                    }
+                    if !chunk.isEmpty {
+                        continuation.yield(.bytes(chunk))
+                    }
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
+                }
+            }
+            continuation.onTermination = { _ in
+                task.cancel()
+            }
+        }
+    }
+}

ios/Packages/HapiKit/Tests/HapiClientTests/SSE/ReconnectPolicyTests.swift

@@ -0,0 +1,93 @@
+import Foundation
+import Testing
+@testable import HapiClient
+
+/// Deterministic seedable RNG (SplitMix64) for jitter tests.
+struct SplitMix64: RandomNumberGenerator {
+    private var state: UInt64
+
+    init(seed: UInt64) {
+        state = seed
+    }
+
+    mutating func next() -> UInt64 {
+        state &+= 0x9E37_79B9_7F4A_7C15
+        var z = state
+        z = (z ^ (z >> 30)) &* 0xBF58_476D_1CE4_E5B9
+        z = (z ^ (z >> 27)) &* 0x94D0_49BB_1331_11EB
+        return z ^ (z >> 31)
+    }
+}
+
+@Suite("Reconnect policy")
+struct ReconnectPolicyTests {
+    @Test func exactScheduleIncludingSlowCeiling() {
+        let policy = ReconnectPolicy()
+        let expected = [
+            0: 0, // first retry is immediate (jitter only)
+            1: 1_000,
+            2: 2_000,
+            3: 4_000,
+            4: 8_000,
+            5: 16_000,
+            6: 30_000, // 32 s capped at 30 s
+            7: 30_000, // 64 s capped at 30 s
+            8: 128_000, // slow ceiling (300 s) active, exponential resumes
+            9: 256_000,
+            10: 300_000, // 512 s capped at the slow ceiling
+            11: 300_000,
+            25: 300_000,
+        ]
+        for (attempt, delay) in expected {
+            #expect(policy.exponentialDelayMs(forAttempt: attempt) == delay, "attempt \(attempt)")
+        }
+    }
+
+    @Test func hugeAttemptCountsDoNotOverflow() {
+        let policy = ReconnectPolicy()
+        #expect(policy.exponentialDelayMs(forAttempt: 100) == 300_000)
+        #expect(policy.exponentialDelayMs(forAttempt: Int(Int32.max)) == 300_000)
+    }
+
+    @Test func negativeAttemptTreatedAsImmediate() {
+        #expect(ReconnectPolicy().exponentialDelayMs(forAttempt: -3) == 0)
+    }
+
+    @Test func jitterStaysInsideBoundsWithSeededRNG() {
+        let policy = ReconnectPolicy()
+        var rng = SplitMix64(seed: 0xDEAD_BEEF)
+        for attempt in 0...20 {
+            for _ in 0..<50 {
+                let base = policy.exponentialDelayMs(forAttempt: attempt)
+                let full = policy.delayMs(forAttempt: attempt, using: &rng)
+                let jitter = full - base
+                #expect(jitter >= 0 && jitter <= 500, "attempt \(attempt) jitter \(jitter)")
+            }
+        }
+    }
+
+    @Test func seededJitterIsDeterministic() {
+        let policy = ReconnectPolicy()
+        var first = SplitMix64(seed: 42)
+        var second = SplitMix64(seed: 42)
+        let a = (0..<20).map { _ in policy.delayMs(forAttempt: 3, using: &first) }
+        let b = (0..<20).map { _ in policy.delayMs(forAttempt: 3, using: &second) }
+        #expect(a == b)
+        // The draws themselves vary (jitter is actually applied).
+        #expect(Set(a).count > 1)
+    }
+
+    @Test func timingConstantsMatchContract() {
+        #expect(SSETimings.heartbeatIntervalMs == 30_000)
+        #expect(SSETimings.stalenessThresholdMs == 90_000)
+        #expect(SSETimings.foregroundResumeStalenessMs == 45_000)
+        #expect(SSETimings.watchdogTickMs == 10_000)
+        #expect(SSETimings.connectTimeoutMs == 10_000)
+        let policy = ReconnectPolicy()
+        #expect(policy.baseDelayMs == 1_000)
+        #expect(policy.maxDelayMs == 30_000)
+        #expect(policy.slowMaxDelayMs == 300_000)
+        #expect(policy.slowAfterAttempts == 8)
+        #expect(policy.jitterRangeMs == 0...500)
+    }
+}

ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSEClientTests.swift

@@ -0,0 +1,419 @@
+import Foundation
+import HapiProtocol
+import Testing
+@testable import HapiClient
+
+/// State-machine tests for `SSEClient`, all against a fake transport and a
+/// manually advanced clock (no real timers, no sockets). Real-time waits
+/// appear only in bounded polls that synchronize with the actor's async
+/// processing.
+@Suite("SSEClient state machine")
+struct SSEClientTests {
+    typealias Harness = (
+        client: SSEClient,
+        transport: FakeTransport,
+        clock: ManualClock,
+        tokens: TokenSource,
+        box: EventBox
+    )
+
+    private func makeHarness(
+        scope: SSEScope = .global,
+        visibility: String = "visible",
+        tokens: [String?] = ["tok-1"],
+        pathObserver: FakePathObserver? = nil,
+        transport: FakeTransport = FakeTransport(),
+        clock: ManualClock = ManualClock()
+    ) async -> Harness {
+        let tokenSource = TokenSource(tokens)
+        let configuration = SSEClientConfiguration(
+            baseUrl: URL(string: "https://hub.test")!,
+            tokenProvider: { tokenSource.next() },
+            scope: scope,
+            visibility: visibility,
+            jitterMs: { 0 }
+        )
+        let client = SSEClient(
+            configuration: configuration,
+            transport: transport,
+            clock: clock,
+            pathObserver: pathObserver
+        )
+        let stream = await client.start()
+        let (box, _) = collect(stream)
+        return (client, transport, clock, tokenSource, box)
+    }
+
+    /// Advances the manual clock once the watchdog (or any sleeper) has
+    /// actually parked — the watchdog registers its sleep asynchronously, so
+    /// blind advances could set a deadline the test never crosses.
+    private func advanceOnceSleeperParked(_ clock: ManualClock, byMs delta: Int) async {
+        #expect(await waitUntil { clock.activeSleeperCount() >= 1 })
+        clock.advance(byMs: delta)
+    }
+
+    // MARK: - Handshake
+
+    @Test func handshakeGatesConnectedState() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        // Query-param auth, exactly like the web reference client.
+        #expect(connection.queryItems["token"] == "tok-1")
+        #expect(connection.queryItems["visibility"] == "visible")
+        #expect(connection.queryItems["all"] == "true")
+        #expect(connection.queryItems["sessionId"] == nil)
+        #expect(connection.queryItems["lastEventId"] == nil)
+        #expect(connection.request.value(forHTTPHeaderField: "Accept") == "text/event-stream")
+        #expect(connection.request.value(forHTTPHeaderField: "Accept-Encoding") == nil)
+
+        h.clock.advance(byMs: 5)
+        connection.open()
+        #expect(await waitUntil { await h.client.transportOpen })
+        // A 200 alone is not "up": the state machine stays .connecting until
+        // the hub's connection-changed handshake, heartbeats included.
+        h.clock.advance(byMs: 1)
+        connection.sendHeartbeat()
+        #expect(await waitUntil { await h.client.lastActivityAtMs == 6 })
+        #expect(await h.client.state == .connecting)
+        #expect(h.box.handshakes.isEmpty)
+        #expect(!h.box.states.contains(.connected))
+
+        connection.sendHandshake(resume: "ok", subscriptionId: "sub-9")
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+        #expect(h.box.handshakes == [.handshake(resume: .ok, subscriptionId: "sub-9")])
+        #expect(await h.client.state == .connected)
+        #expect(await h.client.subscriptionId == "sub-9")
+        // stateChanged(.connected) precedes the handshake yield.
+        let events = h.box.snapshot
+        let connectedIndex = try #require(events.firstIndex(of: .stateChanged(.connected)))
+        let handshakeIndex = try #require(events.firstIndex(of: .handshake(resume: .ok, subscriptionId: "sub-9")))
+        #expect(connectedIndex < handshakeIndex)
+        await h.client.stop()
+    }
+
+    @Test func resumeVerdictSurfacesGapWhenAbsent() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        connection.open()
+        connection.sendHandshake(resume: nil, subscriptionId: "sub-2")
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+        // Older hubs omit the verdict; absence must read as gap (full resync).
+        #expect(h.box.handshakes == [.handshake(resume: .gap, subscriptionId: "sub-2")])
+        await h.client.stop()
+    }
+
+    // MARK: - Cursor
+
+    @Test func cursorSentOnReconnectAndIsolatedPerSubscription() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        first.sendEvent(
+            "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-9\"}",
+            id: "018f:41:aa"
+        )
+        #expect(await waitUntil { !h.box.syncEvents.isEmpty })
+        #expect(h.box.syncEvents == [.sessionRemoved(namespace: "ns", sessionId: "s-9")])
+        #expect(await h.client.lastEventId == "018f:41:aa")
+
+        // Server closes; attempt counter was reset on open, so the reconnect
+        // is immediate and replays the cursor of THIS subscription.
+        first.finish()
+        let second = try #require(await h.transport.nextConnection())
+        #expect(second.queryItems["lastEventId"] == "018f:41:aa")
+        #expect(second.queryItems["all"] == "true")
+
+        // A different subscription must never inherit that cursor: fresh
+        // client, session scope, same hub.
+        let other = await makeHarness(
+            scope: .session("s-42"),
+            visibility: "hidden",
+            tokens: ["tok-B"],
+            transport: h.transport,
+            clock: h.clock
+        )
+        let otherConnection = try #require(await other.transport.nextConnection())
+        #expect(otherConnection.queryItems["sessionId"] == "s-42")
+        #expect(otherConnection.queryItems["all"] == nil)
+        #expect(otherConnection.queryItems["visibility"] == "hidden")
+        #expect(otherConnection.queryItems["token"] == "tok-B")
+        #expect(otherConnection.queryItems["lastEventId"] == nil)
+        await h.client.stop()
+        await other.client.stop()
+    }
+
+    @Test func heartbeatKeepsWatchdogAliveButDoesNotMoveCursor() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        connection.open()
+        connection.sendHandshake()
+        connection.sendEvent(
+            "{\"type\":\"scheduled-matured\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}",
+            id: "018f:7:aa"
+        )
+        #expect(await waitUntil { await h.client.lastEventId == "018f:7:aa" })
+
+        // 50 s of silence, then a heartbeat: watchdog stays quiet and the
+        // cursor keeps the last real id (heartbeat frames carry no id).
+        await advanceOnceSleeperParked(h.clock, byMs: 50_000)
+        connection.sendHeartbeat()
+        #expect(await waitUntil { await h.client.lastActivityAtMs == 50_000 })
+        await advanceOnceSleeperParked(h.clock, byMs: 50_000) // 100 s total, only 50 s since the heartbeat
+        #expect(await h.transport.expectNoConnection())
+        #expect(h.transport.totalConnections() == 1)
+        #expect(await h.client.state == .connected)
+        #expect(await h.client.lastEventId == "018f:7:aa")
+
+        // Full silence from here: the 90 s threshold tears it down and the
+        // reconnect replays the heartbeat-untouched cursor.
+        await advanceOnceSleeperParked(h.clock, byMs: 90_000)
+        let second = try #require(await h.transport.nextConnection())
+        #expect(second.queryItems["lastEventId"] == "018f:7:aa")
+        #expect(connection.isTerminated)
+        await h.client.stop()
+    }
+
+    // MARK: - Watchdog
+
+    @Test func ninetySecondsOfSilenceTriggersReconnect() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+        await advanceOnceSleeperParked(h.clock, byMs: 90_000)
+        _ = try #require(await h.transport.nextConnection())
+        #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) })
+        #expect(first.isTerminated)
+        await h.client.stop()
+    }
+
+    @Test func hungConnectAttemptIsAbandonedAfterTimeout() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        // Never reaches OPEN; the 10 s connect deadline abandons it.
+        await advanceOnceSleeperParked(h.clock, byMs: 10_000)
+        _ = try #require(await h.transport.nextConnection())
+        #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) })
+        #expect(first.isTerminated)
+        await h.client.stop()
+    }
+
+    // MARK: - Suspend / resume
+
+    @Test func suspendDefersRetriesAndResumeRunsThemImmediately() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.finish() // attempt 0 fails → immediate retry
+        let second = try #require(await h.transport.nextConnection())
+        second.finish() // attempt 1 → 1 s backoff
+        #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 1)) })
+
+        await h.client.suspend()
+        #expect(await waitUntil { await h.client.state == .suspended })
+
+        // Arbitrary time passes; a suspended client schedules nothing.
+        h.clock.advance(byMs: 600_000)
+        #expect(await h.transport.expectNoConnection())
+        #expect(h.transport.totalConnections() == 2)
+
+        await h.client.resume()
+        _ = try #require(await h.transport.nextConnection())
+        #expect(h.transport.totalConnections() == 3)
+        await h.client.stop()
+    }
+
+    @Test func resumeWithStaleConnectionReconnectsImmediately() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+
+        await h.client.suspend()
+        // 50 s pass while suspended — beyond the 45 s foreground threshold.
+        // The watchdog itself must stay quiet (checks skip while suspended).
+        h.clock.advance(byMs: 50_000)
+        #expect(await h.transport.expectNoConnection())
+
+        await h.client.resume()
+        _ = try #require(await h.transport.nextConnection())
+        #expect(first.isTerminated)
+        await h.client.stop()
+    }
+
+    @Test func resumeWithFreshConnectionKeepsIt() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+
+        await h.client.suspend()
+        h.clock.advance(byMs: 10_000) // well under the 45 s threshold
+        await h.client.resume()
+        #expect(await h.transport.expectNoConnection())
+        #expect(!first.isTerminated)
+        #expect(await h.client.state == .connected)
+        await h.client.stop()
+    }
+
+    // MARK: - Event delivery
+
+    @Test func eventsAreYieldedInArrivalOrder() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        connection.open()
+        connection.sendHandshake()
+        connection.send(
+            "id: e-1\ndata: {\"type\":\"session-removed\",\"sessionId\":\"s-1\"}\n\n"
+                + "id: e-2\ndata: {\"type\":\"messages-invalidated\",\"sessionId\":\"s-2\"}\n\n"
+                + "id: e-3\ndata: {\"type\":\"scheduled-matured\",\"sessionId\":\"s-3\"}\n\n"
+        )
+        #expect(await waitUntil { h.box.syncEvents.count == 3 })
+        #expect(h.box.syncEvents == [
+            .sessionRemoved(namespace: nil, sessionId: "s-1"),
+            .messagesInvalidated(namespace: nil, sessionId: "s-2"),
+            .scheduledMatured(namespace: nil, sessionId: "s-3"),
+        ])
+        #expect(await h.client.lastEventId == "e-3")
+        await h.client.stop()
+    }
+
+    @Test func unknownEventTypesPassThroughWithoutKillingTheStream() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        connection.open()
+        connection.sendHandshake()
+        connection.sendEvent(
+            "{\"type\":\"quantum-flux\",\"namespace\":\"ns\",\"data\":{\"weird\":true}}",
+            id: "e-8"
+        )
+        connection.sendEvent(
+            "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}",
+            id: "e-9"
+        )
+        #expect(await waitUntil { h.box.syncEvents.count == 2 })
+        #expect(h.box.syncEvents == [
+            .unknown(type: "quantum-flux", namespace: "ns"),
+            .sessionRemoved(namespace: "ns", sessionId: "s-1"),
+        ])
+        #expect(await h.client.lastEventId == "e-9")
+        #expect(await h.client.state == .connected)
+        await h.client.stop()
+    }
+
+    @Test func malformedKnownEventIsSkippedWithoutAdvancingCursor() async throws {
+        let h = await makeHarness()
+        let connection = try #require(await h.transport.nextConnection())
+        connection.open()
+        connection.sendHandshake()
+        connection.sendEvent(
+            "{\"type\":\"session-removed\",\"namespace\":\"ns\",\"sessionId\":\"s-1\"}",
+            id: "good-1"
+        )
+        #expect(await waitUntil { await h.client.lastEventId == "good-1" })
+        // Known type, missing required field: dropped, stream stays up, and
+        // the cursor stays BEHIND the unhandled frame (at-least-once).
+        connection.sendEvent("{\"type\":\"session-removed\",\"namespace\":\"ns\"}", id: "bad-1")
+        let moved = await waitUntil(timeoutMs: 300) { await h.client.lastEventId == "bad-1" }
+        #expect(!moved)
+        connection.sendEvent("{\"type\":\"messages-invalidated\",\"sessionId\":\"s-2\"}", id: "good-2")
+        #expect(await waitUntil { await h.client.lastEventId == "good-2" })
+        #expect(h.box.syncEvents.count == 2)
+        #expect(await h.client.state == .connected)
+        await h.client.stop()
+    }
+
+    // MARK: - Auth
+
+    @Test func unauthorizedConnectGetsOneTokenRefreshBypassPerCycle() async throws {
+        let h = await makeHarness(tokens: ["tok-1", "tok-2", "tok-3"])
+        let first = try #require(await h.transport.nextConnection())
+        #expect(first.queryItems["token"] == "tok-1")
+        first.open(status: 401)
+
+        // Free retry: the provider is asked again (it refreshes) and the
+        // reconnect is immediate, without counting a backoff attempt.
+        let second = try #require(await h.transport.nextConnection())
+        #expect(second.queryItems["token"] == "tok-2")
+
+        // The bypass is capped at once per connect cycle: a second
+        // consecutive 401 fails into normal backoff.
+        second.open(status: 401)
+        let third = try #require(await h.transport.nextConnection())
+        #expect(third.queryItems["token"] == "tok-3")
+        #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 0)) })
+
+        // Any other non-2xx fails into backoff too (503: hub not ready).
+        third.open(status: 503)
+        #expect(await waitUntil { h.box.states.contains(.backoff(attempt: 1)) })
+
+        // Exactly two backoffs total — none between the first 401 and its
+        // free retry.
+        let backoffs = h.box.states.filter {
+            if case .backoff = $0 { return true }
+            return false
+        }
+        #expect(backoffs == [.backoff(attempt: 0), .backoff(attempt: 1)])
+        #expect(h.tokens.callCount() == 3)
+        await h.client.stop()
+    }
+
+    // MARK: - Network path
+
+    @Test func networkPathChangeWhileConnectedForcesReconnect() async throws {
+        let observer = FakePathObserver()
+        let h = await makeHarness(pathObserver: observer)
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+
+        #expect(await waitUntil { observer.subscriberCount() >= 1 })
+        observer.emitChange(isSatisfied: true) // e.g. wifi → cellular
+        _ = try #require(await h.transport.nextConnection())
+        #expect(first.isTerminated)
+        await h.client.stop()
+    }
+
+    // MARK: - Lifecycle
+
+    @Test func stopTearsDownAndFinishesTheStream() async throws {
+        let h = await makeHarness()
+        let first = try #require(await h.transport.nextConnection())
+        first.open()
+        first.sendHandshake()
+        #expect(await waitUntil { !h.box.handshakes.isEmpty })
+
+        await h.client.stop()
+        #expect(await waitUntil { first.isTerminated })
+        #expect(await h.client.state == .idle)
+        #expect(await waitUntil { h.box.states.last == .idle })
+
+        // start() after stop() yields a finished stream, not a revived client.
+        let revived = await h.client.start()
+        var iterator = revived.makeAsyncIterator()
+        #expect(await iterator.next() == nil)
+    }
+
+    @Test func seededCursorIsSentOnFirstConnect() async throws {
+        let transport = FakeTransport()
+        let clock = ManualClock()
+        let tokenSource = TokenSource(["tok-1"])
+        let configuration = SSEClientConfiguration(
+            baseUrl: URL(string: "https://hub.test")!,
+            tokenProvider: { tokenSource.next() },
+            scope: .global,
+            jitterMs: { 0 }
+        )
+        let client = SSEClient(configuration: configuration, transport: transport, clock: clock)
+        await client.seedCursor("018f:99:zz")
+        let stream = await client.start()
+        _ = collect(stream)
+        let connection = try #require(await transport.nextConnection())
+        #expect(connection.queryItems["lastEventId"] == "018f:99:zz")
+        await client.stop()
+    }
+}

ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSELineParserTests.swift

@@ -0,0 +1,114 @@
+import Foundation
+import Testing
+@testable import HapiClient
+
+@Suite("SSE line parser")
+struct SSELineParserTests {
+    private func frames(_ text: String) -> [SSEFrame] {
+        var parser = SSELineParser()
+        return parser.consume(Data(text.utf8))
+    }
+
+    @Test func joinsMultiLineDataWithNewline() {
+        let result = frames("data: line1\ndata: line2\ndata: line3\n\n")
+        #expect(result == [SSEFrame(data: "line1\nline2\nline3")])
+    }
+
+    @Test func emitsIdOnlyForTheBlockThatCarriedIt() {
+        // A dispatched id is consumed; the next block starts with no id.
+        // Cursor stickiness across id-less frames is the caller's job.
+        let result = frames("id: 018f:1:aa\ndata: first\n\ndata: second\n\n")
+        #expect(result.count == 2)
+        #expect(result[0] == SSEFrame(id: "018f:1:aa", data: "first"))
+        #expect(result[1].id == nil)
+        #expect(result[1].data == "second")
+    }
+
+    @Test func heartbeatShapedFrameHasNilId() {
+        let result = frames("id: 42\ndata: {\"a\":1}\n\ndata: {\"type\":\"heartbeat\"}\n\n")
+        #expect(result.map(\.id) == ["42", nil])
+    }
+
+    @Test func idFromNonDispatchedBlockTagsNextDispatch() {
+        // Mirrors the EventSource last-event-id buffer: an id-only block sets
+        // the id even though nothing dispatches, and the next data block
+        // carries it.
+        let result = frames("id: X\n\ndata: y\n\n")
+        #expect(result == [SSEFrame(id: "X", data: "y")])
+    }
+
+    @Test func handlesCRLFLineEndings() {
+        let result = frames("id: 5\r\nevent: named\r\ndata: hello\r\ndata: world\r\n\r\n")
+        #expect(result == [SSEFrame(id: "5", event: "named", data: "hello\nworld")])
+    }
+
+    @Test func ignoresInterleavedComments() {
+        let result = frames(": keepalive\ndata: a\n: mid-frame comment\ndata: b\n\n: trailing\n")
+        #expect(result == [SSEFrame(data: "a\nb")])
+    }
+
+    @Test func reassemblesFramesSplitAcrossChunks() {
+        let text = "id: 018f:9:bb\r\ndata: {\"type\":\"x\"}\r\n\r\ndata: tail\n\n"
+        let whole = frames(text)
+
+        // Byte-at-a-time feeding must produce the identical frames,
+        // including a CRLF split between the CR and the LF.
+        var parser = SSELineParser()
+        var collected: [SSEFrame] = []
+        for byte in Data(text.utf8) {
+            collected += parser.consume(Data([byte]))
+        }
+        #expect(collected == whole)
+        #expect(whole.count == 2)
+        #expect(whole[0] == SSEFrame(id: "018f:9:bb", data: "{\"type\":\"x\"}"))
+
+        // And an arbitrary mid-line split.
+        var parser2 = SSELineParser()
+        let bytes = Data(text.utf8)
+        let cut = 7
+        var collected2 = parser2.consume(bytes.prefix(cut))
+        collected2 += parser2.consume(bytes.dropFirst(cut))
+        #expect(collected2 == whole)
+    }
+
+    @Test func ignoresRetryAndUnknownFields() {
+        let result = frames("retry: 5000\nfancy-field: nope\ndata: a\n\n")
+        #expect(result == [SSEFrame(data: "a")])
+    }
+
+    @Test func lineWithoutColonIsFieldWithEmptyValue() {
+        // Per spec, "data" alone is the data field with an empty value.
+        let result = frames("data\n\n")
+        #expect(result == [SSEFrame(data: "")])
+    }
+
+    @Test func stripsExactlyOneLeadingSpace() {
+        #expect(frames("data:no-space\n\n") == [SSEFrame(data: "no-space")])
+        #expect(frames("data:  two-spaces\n\n") == [SSEFrame(data: " two-spaces")])
+    }
+
+    @Test func emptyDataFieldDispatchesEmptyString() {
+        #expect(frames("data:\n\n") == [SSEFrame(data: "")])
+    }
+
+    @Test func blankLineWithoutDataDispatchesNothingAndResetsEventType() {
+        let result = frames("event: named\n\ndata: x\n\n")
+        // The first block had no data: no frame, and its event type must not
+        // leak into the following block.
+        #expect(result == [SSEFrame(data: "x")])
+    }
+
+    @Test func idContainingNulIsIgnored() {
+        let result = frames("id: a\u{0000}b\ndata: x\n\n")
+        #expect(result == [SSEFrame(id: nil, data: "x")])
+    }
+
+    @Test func incompleteTrailingBlockIsNotEmitted() {
+        var parser = SSELineParser()
+        let first = parser.consume(Data("data: complete\n\ndata: partial".utf8))
+        #expect(first == [SSEFrame(data: "complete")])
+        // The unterminated tail only dispatches once its blank line arrives.
+        let second = parser.consume(Data("\n\n".utf8))
+        #expect(second == [SSEFrame(data: "partial")])
+    }
+}

ios/Packages/HapiKit/Tests/HapiClientTests/SSE/SSETestSupport.swift

@@ -0,0 +1,332 @@
+import Foundation
+import HapiProtocol
+@testable import HapiClient
+
+// MARK: - Manual clock
+
+/// Deterministic clock: `sleep` parks on a continuation until test code
+/// advances time. Cancellation resumes the sleeper with `CancellationError`,
+/// including the cancel-before-park race.
+final class ManualClock: SSEClock, @unchecked Sendable {
+    private struct Sleeper {
+        let id: Int
+        let deadlineMs: Int
+        let continuation: CheckedContinuation<Void, Error>
+    }
+
+    private let lock = NSLock()
+    private var currentMs = 0
+    private var nextSleeperId = 0
+    private var sleepers: [Sleeper] = []
+    private var cancelledIds: Set<Int> = []
+
+    func nowMs() -> Int {
+        lock.withLock { currentMs }
+    }
+
+    func sleep(ms: Int) async throws {
+        let id: Int = lock.withLock {
+            nextSleeperId += 1
+            return nextSleeperId
+        }
+        try await withTaskCancellationHandler {
+            try await withCheckedThrowingContinuation { (continuation: CheckedContinuation<Void, Error>) in
+                enum Immediate { case resume, cancel, park }
+                let action: Immediate = lock.withLock {
+                    if cancelledIds.remove(id) != nil {
+                        return .cancel
+                    }
+                    if ms <= 0 {
+                        return .resume
+                    }
+                    sleepers.append(Sleeper(id: id, deadlineMs: currentMs + ms, continuation: continuation))
+                    return .park
+                }
+                switch action {
+                case .resume: continuation.resume()
+                case .cancel: continuation.resume(throwing: CancellationError())
+                case .park: break
+                }
+            }
+        } onCancel: {
+            let sleeper: Sleeper? = lock.withLock {
+                if let index = sleepers.firstIndex(where: { $0.id == id }) {
+                    return sleepers.remove(at: index)
+                }
+                cancelledIds.insert(id)
+                return nil
+            }
+            sleeper?.continuation.resume(throwing: CancellationError())
+        }
+    }
+
+    /// Number of currently parked sleepers. Tests gate clock advances on
+    /// this: the client's watchdog registers its sleep asynchronously, and
+    /// advancing before it parked would leave it waiting for a deadline the
+    /// test never reaches.
+    func activeSleeperCount() -> Int {
+        lock.withLock { sleepers.count }
+    }
+
+    func advance(byMs delta: Int) {
+        let due: [Sleeper] = lock.withLock {
+            currentMs += delta
+            let now = currentMs
+            let ready = sleepers.filter { $0.deadlineMs <= now }.sorted { $0.deadlineMs < $1.deadlineMs }
+            sleepers.removeAll { $0.deadlineMs <= now }
+            return ready
+        }
+        for sleeper in due {
+            sleeper.continuation.resume()
+        }
+    }
+}
+
+// MARK: - Fake transport
+
+final class Flag: @unchecked Sendable {
+    private let lock = NSLock()
+    private var value = false
+
+    func set() {
+        lock.withLock { value = true }
+    }
+
+    var isSet: Bool {
+        lock.withLock { value }
+    }
+}
+
+final class FakeConnection: @unchecked Sendable {
+    let request: URLRequest
+    private let continuation: AsyncThrowingStream<SSETransportEvent, Error>.Continuation
+    private let terminated = Flag()
+
+    init(request: URLRequest, continuation: AsyncThrowingStream<SSETransportEvent, Error>.Continuation) {
+        self.request = request
+        self.continuation = continuation
+        let flag = terminated
+        continuation.onTermination = { _ in flag.set() }
+    }
+
+    var isTerminated: Bool {
+        terminated.isSet
+    }
+
+    func open(status: Int = 200) {
+        let response = HTTPURLResponse(
+            url: request.url ?? URL(string: "https://hub.test")!,
+            statusCode: status,
+            httpVersion: "HTTP/1.1",
+            headerFields: nil
+        )!
+        continuation.yield(.connected(response))
+    }
+
+    func send(_ text: String) {
+        continuation.yield(.bytes(Data(text.utf8)))
+    }
+
+    func sendEvent(_ json: String, id: String? = nil) {
+        var block = ""
+        if let id {
+            block += "id: \(id)\n"
+        }
+        block += "data: \(json)\n\n"
+        send(block)
+    }
+
+    func sendHeartbeat(timestamp: Int = 1) {
+        sendEvent("{\"type\":\"heartbeat\",\"namespace\":\"ns\",\"data\":{\"timestamp\":\(timestamp)}}")
+    }
+
+    func sendHandshake(resume: String? = "ok", subscriptionId: String = "sub-1") {
+        let resumeField = resume.map { ",\"resume\":\"\($0)\"" } ?? ""
+        sendEvent("{\"type\":\"connection-changed\",\"data\":{\"status\":\"connected\",\"subscriptionId\":\"\(subscriptionId)\"\(resumeField)}}")
+    }
+
+    func finish() {
+        continuation.finish()
+    }
+
+    func fail(_ error: Error = URLError(.networkConnectionLost)) {
+        continuation.finish(throwing: error)
+    }
+
+    var queryItems: [String: String] {
+        guard let url = request.url,
+              let components = URLComponents(url: url, resolvingAgainstBaseURL: false) else {
+            return [:]
+        }
+        var result: [String: String] = [:]
+        for item in components.queryItems ?? [] {
+            result[item.name] = item.value ?? ""
+        }
+        return result
+    }
+}
+
+final class FakeTransport: SSETransport, @unchecked Sendable {
+    private let lock = NSLock()
+    private var pending: [FakeConnection] = []
+    private var connectCount = 0
+
+    func connect(_ request: URLRequest) -> AsyncThrowingStream<SSETransportEvent, Error> {
+        let (stream, continuation) = AsyncThrowingStream.makeStream(of: SSETransportEvent.self)
+        let connection = FakeConnection(request: request, continuation: continuation)
+        lock.withLock {
+            connectCount += 1
+            pending.append(connection)
+        }
+        return stream
+    }
+
+    func totalConnections() -> Int {
+        lock.withLock { connectCount }
+    }
+
+    /// Real-time bounded wait for the next unclaimed connection.
+    func nextConnection(timeoutMs: Int = 2_000) async -> FakeConnection? {
+        let deadline = Date().addingTimeInterval(Double(timeoutMs) / 1_000)
+        while Date() < deadline {
+            let claimed: FakeConnection? = lock.withLock {
+                pending.isEmpty ? nil : pending.removeFirst()
+            }
+            if let claimed {
+                return claimed
+            }
+            try? await Task.sleep(nanoseconds: 5_000_000)
+        }
+        return nil
+    }
+
+    /// True when no NEW connection shows up within the grace window.
+    func expectNoConnection(forMs ms: Int = 200) async -> Bool {
+        let deadline = Date().addingTimeInterval(Double(ms) / 1_000)
+        while Date() < deadline {
+            let hasPending = lock.withLock { !pending.isEmpty }
+            if hasPending {
+                return false
+            }
+            try? await Task.sleep(nanoseconds: 5_000_000)
+        }
+        return lock.withLock { pending.isEmpty }
+    }
+}
+
+// MARK: - Fake path observer
+
+final class FakePathObserver: NetworkPathObserving, @unchecked Sendable {
+    private let lock = NSLock()
+    private var continuations: [AsyncStream<NetworkPathUpdate>.Continuation] = []
+
+    func pathUpdates() -> AsyncStream<NetworkPathUpdate> {
+        let (stream, continuation) = AsyncStream.makeStream(of: NetworkPathUpdate.self)
+        lock.withLock { continuations.append(continuation) }
+        // NWPathMonitor always fires the current path on subscription.
+        continuation.yield(NetworkPathUpdate(isSatisfied: true))
+        return stream
+    }
+
+    /// Tests gate `emitChange` on this: the client subscribes from an async
+    /// task, and a change emitted before subscription would be lost.
+    func subscriberCount() -> Int {
+        lock.withLock { continuations.count }
+    }
+
+    func emitChange(isSatisfied: Bool = true) {
+        let all = lock.withLock { continuations }
+        for continuation in all {
+            continuation.yield(NetworkPathUpdate(isSatisfied: isSatisfied))
+        }
+    }
+}
+
+// MARK: - Token source
+
+final class TokenSource: @unchecked Sendable {
+    private let lock = NSLock()
+    private var queue: [String?]
+    private(set) var calls = 0
+
+    /// Returns the queued tokens in order, sticking on the last one.
+    init(_ tokens: [String?]) {
+        queue = tokens
+    }
+
+    func next() -> String? {
+        lock.withLock {
+            calls += 1
+            if queue.count > 1 {
+                return queue.removeFirst()
+            }
+            return queue.first ?? nil
+        }
+    }
+
+    func callCount() -> Int {
+        lock.withLock { calls }
+    }
+}
+
+// MARK: - Event collection
+
+final class EventBox: @unchecked Sendable {
+    private let lock = NSLock()
+    private var storage: [SSEClientEvent] = []
+
+    func append(_ event: SSEClientEvent) {
+        lock.withLock { storage.append(event) }
+    }
+
+    var snapshot: [SSEClientEvent] {
+        lock.withLock { storage }
+    }
+
+    var syncEvents: [SyncEvent] {
+        snapshot.compactMap {
+            if case .event(let event) = $0 { return event }
+            return nil
+        }
+    }
+
+    var handshakes: [SSEClientEvent] {
+        snapshot.filter {
+            if case .handshake = $0 { return true }
+            return false
+        }
+    }
+
+    var states: [SSEConnectionState] {
+        snapshot.compactMap {
+            if case .stateChanged(let state) = $0 { return state }
+            return nil
+        }
+    }
+}
+
+func collect(_ stream: AsyncStream<SSEClientEvent>) -> (box: EventBox, task: Task<Void, Never>) {
+    let box = EventBox()
+    let task = Task {
+        for await event in stream {
+            box.append(event)
+        }
+    }
+    return (box, task)
+}
+
+/// Real-time bounded poll for an async condition.
+@discardableResult
+func waitUntil(
+    timeoutMs: Int = 2_000,
+    _ condition: @Sendable () async -> Bool
+) async -> Bool {
+    let deadline = Date().addingTimeInterval(Double(timeoutMs) / 1_000)
+    while Date() < deadline {
+        if await condition() {
+            return true
+        }
+        try? await Task.sleep(nanoseconds: 5_000_000)
+    }
+    return await condition()
+}

ios/README.md

@@ -47,7 +47,7 @@ ios/
                          The chat pipeline + message window logic (ported from
                          web/src/chat/**) land in M2, validated against
                          shared/fixtures/**.
-    HapiClient           Transport layer. As of M1b:
+    HapiClient           Transport layer. As of M1b+M1c:
                            APIClient        typed REST client (Endpoints/*):
                                             Bearer auth, 401 -> refresh ->
                                             retry-once, {error, code} parsing
@@ -64,12 +64,26 @@ ios/
                                             under run.hapi.companion),
                                             HubRegistry (multi-hub + active
                                             hub in UserDefaults).
+                           SSE/             actor SSEClient — handshake-gated
+                                            connect (resume ok/gap surfaced),
+                                            sticky per-subscription cursor with
+                                            at-least-once replay, 10 s connect
+                                            timeout + 90 s staleness watchdog,
+                                            backoff per sse.md (1 s ×2 → 30 s,
+                                            300 s after 8 attempts, 0–500 ms
+                                            jitter), suspend/resume with the
+                                            45 s foreground staleness check,
+                                            NWPath change → immediate reconnect.
+                                            SSELineParser, ReconnectPolicy/
+                                            SSETimings, URLSessionSSETransport
+                                            (gzip streaming-decompression
+                                            verification TODO — fallback flag
+                                            `acceptEncodingIdentity`).
                            MultipartEncoder for the voice-transcription
                                             endpoint (M4c).
-                         SSEClient (M1c), @Observable stores and snapshots
-                         (M2) land next; feature endpoints (git/files,
-                         scratchlist, voice, usage) join Endpoints/ with
-                         their feature packages.
+                         @Observable stores and snapshots (M2) land next;
+                         feature endpoints (git/files, scratchlist, voice,
+                         usage) join Endpoints/ with their feature packages.
 ```
 
 The app target stays thin; features live in `HapiKit` so they are testable