@@ -20,6 +20,7 @@ import soy.iko.opencode.data.network.OpencodeApiClient
20import soy.iko.opencode.util.runCatchingCancellable
20import soy.iko.opencode.util.runCatchingCancellable
21import soy.iko.opencode.util.safeExceptionSummary
21import soy.iko.opencode.util.safeExceptionSummary
22import android.util.Log
22import android.util.Log
23+import kotlinx.coroutines.CoroutineScope
23import kotlinx.coroutines.Dispatchers
24import kotlinx.coroutines.Dispatchers
24import kotlinx.coroutines.Job
25import kotlinx.coroutines.Job
25import kotlinx.coroutines.NonCancellable
26import kotlinx.coroutines.NonCancellable
@@ -46,6 +47,9 @@ open class SessionRepository(
46 // Optional on-disk cache so a conversation renders instantly on open and stays readable
47 // Optional on-disk cache so a conversation renders instantly on open and stays readable
47 // offline. Null in tests (and any path that doesn't wire it), where caching is skipped.
48 // offline. Null in tests (and any path that doesn't wire it), where caching is skipped.
48 private val messageCache: MessageCacheStore? = null,
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 = "",
50 open suspend fun listSessions() = api.listSessions()
54 open suspend fun listSessions() = api.listSessions()
51 open suspend fun createSession(title: String? = null, directory: String? = null) =
55 open suspend fun createSession(title: String? = null, directory: String? = null) =
@@ -78,6 +82,58 @@ open class SessionRepository(
78 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
82 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
79 * event reducer — a slow collector simply sees the most recent snapshot.
83 * event reducer — a slow collector simply sees the most recent snapshot.
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
106+ val now = System.currentTimeMillis()
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) {
116+ NetworkConfig.messageCacheWriteThrottleMs
118+ pending = scope.launch {
119+ if (delayMs > 0) delay(delayMs)
120+ cache.save(profileId, sessionId, snapshot)
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) }
81 open fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
137 open fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
82 val store = MessageStore()
138 val store = MessageStore()
83 val lock = Mutex()
139 val lock = Mutex()
@@ -118,12 +174,12 @@ open class SessionRepository(
118 // reconnect re-seed that somehow raced ahead isn't clobbered; the initial REST seed
174 // reconnect re-seed that somehow raced ahead isn't clobbered; the initial REST seed
119 // (also generation 0) then overrides this with authoritative data when it succeeds.
175 // (also generation 0) then overrides this with authoritative data when it succeeds.
120 messageCache?.let { cache ->
176 messageCache?.let { cache ->
121− val cached = cache.load(sessionId)
177+ val cached = cache.load(profileId, sessionId)
122 if (cached.isNotEmpty()) {
178 if (cached.isNotEmpty()) {
123 val changed = lock.withLock {
179 val changed = lock.withLock {
124 if (seedGeneration == 0) {
180 if (seedGeneration == 0) {
125 cacheSeededIds = cached.mapTo(mutableSetOf()) { it.info.id }
181 cacheSeededIds = cached.mapTo(mutableSetOf()) { it.info.id }
126− store.seed(cached)
182+ store.seed(cached, clearPivot = false)
127 } else false
183 } else false
129 if (changed) publish()
185 if (changed) publish()
@@ -212,18 +268,10 @@ open class SessionRepository(
212 // outside the lock so a slow downstream collector can't stall the drain. This
268 // outside the lock so a slow downstream collector can't stall the drain. This
213 // loop also keeps the flow alive until the collector cancels; the launched jobs
269 // loop also keeps the flow alive until the collector cancels; the launched jobs
214 // above are children of this scope and are torn down with it.
270 // above are children of this scope and are torn down with it.
215− var lastCacheWrite = 0L
271+ // Per-subscription cache writer: throttled snapshot persistence + a NonCancellable
216− var pendingCacheSave: Job? = null
272+ // teardown flush. `this` is the channelFlow scope, so its scheduled saves are torn down
217− // Most recent published snapshot, and the most recent one whose on-disk write has
273+ // with the flow (the teardown flush re-persists a cancelled trailing write).
218− // actually completed. The teardown flush in the finally below persists a pending
274+ val cacheWriter = CacheWriter(this, sessionId)
219− // trailing snapshot exactly once. savedSnapshot is assigned from inside the launched
220− // saves *after* messageCache.save() returns, so a save cancelled mid-IO by teardown is
221− // NOT marked saved and the finally re-persists it under NonCancellable. It's written
222− // from the save coroutines and read from the collector/finally; the only possible race
223− // is a stale read causing one redundant, idempotent re-save (atomic temp-file+rename
224− // makes a duplicate write of identical data harmless) — never data loss.
225− var latestSnapshot: List<MessageWithParts>? = null
226− var savedSnapshot: List<MessageWithParts>? = null
227 // Guarantee at least one emission even when nothing seeds the store: with no on-disk
275 // Guarantee at least one emission even when nothing seeds the store: with no on-disk
228 // cache, a failing initial REST load (swallowed above), and a quiet SSE stream, none of
276 // cache, a failing initial REST load (swallowed above), and a quiet SSE stream, none of
229 // the seed paths publish, so the flow would emit nothing at all. The sole collector
277 // the seed paths publish, so the flow would emit nothing at all. The sole collector
@@ -236,46 +284,13 @@ open class SessionRepository(
236 dirty.consumeAsFlow().collect {
284 dirty.consumeAsFlow().collect {
237 val snapshot = lock.withLock { store.snapshot() }
285 val snapshot = lock.withLock { store.snapshot() }
238 send(snapshot)
286 send(snapshot)
239− // Persist the snapshot to the on-disk cache, throttled so a fast token stream
287+ // Persist to the on-disk cache, throttled so a fast token stream doesn't hammer
240− // doesn't hammer the disk — the cache only needs to be "recent enough" for an
288+ // the disk — the cache only needs to be "recent enough" for an instant/offline
241− // instant/offline first paint; the network corrects it on the next open.
289+ // first paint; the network corrects it on the next open.
242− if (messageCache != null && snapshot.isNotEmpty()) {
290+ cacheWriter.schedule(snapshot)
243− latestSnapshot = snapshot
244− val now = System.currentTimeMillis()
245− pendingCacheSave?.cancel()
246− if (now - lastCacheWrite >= NetworkConfig.messageCacheWriteThrottleMs) {
247− lastCacheWrite = now
248− pendingCacheSave = launch {
249− messageCache.save(sessionId, snapshot)
250− savedSnapshot = snapshot
253− // Trailing-edge write: a burst that ends inside the throttle window (a run's
254− // final tokens right after a write) would otherwise never persist its last
255− // snapshot, so an offline reopen would show the conversation missing its tail.
256− // Write the latest snapshot once the window elapses; a newer snapshot cancels
257− // and replaces this, so continuous streaming still coalesces to one write/window.
258− pendingCacheSave = launch {
259− delay(NetworkConfig.messageCacheWriteThrottleMs)
260− messageCache.save(sessionId, snapshot)
261− savedSnapshot = snapshot
266 } finally {
292 } finally {
267− // FIX: both cache-saves above run as children of this inner flow scope, so when
293+ cacheWriter.flushOnTeardown()
268− // ChatViewModel's flatMapLatest switches (an SSE reconnect changing activeConnection)
269− // or onCleared cancels this flow, a pending trailing save is torn down before its delay
270− // elapses — a run's final tokens then never reach the on-disk cache. Flush the latest
271− // un-persisted snapshot here under NonCancellable so it survives the cancellation.
272− // savedSnapshot tracks the snapshot an immediate write already handled, so the common
273− // case (nothing pending) writes nothing extra — at most one write per teardown, never
274− // one per emission.
275− val pending = latestSnapshot
276− if (messageCache != null && pending != null && pending !== savedSnapshot) {
277− withContext(NonCancellable) { messageCache.save(sessionId, pending) }
280 // Run the whole reducer pipeline — event reduction, list rebuilds, and the O(N)
295 // Run the whole reducer pipeline — event reduction, list rebuilds, and the O(N)
281 // snapshot copy — on Dispatchers.Default rather than the collector's context. The
296 // snapshot copy — on Dispatchers.Default rather than the collector's context. The
@@ -380,7 +395,7 @@ internal class MessageStore {
381 fun snapshot(): List<MessageWithParts> = messages.values.map { it.view() }
396 fun snapshot(): List<MessageWithParts> = messages.values.map { it.view() }
383− fun seed(initial: List<MessageWithParts>, prune: Boolean = false): Boolean {
398+ fun seed(initial: List<MessageWithParts>, prune: Boolean = false, clearPivot: Boolean = true): Boolean {
384 var changed = false
399 var changed = false
385 // On re-seed (after SSE reconnect), remove messages that are no longer in the
400 // On re-seed (after SSE reconnect), remove messages that are no longer in the
386 // server snapshot (e.g. deleted during the disconnection gap). This keeps the
401 // server snapshot (e.g. deleted during the disconnection gap). This keeps the
@@ -439,9 +454,14 @@ internal class MessageStore {
439 if (reorderByTime()) changed = true
454 if (reorderByTime()) changed = true
440 if (evictOldMessages()) changed = true
455 if (evictOldMessages()) changed = true
441 // Reset the pivot: everything just merged is now the baseline, so only parts/info
456 // Reset the pivot: everything just merged is now the baseline, so only parts/info
442− // that stream in *after* this seed can outrank the next reconnect's snapshot.
457+ // that stream in *after* this seed can outrank the next reconnect's snapshot. The
458+ // on-disk cache seed passes clearPivot = false: it runs *before* the authoritative
459+ // initial REST seed, so clearing here would strip pivot protection from SSE parts/info
460+ // that streamed in since subscribe, letting that REST seed regress them.
443− streamedSincePivot.clear()
462+ streamedSincePivot.clear()
444− messageInfoUpdatedSincePivot.clear()
463+ messageInfoUpdatedSincePivot.clear()
445 return changed
465 return changed