git.iko.soy/opencode-android
on 0bc1eda help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
191 lines 8.2 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.OpencodeApiClient
16import android.util.Log
17import kotlinx.coroutines.flow.Flow
18import kotlinx.coroutines.flow.channelFlow
19import kotlinx.coroutines.flow.conflate
20import kotlinx.coroutines.launch
21import kotlinx.coroutines.sync.Mutex
22import kotlinx.coroutines.sync.withLock
24/**
25 * Source of truth for a session's conversation. It seeds from `GET /session/:id/message`
26 * and then reduces [BusEvent]s from the SSE stream into an ordered list of messages,
27 * each keyed by message id with parts keyed by part id (idempotent upserts).
28 */
29class SessionRepository(
30 private val api: OpencodeApiClient,
31 private val eventStream: EventStreamClient,
32) {
33 suspend fun listSessions() = api.listSessions()
34 suspend fun createSession(title: String? = null) = api.createSession(title)
35 suspend fun deleteSession(id: String) = api.deleteSession(id)
36 suspend fun abort(sessionId: String) = api.abort(sessionId)
38 suspend fun sendPrompt(sessionId: String, text: String, model: ModelRef?, agent: String? = null) =
39 api.sendPrompt(sessionId, text, model, agent)
41 suspend fun runCommand(
42 sessionId: String,
43 command: String,
44 arguments: String = "",
45 agent: String? = null,
46 ) = api.runCommand(sessionId, command, arguments = arguments, agent = agent)
48 /**
49 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
50 * stream before the initial REST load so no streamed part is missed, then reduces
51 * events into the in-memory state. The flow is conflated so a fast burst of
52 * streaming tokens (each publishing a fresh snapshot) never back-pressures the
53 * event reducer — a slow collector simply sees the most recent snapshot.
54 */
55 fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
56 val store = MessageStore()
57 val lock = Mutex()
59 // Collect the snapshot under the lock, then publish outside it so a slow
60 // downstream collector can't stall event processing.
61 suspend fun publish() {
62 val snapshot = lock.withLock { store.snapshot() }
63 send(snapshot)
64 }
66 // Subscribe to events first so we don't miss early deltas during the initial load.
67 val job = launch {
68 eventStream.events.collect { event ->
69 val changed = lock.withLock { store.reduce(sessionId, event) }
70 if (changed) publish()
71 }
72 }
74 val initial = runCatching { api.listMessages(sessionId) }
75 .onFailure { Log.w("SessionRepository", "Initial message load failed for $sessionId; relying on SSE", it) }
76 .getOrDefault(emptyList())
77 lock.withLock { store.seed(initial) }
78 publish()
80 // Keep the flow alive until the collector cancels; the launched job is torn down with it.
81 job.join()
82 }.conflate()
84 companion object {
85 /** Convenience: is this event a run-completion signal for [sessionId]? */
86 fun isIdle(event: BusEvent, sessionId: String): Boolean =
87 event is SessionIdle && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
89 fun isError(event: BusEvent, sessionId: String): Boolean =
90 event is SessionError && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
91 }
94/**
95 * In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex.
96 * Exposed as `internal` so the reducer logic can be unit-tested directly.
97 */
98internal class MessageStore {
99 // messageId -> (info + parts), insertion-ordered.
100 private val messages = LinkedHashMap<String, MessageWithParts>()
102 fun snapshot(): List<MessageWithParts> = messages.values.toList()
104 fun seed(initial: List<MessageWithParts>) {
105 for (m in initial) {
106 val existing = messages[m.info.id]
107 if (existing == null) {
108 messages[m.info.id] = m
109 } else {
110 // A part streamed in between subscribe and this initial load may already
111 // have populated this message (see observeMessages). Merge instead of
112 // overwriting so that newer streamed part isn't discarded: take the
113 // snapshot's part order, swap in the streamed version where ids overlap,
114 // append any streamed-only parts, and adopt the authoritative REST info.
115 val streamedById = existing.parts.associateBy { it.id }
116 val snapshotIds = m.parts.mapTo(mutableSetOf()) { it.id }
117 val ordered = m.parts.map { p -> streamedById[p.id] ?: p }.toMutableList()
118 for (p in existing.parts) {
119 if (p.id !in snapshotIds) ordered.add(p)
120 }
121 val info = if (existing.info is UnknownMessage) m.info else existing.info
122 messages[m.info.id] = MessageWithParts(info = info, parts = ordered)
123 }
124 }
125 }
127 /** Returns true if the state changed (and a new snapshot should be published). */
128 fun reduce(sessionId: String, event: BusEvent): Boolean {
129 return when (event) {
130 is MessageUpdated -> {
131 val info = event.properties.info
132 if (info.sessionID != sessionId) {
133 false
134 } else {
135 val existing = messages[info.id]
136 messages[info.id] = existing?.copy(info = info) ?: MessageWithParts(info)
137 true
138 }
139 }
141 is MessagePartUpdated -> {
142 val part = event.properties.part
143 val messageId = part.messageID ?: event.properties.messageID
144 val partSession = part.sessionID ?: event.properties.sessionID
145 if (messageId == null || (partSession != null && partSession != sessionId)) {
146 false
147 } else {
148 upsertPart(messageId, part)
149 }
150 }
152 is MessagePartRemoved -> {
153 val messageId = event.properties.messageID
154 val partId = event.properties.partID
155 if (messageId == null || partId == null) false else removePart(messageId, partId)
156 }
158 is MessageRemoved -> {
159 val id = event.properties.messageID
160 if (id == null) false else messages.remove(id) != null
161 }
163 else -> false
164 }
165 }
167 private fun upsertPart(messageId: String, part: Part): Boolean {
168 val current = messages[messageId]
169 val parts = current?.parts.orEmpty()
170 val idx = parts.indexOfFirst { it.id == part.id }
171 val newParts = if (idx >= 0) {
172 parts.toMutableList().also { it[idx] = part }
173 } else {
174 parts + part
175 }
176 messages[messageId] = current?.copy(parts = newParts)
177 ?: MessageWithParts(
178 info = UnknownMessage(id = messageId, sessionID = part.sessionID ?: ""),
179 parts = newParts,
180 )
181 return true
182 }
184 private fun removePart(messageId: String, partId: String): Boolean {
185 val current = messages[messageId] ?: return false
186 val newParts = current.parts.filterNot { it.id == partId }
187 if (newParts.size == current.parts.size) return false
188 messages[messageId] = current.copy(parts = newParts)
189 return true
190 }