Fix SSE live updates for open conversations (#14)

.gitignore

@@ -44,6 +44,10 @@ captures/
 .externalNativeBuild/
 .cxx/
 
+# Local dev convenience script (machine-specific paths)
+Makefile
+CLAUDE.md
+
 # Temp files
 temp_opencode_analysis/
 # Images (except screenshots/ and app resources)

app/src/main/java/dev/blazelight/p4oc/core/network/OpenCodeEventSource.kt

@@ -17,12 +17,14 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.flow.asSharedFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
 import kotlinx.serialization.json.Json
 import okhttp3.OkHttpClient
@@ -44,6 +46,12 @@ class OpenCodeEventSource(
     companion object {
         private const val TAG = "OpenCodeEventSource"
         private const val MAX_CONSECUTIVE_ERRORS = 15
+
+        // Liveness watchdog: the server emits frequent heartbeat comments, so a Connected socket
+        // that receives no frame (message OR heartbeat) for this long is presumed dead (silent
+        // zombie socket — readTimeout is 0). Conservative vs. the heartbeat cadence. (Issue #14.)
+        private const val SSE_STALE_MS = 60_000L
+        private const val WATCHDOG_INTERVAL_MS = 20_000L
     }
 
     private val eventPumpScope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
@@ -73,6 +81,15 @@ class OpenCodeEventSource(
 
     private val consecutiveErrors = AtomicInteger(0)
 
+    // Timestamp of the last received SSE frame (message or heartbeat comment). 0 = none yet.
+    // Read by the watchdog off the event thread, so @Volatile.
+    @Volatile
+    private var lastFrameAtMs: Long = 0L
+
+    private fun markFrameReceived() {
+        lastFrameAtMs = System.currentTimeMillis()
+    }
+
     init {
         eventPumpScope.launch {
             for (queued in eventChannel) {
@@ -83,8 +100,37 @@ class OpenCodeEventSource(
                 }
             }
         }
+        startLivenessWatchdog()
     }
 
+    /**
+     * Force a reconnect when the socket claims Connected but has gone silent past [SSE_STALE_MS].
+     * With readTimeout(0) the library never surfaces a silently-dead socket, so nothing else would
+     * detect it — the app would show "connected" while receiving zero events (issue #14).
+     */
+    private fun startLivenessWatchdog() {
+        eventPumpScope.launch {
+            while (isActive) {
+                delay(WATCHDOG_INTERVAL_MS)
+                if (isStale(System.currentTimeMillis(), lastFrameAtMs, _connectionState.value)) {
+                    val idle = System.currentTimeMillis() - lastFrameAtMs
+                    AppLog.w(TAG, "SSE idle ${idle}ms > ${SSE_STALE_MS}ms while Connected – forcing reconnect")
+                    reconnect()
+                }
+            }
+        }
+    }
+
+    /**
+     * True when the socket claims [ConnectionState.Connected] but has received no frame
+     * (message or heartbeat) within [SSE_STALE_MS]. `lastFrameAtMs == 0L` means no frame yet
+     * (connect in progress) — never stale. Extracted for deterministic unit testing.
+     */
+    internal fun isStale(nowMs: Long, lastFrameAtMs: Long, state: ConnectionState): Boolean =
+        lastFrameAtMs != 0L &&
+            state is ConnectionState.Connected &&
+            (nowMs - lastFrameAtMs) > SSE_STALE_MS
+
     fun connect() {
         val besRef: BackgroundEventSource
         synchronized(lock) {
@@ -100,6 +146,7 @@ class OpenCodeEventSource(
             AppLog.d(TAG, "connect() – creating BackgroundEventSource")
             _connectionState.value = ConnectionState.Connecting
             consecutiveErrors.set(0)
+            markFrameReceived()
 
             val gen = ++generation
             besRef = createBackgroundEventSource(gen)
@@ -132,6 +179,7 @@ class OpenCodeEventSource(
             toClose = detachLocked()
             _connectionState.value = ConnectionState.Connecting
             consecutiveErrors.set(0)
+            markFrameReceived()
 
             val gen = ++generation
             besRef = createBackgroundEventSource(gen)
@@ -155,6 +203,9 @@ class OpenCodeEventSource(
             _connectionState.value = ConnectionState.Disconnected
         }
         toClose?.closeSafely()
+        // Stop the event pump + liveness watchdog so they don't outlive this instance.
+        eventPumpScope.cancel("OpenCodeEventSource shut down")
+        eventChannel.close()
     }
 
     /**
@@ -279,19 +330,23 @@ class OpenCodeEventSource(
             AppLog.d(TAG, "SSE connected (onOpen)")
             errorFiredSinceOpen = false
             consecutiveErrors.set(0)
+            markFrameReceived()
             _connectionState.value = ConnectionState.Connected
             enqueueEvent(OpenCodeEvent.Connected, gen)
         }
 
         override fun onMessage(event: String, messageEvent: MessageEvent) {
             if (!isActiveGeneration(gen)) return
+            markFrameReceived()
             val data = messageEvent.data
             AppLog.v(TAG, "SSE message received: event=$event, length=${data.length}")
             parseAndEmitEvent(data, gen)
         }
 
         override fun onComment(comment: String) {
-            // Keepalive — nothing to do
+            // Keepalive heartbeat — no event to emit, but it proves the socket is alive.
+            if (!isActiveGeneration(gen)) return
+            markFrameReceived()
         }
 
         override fun onClosed() {

app/src/main/java/dev/blazelight/p4oc/data/session/SessionRepositoryImpl.kt

@@ -191,6 +191,10 @@ class SessionRepositoryImpl(
     override fun acceptEvent(event: OpenCodeEvent) {
         if (event is OpenCodeEvent.Connected) {
             hydrateAfterReconnect()
+            // Message refresh must run on EVERY reconnect, independent of the snapshot-hydrate
+            // inFlight guard above (which only fires once). Messages are a separate concern from
+            // the session-list snapshot, so refetch them directly here (issue #14).
+            scope.launch { refreshOpenSessionMessages() }
             scope.launch { reconcileObservedPendingPermissions() }
             return
         }
@@ -345,6 +349,21 @@ class SessionRepositoryImpl(
         }
     }
 
+    /**
+     * Re-fetch messages for every actively-observed session after an SSE reconnect.
+     * Called directly from the [OpenCodeEvent.Connected] path (NOT gated by the snapshot-hydrate
+     * inFlight guard, which only permits one hydration) so an open conversation self-heals on every
+     * reconnect (issue #14: had to leave and re-enter to see updates).
+     * Reuses the same REST refetch/merge path [loadMessages] that ChatViewModel uses on entry.
+     */
+    private suspend fun refreshOpenSessionMessages() {
+        val openSessions = synchronized(messageStates) { messageStates.keys.toList() }
+        for (id in openSessions) {
+            runCatching { loadMessages(SessionId(id), limit = null) }
+                .onFailure { AppLog.w(TAG, "Post-reconnect message refresh failed for $id: ${it.message}") }
+        }
+    }
+
     override fun messages(sessionId: SessionId): StateFlow<List<MessageWithParts>> = messageState(
         sessionId.value
     ).asStateFlow()

app/src/main/java/dev/blazelight/p4oc/data/session/SessionRepositoryProvider.kt

@@ -1,12 +1,16 @@
 package dev.blazelight.p4oc.data.session
 
+import dev.blazelight.p4oc.core.log.AppLog
 import dev.blazelight.p4oc.core.network.ConnectionManager
+import dev.blazelight.p4oc.core.network.ConnectionState
 import dev.blazelight.p4oc.data.remote.mapper.MessageMapper
 import dev.blazelight.p4oc.data.server.ActiveServerApiProvider
 import dev.blazelight.p4oc.data.workspace.WorkspaceClient
+import dev.blazelight.p4oc.domain.model.OpenCodeEvent
 import dev.blazelight.p4oc.domain.server.ServerGeneration
 import dev.blazelight.p4oc.domain.server.WorkspaceKey
 import dev.blazelight.p4oc.domain.workspace.Workspace
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -41,6 +45,46 @@ class SessionRepositoryProvider(
     private val scope = CoroutineScope(SupervisorJob() + dispatcher)
     private val entries = mutableMapOf<Key, Entry>()
 
+    private companion object {
+        const val TAG = "SessionRepositoryProvider"
+    }
+
+    init {
+        // Deliver SSE reconnects to every live workspace repository so open conversations
+        // self-heal without navigation (issue #14). The synthetic OpenCodeEvent.Connected never
+        // reaches the per-workspace fan-out (it carries no directory), so we broadcast it here on
+        // each non-Connected → Connected transition of the shared connection.
+        scope.launch {
+            var wasConnected = false
+            connectionManager.connectionState.collect { state ->
+                val nowConnected = state is ConnectionState.Connected
+                if (nowConnected && !wasConnected) {
+                    broadcastReconnect()
+                }
+                wasConnected = nowConnected
+            }
+        }
+    }
+
+    @Suppress("TooGenericExceptionCaught") // resilience guard: one bad repo must not stop the broadcast
+    private fun broadcastReconnect() {
+        // Generation is a single global counter in ConnectionManager, so one generation == one
+        // connection == one server: filtering by generation alone correctly targets exactly the
+        // repositories on the reconnected connection, without depending on server-URL normalization.
+        val generation = connectionManager.currentGeneration ?: return
+        val repositories = synchronized(this) {
+            entries.filterKeys { it.generation == generation.value }
+                .map { it.value.repository }
+        }
+        repositories.forEach { repository ->
+            try {
+                repository.acceptEvent(OpenCodeEvent.Connected)
+            } catch (e: Exception) {
+                AppLog.e(TAG, "Reconnect broadcast failed for a workspace: ${e.message}", e)
+            }
+        }
+    }
+
     fun acquire(workspace: Workspace, generation: ServerGeneration): Lease = synchronized(this) {
         val key = workspace.toProviderKey(generation)
         val entry = entries.getOrPut(key) {
@@ -70,6 +114,7 @@ class SessionRepositoryProvider(
         repositoryToClose.repository.close()
     }
 
+    @Suppress("TooGenericExceptionCaught") // resilience guard: one bad event must not kill the collector
     private fun collectWorkspaceEvents(
         workspace: Workspace,
         generation: ServerGeneration,
@@ -80,7 +125,19 @@ class SessionRepositoryProvider(
                 scopedEvent.generation == generation &&
                 scopedEvent.workspaceKey == workspace.key
             ) {
-                repository.acceptEvent(scopedEvent.event)
+                try {
+                    repository.acceptEvent(scopedEvent.event)
+                } catch (ce: CancellationException) {
+                    throw ce
+                } catch (e: Exception) {
+                    // A single malformed/unexpected event must never permanently kill this
+                    // workspace's live event delivery (issue #14: chat froze until re-entry).
+                    AppLog.e(
+                        TAG,
+                        "Dropping event ${scopedEvent.event::class.simpleName} for ${workspace.key}: ${e.message}",
+                        e,
+                    )
+                }
             }
         }
     }

app/src/test/java/dev/blazelight/p4oc/core/network/OpenCodeEventSourceTest.kt

@@ -43,6 +43,35 @@ class OpenCodeEventSourceTest {
         unmockkObject(AppLog)
     }
 
+    @Test
+    fun `isStale flags a silent Connected socket past the threshold`() {
+        val source = newSource()
+        try {
+            val now = 1_000_000L
+            val staleFrame = now - 61_000L // > SSE_STALE_MS (60s)
+            val freshFrame = now - 30_000L
+
+            // Connected + no frame for > 60s => stale (the zombie-socket case, issue #14).
+            assertEquals(true, source.isStale(now, staleFrame, ConnectionState.Connected))
+            // Connected but a recent frame => not stale.
+            assertEquals(false, source.isStale(now, freshFrame, ConnectionState.Connected))
+            // No frame yet (0L) => never stale, even if "Connected".
+            assertEquals(false, source.isStale(now, 0L, ConnectionState.Connected))
+            // Not Connected => watchdog leaves reconnect to the normal error/escalation path.
+            assertEquals(false, source.isStale(now, staleFrame, ConnectionState.Disconnected))
+            assertEquals(false, source.isStale(now, staleFrame, ConnectionState.Connecting))
+        } finally {
+            source.shutdown()
+        }
+    }
+
+    private fun newSource(): OpenCodeEventSource = OpenCodeEventSource(
+        okHttpClient = OkHttpClient(),
+        json = json,
+        baseUrl = "http://127.0.0.1:1",
+        eventMapper = EventMapper(json, MessageMapper(json)),
+    )
+
     @Test
     fun `slow collector receives more than previous delta buffer capacity without loss`() = runTest {
         val source = OpenCodeEventSource(