From a21cc4eff675aeb5f4f855f17561cad2aa2a06d3 Mon Sep 17 00:00:00 2001 From: Val Alexander Date: Sat, 18 Jul 2026 17:55:21 -0500 Subject: [PATCH 1/3] perf(ios/chat): coalesce assistant stream tokens MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Streaming applied every assistant chunk immediately: mutate the message then onChange() on EVERY token. Because messages is an observed array on an @Observable ChatThread, each mutate reassigns messages[idx] and invalidates the whole message list — so a fast stream (tokens arriving faster than a frame) drove a per-token render and scroll storm. Introduce a small @MainActor StreamCoalescer that buffers chunk text and flushes on a ~50ms cadence (~20 UI updates/sec) instead of per token. All terminal paths (.done/.error, end of the send loop, and the resume loop) flush unconditionally, so the final text is always complete; a single coalescer instance is shared across the send + resume streams so a resumed turn keeps the same buffer. Verified: xcodebuild build (scheme CovenCave, iOS Simulator) succeeds. --- .../CovenCave/State/ChatThread.swift | 71 +++++++++++++++++-- 1 file changed, 65 insertions(+), 6 deletions(-) diff --git a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift index 92303b6ef..c97c38644 100644 --- a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift +++ b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift @@ -264,13 +264,15 @@ final class ChatThread: Identifiable, Hashable { // Resume cursor: the last applied frame's SSE id (run-buffer seq). var cursor = 0 var sawDone = false + let coalescer = StreamCoalescer() do { for try await frame in client.sendStream(body) { receivedAnyEvent = true apply(frame.event, into: messageId, familiarId: familiarId, - sawDone: &sawDone, onChange: onChange) + sawDone: &sawDone, coalescer: coalescer, onChange: onChange) if let id = frame.id { cursor = id } } + flush(coalescer, into: messageId, onChange: onChange) mutate(messageId) { $0.streaming = false } } catch { // Transport interruption (network handoff, backgrounding, desktop @@ -281,7 +283,7 @@ final class ChatThread: Identifiable, Hashable { // transcript. let resumed = await resumeInterruptedStream(runId: runId, cursor: cursor, into: messageId, familiarId: familiarId, - sawDone: &sawDone, + sawDone: &sawDone, coalescer: coalescer, client: client, onChange: onChange) var recovered = resumed if !recovered { @@ -313,18 +315,29 @@ final class ChatThread: Identifiable, Hashable { /// Apply one stream event to the thread — shared by the original send /// stream and the mid-turn resume stream so both render identically. private func apply(_ event: StreamEvent, into messageId: String, familiarId: String, - sawDone: inout Bool, onChange: @escaping () -> Void) { + sawDone: inout Bool, coalescer: StreamCoalescer, onChange: @escaping () -> Void) { switch event { case .session(let sid): if !sid.isEmpty { sessionIds[familiarId] = sid } case .assistantChunk(let chunk): - mutate(messageId) { $0.text += chunk } - onChange() + // Coalesce tokens: buffer chunk text and flush to the message on a + // short cadence instead of mutating the (observed) messages array + + // firing onChange() on EVERY token. A fast stream can emit tokens + // faster than a frame, and each mutate reassigns messages[idx] on an + // @Observable class — invalidating the whole list — so per-token + // updates caused a render/scroll storm. Coalescing flushes at most + // ~every 50ms while keeping streaming visibly live. + coalescer.append(chunk) + if coalescer.shouldFlush() { + flush(coalescer, into: messageId, onChange: onChange) + } case .done(let isError, let sid): if let sid, !sid.isEmpty { sessionIds[familiarId] = sid } + flush(coalescer, into: messageId, onChange: onChange) mutate(messageId) { $0.streaming = false; if isError { $0.isError = true } } sawDone = true case .error(let message): + flush(coalescer, into: messageId, onChange: onChange) mutate(messageId) { if $0.text.isEmpty { $0.text = message } $0.isError = true; $0.streaming = false @@ -334,6 +347,16 @@ final class ChatThread: Identifiable, Hashable { } } + /// Drain any buffered stream text into the message and notify observers. + /// Idempotent: a no-op when the buffer is empty, so terminal paths can call + /// it unconditionally. + private func flush(_ coalescer: StreamCoalescer, into messageId: String, + onChange: @escaping () -> Void) { + guard let pending = coalescer.drain() else { return } + mutate(messageId) { $0.text += pending } + onChange() + } + /// Re-attach to the still-live run after a transport drop: replay past /// the cursor, then tail live until the turn ends. A few short-backoff /// attempts ride out the network still settling (Wi-Fi handoff, tunnel @@ -341,6 +364,7 @@ final class ChatThread: Identifiable, Hashable { /// falls back to the post-hoc transcript resync. private func resumeInterruptedStream(runId: String, cursor: Int, into messageId: String, familiarId: String, sawDone: inout Bool, + coalescer: StreamCoalescer, client: CaveClient, onChange: @escaping () -> Void) async -> Bool { var nextCursor = cursor for attempt in 0..<3 { @@ -350,9 +374,10 @@ final class ChatThread: Identifiable, Hashable { do { for try await frame in client.resumeStream(runId: runId, cursor: nextCursor) { apply(frame.event, into: messageId, familiarId: familiarId, - sawDone: &sawDone, onChange: onChange) + sawDone: &sawDone, coalescer: coalescer, onChange: onChange) if let id = frame.id { nextCursor = id } } + flush(coalescer, into: messageId, onChange: onChange) // The resume stream closes when the run finishes. Without a // done event the run may still be live (our tail dropped // again) — retry from the advanced cursor. @@ -455,3 +480,37 @@ final class ChatThread: Identifiable, Hashable { messages[idx] = message } } + +/// Buffers assistant stream chunks so the UI updates on a short cadence rather +/// than once per token. Each `ChatThread` mutation of the observed `messages` +/// array invalidates the whole message list, so flushing per token turned a +/// fast stream into a render/scroll storm. This accumulates text and reports +/// `shouldFlush()` at most ~every 50ms; terminal stream events drain it +/// unconditionally so the final text is always complete. +@MainActor +final class StreamCoalescer { + private var buffer = "" + private var lastFlush = ContinuousClock.now + /// Max time text may sit buffered before the next flush. 50ms keeps the + /// stream visibly live (~20 updates/sec) while collapsing token bursts. + private let interval: Duration = .milliseconds(50) + + func append(_ chunk: String) { buffer += chunk } + + /// True when enough time has elapsed since the last flush and text is + /// pending. Checked after each appended chunk. + func shouldFlush() -> Bool { + guard !buffer.isEmpty else { return false } + return ContinuousClock.now - lastFlush >= interval + } + + /// Returns and clears the buffered text (nil when empty), and resets the + /// flush clock. + func drain() -> String? { + guard !buffer.isEmpty else { return nil } + let pending = buffer + buffer = "" + lastFlush = ContinuousClock.now + return pending + } +} From 2933329da34abef88364af6b67f9644233f7202e Mon Sep 17 00:00:00 2001 From: Timothy Wayne Gregg <5861166+romgenie@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:04:04 -0400 Subject: [PATCH 2/3] fix(ios): flush stream buffer on transport errors --- .../ios/CovenCave/CovenCave/State/ChatThread.swift | 3 +++ scripts/mobile-tailscale-native.test.mjs | 14 ++++++++++++-- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift index c97c38644..d74ff32ed 100644 --- a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift +++ b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift @@ -281,6 +281,7 @@ final class ChatThread: Identifiable, Hashable { // (cave-h40l). Only when no resumable run exists (finished long // ago / server restarted) fall back to adopting the persisted // transcript. + flush(coalescer, into: messageId, onChange: onChange) let resumed = await resumeInterruptedStream(runId: runId, cursor: cursor, into: messageId, familiarId: familiarId, sawDone: &sawDone, coalescer: coalescer, @@ -386,10 +387,12 @@ final class ChatThread: Identifiable, Hashable { return true } } catch is CaveClient.NoResumableRun { + flush(coalescer, into: messageId, onChange: onChange) // Nothing buffered under that run — turn ended long ago or // the server restarted. Post-hoc resync owns recovery. return false } catch { + flush(coalescer, into: messageId, onChange: onChange) // Transport still flaky — back off and retry from the cursor. } } diff --git a/scripts/mobile-tailscale-native.test.mjs b/scripts/mobile-tailscale-native.test.mjs index 5cbe98ec8..da7f26b21 100644 --- a/scripts/mobile-tailscale-native.test.mjs +++ b/scripts/mobile-tailscale-native.test.mjs @@ -57,8 +57,18 @@ assert.match( ); assert.match( swiftChatThread, - /case \.assistantChunk\(let chunk\):\s*\n\s*mutate\(messageId\) \{ \$0\.text \+= chunk \}\s*\n\s*onChange\(\)/, - "native SwiftUI chat should notify/persist after assistant chunks so responses render while streaming", + /case \.assistantChunk\(let chunk\):[\s\S]*?coalescer\.append\(chunk\)[\s\S]*?if coalescer\.shouldFlush\(\) \{\s*\n\s*flush\(coalescer, into: messageId, onChange: onChange\)/, + "native SwiftUI chat should buffer assistant chunks and periodically notify/persist while streaming", +); +assert.match( + swiftChatThread, + /catch \{[\s\S]*?flush\(coalescer, into: messageId, onChange: onChange\)\s*\n\s*let resumed = await resumeInterruptedStream/, + "native SwiftUI chat should flush buffered assistant text before recovering a dropped send stream", +); +assert.match( + swiftChatThread, + /} catch is CaveClient\.NoResumableRun \{\s*\n\s*flush\(coalescer, into: messageId, onChange: onChange\)[\s\S]*?} catch \{\s*\n\s*flush\(coalescer, into: messageId, onChange: onChange\)/, + "native SwiftUI chat should retain buffered text when a resumed stream cannot continue", ); assert.match( swiftChatThread, From d3ca1a37158841a61bd4fe96fb818eaf9ad9545b Mon Sep 17 00:00:00 2001 From: Timothy Wayne Gregg <5861166+romgenie@users.noreply.github.com> Date: Sat, 18 Jul 2026 19:08:43 -0400 Subject: [PATCH 3/3] fix(ios): schedule coalesced stream flushes Ensure a paused stream drains its final buffered assistant chunk within the coalescing interval rather than waiting for a later frame or terminal event. Co-authored-by: Val Alexander <68980965+BunsDev@users.noreply.github.com> --- .../CovenCave/State/ChatThread.swift | 33 ++++++++++++------- scripts/mobile-tailscale-native.test.mjs | 9 +++-- 2 files changed, 28 insertions(+), 14 deletions(-) diff --git a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift index d74ff32ed..2c863fbc0 100644 --- a/apps/ios/CovenCave/CovenCave/State/ChatThread.swift +++ b/apps/ios/CovenCave/CovenCave/State/ChatThread.swift @@ -328,9 +328,9 @@ final class ChatThread: Identifiable, Hashable { // @Observable class — invalidating the whole list — so per-token // updates caused a render/scroll storm. Coalescing flushes at most // ~every 50ms while keeping streaming visibly live. - coalescer.append(chunk) - if coalescer.shouldFlush() { - flush(coalescer, into: messageId, onChange: onChange) + coalescer.append(chunk) { [weak self] in + guard let self else { return } + self.flush(coalescer, into: messageId, onChange: onChange) } case .done(let isError, let sid): if let sid, !sid.isEmpty { sessionIds[familiarId] = sid } @@ -493,27 +493,36 @@ final class ChatThread: Identifiable, Hashable { @MainActor final class StreamCoalescer { private var buffer = "" - private var lastFlush = ContinuousClock.now + private var flushTask: Task? /// Max time text may sit buffered before the next flush. 50ms keeps the /// stream visibly live (~20 updates/sec) while collapsing token bursts. private let interval: Duration = .milliseconds(50) - func append(_ chunk: String) { buffer += chunk } - - /// True when enough time has elapsed since the last flush and text is - /// pending. Checked after each appended chunk. - func shouldFlush() -> Bool { - guard !buffer.isEmpty else { return false } - return ContinuousClock.now - lastFlush >= interval + /// Start one delayed flush for a burst. Scheduling rather than checking + /// elapsed time only when a new chunk arrives also drains the final chunk + /// when a stream pauses without immediately ending. + func append(_ chunk: String, onFlushDue: @escaping @MainActor () -> Void) { + buffer += chunk + guard flushTask == nil else { return } + flushTask = Task { @MainActor [weak self] in + guard let self else { return } + try? await Task.sleep(for: self.interval) + guard !Task.isCancelled else { return } + self.flushTask = nil + onFlushDue() + } } /// Returns and clears the buffered text (nil when empty), and resets the /// flush clock. func drain() -> String? { + flushTask?.cancel() + flushTask = nil guard !buffer.isEmpty else { return nil } let pending = buffer buffer = "" - lastFlush = ContinuousClock.now return pending } + + deinit { flushTask?.cancel() } } diff --git a/scripts/mobile-tailscale-native.test.mjs b/scripts/mobile-tailscale-native.test.mjs index da7f26b21..46ddbb131 100644 --- a/scripts/mobile-tailscale-native.test.mjs +++ b/scripts/mobile-tailscale-native.test.mjs @@ -57,8 +57,13 @@ assert.match( ); assert.match( swiftChatThread, - /case \.assistantChunk\(let chunk\):[\s\S]*?coalescer\.append\(chunk\)[\s\S]*?if coalescer\.shouldFlush\(\) \{\s*\n\s*flush\(coalescer, into: messageId, onChange: onChange\)/, - "native SwiftUI chat should buffer assistant chunks and periodically notify/persist while streaming", + /case \.assistantChunk\(let chunk\):[\s\S]*?coalescer\.append\(chunk\) \{[\s\S]*?self\.flush\(coalescer, into: messageId, onChange: onChange\)/, + "native SwiftUI chat should buffer assistant chunks and schedule a coalesced UI/persistence flush", +); +assert.match( + swiftChatThread, + /private var flushTask: Task\?[\s\S]*?Task\.sleep\(for: self\.interval\)[\s\S]*?onFlushDue\(\)/, + "native SwiftUI chat should flush a paused stream's final buffered chunk without waiting for another event", ); assert.match( swiftChatThread,