git.iko.soy/opencode-android
on e37d1bb help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
160 lines 6.3 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.MessagePartRemoved
6import soy.iko.opencode.data.model.MessagePartUpdated
7import soy.iko.opencode.data.model.MessageRemoved
8import soy.iko.opencode.data.model.MessageUpdated
9import soy.iko.opencode.data.model.MessageWithParts
10import soy.iko.opencode.data.model.ModelRef
11import soy.iko.opencode.data.model.Part
12import soy.iko.opencode.data.model.SessionError
13import soy.iko.opencode.data.model.SessionIdle
14import soy.iko.opencode.data.network.EventStreamClient
15import soy.iko.opencode.data.network.OpencodeApiClient
16import kotlinx.coroutines.flow.Flow
17import kotlinx.coroutines.flow.channelFlow
18import kotlinx.coroutines.launch
19import kotlinx.coroutines.sync.Mutex
20import kotlinx.coroutines.sync.withLock
22/**
23 * Source of truth for a session's conversation. It seeds from `GET /session/:id/message`
24 * and then reduces [BusEvent]s from the SSE stream into an ordered list of messages,
25 * each keyed by message id with parts keyed by part id (idempotent upserts).
26 */
27class SessionRepository(
28 private val api: OpencodeApiClient,
29 private val eventStream: EventStreamClient,
30) {
31 suspend fun listSessions() = api.listSessions()
32 suspend fun createSession(title: String? = null) = api.createSession(title)
33 suspend fun deleteSession(id: String) = api.deleteSession(id)
34 suspend fun abort(sessionId: String) = api.abort(sessionId)
36 suspend fun sendPrompt(sessionId: String, text: String, model: ModelRef?, agent: String? = null) =
37 api.sendPrompt(sessionId, text, model, agent)
39 suspend fun runCommand(sessionId: String, command: String, agent: String? = null) =
40 api.runCommand(sessionId, command, agent = agent)
42 /**
43 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
44 * stream before the initial REST load so no streamed part is missed, then reduces
45 * events into the in-memory state.
46 */
47 fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
48 val store = MessageStore()
49 val lock = Mutex()
51 suspend fun publish() = send(store.snapshot())
53 // Subscribe to events first so we don't miss early deltas during the initial load.
54 val job = launch {
55 eventStream.events.collect { event ->
56 lock.withLock {
57 if (store.reduce(sessionId, event)) publish()
58 }
59 }
60 }
62 val initial = runCatching { api.listMessages(sessionId) }.getOrDefault(emptyList())
63 lock.withLock {
64 store.seed(initial)
65 publish()
66 }
68 // Keep the flow alive until the collector cancels; the launched job is torn down with it.
69 job.join()
70 }
72 companion object {
73 /** Convenience: is this event a run-completion signal for [sessionId]? */
74 fun isIdle(event: BusEvent, sessionId: String): Boolean =
75 event is SessionIdle && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
77 fun isError(event: BusEvent, sessionId: String): Boolean =
78 event is SessionError && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
79 }
82/**
83 * In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex.
84 * Exposed as `internal` so the reducer logic can be unit-tested directly.
85 */
86internal class MessageStore {
87 // messageId -> (info + parts), insertion-ordered.
88 private val messages = LinkedHashMap<String, MessageWithParts>()
90 fun snapshot(): List<MessageWithParts> = messages.values.toList()
92 fun seed(initial: List<MessageWithParts>) {
93 for (m in initial) messages[m.info.id] = m
94 }
96 /** Returns true if the state changed (and a new snapshot should be published). */
97 fun reduce(sessionId: String, event: BusEvent): Boolean {
98 return when (event) {
99 is MessageUpdated -> {
100 val info = event.properties.info
101 if (info.sessionID != sessionId) {
102 false
103 } else {
104 val existing = messages[info.id]
105 messages[info.id] = existing?.copy(info = info) ?: MessageWithParts(info)
106 true
107 }
108 }
110 is MessagePartUpdated -> {
111 val part = event.properties.part
112 val messageId = part.messageID ?: event.properties.messageID
113 val partSession = part.sessionID ?: event.properties.sessionID
114 if (messageId == null || (partSession != null && partSession != sessionId)) {
115 false
116 } else {
117 upsertPart(messageId, part)
118 }
119 }
121 is MessagePartRemoved -> {
122 val messageId = event.properties.messageID
123 val partId = event.properties.partID
124 if (messageId == null || partId == null) false else removePart(messageId, partId)
125 }
127 is MessageRemoved -> {
128 val id = event.properties.messageID
129 if (id == null) false else messages.remove(id) != null
130 }
132 else -> false
133 }
134 }
136 private fun upsertPart(messageId: String, part: Part): Boolean {
137 val current = messages[messageId]
138 val parts = current?.parts.orEmpty()
139 val idx = parts.indexOfFirst { it.id == part.id }
140 val newParts = if (idx >= 0) {
141 parts.toMutableList().also { it[idx] = part }
142 } else {
143 parts + part
144 }
145 messages[messageId] = current?.copy(parts = newParts)
146 ?: MessageWithParts(
147 info = AssistantMessage(id = messageId, sessionID = part.sessionID ?: ""),
148 parts = newParts,
149 )
150 return true
151 }
153 private fun removePart(messageId: String, partId: String): Boolean {
154 val current = messages[messageId] ?: return false
155 val newParts = current.parts.filterNot { it.id == partId }
156 if (newParts.size == current.parts.size) return false
157 messages[messageId] = current.copy(parts = newParts)
158 return true
159 }