7a6768e · opencode-android

git.iko.soy/opencode-android

Fix SSE re-seed dead code, missing session filters on remove events, retryFailed draft loss, terminal catch on messages flow, debounced draft defeating send protection, DraftStore init race, runCommand idempotency, cache stampede, Connecting state on cancellation, shutdown executor drain, nanoTime key collision, loading state conflict, and catalog reload race

commit7a6768e78c50e00a2bd70ca9cf903a49bc3354fc authoriko <mail@iko.soy> dateTue 30 Jun 2026, 11:30 (3 months ago) parent3996399

5 files changed+163−125

compare with master

unifiedsplit

app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt+1 app/src/main/java/soy/iko/opencode/data/network/OpencodeApiClient.kt+40−34 app/src/main/java/soy/iko/opencode/data/repo/DraftStore.kt+10−1 app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt+46−27 app/src/main/java/soy/iko/opencode/ui/chat/ChatViewModel.kt+66−63

app/src/main/java/soy/iko/opencode/data/network/EventStreamClient.kt+1

@@ -164,6 +164,7 @@ class EventStreamClient(

164164 }

165165 }

166166 } catch (c: CancellationException) {

167+ _state.value = ConnectionState.Disconnected

167168 throw c

168169 } catch (e: Exception) {

169170 // If the scope was cancelled (e.g. connection close), the closed-client

app/src/main/java/soy/iko/opencode/data/network/OpencodeApiClient.kt+40−34

@@ -45,7 +45,12 @@ class OpencodeApiClient(private val client: HttpClient) {

4545 // A short-lived in-memory cache avoids redundant network calls when navigating

4646 // between sessions on the same server.

4747 private data class CachedEntry<T>(val value: T, val fetchedAt: Long)

48− private val cacheMutex = Mutex()

48+ // Per-catalog mutexes so a slow fetch of one catalog doesn't block the others.

49+ // The lock is held during the fetch to prevent a thundering-herd stampede when

50+ // multiple coroutines miss the cache simultaneously.

51+ private val providersMutex = Mutex()

52+ private val agentsMutex = Mutex()

53+ private val commandsMutex = Mutex()

4954 private var cachedProviders: CachedEntry<ProvidersResponse>? = null

5055 private var cachedAgents: CachedEntry<List<Agent>>? = null

5156 private var cachedCommands: CachedEntry<List<Command>>? = null

@@ -117,57 +122,58 @@ class OpencodeApiClient(private val client: HttpClient) {

117122 command: String,

118123 arguments: String = "",

119124 agent: String? = null,

120− ): MessageWithParts = withRetry {

121− client.post("session/${encode(sessionId)}/command") {

122− contentType(ContentType.Application.Json)

123− setBody(CommandRequest(command = command, arguments = arguments, agent = agent))

124− }.body()

125+ ): MessageWithParts {

126+ // Generate the idempotency key before entering withRetry so all retry

127+ // attempts share the same key, preventing duplicate command execution if

128+ // the server processed the request but the response was lost.

129+ val idempotencyKey = java.util.UUID.randomUUID().toString()

130+ return withRetry {

131+ client.post("session/${encode(sessionId)}/command") {

132+ contentType(ContentType.Application.Json)

133+ header("Idempotency-Key", idempotencyKey)

134+ setBody(CommandRequest(command = command, arguments = arguments, agent = agent))

135+ }.body()

136+ }

125137 }

126138

127− suspend fun providers(): ProvidersResponse {

139+ suspend fun providers(): ProvidersResponse = providersMutex.withLock {

128140 val now = System.currentTimeMillis()

129− cacheMutex.withLock {

130− val cached = cachedProviders

131− if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

132− return cached.value

133− }

141+ val cached = cachedProviders

142+ if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

143+ return@withLock cached.value

134144 }

135145 val value = withRetry { client.get("config/providers").body<ProvidersResponse>() }

136− cacheMutex.withLock { cachedProviders = CachedEntry(value, System.currentTimeMillis()) }

137− return value

146+ cachedProviders = CachedEntry(value, System.currentTimeMillis())

147+ value

138148 }

139149

140− suspend fun agents(): List<Agent> {

150+ suspend fun agents(): List<Agent> = agentsMutex.withLock {

141151 val now = System.currentTimeMillis()

142− cacheMutex.withLock {

143− val cached = cachedAgents

144− if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

145− return cached.value

146− }

152+ val cached = cachedAgents

153+ if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

154+ return@withLock cached.value

147155 }

148156 val value = withRetry { client.get("agent").body<List<Agent>>() }

149− cacheMutex.withLock { cachedAgents = CachedEntry(value, System.currentTimeMillis()) }

150− return value

157+ cachedAgents = CachedEntry(value, System.currentTimeMillis())

158+ value

151159 }

152160

153− suspend fun commands(): List<Command> {

161+ suspend fun commands(): List<Command> = commandsMutex.withLock {

154162 val now = System.currentTimeMillis()

155− cacheMutex.withLock {

156− val cached = cachedCommands

157− if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

158− return cached.value

159− }

163+ val cached = cachedCommands

164+ if (cached != null && now - cached.fetchedAt < NetworkConfig.catalogCacheTtlMs) {

165+ return@withLock cached.value

160166 }

161167 val value = withRetry { client.get("command").body<List<Command>>() }

162− cacheMutex.withLock { cachedCommands = CachedEntry(value, System.currentTimeMillis()) }

163− return value

168+ cachedCommands = CachedEntry(value, System.currentTimeMillis())

169+ value

164170 }

165171

166172 /** Invalidate the catalog cache (e.g. on server switch). */

167− suspend fun invalidateCache() = cacheMutex.withLock {

168− cachedProviders = null

169− cachedAgents = null

170− cachedCommands = null

173+ suspend fun invalidateCache() {

174+ providersMutex.withLock { cachedProviders = null }

175+ agentsMutex.withLock { cachedAgents = null }

176+ commandsMutex.withLock { cachedCommands = null }

171177 }

172178

173179 /** Respond to a permission request so a paused tool run can proceed. */

app/src/main/java/soy/iko/opencode/data/repo/DraftStore.kt+10−1

@@ -59,7 +59,13 @@ class DraftStore(context: Context, scope: CoroutineScope) {

5959 val snapshot = runCatching {

6060 prefs.all.mapValues { it.value as? String ?: "" }

6161 }.getOrDefault(emptyMap())

62− _drafts.value = snapshot

62+ // Merge instead of overwriting: any in-memory writes that arrived before

63+ // the background load completed (e.g. a share-intent draft injected via

64+ // setImmediate) take priority over the persisted snapshot so they aren't

65+ // clobbered.

66+ _drafts.update { current ->

67+ snapshot.toMutableMap().apply { putAll(current) }

68+ }

6369 _ready.value = true

6470 }

6571 }

@@ -151,5 +157,8 @@ class DraftStore(context: Context, scope: CoroutineScope) {

151157 /** Shut down the background flush executor. Call from AppContainer.shutdown(). */

152158 fun shutdown() {

153159 flushExecutor.shutdown()

160+ // Await pending flushDraft writes so drafts aren't lost if the process exits

161+ // immediately after shutdown (e.g. ANR-triggered process kill).

162+ runCatching { flushExecutor.awaitTermination(2, java.util.concurrent.TimeUnit.SECONDS) }

154163 }

155164}

app/src/main/java/soy/iko/opencode/data/repo/SessionRepository.kt+46−27

@@ -74,31 +74,37 @@ class SessionRepository(

7474 }

7575 }

7676

77− // Re-seed from REST when the SSE stream reconnects after a drop. Events emitted

78− // by the server during the disconnection gap are lost from the in-memory state;

79− // re-fetching the current snapshot fills those holes. The seed merge logic

80− // (see MessageStore.seed) handles interleaving with concurrently-arrived events.

81− //

82− // The REST fetch is performed *outside* the lock so events arriving via SSE

83− // during the fetch are not blocked — holding the lock during a network call

84− // would stall the event reducer and overflow the SharedFlow buffer (DROP_OLDEST),

85− // silently losing events. The seed itself is under the lock; an interim seed

86− // token is checked so a concurrent re-seed doesn't clobber a newer fetch.

77+ // Re-seed from REST when the SSE stream reconnects after a drop. The REST fetch

78+ // is performed *outside* the lock so events arriving via SSE during the fetch are

79+ // not blocked. The seed itself is under the lock; a seed generation counter ensures

80+ // a stale fetch (from an earlier reconnect) doesn't clobber a newer one.

8781 launch {

88− var wasConnected = false

82+ // Re-seed from REST when the SSE stream reconnects after a drop. The

83+ // hasConnectedBefore flag is NOT reset on disconnect — it stays true after

84+ // the first successful connection so a subsequent reconnect triggers the

85+ // re-seed. Resetting it on disconnect would make the condition always false

86+ // after the first cycle, turning the re-seed into dead code.

87+ //

88+ // A seed generation counter guards against a stale re-seed clobbering a

89+ // newer one: if two reconnects fire in quick succession, the older fetch

90+ // (which returns later) is discarded because its generation no longer matches.

91+ var hasConnectedBefore = false

92+ var seedGeneration = 0

8993 eventStream.state.collect { state ->

9094 if (state == EventStreamClient.ConnectionState.Connected) {

91− if (wasConnected) {

95+ if (hasConnectedBefore) {

96+ val generation = ++seedGeneration

9297 val fresh = runCatchingCancellable { api.listMessages(sessionId) }

9398 .onFailure { Log.w("SessionRepository", "Re-seed message load failed for $sessionId; relying on SSE: ${safeExceptionSummary(it)}") }

9499 .getOrDefault(emptyList())

95− lock.withLock { store.seed(fresh, prune = true) }

100+ lock.withLock {

101+ if (generation == seedGeneration) {

102+ store.seed(fresh, prune = true)

103+ }

104+ }

96105 publish()

97106 }

98− wasConnected = true

99− } else if (state == EventStreamClient.ConnectionState.Disconnected ||

100− state == EventStreamClient.ConnectionState.Failed) {

101− wasConnected = false

107+ hasConnectedBefore = true

102108 }

103109 }

104110 }

@@ -133,6 +139,9 @@ internal class MessageStore {

133139 // messageId -> (info + parts), insertion-ordered.

134140 private val messages = LinkedHashMap<String, MessageWithParts>()

135141

142+ /** Monotonic counter for synthetic UnknownMessage keys, avoiding nanoTime collisions. */

143+ private val unknownCounter = java.util.concurrent.atomic.AtomicLong(0)

144+

136145 /** Maximum number of messages to keep in memory; oldest are evicted when exceeded. */

137146 internal val maxMessages = NetworkConfig.maxInMemoryMessages

138147

@@ -185,7 +194,7 @@ internal class MessageStore {

185194 // messages in the map. Generate a unique synthetic key so each

186195 // unknown message gets its own entry instead of overwriting others.

187196 val key = if (info.id.isEmpty() && info is UnknownMessage) {

188− "unknown-${System.nanoTime()}"

197+ "unknown-${unknownCounter.incrementAndGet()}"

189198 } else {

190199 info.id

191200 }

@@ -207,21 +216,31 @@ internal class MessageStore {

207216 }

208217 }

209218

210− is MessagePartRemoved -> {

211− val messageId = event.properties.messageID

212− val partId = event.properties.partID

213− if (messageId == null || partId == null) false else removePart(messageId, partId)

214− }

219+ is MessagePartRemoved -> handlePartRemoved(sessionId, event)

215220

216− is MessageRemoved -> {

217− val id = event.properties.messageID

218− if (id == null) false else messages.remove(id) != null

219− }

221+ is MessageRemoved -> handleMessageRemoved(sessionId, event)

220222

221223 else -> false

222224 }

223225 }

224226

227+ /** Returns false if [eventSession] is non-null and doesn't match [sessionId]. */

228+ private fun matchesSession(eventSession: String?, sessionId: String): Boolean =

229+ eventSession == null || eventSession == sessionId

230+

231+ private fun handlePartRemoved(sessionId: String, event: MessagePartRemoved): Boolean {

232+ val messageId = event.properties.messageID ?: return false

233+ val partId = event.properties.partID ?: return false

234+ if (!matchesSession(event.properties.sessionID, sessionId)) return false

235+ return removePart(messageId, partId)

236+ }

237+

238+ private fun handleMessageRemoved(sessionId: String, event: MessageRemoved): Boolean {

239+ val id = event.properties.messageID ?: return false

240+ if (!matchesSession(event.properties.sessionID, sessionId)) return false

241+ return messages.remove(id) != null

242+ }

243+

225244 private fun upsertPart(messageId: String, part: Part): Boolean {

226245 val current = messages[messageId]

227246 val parts = current?.parts.orEmpty()

app/src/main/java/soy/iko/opencode/ui/chat/ChatViewModel.kt+66−63

@@ -27,7 +27,6 @@ import kotlinx.coroutines.flow.MutableStateFlow

2727import kotlinx.coroutines.flow.SharingStarted

2828import kotlinx.coroutines.flow.StateFlow

2929import kotlinx.coroutines.flow.asStateFlow

30−import kotlinx.coroutines.flow.catch

3130import kotlinx.coroutines.CancellationException

3231import kotlinx.coroutines.flow.collectLatest

3332import kotlinx.coroutines.flow.debounce

@@ -35,11 +34,14 @@ import kotlinx.coroutines.flow.drop

3534import kotlinx.coroutines.flow.first

3635import kotlinx.coroutines.flow.flatMapLatest

3736import kotlinx.coroutines.flow.flowOf

37+import kotlinx.coroutines.flow.merge

3838import kotlinx.coroutines.flow.onEach

39+import kotlinx.coroutines.flow.retryWhen

3940import kotlinx.coroutines.flow.stateIn

41+import kotlinx.coroutines.delay

4042import kotlinx.coroutines.launch

4143

42−@OptIn(ExperimentalCoroutinesApi::class)

44+@OptIn(ExperimentalCoroutinesApi::class, kotlinx.coroutines.FlowPreview::class)

4345class ChatViewModel(

4446 private val container: AppContainer,

4547 private val sessionId: String,

@@ -52,15 +54,35 @@ class ChatViewModel(

5254 private val _loading = MutableStateFlow(true)

5355 val loading: StateFlow<Boolean> = _loading.asStateFlow()

5456

57+ /** Separate from [loading]: tracks the manual reconnect() flow so its spinner

58+ * doesn't conflict with the messages flow's loading state. */

59+ private val _reconnecting = MutableStateFlow(false)

60+ val reconnecting: StateFlow<Boolean> = _reconnecting.asStateFlow()

61+

62+ /** When true, the debounced draft collector skips persisting an empty draft so

63+ * [send]'s deliberate non-persistence of the cleared draft isn't undone by the

64+ * debounce timer firing before the send completes. */

65+ private val suppressDraftPersist = java.util.concurrent.atomic.AtomicBoolean(false)

66+

67+ /** Per-catalog reload triggers: incrementing one causes [observeCatalog]'s

68+ * collectLatest to cancel any in-flight fetch and start a fresh one, so a

69+ * manual reload supersedes a stale observeCatalog fetch. */

70+ private val _modelsReload = MutableStateFlow(0)

71+ private val _agentsReload = MutableStateFlow(0)

72+ private val _commandsReload = MutableStateFlow(0)

73+

5574 val messages: StateFlow<List<MessageWithParts>> =

5675 container.activeConnection

5776 .flatMapLatest { conn ->

5877 conn?.repository?.observeMessages(sessionId) ?: flowOf(emptyList())

5978 }

6079 .onEach { _loading.value = false }

61− .catch {

80+ .retryWhen { cause, _ ->

6281 _loading.value = false

63− _error.value = container.friendlyError(it)

82+ _error.value = container.friendlyError(cause)

83+ // Delay before retrying to avoid a tight loop on persistent errors.

84+ delay(NetworkConfig.retryInitialDelayMs)

85+ true

6486 }

6587 .stateIn(

6688 scope = viewModelScope,

@@ -155,73 +177,36 @@ class ChatViewModel(

155177

156178 fun reloadModels() {

157179 val conn = connection ?: return

158− reloadCatalog(

159− conn = conn,

160− tag = "model catalog",

161− loading = _modelsLoading,

162− error = _modelsError,

163− fetch = { it.providers() },

164− onSuccess = { resp ->

165− val options = resp.toOptions()

166− _models.value = options

167− _selectedModel.value = resp.defaultOption(options)

168− },

169− )

180+ viewModelScope.launch {

181+ conn.api.invalidateCache()

182+ _modelsReload.value++

183+ }

170184 }

171185

172186 fun reloadAgents() {

173187 val conn = connection ?: return

174− reloadCatalog(

175− conn = conn,

176− tag = "agent catalog",

177− loading = _agentsLoading,

178− error = _agentsError,

179− fetch = { it.agents() },

180− onSuccess = { _agents.value = it },

181− )

188+ viewModelScope.launch {

189+ conn.api.invalidateCache()

190+ _agentsReload.value++

191+ }

182192 }

183193

184194 fun reloadCommands() {

185195 val conn = connection ?: return

186− reloadCatalog(

187− conn = conn,

188− tag = "command catalog",

189− loading = _commandsLoading,

190− error = _commandsError,

191− fetch = { it.commands() },

192− onSuccess = { _commands.value = it },

193− )

194− }

195−

196− /**

197− * Shared helper for reloadXxx(): invalidates the API cache, sets loading/error

198− * state, and applies [onSuccess] to the fetched result. Reduces triplicated

199− * boilerplate that was easy to get out of sync.

200− */

201− private fun <T> reloadCatalog(

202− conn: soy.iko.opencode.di.OpencodeConnection,

203− tag: String,

204− loading: MutableStateFlow<Boolean>,

205− error: MutableStateFlow<Boolean>,

206− fetch: suspend (soy.iko.opencode.data.network.OpencodeApiClient) -> T,

207− onSuccess: (T) -> Unit,

208− ) {

209196 viewModelScope.launch {

210197 conn.api.invalidateCache()

211− loading.value = true

212− error.value = false

213− runCatchingCancellable { fetch(conn.api) }

214− .onFailure { Log.w("ChatViewModel", "Failed to reload $tag", it); error.value = true }

215− .getOrNull()?.let(onSuccess)

216− loading.value = false

198+ _commandsReload.value++

217199 }

218200 }

219201

220202 /**

221203 * Shared helper for init-block catalog observers: re-fetches whenever the active

222− * connection changes, invoking [onNull] to clear state on null so stale data from

223− * the old server doesn't persist. Failure is non-fatal — the error flag is surfaced

224− * to the UI.

204+ * connection or the catalog's reload trigger changes, invoking [onNull] to clear

205+ * state on null so stale data from the old server doesn't persist. The

206+ * [reloadTrigger] is merged with [activeConnection] so a manual reload (via

207+ * reloadModels/reloadAgents/reloadCommands) cancels any in-flight observe fetch

208+ * via collectLatest, preventing a stale observe result from overwriting a fresh

209+ * reload result. Failure is non-fatal — the error flag is surfaced to the UI.

225210 */

226211 private fun <T> observeCatalog(

227212 tag: String,

@@ -230,9 +215,11 @@ class ChatViewModel(

230215 fetch: suspend (soy.iko.opencode.data.network.OpencodeApiClient) -> T,

231216 onSuccess: (T) -> Unit,

232217 onNull: () -> Unit,

218+ reloadTrigger: StateFlow<Int>,

233219 ) {

234220 viewModelScope.launch {

235− container.activeConnection.collectLatest { conn ->

221+ merge(container.activeConnection, reloadTrigger).collectLatest { _ ->

222+ val conn = container.activeConnection.value

236223 if (conn == null) { onNull(); loading.value = false; error.value = false; return@collectLatest }

237224 loading.value = true

238225 error.value = false

@@ -250,8 +237,13 @@ class ChatViewModel(

250237

251238 init {

252239 // Debounce draft persistence so we don't write to disk on every keystroke.

240+ // Skip persisting empty drafts that were set by send() (suppressed via

241+ // suppressDraftPersist) — send() clears the in-memory draft immediately for

242+ // UI feedback but deliberately doesn't persist the clear until the send

243+ // succeeds, so a failed send can restore the draft.

253244 viewModelScope.launch {

254245 _draft.drop(1).debounce(NetworkConfig.draftDebounceMs).collect { text ->

246+ if (text.isBlank() && suppressDraftPersist.get()) return@collect

255247 runCatchingCancellable { container.draftStore.set(sessionId, text) }

256248 .onFailure { Log.w("ChatViewModel", "Failed to persist draft", it) }

257249 }

@@ -308,6 +300,7 @@ class ChatViewModel(

308300 _selectedModel.value = resp.defaultOption(options)

309301 },

310302 onNull = { _models.value = emptyList(); _selectedModel.value = null },

303+ reloadTrigger = _modelsReload,

311304 )

312305 // Load the agent catalog (non-fatal).

313306 observeCatalog(

@@ -317,6 +310,7 @@ class ChatViewModel(

317310 fetch = { it.agents() },

318311 onSuccess = { _agents.value = it },

319312 onNull = { _agents.value = emptyList() },

313+ reloadTrigger = _agentsReload,

320314 )

321315 // Load the command catalog (non-fatal).

322316 observeCatalog(

@@ -326,6 +320,7 @@ class ChatViewModel(

326320 fetch = { it.commands() },

327321 onSuccess = { _commands.value = it },

328322 onNull = { _commands.value = emptyList() },

323+ reloadTrigger = _commandsReload,

329324 )

330325 // Resolve the human-readable session title for the app bar (non-fatal).

331326 viewModelScope.launch {

@@ -368,6 +363,9 @@ class ChatViewModel(

368363 // Clear the in-memory draft for the UI immediately, but don't persist the clear

369364 // yet — if the send fails and the process dies before we restore, the draft

370365 // would be lost forever. The persisted draft is cleared only on success.

366+ // suppressDraftPersist prevents the debounced collector from persisting the

367+ // empty draft before the send resolves.

368+ suppressDraftPersist.set(true)

371369 _draft.value = ""

372370 viewModelScope.launch {

373371 runCatchingCancellable {

@@ -378,12 +376,14 @@ class ChatViewModel(

378376 agent = _selectedAgent.value,

379377 )

380378 }.onFailure {

379+ suppressDraftPersist.set(false)

381380 _failedDraft.value = trimmed

382381 // Only restore the draft if the user hasn't typed anything new since.

383382 if (_draft.value.isBlank()) updateDraft(trimmed)

384383 _error.value = container.friendlyError(it)

385384 _running.value = false

386385 }.onSuccess {

386+ suppressDraftPersist.set(false)

387387 // Send succeeded — now it's safe to persist the empty draft.

388388 container.draftStore.set(sessionId, "")

389389 // Don't reset _running here: the agent continues streaming via SSE.

@@ -397,8 +397,11 @@ class ChatViewModel(

397397 /** Re-send the last draft whose send failed, if any. */

398398 fun retryFailed() {

399399 val draft = _failedDraft.value ?: return

400− _failedDraft.value = null

401− send(draft)

400+ // Don't clear _failedDraft until send() accepts the text — if _running is

401+ // already true, send() returns false and the draft would be lost forever.

402+ if (send(draft)) {

403+ _failedDraft.value = null

404+ }

402405 }

403406

404407 /** Invoke a slash-command by name via the server's /command endpoint. */

@@ -429,19 +432,19 @@ class ChatViewModel(

429432 fun reconnect() {

430433 if (container.activeConnection.value != null) return

431434 viewModelScope.launch {

432− _loading.value = true

435+ _reconnecting.value = true

433436 val recent = runCatchingCancellable { container.profileStore.profiles.first() }

434437 .getOrDefault(emptyList())

435− .firstOrNull() ?: run { _loading.value = false; return@launch }

438+ .firstOrNull() ?: run { _reconnecting.value = false; return@launch }

436439 runCatchingCancellable {

437440 val conn = container.connect(recent)

438441 conn.api.ping()

439442 }.onSuccess {

440− _loading.value = false

443+ _reconnecting.value = false

441444 }.onFailure {

442445 container.disconnect()

443446 _error.value = container.friendlyError(it)

444− _loading.value = false

447+ _reconnecting.value = false

445448 }

446449 }

447450 }