@@ -17,8 +17,10 @@ import soy.iko.opencode.data.network.OpencodeApiClient
1717import soy.iko.opencode.util.runCatchingCancellable
1818import soy.iko.opencode.util.safeExceptionSummary
1919import android.util.Log
20+import kotlinx.coroutines.channels.Channel
2021import kotlinx.coroutines.flow.Flow
2122import kotlinx.coroutines.flow.channelFlow
23+import kotlinx.coroutines.flow.consumeAsFlow
2224import kotlinx.coroutines.flow.conflate
2325import kotlinx.coroutines.launch
2426import kotlinx.coroutines.sync.Mutex
@@ -58,25 +60,23 @@ open class SessionRepository(
5860 open fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
5961 val store = MessageStore()
6062 val lock = Mutex()
63+ // Conflated dirty signal: a burst of reduce()/seed() calls coalesces into a
64+ // single snapshot+send per demand cycle. Previously every changed event called
65+ // snapshot() — allocating an O(N) copy of the whole message list — and ~97% of
66+ // those copies were immediately discarded by the outer .conflate() during fast
67+ // streaming (hundreds of tokens/sec). Marking dirty and snapshotting in one
68+ // drain loop means N events in a burst allocate ~1 snapshot, not N.
69+ val dirty = Channel<Unit>(Channel.CONFLATED)
6170
62− // Collect the snapshot under the lock, then publish outside it so a slow
63− // downstream collector can't stall event processing.
64− suspend fun publish() {
65− val snapshot = lock.withLock { store.snapshot() }
66− send(snapshot)
67− }
71+ // Mark the store dirty; the drain loop below snapshots and publishes. CONFLATED
72+ // collapses repeated marks into a single pending signal, so a tight burst of
73+ // events doesn't even allocate N marks.
74+ fun publish() { dirty.trySend(Unit) }
6875
6976 // Subscribe to events first so we don't miss early deltas during the initial load.
70− val job = launch {
77+ launch {
7178 eventStream.events.collect { event ->
72− // Reduce and snapshot in a single lock acquisition: during fast streaming
73− // (hundreds of tokens/sec) this halves mutex contention vs. acquiring the
74− // lock once for reduce and again for snapshot. send() is called outside the
75− // lock so a slow downstream collector can't stall event processing.
76− val snapshot = lock.withLock {
77− if (store.reduce(sessionId, event)) store.snapshot() else null
78− }
79− if (snapshot != null) send(snapshot)
79+ if (lock.withLock { store.reduce(sessionId, event) }) publish()
8080 }
8181 }
8282
@@ -128,8 +128,15 @@ open class SessionRepository(
128128 lock.withLock { store.seed(initial) }
129129 publish()
130130
131− // Keep the flow alive until the collector cancels; the launched job is torn down with it.
132− job.join()
131+ // Single drain: one snapshot+send per conflated dirty signal. The snapshot is
132+ // taken under the lock so a reader never sees a half-reduced state; send() runs
133+ // outside the lock so a slow downstream collector can't stall the drain. This
134+ // loop also keeps the flow alive until the collector cancels; the launched jobs
135+ // above are children of this scope and are torn down with it.
136+ dirty.consumeAsFlow().collect {
137+ val snapshot = lock.withLock { store.snapshot() }
138+ send(snapshot)
139+ }
133140 }.conflate()
134141
135142 companion object {