Fix SSE live updates for open conversations (#14)

[Melloss]

Summary

Fixes #14 — "the app isn't refreshing / i have to get out of the
conversation and enter again to see it."

Root cause: leaving and re-entering a chat screen "fixed" it only
because it constructs a new ChatViewModel, which does a one-shot REST
refetch. The actual SSE live-update pipeline was never being repaired —
it stayed broken, so the conversation went stale again shortly after.

Three independent, compounding defects in the SSE/session layer:

1. Unguarded exception silently kills live updates.
   SessionRepositoryProvider.collectWorkspaceEvents called
   repository.acceptEvent(event) inside a collect {} with no
   try/catch. Since the collector runs under a SupervisorJob, one
   throwing event permanently and silently terminated only that
   workspace's event job — no log, no crash, no UI signal, and it was
   never recreated by chat-screen navigation (only by closing the tab).

2. Zombie SSE sockets go undetected. OpenCodeEventSource used
   readTimeout(0), so a socket that died silently (NAT/proxy idle-kill,
   Doze-related network changes) never fired onError/onClosed.
   connectionState stayed Connected indefinitely with zero events
   flowing, and nothing checked for it.

3. Reconnect self-heal was dead code, and incomplete even when live.
   The synthetic OpenCodeEvent.Connected never reached the
   per-workspace fan-out (it has no directory, so it never entered
   ConnectionManager.scopedEvents), meaning
   SessionRepositoryImpl.acceptEvent's reconnect → re-hydrate path
   (hydrateAfterReconnect()) never fired in production. Even when it
   did fire it only rebuilt the session/project snapshot — it never
   refreshed a conversation's messages.

Changes

• SessionRepositoryProvider.kt — wrap acceptEvent in try/catch
  (rethrows CancellationException, logs everything else) so one bad
  event can't kill the collector. Also broadcasts a synthetic
  Connected event to every live workspace repository (filtered by
  connection generation) on each non-Connected → Connected transition,
  since the original per-directory routing could never reach it.
• OpenCodeEventSource.kt — liveness watchdog: tracks the last received
  frame (message or heartbeat) and forces reconnect() if the socket
  claims Connected but has been silent > 60s. Also fixes a pre-existing
  leak where shutdown() didn't cancel the event-pump scope.
• SessionRepositoryImpl.kt — on every Connected event, re-fetches
  messages (via the existing loadMessages REST path) for every
  actively-observed session, independent of the one-shot snapshot-hydrate
  guard (inFlight), so it runs on every reconnect, not just the first.
• docs/design-locks/B-sse-hydrate-race.md — extends item 11 with a new
  item 12 + worked examples documenting that reconnect self-heal now
  covers message state, not just the session/project snapshot.

Test plan

• ./gradlew :app:testDebugUnitTest — full suite passes, including
  new regression tests:
  • collector survives a throwing event and keeps processing subsequent
    events (SessionRepositoryProviderTest)
  • messages refresh on every reconnect, not just the first
    (SessionRepositoryProviderTest — this one initially caught a bug
    in my own first attempt, where the refresh was accidentally gated
    behind the one-shot hydrate guard)
  • watchdog staleness decision logic (OpenCodeEventSourceTest)
• ./gradlew :app:detekt — clean
• ./scripts/check_theme_violations.sh — clean
• Manual, on-device: stopped the OpenCode server while a conversation
  was open, sent a message from the desktop TUI, restarted the server —
  the open conversation on the phone caught up without navigating away
  and back.

[zcmx]

Thanks for investigating this — the watchdog and reconnect-triggered recovery are useful directions. While adapting the approach downstream, I found a few improvements that may make the fix safer and more complete:

1. Use authoritative reconnect reconciliation guarded by a session revision

The reconnect refetch currently goes through loadMessages() / mergeLoadedMessages(). That merge preserves local-only messages and parts and prefers current state for matching IDs, so it cannot fully repair missed removals or stale part contents.

Consider a reconnect-specific path that:

1. captures a per-session message revision before the REST request;
2. fetches the current server snapshot;
3. replaces the loaded window authoritatively only if the revision is unchanged;
4. retries a small bounded number of times if an SSE mutation arrived while REST was in flight.

This allows REST to repair missed updates and removals without overwriting newer live events.

2. Track active sessions explicitly and keep reconnect fetches bounded

messageStates.keys does not necessarily mean “actively observed”: entries remain cached after the UI stops observing them, until the repository closes. Reconnects can therefore refetch every previously opened conversation. Also, limit = null may turn each reconnect into an unbounded history download.

A safer approach is to use a session lease/ref-count, reconcile only sessions with active consumers, remember the currently loaded history window for each active session, and use a bounded default (for example 100 messages).

3. Use monotonic elapsed time for liveness

The watchdog currently uses System.currentTimeMillis(). Wall-clock changes can cause false reconnects or delay stale detection. System.nanoTime() converted to milliseconds (or another monotonic elapsed-time source) is safer for timeout calculations.

4. Add watchdog lifecycle tests

The current test covers the isStale predicate, but not the watchdog behavior itself. It would be useful to verify that:

• a stale connected source invokes reconnect;
• repeated checks do not create a reconnect storm;
• a message or heartbeat/comment refreshes liveness;
• shutdown() cancels the watchdog.

5. Scope reconnect delivery by server identity and generation

The provider currently observes one application-global ConnectionManager and filters repositories only by generation. This becomes ambiguous if multiple server connections exist. If multi-server support is planned, reconnect observation and delivery should be scoped by both immutable server identity and connection generation.

The collector exception isolation already included in this PR is a good improvement and should be retained.

[zcmx]

Follow-up with concrete Kotlin sketches based on the downstream implementation. These are intentionally adapted to this PR's single-ConnectionManager architecture rather than copied verbatim from the downstream multi-server branch, so a few names may need adjustment.

1. Track active consumers and message revisions

In SessionRepositoryImpl:

private val messageReconciliationLock = Any()
private val messageRevisions = mutableMapOf<String, Long>()
private val messageHistoryLimits = mutableMapOf<String, Int>()
private val sessionConsumerCounts = mutableMapOf<String, Int>()

fun acquireSession(sessionId: SessionId): AutoCloseable {
    synchronized(sessionConsumerCounts) {
        sessionConsumerCounts[sessionId.value] =
            sessionConsumerCounts.getOrDefault(sessionId.value, 0) + 1
    }

    val released = AtomicBoolean(false)
    return AutoCloseable {
        if (released.compareAndSet(false, true)) {
            releaseSession(sessionId.value)
        }
    }
}

private fun releaseSession(sessionId: String) {
    synchronized(sessionConsumerCounts) {
        val remaining = (sessionConsumerCounts[sessionId] ?: return) - 1
        if (remaining > 0) {
            sessionConsumerCounts[sessionId] = remaining
            return
        }
        sessionConsumerCounts.remove(sessionId)

        synchronized(messageReconciliationLock) {
            messageRevisions.remove(sessionId)
            messageHistoryLimits.remove(sessionId)
            synchronized(messageStates) {
                messageStates.remove(sessionId)?.value = emptyList()
            }
        }
    }
}

The chat owner/ViewModel acquires one lease for its lifetime and closes it from onCleared(). This makes “open session” explicit rather than deriving it from messageStates.keys.

2. Record mutations that race with REST

All message-state mutations should share one revision boundary. For example:

private fun incrementMessageRevisionLocked(sessionId: String) {
    messageRevisions[sessionId] = (messageRevisions[sessionId] ?: 0L) + 1L
}

private fun upsertMessage(message: Message) {
    synchronized(messageReconciliationLock) {
        messageState(message.sessionID).update { messages ->
            val existing = messages.firstOrNull { it.message.id == message.id }
            val updated = if (existing == null) {
                messages + MessageWithParts(message, emptyList())
            } else {
                messages.map { current ->
                    if (current.message.id == message.id) current.copy(message = message) else current
                }
            }
            updated.sortedBy { it.message.createdAt }
        }
        incrementMessageRevisionLocked(message.sessionID)
    }
}

The same pattern should wrap upsertPart, part deltas, removeMessage, removePart, and streaming-flag changes. Normal user pagination can keep its current merge semantics, but should also increment the revision after committing.

3. Authoritative reconnect replacement with bounded retries

private suspend fun reconcileActiveSessionMessages() {
    val sessionIds = synchronized(sessionConsumerCounts) {
        sessionConsumerCounts.keys.toList()
    }

    sessionIds.forEach { sessionId ->
        runCatching { reconcileSessionMessages(sessionId) }
            .onFailure { error ->
                if (error is CancellationException) throw error
                AppLog.w(
                    TAG,
                    "Post-reconnect message reconciliation failed: ${error.javaClass.simpleName}",
                )
            }
    }
}

private suspend fun reconcileSessionMessages(sessionId: String) {
    repeat(MESSAGE_RECONCILIATION_ATTEMPTS) { attempt ->
        val stillActive = synchronized(sessionConsumerCounts) {
            sessionConsumerCounts.containsKey(sessionId)
        }
        if (!stillActive) return

        val (expectedRevision, historyLimit) = synchronized(messageReconciliationLock) {
            (messageRevisions[sessionId] ?: 0L) to
                (messageHistoryLimits[sessionId] ?: DEFAULT_MESSAGE_RECONCILIATION_LIMIT)
        }

        val loaded = client.getMessages(sessionId, historyLimit)
            .map(messageMapper::mapWrapperToDomain)

        if (replaceMessagesIfUnchanged(sessionId, expectedRevision, loaded)) return
        if (attempt + 1 < MESSAGE_RECONCILIATION_ATTEMPTS) {
            delay(MESSAGE_RECONCILIATION_RETRY_MS)
        }
    }
}

private fun replaceMessagesIfUnchanged(
    sessionId: String,
    expectedRevision: Long,
    loaded: List<MessageWithParts>,
): Boolean = synchronized(sessionConsumerCounts) {
    if (!sessionConsumerCounts.containsKey(sessionId)) return false

    synchronized(messageReconciliationLock) {
        if ((messageRevisions[sessionId] ?: 0L) != expectedRevision) return false

        // Authoritative for exactly the requested history window: this repairs missed removals
        // and stale parts instead of retaining local-only entries forever.
        messageState(sessionId).value = loaded.sortedBy { it.message.createdAt }
        messageRevisions[sessionId] = expectedRevision + 1L
        true
    }
}

private companion object {
    const val DEFAULT_MESSAGE_RECONCILIATION_LIMIT = 100
    const val MESSAGE_RECONCILIATION_ATTEMPTS = 3
    const val MESSAGE_RECONCILIATION_RETRY_MS = 100L
}

When normal history is loaded, remember the largest visible window:

synchronized(messageReconciliationLock) {
    messageHistoryLimits[sessionId.value] = maxOf(
        messageHistoryLimits[sessionId.value] ?: 0,
        limit,
    )
}

Then trigger message recovery independently of snapshot hydration, so an existing snapshot inFlight does not suppress it:

if (event is OpenCodeEvent.Connected) {
    hydrateAfterReconnect()
    scope.launch { reconcileActiveSessionMessages() }
    scope.launch { reconcileObservedPendingPermissions() }
    return
}

4. Monotonic watchdog with throttling

In OpenCodeEventSource:

private val nowMs: () -> Long = { System.nanoTime() / 1_000_000L }

@Volatile
private var lastFrameAtMs: Long = 0L

private fun recordFrame(receivedAtMs: Long = nowMs()) {
    lastFrameAtMs = receivedAtMs
}

internal fun isStale(
    nowMs: Long,
    lastFrameAtMs: Long,
    state: ConnectionState,
): Boolean = lastFrameAtMs != 0L &&
    state is ConnectionState.Connected &&
    nowMs - lastFrameAtMs > SSE_STALE_MS

private fun startLivenessWatchdog() {
    eventPumpScope.launch {
        while (isActive) {
            delay(WATCHDOG_INTERVAL_MS)
            val checkedAt = nowMs()
            if (!isStale(checkedAt, lastFrameAtMs, connectionState.value)) continue

            // Throttle before reconnecting so repeated checks cannot create a reconnect storm.
            recordFrame(checkedAt)
            AppLog.w(TAG, "SSE became unresponsive; restarting live updates")
            reconnect()
        }
    }
}

Call recordFrame() from connect(), reconnect(), onOpen(), onMessage(), and onComment(). Starting/resetting the timestamp before the socket opens also gives each connection attempt a fresh stale window. The watchdog naturally stops when eventPumpScope is cancelled by shutdown().

Suggested tests

@Test
fun `authoritative replacement rejects snapshot after concurrent SSE mutation`() {
    // acquire lease; capture revision; apply SSE mutation; assert replacement returns false
}

@Test
fun `authoritative replacement removes server-deleted message and part`() {
    // seed local-only message/part; replace at unchanged revision; assert they disappear
}

@Test
fun `watchdog reconnects only once per stale window`() {
    // inject fake monotonic clock and reconnect callback; run check twice; assert one reconnect
}

@Test
fun `heartbeat keeps connected source live`() {
    // advance fake clock, record a comment frame, verify no reconnect
}

The key distinction is that normal history loading remains merge-based for SSE safety, while reconnect recovery uses an authoritative, revision-guarded replacement specifically to repair missed state.