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