git.iko.soy/opencode-android
on 66d239c help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
236 lines 10 kB history raw
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 android.util.Log
18import kotlinx.coroutines.flow.Flow
19import kotlinx.coroutines.flow.channelFlow
20import kotlinx.coroutines.flow.conflate
21import kotlinx.coroutines.launch
22import kotlinx.coroutines.sync.Mutex
23import kotlinx.coroutines.sync.withLock
25/**
26 * Source of truth for a session's conversation. It seeds from `GET /session/:id/message`
27 * and then reduces [BusEvent]s from the SSE stream into an ordered list of messages,
28 * each keyed by message id with parts keyed by part id (idempotent upserts).
29 */
30class SessionRepository(
31 private val api: OpencodeApiClient,
32 private val eventStream: EventStreamClient,
33) {
34 suspend fun listSessions() = api.listSessions()
35 suspend fun createSession(title: String? = null) = api.createSession(title)
36 suspend fun deleteSession(id: String) = api.deleteSession(id)
37 suspend fun abort(sessionId: String) = api.abort(sessionId)
39 suspend fun sendPrompt(sessionId: String, text: String, model: ModelRef?, agent: String? = null) =
40 api.sendPrompt(sessionId, text, model, agent)
42 suspend fun runCommand(
43 sessionId: String,
44 command: String,
45 arguments: String = "",
46 agent: String? = null,
47 ) = api.runCommand(sessionId, command, arguments = arguments, agent = agent)
49 /**
50 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
51 * stream before the initial REST load so no streamed part is missed, then reduces
52 * events into the in-memory state. The flow is conflated so a fast burst of
53 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
54 * event reducer — a slow collector simply sees the most recent snapshot.
55 */
56 fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
57 val store = MessageStore()
58 val lock = Mutex()
60 // Collect the snapshot under the lock, then publish outside it so a slow
61 // downstream collector can't stall event processing.
62 suspend fun publish() {
63 val snapshot = lock.withLock { store.snapshot() }
64 send(snapshot)
65 }
67 // Subscribe to events first so we don't miss early deltas during the initial load.
68 val job = launch {
69 eventStream.events.collect { event ->
70 val changed = lock.withLock { store.reduce(sessionId, event) }
71 if (changed) publish()
72 }
73 }
75 // Re-seed from REST when the SSE stream reconnects after a drop. Events emitted
76 // by the server during the disconnection gap are lost from the in-memory state;
77 // re-fetching the current snapshot fills those holes. The seed merge logic
78 // (see MessageStore.seed) handles interleaving with concurrently-arrived events.
79 launch {
80 var wasConnected = false
81 eventStream.state.collect { state ->
82 if (state == EventStreamClient.ConnectionState.Connected) {
83 if (wasConnected) {
84 val fresh = runCatching { api.listMessages(sessionId) }.getOrDefault(emptyList())
85 lock.withLock { store.seed(fresh, prune = true) }
86 publish()
87 }
88 wasConnected = true
89 } else if (state == EventStreamClient.ConnectionState.Disconnected) {
90 wasConnected = false
91 }
92 }
93 }
95 val initial = runCatching { api.listMessages(sessionId) }
96 .onFailure { Log.w("SessionRepository", "Initial message load failed for $sessionId; relying on SSE", it) }
97 .getOrDefault(emptyList())
98 lock.withLock { store.seed(initial) }
99 publish()
101 // Keep the flow alive until the collector cancels; the launched job is torn down with it.
102 job.join()
103 }.conflate()
105 companion object {
106 /** Convenience: is this event a run-completion signal for [sessionId]? */
107 fun isIdle(event: BusEvent, sessionId: String): Boolean =
108 event is SessionIdle && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
110 fun isError(event: BusEvent, sessionId: String): Boolean =
111 event is SessionError && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
112 }
115/**
116 * In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex.
117 * Exposed as `internal` so the reducer logic can be unit-tested directly.
118 */
119internal class MessageStore {
120 // messageId -> (info + parts), insertion-ordered.
121 private val messages = LinkedHashMap<String, MessageWithParts>()
123 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */
124 internal val maxMessages = NetworkConfig.maxInMemoryMessages
126 fun snapshot(): List<MessageWithParts> = messages.values.toList()
128 fun seed(initial: List<MessageWithParts>, prune: Boolean = false) {
129 // On re-seed (after SSE reconnect), remove messages that are no longer in the
130 // server snapshot (e.g. deleted during the disconnection gap). This keeps the
131 // in-memory state in sync with the authoritative REST snapshot rather than
132 // accumulating stale messages forever. Skipped on the initial seed to avoid
133 // racing with just-arrived SSE events whose messages may not yet be in REST.
134 if (prune) {
135 val incomingIds = initial.mapTo(mutableSetOf()) { it.info.id }
136 messages.keys.retainAll(incomingIds)
137 }
139 for (m in initial) {
140 val existing = messages[m.info.id]
141 if (existing == null) {
142 messages[m.info.id] = m
143 } else {
144 // A part streamed in between subscribe and this initial load may already
145 // have populated this message (see observeMessages). Merge instead of
146 // overwriting so that newer streamed part isn't discarded: take the
147 // snapshot's part order, swap in the streamed version where ids overlap,
148 // append any streamed-only parts, and adopt the authoritative REST info.
149 val streamedById = existing.parts.associateBy { it.id }
150 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
151 val ordered = m.parts.map { p -> streamedById[p.id] ?: p }.toMutableList()
152 for (p in existing.parts) {
153 if (p.id !in snapshotIds) ordered.add(p)
154 }
155 val info = if (existing.info is UnknownMessage) m.info else existing.info
156 messages[m.info.id] = MessageWithParts(info = info, parts = ordered)
157 }
158 }
159 evictOldMessages()
160 }
162 /** Returns true if the state changed (and a new snapshot should be published). */
163 fun reduce(sessionId: String, event: BusEvent): Boolean {
164 return when (event) {
165 is MessageUpdated -> {
166 val info = event.properties.info
167 if (info.sessionID != sessionId) {
168 false
169 } else {
170 val existing = messages[info.id]
171 messages[info.id] = existing?.copy(info = info) ?: MessageWithParts(info)
172 if (existing == null) evictOldMessages()
173 true
174 }
175 }
177 is MessagePartUpdated -> {
178 val part = event.properties.part
179 val messageId = part.messageID ?: event.properties.messageID
180 val partSession = part.sessionID ?: event.properties.sessionID
181 if (messageId == null || (partSession != null && partSession != sessionId)) {
182 false
183 } else {
184 upsertPart(messageId, part)
185 }
186 }
188 is MessagePartRemoved -> {
189 val messageId = event.properties.messageID
190 val partId = event.properties.partID
191 if (messageId == null || partId == null) false else removePart(messageId, partId)
192 }
194 is MessageRemoved -> {
195 val id = event.properties.messageID
196 if (id == null) false else messages.remove(id) != null
197 }
199 else -> false
200 }
201 }
203 private fun upsertPart(messageId: String, part: Part): Boolean {
204 val current = messages[messageId]
205 val parts = current?.parts.orEmpty()
206 val idx = parts.indexOfFirst { it.id == part.id }
207 val newParts = if (idx >= 0) {
208 parts.toMutableList().also { it[idx] = part }
209 } else {
210 parts + part
211 }
212 messages[messageId] = current?.copy(parts = newParts)
213 ?: MessageWithParts(
214 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
215 parts = newParts,
216 )
217 evictOldMessages()
218 return true
219 }
221 private fun removePart(messageId: String, partId: String): Boolean {
222 val current = messages[messageId] ?: return false
223 val newParts = current.parts.filterNot { it.id == partId }
224 if (newParts.size == current.parts.size) return false
225 messages[messageId] = current.copy(parts = newParts)
226 return true
227 }
229 /** Evict the oldest messages when the store exceeds [maxMessages], keeping memory bounded. */
230 private fun evictOldMessages() {
231 while (messages.size > maxMessages) {
232 val oldestKey = messages.keys.iterator().next()
233 messages.remove(oldestKey)
234 }
235 }