git.iko.soy/opencode-android
on 366f92e help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
647 lines 37 kB history raw
1package soy.iko.opencode.data.repo
3import soy.iko.opencode.data.model.AssistantMessage
4import soy.iko.opencode.data.model.BusEvent
5import soy.iko.opencode.data.model.FilePromptPart
6import soy.iko.opencode.data.model.MessageInfo
7import soy.iko.opencode.data.model.MessagePartRemoved
8import soy.iko.opencode.data.model.MessagePartUpdated
9import soy.iko.opencode.data.model.MessageRemoved
10import soy.iko.opencode.data.model.MessageUpdated
11import soy.iko.opencode.data.model.MessageWithParts
12import soy.iko.opencode.data.model.ModelRef
13import soy.iko.opencode.data.model.Part
14import soy.iko.opencode.data.model.SessionError
15import soy.iko.opencode.data.model.SessionIdle
16import soy.iko.opencode.data.model.UnknownMessage
17import soy.iko.opencode.data.network.EventStreamClient
18import soy.iko.opencode.data.network.NetworkConfig
19import soy.iko.opencode.data.network.OpencodeApiClient
20import soy.iko.opencode.util.runCatchingCancellable
21import soy.iko.opencode.util.safeExceptionSummary
22import android.util.Log
23import kotlinx.coroutines.CoroutineScope
24import kotlinx.coroutines.Dispatchers
25import kotlinx.coroutines.Job
26import kotlinx.coroutines.NonCancellable
27import kotlinx.coroutines.delay
28import kotlinx.coroutines.withContext
29import kotlinx.coroutines.channels.Channel
30import kotlinx.coroutines.flow.Flow
31import kotlinx.coroutines.flow.channelFlow
32import kotlinx.coroutines.flow.consumeAsFlow
33import kotlinx.coroutines.flow.conflate
34import kotlinx.coroutines.flow.flowOn
35import kotlinx.coroutines.launch
36import kotlinx.coroutines.sync.Mutex
37import kotlinx.coroutines.sync.withLock
39/**
40 * Source of truth for a session's conversation. It seeds from `GET /session/:id/message`
41 * and then reduces [BusEvent]s from the SSE stream into an ordered list of messages,
42 * each keyed by message id with parts keyed by part id (idempotent upserts).
43 */
44open class SessionRepository(
45 private val api: OpencodeApiClient,
46 private val eventStream: EventStreamClient,
47 // Optional on-disk cache so a conversation renders instantly on open and stays readable
48 // offline. Null in tests (and any path that doesn't wire it), where caching is skipped.
49 private val messageCache: MessageCacheStore? = null,
50 // Profile the cache is namespaced under, so two servers with a colliding session id don't
51 // share a cache file. Empty in tests / when no cache is wired.
52 private val profileId: String = "",
53) {
54 open suspend fun listSessions() = api.listSessions()
55 open suspend fun createSession(title: String? = null, directory: String? = null) =
56 api.createSession(title, directory)
57 open suspend fun deleteSession(id: String) = api.deleteSession(id)
58 open suspend fun abort(sessionId: String) = api.abort(sessionId)
60 open suspend fun sendPrompt(
61 sessionId: String,
62 text: String,
63 attachments: List<FilePromptPart> = emptyList(),
64 model: ModelRef?,
65 agent: String? = null,
66 // Stable key for retriable/re-sendable messages (the offline outbox). See
67 // [OpencodeApiClient.sendPrompt]. Null generates a fresh key per call.
68 idempotencyKey: String? = null,
69 ) = api.sendPrompt(sessionId, text, attachments, model, agent, idempotencyKey)
71 open suspend fun runCommand(
72 sessionId: String,
73 command: String,
74 arguments: String = "",
75 agent: String? = null,
76 ) = api.runCommand(sessionId, command, arguments = arguments, agent = agent)
78 /**
79 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
80 * stream before the initial REST load so no streamed part is missed, then reduces
81 * events into the in-memory state. The flow is conflated so a fast burst of
82 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
83 * event reducer — a slow collector simply sees the most recent snapshot.
84 */
85 /** Per-subscription helper that persists reduced snapshots to [messageCache], throttled so a
86 * fast token stream doesn't hammer the disk, plus a NonCancellable teardown flush. Extracted
87 * from [observeMessages] so this cache-write branching lives in its own methods. [scope] is
88 * the channelFlow scope, so scheduled saves are torn down with the flow — the teardown flush
89 * re-persists a trailing write the cancellation dropped. */
90 private inner class CacheWriter(private val scope: CoroutineScope, private val sessionId: String) {
91 private var lastWrite = 0L
92 private var pending: Job? = null
93 // Most recent snapshot handed to [schedule], and the most recent one whose on-disk write
94 // completed. `saved` is assigned from inside the launched save *after* it returns, so a
95 // save cancelled mid-IO by teardown is NOT marked saved and [flushOnTeardown] re-persists
96 // it. Written from the save coroutines, read from the collector/teardown; the only race is
97 // a stale read causing one redundant, idempotent re-save (atomic temp+rename makes a
98 // duplicate write of identical data harmless) — never data loss.
99 private var latest: List<MessageWithParts>? = null
100 private var saved: List<MessageWithParts>? = null
102 fun schedule(snapshot: List<MessageWithParts>) {
103 val cache = messageCache ?: return
104 if (snapshot.isEmpty()) return
105 latest = snapshot
106 val now = System.currentTimeMillis()
107 pending?.cancel()
108 // Immediate write once the throttle window has elapsed; otherwise a trailing-edge
109 // write so a burst that ends inside the window (a run's final tokens right after a
110 // write) still persists its last snapshot. A newer snapshot cancels and replaces a
111 // pending trailing write, so continuous streaming coalesces to one write per window.
112 val delayMs = if (now - lastWrite >= NetworkConfig.messageCacheWriteThrottleMs) {
113 lastWrite = now
114 0L
115 } else {
116 NetworkConfig.messageCacheWriteThrottleMs
117 }
118 pending = scope.launch {
119 if (delayMs > 0) delay(delayMs)
120 cache.save(profileId, sessionId, snapshot)
121 saved = snapshot
122 }
123 }
125 suspend fun flushOnTeardown() {
126 // Scheduled saves are children of [scope], so teardown cancels a pending trailing
127 // write before its delay elapses — a run's final tokens would never reach disk. Flush
128 // the latest un-saved snapshot under NonCancellable so it survives the cancellation.
129 val cache = messageCache ?: return
130 val pendingSnapshot = latest ?: return
131 if (pendingSnapshot !== saved) {
132 withContext(NonCancellable) { cache.save(profileId, sessionId, pendingSnapshot) }
133 }
134 }
135 }
137 open fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
138 val store = MessageStore()
139 val lock = Mutex()
140 // Conflated dirty signal: a burst of reduce()/seed() calls coalesces into a
141 // single snapshot+send per demand cycle. Previously every changed event called
142 // snapshot() — allocating an O(N) copy of the whole message list — and ~97% of
143 // those copies were immediately discarded by the outer .conflate() during fast
144 // streaming (hundreds of tokens/sec). Marking dirty and snapshotting in one
145 // drain loop means N events in a burst allocate ~1 snapshot, not N.
146 val dirty = Channel<Unit>(Channel.CONFLATED)
148 // Mark the store dirty; the drain loop below snapshots and publishes. CONFLATED
149 // collapses repeated marks into a single pending signal, so a tight burst of
150 // events doesn't even allocate N marks.
151 fun publish() { dirty.trySend(Unit) }
153 // Subscribe to events first so we don't miss early deltas during the initial load.
154 launch {
155 eventStream.events.collect { event ->
156 if (lock.withLock { store.reduce(sessionId, event) }) publish()
157 }
158 }
160 // Seed generation counter, shared between the initial load and the reconnect
161 // re-seed. Guards against a stale fetch (from an earlier reconnect, or from a slow
162 // initial load) clobbering a newer seed. Only mutated/read under [lock].
163 var seedGeneration = 0
165 // Message ids introduced by the on-disk cache seed below. The initial REST seed uses
166 // these to prune ghosts: a cache-seeded message absent from the authoritative REST
167 // snapshot was deleted server-side while the app was closed. Restricting the prune to
168 // cache-seeded ids means live SSE messages added since subscribe (not yet indexed by
169 // REST) are never dropped. Written/read only in this collector coroutine.
170 var cacheSeededIds: Set<String> = emptySet()
172 // Seed from the on-disk cache first so the conversation renders instantly — and stays
173 // readable offline — before the network responds. Guarded on generation 0 so a
174 // reconnect re-seed that somehow raced ahead isn't clobbered; the initial REST seed
175 // (also generation 0) then overrides this with authoritative data when it succeeds.
176 messageCache?.let { cache ->
177 val cached = cache.load(profileId, sessionId)
178 if (cached.isNotEmpty()) {
179 val changed = lock.withLock {
180 if (seedGeneration == 0) {
181 cacheSeededIds = cached.mapTo(mutableSetOf()) { it.info.id }
182 store.seed(cached, clearPivot = false)
183 } else false
184 }
185 if (changed) publish()
186 }
187 }
189 // Re-seed from REST when the SSE stream reconnects after a drop. The REST fetch
190 // is performed *outside* the lock so events arriving via SSE during the fetch are
191 // not blocked. The seed itself is under the lock; a seed generation counter ensures
192 // a stale fetch (from an earlier reconnect) doesn't clobber a newer one.
193 launch {
194 // Re-seed from REST when the SSE stream reconnects after a drop. The
195 // hasConnectedBefore flag is NOT reset on disconnect — it stays true after
196 // the first successful connection so a subsequent reconnect triggers the
197 // re-seed. Resetting it on disconnect would make the condition always false
198 // after the first cycle, turning the re-seed into dead code.
199 //
200 // A seed generation counter guards against a stale re-seed clobbering a
201 // newer one: if two reconnects fire in quick succession, the older fetch
202 // (which returns later) is discarded because its generation no longer matches.
203 var hasConnectedBefore = false
204 eventStream.state.collect { state ->
205 if (state == EventStreamClient.ConnectionState.Connected) {
206 if (hasConnectedBefore) {
207 // Pivot before the fetch: parts already in memory predate this
208 // snapshot and must yield to it, while parts that stream in during
209 // the fetch (added to streamedSincePivot via reduce) outrank it.
210 // Bump the generation under the same lock so the compare in the
211 // initial seed sees a consistent value.
212 val generation = lock.withLock {
213 store.beginReseed()
214 ++seedGeneration
215 }
216 val fresh = runCatchingCancellable { api.listMessages(sessionId) }
217 .onFailure { Log.w("SessionRepository", "Re-seed message load failed for $sessionId; relying on SSE: ${safeExceptionSummary(it)}") }
218 .getOrDefault(emptyList())
219 var seedChanged = false
220 lock.withLock {
221 if (generation == seedGeneration) {
222 // Merge without pruning: SSE events arriving during the
223 // REST fetch may have added messages not yet indexed by
224 // the REST endpoint. Pruning would silently delete them.
225 // Stale messages from deletions during the disconnect
226 // gap are eventually removed via MessageRemoved events
227 // or on the next app restart.
228 seedChanged = store.seed(fresh, prune = false)
229 }
230 }
231 if (seedChanged) publish()
232 }
233 hasConnectedBefore = true
234 }
235 }
236 }
238 // Load the authoritative message list concurrently with the drain loop below so the
239 // cache seed and any early SSE-reduced parts are emitted immediately. Running it inline
240 // here (before the drain starts) would withhold the already-reduced cached conversation
241 // until this fetch resolved — a blank screen for the whole withRetry backoff when offline.
242 launch {
243 val initialResult = runCatchingCancellable { api.listMessages(sessionId) }
244 .onFailure { Log.w("SessionRepository", "Initial message load failed for $sessionId; relying on cache/SSE: ${safeExceptionSummary(it)}") }
245 // Apply the initial seed only if no reconnect re-seed has run yet (generation still 0).
246 // A re-seed is always a fresher full snapshot, so if the (possibly slow) initial fetch
247 // returns after one has applied, seeding this older REST data would clobber the newer
248 // state — reverting a completed run to "running", regressing cost/token/part state —
249 // until the next SSE event or reconnect healed it. Only seed on SUCCESS: on failure
250 // (e.g. offline) keep whatever's shown (the cache seed above) rather than clearing it.
251 val seeded = lock.withLock {
252 val list = initialResult.getOrNull()
253 if (seedGeneration == 0 && list != null) {
254 // Prune ghosts first: cache-seeded messages absent from this authoritative
255 // REST snapshot were deleted server-side while the app was closed. Without
256 // this they'd survive restart forever (re-seeded from cache, never pruned by
257 // the reconnect path, and re-persisted by the drain loop).
258 val pruned = store.pruneStaleCacheSeeded(cacheSeededIds, list)
259 val changed = store.seed(list)
260 pruned || changed
261 } else false
262 }
263 if (seeded) publish()
264 }
266 // Single drain: one snapshot+send per conflated dirty signal. The snapshot is
267 // taken under the lock so a reader never sees a half-reduced state; send() runs
268 // outside the lock so a slow downstream collector can't stall the drain. This
269 // loop also keeps the flow alive until the collector cancels; the launched jobs
270 // above are children of this scope and are torn down with it.
271 // Per-subscription cache writer: throttled snapshot persistence + a NonCancellable
272 // teardown flush. `this` is the channelFlow scope, so its scheduled saves are torn down
273 // with the flow (the teardown flush re-persists a cancelled trailing write).
274 val cacheWriter = CacheWriter(this, sessionId)
275 // Guarantee at least one emission even when nothing seeds the store: with no on-disk
276 // cache, a failing initial REST load (swallowed above), and a quiet SSE stream, none of
277 // the seed paths publish, so the flow would emit nothing at all. The sole collector
278 // (ChatViewModel) only clears its loading spinner on an emission, so without this a
279 // first-open of an erroring/empty session spins forever with no way to reach the
280 // "Failed to load / Retry" affordance. CONFLATED dirty coalesces this with any seed
281 // publish into a single send.
282 publish()
283 try {
284 dirty.consumeAsFlow().collect {
285 val snapshot = lock.withLock { store.snapshot() }
286 send(snapshot)
287 // Persist to the on-disk cache, throttled so a fast token stream doesn't hammer
288 // the disk — the cache only needs to be "recent enough" for an instant/offline
289 // first paint; the network corrects it on the next open.
290 cacheWriter.schedule(snapshot)
291 }
292 } finally {
293 cacheWriter.flushOnTeardown()
294 }
295 // Run the whole reducer pipeline — event reduction, list rebuilds, and the O(N)
296 // snapshot copy — on Dispatchers.Default rather than the collector's context. The
297 // sole consumer collects via stateIn(viewModelScope), i.e. Main.immediate, so without
298 // this every streamed token's reduce()/snapshot() would execute on the UI thread
299 // (hundreds/sec on a fast run). The seed REST calls here only suspend, so moving them
300 // off Main is harmless. JSON decoding already runs off-Main upstream in EventStreamClient.
301 }.flowOn(Dispatchers.Default).conflate()
303 companion object {
304 /** Convenience: is this event a run-completion signal for [sessionId]?
305 * A null sessionID is NOT treated as a wildcard — an idle event with no
306 * session id must not reset the running state of every open chat. */
307 fun isIdle(event: BusEvent, sessionId: String): Boolean =
308 event is SessionIdle && event.properties.sessionID == sessionId
310 fun isError(event: BusEvent, sessionId: String): Boolean =
311 event is SessionError && event.properties.sessionID == sessionId
313 /** Does this event indicate an in-progress run for [sessionId]? A live streamed part or
314 * an assistant message that isn't yet complete/errored means the agent is still working.
315 * Used to restore the "running" indicator after an SSE reconnect: if the run is still
316 * going, such live events keep arriving; a run that finished during the outage produces
317 * none (its completion arrives via the REST re-seed, not the event stream). */
318 fun isRunActivity(event: BusEvent, sessionId: String): Boolean = when (event) {
319 // The session id rides on the part for message.part.updated; properties.sessionID
320 // is null on the wire. Mirror reduce()'s `part.sessionID ?: properties.sessionID`
321 // so a run that's only streaming parts still re-lights the indicator on reconnect.
322 is MessagePartUpdated -> (event.properties.part.sessionID ?: event.properties.sessionID) == sessionId
323 is MessageUpdated -> {
324 val info = event.properties.info
325 info.sessionID == sessionId && info is AssistantMessage && !info.isComplete && info.error == null
326 }
327 else -> false
328 }
329 }
332/**
333 * In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex.
334 * Exposed as `internal` so the reducer logic can be unit-tested directly.
335 */
336internal class MessageStore {
337 // messageId -> mutable holder (info + insertion-ordered parts); the map itself is
338 // insertion-ordered.
339 private val messages = LinkedHashMap<String, Holder>()
341 /**
342 * Mutable per-message holder. Parts live in an insertion-ordered map keyed by part id, so a
343 * streamed part update is an O(1) put instead of an O(P) list rebuild. The immutable
344 * [MessageWithParts] view is materialized lazily and cached until the next mutation — so a
345 * burst of streamed token updates rebuilds a message's part list at most once per published
346 * [snapshot], not once per token. The cached view is also reference-stable across snapshots
347 * while unchanged, so Compose skips recomposing bubbles that didn't change.
348 */
349 private class Holder(var info: MessageInfo, val parts: LinkedHashMap<String, Part>) {
350 private var cached: MessageWithParts? = null
351 fun view(): MessageWithParts =
352 cached ?: MessageWithParts(info, parts.values.toList()).also { cached = it }
353 /** Adopt an existing immutable instance as the cache — avoids an immediate rebuild and
354 * preserves its identity (e.g. when seeding an authoritative REST snapshot). */
355 fun adopt(m: MessageWithParts) { cached = m }
356 fun invalidate() { cached = null }
357 }
359 /** Build a [Holder] from an immutable message, reusing [m] itself as the cached view. */
360 private fun holderOf(m: MessageWithParts): Holder {
361 val parts = LinkedHashMap<String, Part>(m.parts.size)
362 for (p in m.parts) parts[p.id] = p
363 return Holder(m.info, parts).also { it.adopt(m) }
364 }
366 /** Part ids upserted via [reduce] since the last reseed pivot ([beginReseed]) or seed.
367 * On a reconnect re-seed these are the parts that streamed in *during* the REST fetch,
368 * so they're newer than the snapshot and must win over it; every other part is taken
369 * from the authoritative REST snapshot. Without this distinction the merge kept every
370 * overlapping in-memory part, so a part that changed server-side during the disconnect
371 * (e.g. a tool going running -> completed) stayed stale until app restart. */
372 private val streamedSincePivot = mutableSetOf<String>()
374 /** Message ids whose `info` was updated via a live SSE [reduce] since the last reseed
375 * pivot ([beginReseed]) or seed. Parallels [streamedSincePivot] but for message info:
376 * on a reconnect re-seed these are the messages whose info changed *during* the REST
377 * fetch, so the in-memory info is newer than the snapshot and must win; every other
378 * message's info is taken from the authoritative REST snapshot. Without this the merge
379 * kept the in-memory info, so a run that finished server-side during the disconnect
380 * (e.g. cost/token totals, completion time) stayed stale until app restart. */
381 private val messageInfoUpdatedSincePivot = mutableSetOf<String>()
383 /** Mark the point a reconnect re-seed's REST fetch begins: parts/info streamed from here
384 * on are newer than the fetched snapshot. Call under the same lock as [seed]/[reduce]. */
385 fun beginReseed() {
386 streamedSincePivot.clear()
387 messageInfoUpdatedSincePivot.clear()
388 }
390 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */
391 internal val maxMessages = NetworkConfig.maxInMemoryMessages
393 fun snapshot(): List<MessageWithParts> = messages.values.map { it.view() }
395 fun seed(initial: List<MessageWithParts>, prune: Boolean = false, clearPivot: Boolean = true): Boolean {
396 var changed = false
397 // On re-seed (after SSE reconnect), remove messages that are no longer in the
398 // server snapshot (e.g. deleted during the disconnection gap). This keeps the
399 // in-memory state in sync with the authoritative REST snapshot rather than
400 // accumulating stale messages forever. Skipped on the initial seed to avoid
401 // racing with just-arrived SSE events whose messages may not yet be in REST.
402 if (prune) {
403 val incomingIds = initial.mapTo(mutableSetOf()) { it.info.id }
404 val before = messages.size
405 messages.keys.retainAll(incomingIds)
406 if (messages.size != before) changed = true
407 }
409 for (m in initial) {
410 val existing = messages[m.info.id]
411 if (existing == null) {
412 messages[m.info.id] = holderOf(m)
413 changed = true
414 } else {
415 // A part streamed in between subscribe and this initial load may already
416 // have populated this message (see observeMessages). Merge instead of
417 // overwriting so that newer streamed part isn't discarded: take the
418 // snapshot's part order, swap in the streamed version where ids overlap,
419 // append any streamed-only parts, and adopt the authoritative REST info.
420 // existing.parts is already keyed by part id (insertion-ordered).
421 val streamedById = existing.parts
422 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
423 // Prefer the in-memory version of an overlapping part ONLY if it streamed
424 // in since the reseed pivot (i.e. during this fetch), so it's newer than
425 // the snapshot. Otherwise the REST snapshot is authoritative and wins,
426 // discarding an in-memory part that went stale during a disconnect.
427 val ordered = m.parts.map { p ->
428 if (p.id in streamedSincePivot) streamedById[p.id] ?: p else p
429 }.toMutableList()
430 // Append only parts that streamed in *since the pivot* and are absent from the
431 // snapshot — i.e. genuinely newer than this REST fetch. A part in memory from
432 // before the pivot that the snapshot omits was removed server-side (its
433 // MessagePartRemoved missed during a disconnect gap, or it was deleted while the
434 // app was closed and only survives from the cache seed); the authoritative REST
435 // snapshot must be allowed to drop it, else it reappears as a ghost — and, once
436 // the drain loop persists this merge, gets baked into the on-disk cache forever
437 // (the message still exists, so pruneStaleCacheSeeded never removes it). This
438 // mirrors the overlapping-part branch above and the info branch below, which both
439 // already gate on the pivot.
440 for (p in existing.parts.values) {
441 if (p.id !in snapshotIds && p.id in streamedSincePivot) ordered.add(p)
442 }
443 // Adopt the authoritative REST info unless the in-memory info was itself
444 // updated live since the pivot (i.e. during this fetch), in which case it's
445 // newer than the snapshot. Otherwise the REST snapshot wins, discarding
446 // in-memory info that went stale during a disconnect.
447 val info = if (existing.info is UnknownMessage || m.info.id !in messageInfoUpdatedSincePivot) m.info else existing.info
448 val merged = MessageWithParts(info = info, parts = ordered)
449 if (merged != existing.view()) {
450 messages[m.info.id] = holderOf(merged)
451 changed = true
452 }
453 }
454 }
455 // A part that streamed in *before* this initial load (observeMessages subscribes to
456 // SSE before fetching) inserts its message at the front of the insertion-ordered map,
457 // and the merge above keeps that position — pinning a running message above the older
458 // history that seed() inserts afterward. Restore chronological order by creation time
459 // FIRST, so eviction below drops the genuinely-oldest entries: evictOldMessages()
460 // removes from the map's front, which before reordering is the newest streamed message.
461 if (reorderByTime()) changed = true
462 if (evictOldMessages()) changed = true
463 // Reset the pivot: everything just merged is now the baseline, so only parts/info
464 // that stream in *after* this seed can outrank the next reconnect's snapshot. The
465 // on-disk cache seed passes clearPivot = false: it runs *before* the authoritative
466 // initial REST seed, so clearing here would strip pivot protection from SSE parts/info
467 // that streamed in since subscribe, letting that REST seed regress them.
468 if (clearPivot) {
469 streamedSincePivot.clear()
470 messageInfoUpdatedSincePivot.clear()
471 }
472 return changed
473 }
475 /** Remove messages that were seeded from the on-disk cache ([cacheSeededIds]) but are
476 * absent from the authoritative REST snapshot [rest] — i.e. deleted server-side while the
477 * app was closed. Restricted to cache-seeded ids so a message added by a live SSE event
478 * since subscribe (which the REST endpoint may not have indexed yet) is never pruned.
479 * This is what actually removes ghosts on restart; the reconnect re-seed deliberately
480 * keeps prune=false to avoid racing just-arrived SSE. Call under the same lock as [seed].
481 * Returns true if anything was removed. */
482 fun pruneStaleCacheSeeded(cacheSeededIds: Set<String>, rest: List<MessageWithParts>): Boolean {
483 if (cacheSeededIds.isEmpty()) return false
484 val restIds = rest.mapTo(mutableSetOf()) { it.info.id }
485 var changed = false
486 for (id in cacheSeededIds) {
487 if (id !in restIds && messages.remove(id) != null) changed = true
488 }
489 return changed
490 }
492 /** Reorder the message map by message creation time (stable). Ties and entries with no
493 * server time yet (e.g. a brand-new message still streaming and not in REST) keep their
494 * relative order, so such a message stays last. Preserves each entry's map key so the
495 * synthetic UnknownMessage keys aren't collapsed. Returns true if the order changed. */
496 private fun reorderByTime(): Boolean {
497 val entries = messages.entries.map { it.key to it.value }
498 val sorted = entries.sortedBy { it.second.info.time?.created ?: Long.MAX_VALUE }
499 // Detect a reordering by comparing keys positionally, instead of allocating two more
500 // N-sized key lists just to compare them.
501 var changed = false
502 for (i in sorted.indices) {
503 if (sorted[i].first != entries[i].first) { changed = true; break }
504 }
505 if (!changed) return false
506 messages.clear()
507 for ((k, v) in sorted) messages[k] = v
508 return true
509 }
511 /** Returns true if the state changed (and a new snapshot should be published). */
512 fun reduce(sessionId: String, event: BusEvent): Boolean {
513 return when (event) {
514 is MessageUpdated -> handleMessageUpdated(sessionId, event)
516 is MessagePartUpdated -> {
517 val part = event.properties.part
518 val messageId = part.messageID ?: event.properties.messageID
519 val partSession = part.sessionID ?: event.properties.sessionID
520 when {
521 messageId == null -> false
522 partSession != null -> if (partSession == sessionId) upsertPart(messageId, part) else false
523 // No session id anywhere on the event: a null session is NOT a wildcard
524 // (mirroring isIdle/isError). Every open session's store sees the shared
525 // event stream, so accepting it unconditionally would leak the part into
526 // unrelated conversations (e.g. both panes in two-pane mode). Attribute it
527 // only if this store already holds the parent message — MessageUpdated,
528 // which always carries a session id, will have created it here first.
529 else -> if (messages.containsKey(messageId)) upsertPart(messageId, part) else false
530 }
531 }
533 is MessagePartRemoved -> handlePartRemoved(sessionId, event)
535 is MessageRemoved -> handleMessageRemoved(sessionId, event)
537 else -> false
538 }
539 }
541 private fun handleMessageUpdated(sessionId: String, event: MessageUpdated): Boolean {
542 val info = event.properties.info
543 if (info.sessionID != sessionId) return false
544 // An UnknownMessage with an empty id (e.g. from an unrecognized server role with no id
545 // field) would collide with other such messages in the map. It carries no id, only
546 // sessionID/time, so derive a synthetic key. Prefer the immutable creation timestamp: it's
547 // stable across updates, so a later update to the SAME unknown message (whose `time.updated`
548 // or `time.completed` changes) coalesces onto its holder via the existing==info guard below
549 // instead of forking a duplicate — the failure mode a full-content hash has, since any
550 // content change moves the hash. Distinct unknown messages still separate by their distinct
551 // creation times; only when even `created` is absent do we fall back to the content hash.
552 val key = if (info.id.isEmpty() && info is UnknownMessage) {
553 info.time?.created?.let { "unknown-c$it" } ?: "unknown-h${info.hashCode()}"
554 } else {
555 info.id
556 }
557 val existing = messages[key]
558 // A server re-sending byte-identical info is a no-op: skip it so we don't publish a
559 // redundant snapshot (mirrors upsertPart's identical-part guard).
560 if (existing != null && existing.info == info) return false
561 if (existing == null) {
562 messages[key] = Holder(info, LinkedHashMap())
563 } else {
564 existing.info = info
565 existing.invalidate()
566 }
567 // Record that this message's info streamed in live, so a re-seed after a reconnect knows
568 // it's newer than the REST snapshot and keeps it (see seed()/beginReseed()).
569 messageInfoUpdatedSincePivot.add(key)
570 if (existing == null) evictOldMessages()
571 return true
572 }
574 /** Returns false if [eventSession] is non-null and doesn't match [sessionId]. */
575 private fun matchesSession(eventSession: String?, sessionId: String): Boolean =
576 eventSession == null || eventSession == sessionId
578 private fun handlePartRemoved(sessionId: String, event: MessagePartRemoved): Boolean {
579 val messageId = event.properties.messageID ?: return false
580 val partId = event.properties.partID ?: return false
581 if (!matchesSession(event.properties.sessionID, sessionId)) return false
582 return removePart(messageId, partId)
583 }
585 private fun handleMessageRemoved(sessionId: String, event: MessageRemoved): Boolean {
586 val id = event.properties.messageID ?: return false
587 if (!matchesSession(event.properties.sessionID, sessionId)) return false
588 val removed = messages.remove(id) ?: return false
589 messageInfoUpdatedSincePivot.remove(id)
590 removed.parts.keys.forEach { streamedSincePivot.remove(it) }
591 return true
592 }
594 private fun upsertPart(messageId: String, part: Part): Boolean {
595 val current = messages[messageId]
596 if (current != null) {
597 // O(1) map lookup + put, vs the former O(P) indexOfFirst scan and full-list copy
598 // on every streamed token. A LinkedHashMap re-put keeps the part's original
599 // position (replace-in-place); a new id is appended — matching the old list order.
600 if (current.parts[part.id] == part) return false
601 current.parts[part.id] = part
602 current.invalidate()
603 } else {
604 val parts = LinkedHashMap<String, Part>()
605 parts[part.id] = part
606 messages[messageId] = Holder(
607 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
608 parts = parts,
609 )
610 }
611 // Record that this part streamed in, so a re-seed after a reconnect knows it's
612 // newer than the REST snapshot and keeps it (see seed()/beginReseed()).
613 streamedSincePivot.add(part.id)
614 if (current == null) evictOldMessages()
615 return true
616 }
618 private fun removePart(messageId: String, partId: String): Boolean {
619 val current = messages[messageId] ?: return false
620 if (current.parts.remove(partId) == null) return false
621 // Drop the pivot entry too so the set doesn't accumulate ids of removed parts for
622 // the rest of the session (it's otherwise only cleared on reseed/eviction). No
623 // correctness impact today — a stale entry just falls back to the REST snapshot's
624 // version in seed() — but keeping it symmetric with handleMessageRemoved/eviction
625 // bounds the set's growth over a long-lived session.
626 streamedSincePivot.remove(partId)
627 current.invalidate()
628 return true
629 }
631 /** Evict the oldest messages when the store exceeds [maxMessages], keeping memory bounded.
632 * Returns true if any messages were evicted. */
633 private fun evictOldMessages(): Boolean {
634 var evicted = false
635 while (messages.size > maxMessages) {
636 val oldestKey = messages.keys.iterator().next()
637 val removed = messages.remove(oldestKey)
638 // Prune the evicted message's pivot bookkeeping so these sets don't grow
639 // unbounded over a long-running session that never reconnects (they're only
640 // otherwise cleared on reseed/seed).
641 messageInfoUpdatedSincePivot.remove(oldestKey)
642 removed?.parts?.keys?.forEach { streamedSincePivot.remove(it) }
643 evicted = true
644 }
645 return evicted
646 }