git.iko.soy/opencode-android
on 8865a77 help sign in
opencode-android/app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt
154 lines 6.4 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 /**
40 * A live, ordered view of [sessionId]'s messages. Begins collecting the event
41 * stream before the initial REST load so no streamed part is missed, then reduces
42 * events into the in-memory state.
43 */
44 fun observeMessages(sessionId: String): Flow<List<MessageWithParts>> = channelFlow {
45 val store = MessageStore()
46 val lock = Mutex()
48 suspend fun publish() = send(store.snapshot())
50 // Subscribe to events first so we don't miss early deltas during the initial load.
51 val job = launch {
52 eventStream.events.collect { event ->
53 lock.withLock {
54 if (store.reduce(sessionId, event)) publish()
55 }
56 }
57 }
59 val initial = runCatching { api.listMessages(sessionId) }.getOrDefault(emptyList())
60 lock.withLock {
61 store.seed(initial)
62 publish()
63 }
65 // Keep the flow alive until the collector cancels; the launched job is torn down with it.
66 job.join()
67 }
69 /** In-memory reduction state for one observed session. Not thread-safe; guard with a Mutex. */
70 private class MessageStore {
71 // messageId -> (info + parts), insertion-ordered.
72 private val messages = LinkedHashMap<String, MessageWithParts>()
74 fun snapshot(): List<MessageWithParts> = messages.values.toList()
76 fun seed(initial: List<MessageWithParts>) {
77 for (m in initial) messages[m.info.id] = m
78 }
80 /** Returns true if the state changed (and a new snapshot should be published). */
81 fun reduce(sessionId: String, event: BusEvent): Boolean {
82 return when (event) {
83 is MessageUpdated -> {
84 val info = event.properties.info
85 if (info.sessionID != sessionId) {
86 false
87 } else {
88 val existing = messages[info.id]
89 messages[info.id] = existing?.copy(info = info) ?: MessageWithParts(info)
90 true
91 }
92 }
94 is MessagePartUpdated -> {
95 val part = event.properties.part
96 val messageId = part.messageID ?: event.properties.messageID
97 val partSession = part.sessionID ?: event.properties.sessionID
98 if (messageId == null || (partSession != null && partSession != sessionId)) {
99 false
100 } else {
101 upsertPart(messageId, part)
102 }
103 }
105 is MessagePartRemoved -> {
106 val messageId = event.properties.messageID
107 val partId = event.properties.partID
108 if (messageId == null || partId == null) false else removePart(messageId, partId)
109 }
111 is MessageRemoved -> {
112 val id = event.properties.messageID
113 if (id == null) false else messages.remove(id) != null
114 }
116 else -> false
117 }
118 }
120 private fun upsertPart(messageId: String, part: Part): Boolean {
121 val current = messages[messageId]
122 val parts = current?.parts.orEmpty()
123 val idx = parts.indexOfFirst { it.id == part.id }
124 val newParts = if (idx >= 0) {
125 parts.toMutableList().also { it[idx] = part }
126 } else {
127 parts + part
128 }
129 messages[messageId] = current?.copy(parts = newParts)
130 ?: MessageWithParts(
131 info = AssistantMessage(id = messageId, sessionID = part.sessionID ?: ""),
132 parts = newParts,
133 )
134 return true
135 }
137 private fun removePart(messageId: String, partId: String): Boolean {
138 val current = messages[messageId] ?: return false
139 val newParts = current.parts.filterNot { it.id == partId }
140 if (newParts.size == current.parts.size) return false
141 messages[messageId] = current.copy(parts = newParts)
142 return true
143 }
144 }
146 companion object {
147 /** Convenience: is this event a run-completion signal for [sessionId]? */
148 fun isIdle(event: BusEvent, sessionId: String): Boolean =
149 event is SessionIdle && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
151 fun isError(event: BusEvent, sessionId: String): Boolean =
152 event is SessionError && (event.properties.sessionID == null || event.properties.sessionID == sessionId)
153 }