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

7a6768eiko3 months ago

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

compare with master

unifiedsplit

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

164 }

165 }

166 } catch (c: CancellationException) {

167+ _state.value = ConnectionState.Disconnected

167 throw c

168 throw c

168 } catch (e: Exception) {

169 } catch (e: Exception) {

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

170 // 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) {

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

46 // between sessions on the same server.

47 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()

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

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

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

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

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

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

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

117 command: String,

122 command: String,

118 arguments: String = "",

123 arguments: String = "",

119 agent: String? = null,

124 agent: String? = null,

120− ): MessageWithParts = withRetry {

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 {

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

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

122− contentType(ContentType.Application.Json)

132+ contentType(ContentType.Application.Json)

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

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

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

124− }.body()

135+ }.body()

136+ }

125 }

137 }

126

138

127− suspend fun providers(): ProvidersResponse {

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

128 val now = System.currentTimeMillis()

140 val now = System.currentTimeMillis()

129− cacheMutex.withLock {

130− val cached = cachedProviders

141+ val cached = cachedProviders

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

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

132− return cached.value

143+ return@withLock cached.value

133− }

134 }

144 }

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

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

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

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

137− return value

147+ value

138 }

148 }

139

149

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

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

141 val now = System.currentTimeMillis()

151 val now = System.currentTimeMillis()

142− cacheMutex.withLock {

143− val cached = cachedAgents

152+ val cached = cachedAgents

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

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

145− return cached.value

154+ return@withLock cached.value

146− }

147 }

155 }

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

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

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

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

150− return value

158+ value

151 }

159 }

152

160

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

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

154 val now = System.currentTimeMillis()

162 val now = System.currentTimeMillis()

155− cacheMutex.withLock {

156− val cached = cachedCommands

163+ val cached = cachedCommands

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

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

158− return cached.value

165+ return@withLock cached.value

159− }

160 }

166 }

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

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

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

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

163− return value

169+ value

164 }

170 }

165

171

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

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

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

173+ suspend fun invalidateCache() {

168− cachedProviders = null

174+ providersMutex.withLock { cachedProviders = null }

169− cachedAgents = null

175+ agentsMutex.withLock { cachedAgents = null }

170− cachedCommands = null

176+ commandsMutex.withLock { cachedCommands = null }

171 }

177 }

172

178

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

179 /** 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) {

59 val snapshot = runCatching {

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

61 }.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+ }

63 _ready.value = true

69 _ready.value = true

64 }

70 }

65 }

71 }

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

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

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

152 fun shutdown() {

158 fun shutdown() {

153 flushExecutor.shutdown()

159 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) }

154 }

163 }

155}

164}

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

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

74 }

75 }

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

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

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

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

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

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

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

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

87 launch {

81 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

89 eventStream.state.collect { state ->

93 eventStream.state.collect { state ->

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

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

91− if (wasConnected) {

95+ if (hasConnectedBefore) {

96+ val generation = ++seedGeneration

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

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

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

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

94 .getOrDefault(emptyList())

99 .getOrDefault(emptyList())

100+ lock.withLock {

101+ if (generation == seedGeneration) {

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

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

103+ }

104+ }

96 publish()

105 publish()

97 }

106 }

98− wasConnected = true

107+ hasConnectedBefore = true

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

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

101− wasConnected = false

102 }

108 }

103 }

109 }

104 }

110 }

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

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

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

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

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

135

141

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

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

144+

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

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

137 internal val maxMessages = NetworkConfig.maxInMemoryMessages

146 internal val maxMessages = NetworkConfig.maxInMemoryMessages

138

147

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

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

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

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

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

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

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

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

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

189 } else {

198 } else {

190 info.id

199 info.id

191 }

200 }

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

207 }

216 }

208 }

217 }

209

218

210− is MessagePartRemoved -> {

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

211− val messageId = event.properties.messageID

212− val partId = event.properties.partID

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

214− }

215

220

216− is MessageRemoved -> {

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

217− val id = event.properties.messageID

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

219− }

220

222

221 else -> false

223 else -> false

222 }

224 }

223 }

225 }

224

226

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+

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

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

226 val current = messages[messageId]

245 val current = messages[messageId]

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

246 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

27import kotlinx.coroutines.flow.SharingStarted

28import kotlinx.coroutines.flow.StateFlow

29import kotlinx.coroutines.flow.asStateFlow

30−import kotlinx.coroutines.flow.catch

31import kotlinx.coroutines.CancellationException

30import kotlinx.coroutines.CancellationException

32import kotlinx.coroutines.flow.collectLatest

31import kotlinx.coroutines.flow.collectLatest

33import kotlinx.coroutines.flow.debounce

32import kotlinx.coroutines.flow.debounce

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

35import kotlinx.coroutines.flow.first

34import kotlinx.coroutines.flow.first

36import kotlinx.coroutines.flow.flatMapLatest

35import kotlinx.coroutines.flow.flatMapLatest

37import kotlinx.coroutines.flow.flowOf

36import kotlinx.coroutines.flow.flowOf

37+import kotlinx.coroutines.flow.merge

38import kotlinx.coroutines.flow.onEach

39+import kotlinx.coroutines.flow.retryWhen

39import kotlinx.coroutines.flow.stateIn

40import kotlinx.coroutines.flow.stateIn

41+import kotlinx.coroutines.delay

40import kotlinx.coroutines.launch

42import kotlinx.coroutines.launch

42−@OptIn(ExperimentalCoroutinesApi::class)

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

43class ChatViewModel(

45class ChatViewModel(

44 private val container: AppContainer,

46 private val container: AppContainer,

45 private val sessionId: String,

47 private val sessionId: String,

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

52 private val _loading = MutableStateFlow(true)

54 private val _loading = MutableStateFlow(true)

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

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

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+

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

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

56 container.activeConnection

75 container.activeConnection

57 .flatMapLatest { conn ->

76 .flatMapLatest { conn ->

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

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

59 }

78 }

60 .onEach { _loading.value = false }

79 .onEach { _loading.value = false }

61− .catch {

80+ .retryWhen { cause, _ ->

62 _loading.value = false

81 _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

64 }

86 }

65 .stateIn(

87 .stateIn(

66 scope = viewModelScope,

88 scope = viewModelScope,

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

155

177

156 fun reloadModels() {

178 fun reloadModels() {

157 val conn = connection ?: return

179 val conn = connection ?: return

158− reloadCatalog(

180+ viewModelScope.launch {

159− conn = conn,

181+ conn.api.invalidateCache()

160− tag = "model catalog",

182+ _modelsReload.value++

161− loading = _modelsLoading,

183+ }

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− )

170 }

184 }

171

185

172 fun reloadAgents() {

186 fun reloadAgents() {

173 val conn = connection ?: return

187 val conn = connection ?: return

174− reloadCatalog(

188+ viewModelScope.launch {

175− conn = conn,

189+ conn.api.invalidateCache()

176− tag = "agent catalog",

190+ _agentsReload.value++

177− loading = _agentsLoading,

191+ }

178− error = _agentsError,

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

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

181− )

182 }

192 }

183

193

184 fun reloadCommands() {

194 fun reloadCommands() {

185 val conn = connection ?: return

195 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− ) {

209 viewModelScope.launch {

196 viewModelScope.launch {

210 conn.api.invalidateCache()

197 conn.api.invalidateCache()

211− loading.value = true

198+ _commandsReload.value++

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

217 }

199 }

218 }

200 }

219

201

220 /**

202 /**

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

203 * 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

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

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

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

224− * to the UI.

225 */

