diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt new file mode 100644 index 00000000..402fb129 --- /dev/null +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt @@ -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) + + 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 + } +} diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducerTest.kt new file mode 100644 index 00000000..70d44381 --- /dev/null +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducerTest.kt @@ -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() + + 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()) + } +}