git.iko.soy/opencode-android
on e9b2cc3 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
157 lines 6.5 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 /** In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex. */
73 private class MessageStore {
74 // messageId -> (info + parts), insertion-ordered.
75 private val messages = LinkedHashMap<String, MessageWithParts>()
77 fun snapshot(): List<MessageWithParts> = messages.values.toList()
79 fun seed(initial: List<MessageWithParts>) {
80 for (m in initial) messages[m.info.id] = m
81 }
83 /** Returns true if the state changed (and a new snapshot should be published). */
84 fun reduce(sessionId: String, event: BusEvent): Boolean {
85 return when (event) {
86 is MessageUpdated -> {
87 val info = event.properties.info
88 if (info.sessionID != sessionId) {
89 false
90 } else {
91 val existing = messages[info.id]
92 messages[info.id] = existing?.copy(info = info) ?: MessageWithParts(info)
93 true
94 }
95 }
97 is MessagePartUpdated -> {
98 val part = event.properties.part
99 val messageId = part.messageID ?: event.properties.messageID
100 val partSession = part.sessionID ?: event.properties.sessionID
101 if (messageId == null || (partSession != null && partSession != sessionId)) {
102 false
103 } else {
104 upsertPart(messageId, part)
105 }
106 }
108 is MessagePartRemoved -> {
109 val messageId = event.properties.messageID
110 val partId = event.properties.partID
111 if (messageId == null || partId == null) false else removePart(messageId, partId)
112 }
114 is MessageRemoved -> {
115 val id = event.properties.messageID
116 if (id == null) false else messages.remove(id) != null
117 }
119 else -> false
120 }
121 }
123 private fun upsertPart(messageId: String, part: Part): Boolean {
124 val current = messages[messageId]
125 val parts = current?.parts.orEmpty()
126 val idx = parts.indexOfFirst { it.id == part.id }
127 val newParts = if (idx >= 0) {
128 parts.toMutableList().also { it[idx] = part }
129 } else {
130 parts + part
131 }
132 messages[messageId] = current?.copy(parts = newParts)
133 ?: MessageWithParts(
134 info = AssistantMessage(id = messageId, sessionID = part.sessionID ?: ""),
135 parts = newParts,
136 )
137 return true
138 }
140 private fun removePart(messageId: String, partId: String): Boolean {
141 val current = messages[messageId] ?: return false
142 val newParts = current.parts.filterNot { it.id == partId }
143 if (newParts.size == current.parts.size) return false
144 messages[messageId] = current.copy(parts = newParts)
145 return true
146 }
147 }
149 companion object {
150 /** Convenience: is this event a run-completion signal for [sessionId]? */
151 fun isIdle(event: BusEvent, sessionId: String): Boolean =
152 event is SessionIdle && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
154 fun isError(event: BusEvent, sessionId: String): Boolean =
155 event is SessionError && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
156 }