210 */

226 private fun <T> observeCatalog(

211 private fun <T> observeCatalog(

227 tag: String,

212 tag: String,

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

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

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

231 onSuccess: (T) -> Unit,

216 onSuccess: (T) -> Unit,

232 onNull: () -> Unit,

217 onNull: () -> Unit,

218+ reloadTrigger: StateFlow<Int>,

233 ) {

219 ) {

234 viewModelScope.launch {

220 viewModelScope.launch {

235− container.activeConnection.collectLatest { conn ->

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

222+ val conn = container.activeConnection.value

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

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

237 loading.value = true

224 loading.value = true

238 error.value = false

225 error.value = false

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

250

237

251 init {

238 init {

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

239 // 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.

253 viewModelScope.launch {

244 viewModelScope.launch {

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

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

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

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

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

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

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

257 }

249 }

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

308 _selectedModel.value = resp.defaultOption(options)

300 _selectedModel.value = resp.defaultOption(options)

309 },

301 },

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

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

303+ reloadTrigger = _modelsReload,

311 )

304 )

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

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

313 observeCatalog(

306 observeCatalog(

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

317 fetch = { it.agents() },

310 fetch = { it.agents() },

318 onSuccess = { _agents.value = it },

311 onSuccess = { _agents.value = it },

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

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

313+ reloadTrigger = _agentsReload,

320 )

314 )

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

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

322 observeCatalog(

316 observeCatalog(

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

326 fetch = { it.commands() },

320 fetch = { it.commands() },

327 onSuccess = { _commands.value = it },

321 onSuccess = { _commands.value = it },

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

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

323+ reloadTrigger = _commandsReload,

329 )

324 )

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

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

331 viewModelScope.launch {

326 viewModelScope.launch {

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

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

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

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

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

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

365 // 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)

371 _draft.value = ""

369 _draft.value = ""

372 viewModelScope.launch {

370 viewModelScope.launch {

373 runCatchingCancellable {

371 runCatchingCancellable {

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

378 agent = _selectedAgent.value,

376 agent = _selectedAgent.value,

379 )

377 )

380 }.onFailure {

378 }.onFailure {

379+ suppressDraftPersist.set(false)

381 _failedDraft.value = trimmed

380 _failedDraft.value = trimmed

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

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

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

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

384 _error.value = container.friendlyError(it)

383 _error.value = container.friendlyError(it)

385 _running.value = false

384 _running.value = false

386 }.onSuccess {

385 }.onSuccess {

386+ suppressDraftPersist.set(false)

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

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

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

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

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

398 fun retryFailed() {

399 val draft = _failedDraft.value ?: return

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)) {

400− _failedDraft.value = null

403+ _failedDraft.value = null

401− send(draft)

404+ }

402 }

405 }

403

406

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

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

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

429 fun reconnect() {

432 fun reconnect() {

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

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

431 viewModelScope.launch {

434 viewModelScope.launch {

432− _loading.value = true

435+ _reconnecting.value = true

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

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

434 .getOrDefault(emptyList())

437 .getOrDefault(emptyList())

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

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

436 runCatchingCancellable {

439 runCatchingCancellable {

437 val conn = container.connect(recent)

440 val conn = container.connect(recent)

438 conn.api.ping()

441 conn.api.ping()

439 }.onSuccess {

442 }.onSuccess {

440− _loading.value = false

443+ _reconnecting.value = false

441 }.onFailure {

444 }.onFailure {

442 container.disconnect()

445 container.disconnect()

443 _error.value = container.friendlyError(it)

446 _error.value = container.friendlyError(it)

444− _loading.value = false

447+ _reconnecting.value = false

445 }

448 }

446 }

449 }

447 }

450 }