feat(tui-03): add SnapshotPhaseReducer with buffering and dedup logic

Buffers live events during snapshot phase, drains on SnapshotComplete, deduplicates live events by per-session cursor, and resets on disconnect.
2026-05-25 11:32:03 +04:00
parent b50d5fc6ee
commit d841e1c4b2
2 changed files with 121 additions and 0 deletions
@@ -0,0 +1,87 @@
+package com.correx.apps.tui.reducer
+
+import com.correx.apps.server.protocol.ServerMessage
+import com.correx.apps.tui.input.Action
+import com.correx.apps.tui.state.TuiState
+import org.slf4j.LoggerFactory
+
+object SnapshotPhaseReducer {
+    private val log = LoggerFactory.getLogger(SnapshotPhaseReducer::class.java)
+
+    data class Result(val state: TuiState, val actionsToDispatch: List<Action>)
+
+    fun process(state: TuiState, action: Action): Result = when (action) {
+        is Action.Disconnected -> Result(
+            state.copy(snapshotPhase = true, pendingEvents = emptyList(), cursors = emptyMap()),
+            listOf(action),
+        )
+        is Action.ServerEventReceived -> handleServerEvent(state, action)
+        else -> Result(state, listOf(action))
+    }
+
+    private fun handleServerEvent(state: TuiState, action: Action.ServerEventReceived): Result {
+        val msg = action.message
+        return when {
+            msg is ServerMessage.SessionSnapshot -> {
+                val newCursors = state.cursors + (msg.sessionId.value to msg.lastSessionSequence)
+                Result(state.copy(cursors = newCursors), listOf(action))
+            }
+            msg is ServerMessage.SnapshotComplete -> {
+                val replayActions = state.pendingEvents.map { Action.ServerEventReceived(it) }
+                Result(
+                    state.copy(snapshotPhase = false, pendingEvents = emptyList()),
+                    replayActions,
+                )
+            }
+            state.snapshotPhase -> {
+                Result(state.copy(pendingEvents = state.pendingEvents + msg), emptyList())
+            }
+            else -> dedupeLive(state, action, msg)
+        }
+    }
+
+    private fun dedupeLive(
+        state: TuiState,
+        action: Action.ServerEventReceived,
+        msg: ServerMessage,
+    ): Result {
+        val sid = sessionIdOf(msg) ?: return Result(state, listOf(action))
+        val seq = sessionSequenceOf(msg) ?: return Result(state, listOf(action))
+        val cursor = state.cursors[sid] ?: 0L
+        return when {
+            seq <= cursor -> {
+                log.debug("dropping duplicate sessionSequence={} for sid={} (cursor={})", seq, sid, cursor)
+                Result(state, emptyList())
+            }
+            seq > cursor + 1 -> {
+                log.warn("gap detected for sid={}: cursor={} -> incoming seq={} (applying anyway)", sid, cursor, seq)
+                Result(state.copy(cursors = state.cursors + (sid to seq)), listOf(action))
+            }
+            else -> Result(state.copy(cursors = state.cursors + (sid to seq)), listOf(action))
+        }
+    }
+
+    internal fun sessionIdOf(msg: ServerMessage): String? = when (msg) {
+        is ServerMessage.SessionStarted -> msg.sessionId.value
+        is ServerMessage.SessionPaused -> msg.sessionId.value
+        is ServerMessage.SessionCompleted -> msg.sessionId.value
+        is ServerMessage.SessionFailed -> msg.sessionId.value
+        is ServerMessage.StageStarted -> msg.sessionId.value
+        is ServerMessage.StageCompleted -> msg.sessionId.value
+        is ServerMessage.StageFailed -> msg.sessionId.value
+        is ServerMessage.InferenceStarted -> msg.sessionId.value
+        is ServerMessage.InferenceCompleted -> msg.sessionId.value
+        is ServerMessage.InferenceTimedOut -> msg.sessionId.value
+        is ServerMessage.ToolStarted -> msg.sessionId.value
+        is ServerMessage.ToolCompleted -> msg.sessionId.value
+        is ServerMessage.ToolFailed -> msg.sessionId.value
+        is ServerMessage.ToolRejected -> msg.sessionId.value
+        is ServerMessage.ApprovalRequired -> msg.sessionId.value
+        else -> null
+    }
+
+    internal fun sessionSequenceOf(msg: ServerMessage): Long? = when (msg) {
+        is ServerMessage.SessionMessage -> msg.sessionSequence
+        else -> null
+    }
+}
@@ -0,0 +1,34 @@
+package com.correx.apps.tui.reducer
+
+import com.correx.apps.server.protocol.ServerMessage
+import com.correx.apps.tui.input.Action
+import com.correx.apps.tui.state.TuiState
+import com.correx.core.events.types.SessionId
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class SnapshotPhaseReducerTest {
+
+    @Test
+    fun `during snapshotPhase, three live ServerEventReceived actions are buffered and sessions unchanged`() {
+        var state = TuiState(snapshotPhase = true)
+        val allDispatched = mutableListOf<Action>()
+
+        repeat(3) {
+            val msg = ServerMessage.SessionStarted(
+                sessionId = SessionId("s1"),
+                workflowId = "wf",
+                sequence = 1L,
+                sessionSequence = 1L,
+            )
+            val result = SnapshotPhaseReducer.process(state, Action.ServerEventReceived(msg))
+            state = result.state
+            allDispatched += result.actionsToDispatch
+        }
+
+        assertEquals(3, state.pendingEvents.size)
+        assertTrue(allDispatched.isEmpty())
+        assertTrue(state.sessions.sessions.isEmpty())
+    }
+}