1package soy.iko.opencode.data.repo
3import soy.iko.opencode.data.model.BusEvent
4import soy.iko.opencode.data.model.MessagePartRemoved
5import soy.iko.opencode.data.model.MessagePartUpdated
6import soy.iko.opencode.data.model.MessageRemoved
7import soy.iko.opencode.data.model.MessageUpdated
8import soy.iko.opencode.data.model.MessageWithParts
9import soy.iko.opencode.data.model.ModelRef
10import soy.iko.opencode.data.model.Part
11import soy.iko.opencode.data.model.SessionError
12import soy.iko.opencode.data.model.SessionIdle
13import soy.iko.opencode.data.model.UnknownMessage
14import soy.iko.opencode.data.network.EventStreamClient
15import soy.iko.opencode.data.network.NetworkConfig
16import soy.iko.opencode.data.network.OpencodeApiClient
17import soy.iko.opencode.util.runCatchingCancellable
18import soy.iko.opencode.util.safeExceptionSummary
19import android.util.Log
20import kotlinx.coroutines.channels.Channel
21import kotlinx.coroutines.flow.Flow
22import kotlinx.coroutines.flow.channelFlow
23import kotlinx.coroutines.flow.consumeAsFlow
24import kotlinx.coroutines.flow.conflate
25import kotlinx.coroutines.launch
26import kotlinx.coroutines.sync.Mutex
27import kotlinx.coroutines.sync.withLock
30 * Source of truth for a session's conversation. It seeds from `GET /session/:id/message`
31 * and then reduces [BusEvent]s from the SSE stream into an ordered list of messages,
32 * each keyed by message id with parts keyed by part id (idempotent upserts).
34open class SessionRepository(
35 private val api: OpencodeApiClient,
36 private val eventStream: EventStreamClient,
38 open suspend fun listSessions() = api.listSessions()
39 open suspend fun createSession(title: String? = null) = api.createSession(title)
40 open suspend fun deleteSession(id: String) = api.deleteSession(id)
41 open suspend fun abort(sessionId: String) = api.abort(sessionId)
43 open suspend fun sendPrompt(sessionId: String, text: String, model: ModelRef?, agent: String? = null) =
44 api.sendPrompt(sessionId, text, model, agent)
46 open suspend fun runCommand(
49 arguments: String = "",
50 agent: String? = null,
51 ) = api.runCommand(sessionId, command, arguments = arguments, agent = agent)
54 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
55 * stream before the initial REST load so no streamed part is missed, then reduces
56 * events into the in-memory state. The flow is conflated so a fast burst of
57 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
58 * event reducer — a slow collector simply sees the most recent snapshot.
60 open fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
61 val store = MessageStore()
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)
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) }
76 // Subscribe to events first so we don't miss early deltas during the initial load.
78 eventStream.events.collect { event ->
79 if (lock.withLock { store.reduce(sessionId, event) }) publish()
83 // Seed generation counter, shared between the initial load and the reconnect
84 // re-seed. Guards against a stale fetch (from an earlier reconnect, or from a slow
85 // initial load) clobbering a newer seed. Only mutated/read under [lock].
86 var seedGeneration = 0
88 // Re-seed from REST when the SSE stream reconnects after a drop. The REST fetch
89 // is performed *outside* the lock so events arriving via SSE during the fetch are
90 // not blocked. The seed itself is under the lock; a seed generation counter ensures
91 // a stale fetch (from an earlier reconnect) doesn't clobber a newer one.
93 // Re-seed from REST when the SSE stream reconnects after a drop. The
94 // hasConnectedBefore flag is NOT reset on disconnect — it stays true after
95 // the first successful connection so a subsequent reconnect triggers the
96 // re-seed. Resetting it on disconnect would make the condition always false
97 // after the first cycle, turning the re-seed into dead code.
99 // A seed generation counter guards against a stale re-seed clobbering a
100 // newer one: if two reconnects fire in quick succession, the older fetch
101 // (which returns later) is discarded because its generation no longer matches.
102 var hasConnectedBefore = false
103 eventStream.state.collect { state ->
104 if (state == EventStreamClient.ConnectionState.Connected) {
105 if (hasConnectedBefore) {
106 // Pivot before the fetch: parts already in memory predate this
107 // snapshot and must yield to it, while parts that stream in during
108 // the fetch (added to streamedSincePivot via reduce) outrank it.
109 // Bump the generation under the same lock so the compare in the
110 // initial seed sees a consistent value.
111 val generation = lock.withLock {
115 val fresh = runCatchingCancellable { api.listMessages(sessionId) }
116 .onFailure { Log.w("SessionRepository", "Re-seed message load failed for $sessionId; relying on SSE: ${safeExceptionSummary(it)}") }
117 .getOrDefault(emptyList())
118 var seedChanged = false
120 if (generation == seedGeneration) {
121 // Merge without pruning: SSE events arriving during the
122 // REST fetch may have added messages not yet indexed by
123 // the REST endpoint. Pruning would silently delete them.
124 // Stale messages from deletions during the disconnect
125 // gap are eventually removed via MessageRemoved events
126 // or on the next app restart.
127 seedChanged = store.seed(fresh, prune = false)
130 if (seedChanged) publish()
132 hasConnectedBefore = true
137 val initial = runCatchingCancellable { api.listMessages(sessionId) }
138 .onFailure { Log.w("SessionRepository", "Initial message load failed for $sessionId; relying on SSE: ${safeExceptionSummary(it)}") }
139 .getOrDefault(emptyList())
140 // Apply the initial seed only if no reconnect re-seed has run yet (generation still 0).
141 // A re-seed is always a fresher full snapshot, so if the (possibly slow) initial fetch
142 // returns after one has applied, seeding this older REST data would clobber the newer
143 // state — reverting a completed run to "running", regressing cost/token/part state —
144 // until the next SSE event or reconnect healed it.
145 val seeded = lock.withLock {
146 if (seedGeneration == 0) {
151 if (seeded) publish()
153 // Single drain: one snapshot+send per conflated dirty signal. The snapshot is
154 // taken under the lock so a reader never sees a half-reduced state; send() runs
155 // outside the lock so a slow downstream collector can't stall the drain. This
156 // loop also keeps the flow alive until the collector cancels; the launched jobs
157 // above are children of this scope and are torn down with it.
158 dirty.consumeAsFlow().collect {
159 val snapshot = lock.withLock { store.snapshot() }
165 /** Convenience: is this event a run-completion signal for [sessionId]?
166 * A null sessionID is NOT treated as a wildcard — an idle event with no
167 * session id must not reset the running state of every open chat. */
168 fun isIdle(event: BusEvent, sessionId: String): Boolean =
169 event is SessionIdle && event.properties.sessionID == sessionId
171 fun isError(event: BusEvent, sessionId: String): Boolean =
172 event is SessionError && event.properties.sessionID == sessionId
177 * In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex.
178 * Exposed as `internal` so the reducer logic can be unit-tested directly.
180internal class MessageStore {
181 // messageId -> (info + parts), insertion-ordered.
182 private val messages = LinkedHashMap<String, MessageWithParts>()
184 /** Part ids upserted via [reduce] since the last reseed pivot ([beginReseed]) or seed.
185 * On a reconnect re-seed these are the parts that streamed in *during* the REST fetch,
186 * so they're newer than the snapshot and must win over it; every other part is taken
187 * from the authoritative REST snapshot. Without this distinction the merge kept every
188 * overlapping in-memory part, so a part that changed server-side during the disconnect
189 * (e.g. a tool going running -> completed) stayed stale until app restart. */
190 private val streamedSincePivot = mutableSetOf<String>()
192 /** Message ids whose `info` was updated via a live SSE [reduce] since the last reseed
193 * pivot ([beginReseed]) or seed. Parallels [streamedSincePivot] but for message info:
194 * on a reconnect re-seed these are the messages whose info changed *during* the REST
195 * fetch, so the in-memory info is newer than the snapshot and must win; every other
196 * message's info is taken from the authoritative REST snapshot. Without this the merge
197 * kept the in-memory info, so a run that finished server-side during the disconnect
198 * (e.g. cost/token totals, completion time) stayed stale until app restart. */
199 private val messageInfoUpdatedSincePivot = mutableSetOf<String>()
201 /** Mark the point a reconnect re-seed's REST fetch begins: parts/info streamed from here
202 * on are newer than the fetched snapshot. Call under the same lock as [seed]/[reduce]. */
204 streamedSincePivot.clear()
205 messageInfoUpdatedSincePivot.clear()
208 /** Monotonic counter for synthetic UnknownMessage keys, avoiding nanoTime collisions. */
209 private val unknownCounter = java.util.concurrent.atomic.AtomicLong(0)
211 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */
212 internal val maxMessages = NetworkConfig.maxInMemoryMessages
214 fun snapshot(): List<MessageWithParts> = messages.values.toList()
216 fun seed(initial: List<MessageWithParts>, prune: Boolean = false): Boolean {
218 // On re-seed (after SSE reconnect), remove messages that are no longer in the
219 // server snapshot (e.g. deleted during the disconnection gap). This keeps the
220 // in-memory state in sync with the authoritative REST snapshot rather than
221 // accumulating stale messages forever. Skipped on the initial seed to avoid
222 // racing with just-arrived SSE events whose messages may not yet be in REST.
224 val incomingIds = initial.mapTo(mutableSetOf()) { it.info.id }
225 val before = messages.size
226 messages.keys.retainAll(incomingIds)
227 if (messages.size != before) changed = true
231 val existing = messages[m.info.id]
232 if (existing == null) {
233 messages[m.info.id] = m
236 // A part streamed in between subscribe and this initial load may already
237 // have populated this message (see observeMessages). Merge instead of
238 // overwriting so that newer streamed part isn't discarded: take the
239 // snapshot's part order, swap in the streamed version where ids overlap,
240 // append any streamed-only parts, and adopt the authoritative REST info.
241 val streamedById = existing.parts.associateBy { it.id }
242 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
243 // Prefer the in-memory version of an overlapping part ONLY if it streamed
244 // in since the reseed pivot (i.e. during this fetch), so it's newer than
245 // the snapshot. Otherwise the REST snapshot is authoritative and wins,
246 // discarding an in-memory part that went stale during a disconnect.
247 val ordered = m.parts.map { p ->
248 if (p.id in streamedSincePivot) streamedById[p.id] ?: p else p
250 for (p in existing.parts) {
251 if (p.id !in snapshotIds) ordered.add(p)
253 // Adopt the authoritative REST info unless the in-memory info was itself
254 // updated live since the pivot (i.e. during this fetch), in which case it's
255 // newer than the snapshot. Otherwise the REST snapshot wins, discarding
256 // in-memory info that went stale during a disconnect.
257 val info = if (existing.info is UnknownMessage || m.info.id !in messageInfoUpdatedSincePivot) m.info else existing.info
258 val merged = MessageWithParts(info = info, parts = ordered)
259 if (merged != existing) {
260 messages[m.info.id] = merged
265 // A part that streamed in *before* this initial load (observeMessages subscribes to
266 // SSE before fetching) inserts its message at the front of the insertion-ordered map,
267 // and the merge above keeps that position — pinning a running message above the older
268 // history that seed() inserts afterward. Restore chronological order by creation time
269 // FIRST, so eviction below drops the genuinely-oldest entries: evictOldMessages()
270 // removes from the map's front, which before reordering is the newest streamed message.
271 if (reorderByTime()) changed = true
272 if (evictOldMessages()) changed = true
273 // Reset the pivot: everything just merged is now the baseline, so only parts/info
274 // that stream in *after* this seed can outrank the next reconnect's snapshot.
275 streamedSincePivot.clear()
276 messageInfoUpdatedSincePivot.clear()
280 /** Reorder the message map by message creation time (stable). Ties and entries with no
281 * server time yet (e.g. a brand-new message still streaming and not in REST) keep their
282 * relative order, so such a message stays last. Preserves each entry's map key so the
283 * synthetic UnknownMessage keys aren't collapsed. Returns true if the order changed. */
284 private fun reorderByTime(): Boolean {
285 val entries = messages.entries.map { it.key to it.value }
286 val sorted = entries.sortedBy { it.second.info.time?.created ?: Long.MAX_VALUE }
287 if (sorted.map { it.first } == entries.map { it.first }) return false
289 for ((k, v) in sorted) messages[k] = v
293 /** Returns true if the state changed (and a new snapshot should be published). */
294 fun reduce(sessionId: String, event: BusEvent): Boolean {
295 return when (event) {
296 is MessageUpdated -> {
297 val info = event.properties.info
298 if (info.sessionID != sessionId) {
301 // An UnknownMessage with an empty id (e.g. from an unrecognized
302 // server role with no id field) would collide with other such
303 // messages in the map. Generate a unique synthetic key so each
304 // unknown message gets its own entry instead of overwriting others.
305 val key = if (info.id.isEmpty() && info is UnknownMessage) {
306 "unknown-${unknownCounter.incrementAndGet()}"
310 val existing = messages[key]
311 messages[key] = existing?.copy(info = info) ?: MessageWithParts(info)
312 // Record that this message's info streamed in live, so a re-seed after a
313 // reconnect knows it's newer than the REST snapshot and keeps it (see
314 // seed()/beginReseed()).
315 messageInfoUpdatedSincePivot.add(key)
316 if (existing == null) evictOldMessages()
321 is MessagePartUpdated -> {
322 val part = event.properties.part
323 val messageId = part.messageID ?: event.properties.messageID
324 val partSession = part.sessionID ?: event.properties.sessionID
326 messageId == null -> false
327 partSession != null -> if (partSession == sessionId) upsertPart(messageId, part) else false
328 // No session id anywhere on the event: a null session is NOT a wildcard
329 // (mirroring isIdle/isError). Every open session's store sees the shared
330 // event stream, so accepting it unconditionally would leak the part into
331 // unrelated conversations (e.g. both panes in two-pane mode). Attribute it
332 // only if this store already holds the parent message — MessageUpdated,
333 // which always carries a session id, will have created it here first.
334 else -> if (messages.containsKey(messageId)) upsertPart(messageId, part) else false
338 is MessagePartRemoved -> handlePartRemoved(sessionId, event)
340 is MessageRemoved -> handleMessageRemoved(sessionId, event)
346 /** Returns false if [eventSession] is non-null and doesn't match [sessionId]. */
347 private fun matchesSession(eventSession: String?, sessionId: String): Boolean =
348 eventSession == null || eventSession == sessionId
350 private fun handlePartRemoved(sessionId: String, event: MessagePartRemoved): Boolean {
351 val messageId = event.properties.messageID ?: return false
352 val partId = event.properties.partID ?: return false
353 if (!matchesSession(event.properties.sessionID, sessionId)) return false
354 return removePart(messageId, partId)
357 private fun handleMessageRemoved(sessionId: String, event: MessageRemoved): Boolean {
358 val id = event.properties.messageID ?: return false
359 if (!matchesSession(event.properties.sessionID, sessionId)) return false
360 val removed = messages.remove(id) ?: return false
361 messageInfoUpdatedSincePivot.remove(id)
362 removed.parts.forEach { streamedSincePivot.remove(it.id) }
366 private fun upsertPart(messageId: String, part: Part): Boolean {
367 val current = messages[messageId]
368 val parts = current?.parts.orEmpty()
369 val idx = parts.indexOfFirst { it.id == part.id }
370 if (idx >= 0 && parts[idx] == part) return false
371 val newParts = if (idx >= 0) {
372 parts.toMutableList().also { it[idx] = part }
376 messages[messageId] = current?.copy(parts = newParts)
378 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
381 // Record that this part streamed in, so a re-seed after a reconnect knows it's
382 // newer than the REST snapshot and keeps it (see seed()/beginReseed()).
383 streamedSincePivot.add(part.id)
384 if (current == null) evictOldMessages()
388 private fun removePart(messageId: String, partId: String): Boolean {
389 val current = messages[messageId] ?: return false
390 val newParts = current.parts.filterNot { it.id == partId }
391 if (newParts.size == current.parts.size) return false
392 messages[messageId] = current.copy(parts = newParts)
396 /** Evict the oldest messages when the store exceeds [maxMessages], keeping memory bounded.
397 * Returns true if any messages were evicted. */
398 private fun evictOldMessages(): Boolean {
400 while (messages.size > maxMessages) {
401 val oldestKey = messages.keys.iterator().next()
402 val removed = messages.remove(oldestKey)
403 // Prune the evicted message's pivot bookkeeping so these sets don't grow
404 // unbounded over a long-running session that never reconnects (they're only
405 // otherwise cleared on reseed/seed).
406 messageInfoUpdatedSincePivot.remove(oldestKey)
407 removed?.parts?.forEach { streamedSincePivot.remove(it.id) }
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list