From 0cfb7841878375604438cf65b10d5fedc53241b6 Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 24 May 2026 21:47:30 +0400 Subject: [PATCH] feat(server-01): extend wire protocol with sequence cursors and SnapshotComplete Every event-derived ServerMessage gains sequence (global) and sessionSequence (per-session) cursor fields. SessionSnapshot adds lastSequence and lastSessionSequence at projection-read time. SnapshotComplete data object terminates the snapshot phase and is registered in the polymorphic module. replaySnapshot() now emits SnapshotComplete after all SessionSnapshot messages. SessionStarted audit deferred with TODO comment. DomainEventMapper propagates both cursors to all mapped messages including InferenceCompleted. --- .../server/approval/ApprovalCoordinator.kt | 19 +- .../apps/server/bridge/DomainEventMapper.kt | 99 +++++++-- .../apps/server/bridge/SessionEventBridge.kt | 16 +- .../com/correx/apps/server/protocol/Dtos.kt | 13 ++ .../apps/server/protocol/ServerMessage.kt | 205 +++++++++++++++--- .../apps/server/ws/GlobalStreamHandler.kt | 31 ++- .../server/bridge/DomainEventMapperTest.kt | 119 +++++++--- .../server/bridge/SessionEventBridgeTest.kt | 60 +++-- .../ServerMessageSerializationTest.kt | 82 +++++++ .../apps/tui/reducer/ApprovalReducerTest.kt | 2 + .../apps/tui/reducer/RootReducerTest.kt | 4 + .../apps/tui/reducer/SessionsReducerTest.kt | 22 +- .../approvals/DefaultApprovalReducerTest.kt | 1 + .../core/tools/DefaultToolReducerTest.kt | 1 + 14 files changed, 569 insertions(+), 105 deletions(-) create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt index 4bf4e91d..dc9554a6 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -57,6 +57,8 @@ class ApprovalCoordinator( ), toolName = null, preview = null, + sequence = 0L, + sessionSequence = 0L, ) broadcast(event.sessionId, msg) scheduleTimeout(event.requestId, event.sessionId, event.stageId, event.tier) @@ -64,12 +66,25 @@ class ApprovalCoordinator( fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { if (resolved.putIfAbsent(msg.requestId, true) != null) { - return ServerMessage.ProtocolError("Approval request ${msg.requestId.value} already resolved") + return ServerMessage.ProtocolError( + message = "Approval request ${msg.requestId.value} already resolved", + sequence = null, + sessionSequence = null, + ) } timeoutJobs.remove(msg.requestId)?.cancel() val domain = msg.toDomainDecision(sessionId, null, Tier.T2) return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) } - .fold(onSuccess = { null }, onFailure = { ServerMessage.ProtocolError(it.message ?: "Unknown error") }) + .fold( + onSuccess = { null }, + onFailure = { + ServerMessage.ProtocolError( + message = it.message ?: "Unknown error", + sequence = null, + sessionSequence = null, + ) + }, + ) } private suspend fun broadcast(sessionId: SessionId, msg: ServerMessage) { diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt index c5a4715f..f0e81ca0 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt @@ -6,6 +6,7 @@ import com.correx.apps.server.protocol.ServerMessage import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.WorkflowStartedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceTimeoutEvent import com.correx.core.events.events.OrchestrationPausedEvent @@ -33,20 +34,47 @@ private object NoopArtifactStore : ArtifactStore { } @Suppress("CyclomaticComplexMethod") -suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: ArtifactStore): ServerMessage? = - when (val p = event.payload) { - is WorkflowCompletedEvent -> ServerMessage.SessionCompleted(sessionId = p.sessionId) - is WorkflowFailedEvent -> ServerMessage.SessionFailed(sessionId = p.sessionId, reason = p.reason) +suspend fun domainEventToServerMessage( + event: StoredEvent, + artifactStore: ArtifactStore, + sessionSequence: Long = 0L, +): ServerMessage? { + val seq = event.sequence + return when (val p = event.payload) { + is WorkflowStartedEvent -> ServerMessage.SessionStarted( + sessionId = p.sessionId, + workflowId = p.workflowId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowCompletedEvent -> ServerMessage.SessionCompleted( + sessionId = p.sessionId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is WorkflowFailedEvent -> ServerMessage.SessionFailed( + sessionId = p.sessionId, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, + ) + is TransitionExecutedEvent -> ServerMessage.StageStarted( sessionId = p.sessionId, stageId = p.to, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, ) is StageCompletedEvent -> ServerMessage.StageCompleted( sessionId = p.sessionId, stageId = p.stageId, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, ) is StageFailedEvent -> ServerMessage.StageFailed( @@ -54,17 +82,33 @@ suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: Artifa stageId = p.stageId, reason = p.reason, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, ) - is OrchestrationPausedEvent -> mapOrchestrationPaused(p) - is InferenceStartedEvent -> ServerMessage.InferenceStarted(sessionId = p.sessionId, stageId = p.stageId) - is InferenceCompletedEvent -> mapInferenceCompleted(event, p, artifactStore) + is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence) + is InferenceStartedEvent -> ServerMessage.InferenceStarted( + sessionId = p.sessionId, + stageId = p.stageId, + sequence = seq, + sessionSequence = sessionSequence, + ) + + is InferenceCompletedEvent -> mapInferenceCompleted(event, p, artifactStore, sessionSequence) is InferenceTimeoutEvent -> ServerMessage.InferenceTimedOut( - sessionId = p.sessionId, stageId = p.stageId, elapsedMs = p.timeoutMs, + sessionId = p.sessionId, + stageId = p.stageId, + elapsedMs = p.timeoutMs, + sequence = seq, + sessionSequence = sessionSequence, ) is ToolInvocationRequestedEvent -> ServerMessage.ToolStarted( - sessionId = p.sessionId, toolName = p.toolName, tier = p.tier, + sessionId = p.sessionId, + toolName = p.toolName, + tier = p.tier, + sequence = seq, + sessionSequence = sessionSequence, ) is ToolExecutionCompletedEvent -> ServerMessage.ToolCompleted( @@ -72,6 +116,8 @@ suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: Artifa toolName = p.toolName, outputSummary = p.receipt.outputSummary, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, ) is ToolExecutionFailedEvent -> ServerMessage.ToolFailed( @@ -79,25 +125,42 @@ suspend fun domainEventToServerMessage(event: StoredEvent, artifactStore: Artifa toolName = p.toolName, reason = p.reason, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = seq, + sessionSequence = sessionSequence, ) is ToolExecutionRejectedEvent -> ServerMessage.ToolRejected( - sessionId = p.sessionId, toolName = p.toolName, reason = p.reason, + sessionId = p.sessionId, + toolName = p.toolName, + reason = p.reason, + sequence = seq, + sessionSequence = sessionSequence, ) - is ApprovalRequestedEvent -> mapApprovalRequested(p) + is ApprovalRequestedEvent -> mapApprovalRequested(p, seq, sessionSequence) else -> null } +} -private fun mapOrchestrationPaused(p: OrchestrationPausedEvent): ServerMessage { +private fun mapOrchestrationPaused( + p: OrchestrationPausedEvent, + seq: Long, + sessionSequence: Long, +): ServerMessage { val reason = if (p.reason == "APPROVAL_PENDING") PauseReason.APPROVAL_PENDING else PauseReason.USER_REQUESTED - return ServerMessage.SessionPaused(sessionId = p.sessionId, reason = reason) + return ServerMessage.SessionPaused( + sessionId = p.sessionId, + reason = reason, + sequence = seq, + sessionSequence = sessionSequence, + ) } private suspend fun mapInferenceCompleted( event: StoredEvent, p: InferenceCompletedEvent, artifactStore: ArtifactStore, + sessionSequence: Long, ): ServerMessage = runCatching { artifactStore.get(p.responseArtifactId)?.toString(Charsets.UTF_8) ?: "" }.getOrElse { "" }.run { @@ -107,10 +170,16 @@ private suspend fun mapInferenceCompleted( outputSummary = this, responseText = this, occurredAt = event.metadata.timestamp.toEpochMilliseconds(), + sequence = event.sequence, + sessionSequence = sessionSequence, ) } -private fun mapApprovalRequested(p: ApprovalRequestedEvent): ServerMessage = +private fun mapApprovalRequested( + p: ApprovalRequestedEvent, + seq: Long, + sessionSequence: Long, +): ServerMessage = ServerMessage.ApprovalRequired( sessionId = p.sessionId, requestId = p.requestId, @@ -118,4 +187,6 @@ private fun mapApprovalRequested(p: ApprovalRequestedEvent): ServerMessage = riskSummary = RiskSummaryDto("unknown", emptyList(), ""), toolName = p.toolName, preview = p.preview, + sequence = seq, + sessionSequence = sessionSequence, ) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt index 69d374dc..b29ab297 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt @@ -1,6 +1,7 @@ package com.correx.apps.server.bridge import com.correx.apps.server.protocol.ServerMessage +import com.correx.apps.server.protocol.SessionStateDto import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.orchestration.OrchestrationStatus @@ -29,6 +30,7 @@ class SessionEventBridge( approvalState.decisions.values.none { it.requestId == req.id } } + val lastSeq = eventStore.lastSequence(sessionId) ?: 0L send(ServerMessage.SessionSnapshot( sessionId = sessionId, workflowId = orchState.workflowId, @@ -40,13 +42,25 @@ class SessionEventBridge( approvalTier = pendingRequest?.tier?.name, approvalToolName = pendingRequest?.toolName, approvalPreview = pendingRequest?.preview, + state = SessionStateDto( + status = orchState.status.name, + currentStageId = orchState.currentStageId?.value, + pauseReason = orchState.pauseReason, + ), + pendingApprovals = emptyList(), + lastSequence = lastSeq, + lastSessionSequence = 0L, )) } + send(ServerMessage.SnapshotComplete) } suspend fun streamLive(sessionId: SessionId) { + var sessionSequence = 0L eventStore.subscribe(sessionId).collect { event -> - domainEventToServerMessage(event, artifactStore)?.let { send(it) } + sessionSequence++ + domainEventToServerMessage(event, artifactStore, sessionSequence = sessionSequence) + ?.let { send(it) } } } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt index d53000d1..7ef407cc 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt @@ -22,6 +22,19 @@ data class SessionConfigDto( val retryPolicy: String?, ) +@Serializable +data class SessionStateDto( + val status: String, + val currentStageId: String?, + val pauseReason: String?, +) + +@Serializable +data class ApprovalDto( + val requestId: String, + val tier: String, +) + @Serializable enum class PauseReason { APPROVAL_PENDING, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 672d76be..80b06214 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -8,18 +8,69 @@ import kotlinx.serialization.SerialName import kotlinx.serialization.Serializable @Serializable -sealed class ServerMessage { +sealed interface ServerMessage { + /** Global monotonic sequence cursor. Non-null for event-derived messages. */ + val sequence: Long? + + /** Per-session monotonic sequence cursor. Non-null for event-derived messages. */ + val sessionSequence: Long? + + /** Marker for event-derived messages — cursors must be non-null. */ @Serializable - data class SessionStarted(val sessionId: SessionId, val workflowId: String) : ServerMessage() + sealed interface SessionMessage : ServerMessage { + override val sequence: Long + override val sessionSequence: Long + } + + /** Marker for control / infra messages — cursors are always null. */ + @Serializable + sealed interface NonEventMessage : ServerMessage { + override val sequence: Long? get() = null + override val sessionSequence: Long? get() = null + } + + // -- Session lifecycle -- + + /** + * @see SessionStartedEvent — emitted when a session begins. + * TODO(cleanup-NN): retire SessionStarted once TUI migrates to event-derived ordering. + */ + @Serializable + @SerialName("session.started") + data class SessionStarted( + val sessionId: SessionId, + val workflowId: String, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class SessionPaused(val sessionId: SessionId, val reason: PauseReason) : ServerMessage() + @SerialName("session.paused") + data class SessionPaused( + val sessionId: SessionId, + val reason: PauseReason, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class SessionCompleted(val sessionId: SessionId) : ServerMessage() + @SerialName("session.completed") + data class SessionCompleted( + val sessionId: SessionId, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class SessionFailed(val sessionId: SessionId, val reason: String) : ServerMessage() + @SerialName("session.failed") + data class SessionFailed( + val sessionId: SessionId, + val reason: String, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + // -- State snapshot -- @Serializable @SerialName("session_snapshot") @@ -34,55 +85,128 @@ sealed class ServerMessage { val approvalTier: String?, val approvalToolName: String?, val approvalPreview: String?, - ) : ServerMessage() + val state: SessionStateDto, + val pendingApprovals: List, + val lastSequence: Long, + val lastSessionSequence: Long, + override val sequence: Long? = null, + override val sessionSequence: Long? = null, + ) : ServerMessage, NonEventMessage + + // -- Stage lifecycle -- @Serializable - data class StageStarted(val sessionId: SessionId, val stageId: StageId, val occurredAt: Long) : ServerMessage() + @SerialName("stage.started") + data class StageStarted( + val sessionId: SessionId, + val stageId: StageId, + val occurredAt: Long, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class StageCompleted(val sessionId: SessionId, val stageId: StageId, val occurredAt: Long) : ServerMessage() + @SerialName("stage.completed") + data class StageCompleted( + val sessionId: SessionId, + val stageId: StageId, + val occurredAt: Long, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class StageFailed(val sessionId: SessionId, val stageId: StageId, val reason: String, val occurredAt: Long) : - ServerMessage() + @SerialName("stage.failed") + data class StageFailed( + val sessionId: SessionId, + val stageId: StageId, + val reason: String, + val occurredAt: Long, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + // -- Inference -- @Serializable - data class InferenceStarted(val sessionId: SessionId, val stageId: StageId) : ServerMessage() + @SerialName("inference.started") + data class InferenceStarted( + val sessionId: SessionId, + val stageId: StageId, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable + @SerialName("inference.completed") data class InferenceCompleted( val sessionId: SessionId, val stageId: StageId, val outputSummary: String, val responseText: String = "", val occurredAt: Long, - ) : ServerMessage() + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class InferenceTimedOut(val sessionId: SessionId, val stageId: StageId, val elapsedMs: Long) : - ServerMessage() + @SerialName("inference.timed_out") + data class InferenceTimedOut( + val sessionId: SessionId, + val stageId: StageId, + val elapsedMs: Long, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + // -- Tool execution -- @Serializable - data class ToolStarted(val sessionId: SessionId, val toolName: String, val tier: Tier) : - ServerMessage() + @SerialName("tool.started") + data class ToolStarted( + val sessionId: SessionId, + val toolName: String, + val tier: Tier, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable + @SerialName("tool.completed") data class ToolCompleted( val sessionId: SessionId, val toolName: String, val outputSummary: String, val occurredAt: Long, - ) : ServerMessage() + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class ToolFailed(val sessionId: SessionId, val toolName: String, val reason: String, val occurredAt: Long) : - ServerMessage() + @SerialName("tool.failed") + data class ToolFailed( + val sessionId: SessionId, + val toolName: String, + val reason: String, + val occurredAt: Long, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage @Serializable - data class ToolRejected(val sessionId: SessionId, val toolName: String, val reason: String) : - ServerMessage() + @SerialName("tool.rejected") + data class ToolRejected( + val sessionId: SessionId, + val toolName: String, + val reason: String, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + // -- Approval -- @Serializable + @SerialName("approval.required") data class ApprovalRequired( val sessionId: SessionId, val requestId: ApprovalRequestId, @@ -90,19 +214,48 @@ sealed class ServerMessage { val riskSummary: RiskSummaryDto, val toolName: String?, val preview: String?, - ) : ServerMessage() + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + + // -- System / infra -- @Serializable - data class ProviderStatusChanged(val providerId: String, val status: ProviderHealthDto) : - ServerMessage() + @SerialName("provider.status_changed") + data class ProviderStatusChanged( + val providerId: String, + val status: ProviderHealthDto, + override val sequence: Long? = null, + override val sessionSequence: Long? = null, + ) : ServerMessage, NonEventMessage @Serializable - data class ProtocolError(val message: String) : ServerMessage() + @SerialName("protocol_error") + data class ProtocolError( + val message: String, + override val sequence: Long? = null, + override val sessionSequence: Long? = null, + ) : ServerMessage, NonEventMessage + + // -- Router -- @Serializable + @SerialName("router.response") data class RouterResponseMessage( val sessionId: SessionId, val content: String, val steeringEmitted: Boolean, - ) : ServerMessage() + override val sequence: Long? = null, + override val sessionSequence: Long? = null, + ) : ServerMessage, NonEventMessage + + // -- Snapshot phase markers -- + + /** Marker indicating the snapshot phase has completed. No cursor fields. */ + @Serializable + @SerialName("snapshot_complete") + data object SnapshotComplete : ServerMessage, NonEventMessage { + override val sequence: Long? get() = null + override val sessionSequence: Long? get() = null + } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt index fa20c2b0..f81effc4 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt @@ -61,7 +61,11 @@ class GlobalStreamHandler(private val module: ServerModule) { } .onFailure { log.warn("decode error: {}", it.message) - val error = ServerMessage.ProtocolError("Unknown message: ${it.message}") + val error = ServerMessage.ProtocolError( + message = "Unknown message: ${it.message}", + sequence = null, + sessionSequence = null, + ) session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) } } @@ -88,6 +92,8 @@ class GlobalStreamHandler(private val module: ServerModule) { val msg = ServerMessage.ProviderStatusChanged( providerId = providerId.value, status = ProviderHealthDto(providerId.value, status, null), + sequence = null, + sessionSequence = null, ) session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(msg))) } @@ -112,7 +118,15 @@ class GlobalStreamHandler(private val module: ServerModule) { } private fun encodeError(message: String): Frame.Text = - Frame.Text(ProtocolSerializer.encodeServerMessage(ServerMessage.ProtocolError(message))) + Frame.Text( + ProtocolSerializer.encodeServerMessage( + ServerMessage.ProtocolError( + message = message, + sequence = null, + sessionSequence = null, + ), + ), + ) private suspend fun handleStartSession( session: DefaultWebSocketServerSession, @@ -123,7 +137,11 @@ class GlobalStreamHandler(private val module: ServerModule) { val graph = module.workflowRegistry.find(msg.workflowId) log.info("find returned: {}", graph) if (graph == null) { - val error = ServerMessage.ProtocolError("Unknown workflowId: ${msg.workflowId}") + val error = ServerMessage.ProtocolError( + message = "Unknown workflowId: ${msg.workflowId}", + sequence = null, + sessionSequence = null, + ) session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(error))) return } @@ -166,7 +184,12 @@ class GlobalStreamHandler(private val module: ServerModule) { bridge.streamLive(sessionId) }, )?.cancel() - val started = ServerMessage.SessionStarted(sessionId = sessionId, workflowId = msg.workflowId) + val started = ServerMessage.SessionStarted( + sessionId = sessionId, + workflowId = msg.workflowId, + sequence = 0L, + sessionSequence = 0L, + ) session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt index c9571520..3a48976d 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt @@ -65,6 +65,7 @@ class DomainEventMapperTest { correlationId = null, ), sequence = 1L, + sessionSequence = 1L, payload = payload, ) @@ -72,8 +73,11 @@ class DomainEventMapperTest { fun `WorkflowStartedEvent maps to SessionStarted`(): Unit = runTest { val event = storedEvent(WorkflowStartedEvent(sessionId = sessionId, startStageId = stageId, workflowId = workflowId)) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.SessionStarted(sessionId, workflowId, event.sequence, 0L), + result, + ) } @Test @@ -85,8 +89,11 @@ class DomainEventMapperTest { totalStages = 1, ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.SessionCompleted(sessionId), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.SessionCompleted(sessionId, event.sequence, 0L), + result, + ) } @Test @@ -94,8 +101,11 @@ class DomainEventMapperTest { val event = storedEvent( WorkflowFailedEvent(sessionId = sessionId, stageId = stageId, reason = "boom", retryExhausted = false), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.SessionFailed(sessionId, "boom"), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.SessionFailed(sessionId, "boom", event.sequence, 0L), + result, + ) } @Test @@ -105,8 +115,11 @@ class DomainEventMapperTest { val event = storedEvent( TransitionExecutedEvent(sessionId = sessionId, from = from, to = to, transitionId = TransitionId("t1")), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.StageStarted(sessionId, to, occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.StageStarted(sessionId, to, occurredAt, event.sequence, 0L), + result, + ) } @Test @@ -114,19 +127,25 @@ class DomainEventMapperTest { val event = storedEvent( StageCompletedEvent(sessionId = sessionId, stageId = stageId, transitionId = TransitionId("t1")), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.StageCompleted(sessionId, stageId, occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.StageCompleted(sessionId, stageId, occurredAt, event.sequence, 0L), + result, + ) } - @Test + @Test fun `StageFailedEvent maps to StageFailed`(): Unit = runTest { val event = storedEvent( StageFailedEvent( sessionId = sessionId, stageId = stageId, transitionId = TransitionId("t1"), reason = "err", ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.StageFailed(sessionId, stageId, "err", occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.StageFailed(sessionId, stageId, "err", occurredAt, event.sequence, 0L), + result, + ) } @Test @@ -134,8 +153,11 @@ class DomainEventMapperTest { val event = storedEvent( OrchestrationPausedEvent(sessionId = sessionId, stageId = stageId, reason = "APPROVAL_PENDING"), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.SessionPaused(sessionId, PauseReason.APPROVAL_PENDING), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.SessionPaused(sessionId, PauseReason.APPROVAL_PENDING, event.sequence, 0L), + result, + ) } @Test @@ -143,8 +165,11 @@ class DomainEventMapperTest { val event = storedEvent( OrchestrationPausedEvent(sessionId = sessionId, stageId = stageId, reason = "USER_REQUESTED"), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.SessionPaused(sessionId, PauseReason.USER_REQUESTED), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.SessionPaused(sessionId, PauseReason.USER_REQUESTED, event.sequence, 0L), + result, + ) } @Test @@ -158,8 +183,11 @@ class DomainEventMapperTest { promptArtifactId = ArtifactId("art-prompt"), ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.InferenceStarted(sessionId, stageId), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.InferenceStarted(sessionId, stageId, event.sequence, 0L), + result, + ) } @Test @@ -184,9 +212,11 @@ class DomainEventMapperTest { responseArtifactId = artifactId, ), ) - val result = domainEventToServerMessage(event, store) + val result = domainEventToServerMessage(event, store, sessionSequence = 0L) assertEquals( - ServerMessage.InferenceCompleted(sessionId, stageId, responseText, responseText, occurredAt), + ServerMessage.InferenceCompleted( + sessionId, stageId, responseText, responseText, occurredAt, event.sequence, 0L, + ), result, ) } @@ -204,8 +234,11 @@ class DomainEventMapperTest { responseArtifactId = ArtifactId("missing"), ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.InferenceCompleted(sessionId, stageId, "", "", occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.InferenceCompleted(sessionId, stageId, "", "", occurredAt, event.sequence, 0L), + result, + ) } @Test @@ -219,8 +252,11 @@ class DomainEventMapperTest { timeoutMs = 5000L, ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.InferenceTimedOut(sessionId, stageId, 5000L), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.InferenceTimedOut(sessionId, stageId, 5000L, event.sequence, 0L), + result, + ) } @Test @@ -238,8 +274,11 @@ class DomainEventMapperTest { ), ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.ToolStarted(sessionId, "file_write", Tier.T2), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.ToolStarted(sessionId, "file_write", Tier.T2, event.sequence, 0L), + result, + ) } @Test @@ -259,8 +298,11 @@ class DomainEventMapperTest { invocationId = invId, sessionId = sessionId, toolName = "file_write", receipt = receipt, ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.ToolCompleted(sessionId, "file_write", "wrote 3 lines", occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.ToolCompleted(sessionId, "file_write", "wrote 3 lines", occurredAt, event.sequence, 0L), + result, + ) } @Test @@ -273,8 +315,11 @@ class DomainEventMapperTest { reason = "disk full", ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.ToolFailed(sessionId, "file_write", "disk full", occurredAt), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.ToolFailed(sessionId, "file_write", "disk full", occurredAt, event.sequence, 0L), + result, + ) } @Test @@ -288,8 +333,11 @@ class DomainEventMapperTest { reason = "policy denied", ), ) - val result = domainEventToServerMessage(event, noopStore) - assertEquals(ServerMessage.ToolRejected(sessionId, "file_write", "policy denied"), result) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) + assertEquals( + ServerMessage.ToolRejected(sessionId, "file_write", "policy denied", event.sequence, 0L), + result, + ) } @Test @@ -306,17 +354,18 @@ class DomainEventMapperTest { projectId = null, ), ) - val result = domainEventToServerMessage(event, noopStore) as ServerMessage.ApprovalRequired + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) as ServerMessage.ApprovalRequired assertEquals(requestId, result.requestId) assertEquals(Tier.T3, result.tier) assertNull(result.toolName) assertNull(result.preview) + assertEquals(event.sequence, result.sequence) } @Test fun `unmapped event returns null`(): Unit = runTest { val event = storedEvent(SessionStartedEvent(sessionId = sessionId)) - val result = domainEventToServerMessage(event, noopStore) + val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L) assertNull(result) } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt index c3f6b5c9..20827a45 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt @@ -12,15 +12,16 @@ import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.orchestration.OrchestrationState +import com.correx.core.events.orchestration.OrchestrationStatus import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId -import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer -import com.correx.core.kernel.orchestration.OrchestrationProjector import com.correx.core.kernel.orchestration.OrchestrationRepository import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import com.correx.core.sessions.projections.replay.EventReplayer import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.flow @@ -54,6 +55,7 @@ class SessionEventBridgeTest { correlationId = null, ), sequence = seq, + sessionSequence = seq, payload = payload, ) @@ -65,17 +67,26 @@ class SessionEventBridgeTest { override suspend fun appendAll(events: List): List = error("unused") override fun read(sessionId: SessionId): List = allEventsList override fun readFrom(sessionId: SessionId, fromSequence: Long): List = allEventsList - override fun lastSequence(sessionId: SessionId): Long? = null + override fun lastSequence(sessionId: SessionId): Long? = + allEventsList.maxOfOrNull { it.sequence } ?: 0L override fun subscribe(sessionId: SessionId): Flow = liveFlow override fun allEvents(): Sequence = allEventsList.asSequence() override fun allSessionIds(): Set = allEventsList.map { it.metadata.sessionId }.toSet() + override fun subscribeAll(): Flow = TODO("Not needed in this test context") + override suspend fun lastGlobalSequence(): Long = TODO("Not needed in this test context") } - private val orchestrationRepository = OrchestrationRepository( - DefaultEventReplayer( - fakeEventStore(), - OrchestrationProjector(DefaultOrchestrationReducer()), - ), + private fun activeOrchestrationRepository( + status: OrchestrationStatus = OrchestrationStatus.RUNNING, + ): OrchestrationRepository = OrchestrationRepository( + object : EventReplayer { + override fun rebuild(sessionId: SessionId): OrchestrationState = OrchestrationState( + workflowId = workflowId, + status = status, + currentStageId = stageId, + pendingApproval = false, + ) + }, ) private val approvalRepository = DefaultApprovalRepository( @@ -92,19 +103,23 @@ class SessionEventBridgeTest { storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L), ) val store = fakeEventStore(allEventsList = events) + val orchRepo = activeOrchestrationRepository() val sent = mutableListOf() val bridge = SessionEventBridge( store, noopArtifactStore, - orchestrationRepository, + orchRepo, approvalRepository, ) { sent.add(it) } bridge.replaySnapshot() assertEquals(2, sent.size) - assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), sent[0]) - assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1]) + val snapshot = sent[0] as ServerMessage.SessionSnapshot + assertEquals(sessionId, snapshot.sessionId) + assertEquals(2L, snapshot.lastSequence) + assertEquals(0L, snapshot.lastSessionSequence) + assertEquals(ServerMessage.SnapshotComplete, sent[1]) } @Test @@ -114,18 +129,21 @@ class SessionEventBridgeTest { storedEvent(WorkflowCompletedEvent(sessionId, stageId, 1), seq = 2L), ) val store = fakeEventStore(allEventsList = events) + val orchRepo = activeOrchestrationRepository() val sent = mutableListOf() val bridge = SessionEventBridge( store, noopArtifactStore, - orchestrationRepository, + orchRepo, approvalRepository, ) { sent.add(it) } bridge.replaySnapshot() - assertEquals(1, sent.size) - assertEquals(ServerMessage.SessionCompleted(sessionId), sent[0]) + assertEquals(2, sent.size) + val snapshot = sent[0] as ServerMessage.SessionSnapshot + assertEquals(sessionId, snapshot.sessionId) + assertEquals(ServerMessage.SnapshotComplete, sent[1]) } @Test @@ -140,15 +158,21 @@ class SessionEventBridgeTest { val bridge = SessionEventBridge( store, noopArtifactStore, - orchestrationRepository, + activeOrchestrationRepository(), approvalRepository, ) { sent.add(it) } bridge.streamLive(sessionId) assertEquals(2, sent.size) - assertEquals(ServerMessage.SessionStarted(sessionId, workflowId), sent[0]) - assertEquals(ServerMessage.SessionCompleted(sessionId), sent[1]) + assertEquals( + ServerMessage.SessionStarted(sessionId, workflowId, 1L, 1L), + sent[0], + ) + assertEquals( + ServerMessage.SessionCompleted(sessionId, 2L, 2L), + sent[1], + ) } @Test @@ -162,7 +186,7 @@ class SessionEventBridgeTest { val bridge = SessionEventBridge( store, noopArtifactStore, - orchestrationRepository, + activeOrchestrationRepository(), approvalRepository, ) { sent.add(it) } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt new file mode 100644 index 00000000..cc3ca40c --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt @@ -0,0 +1,82 @@ +package com.correx.apps.server.protocol + +import com.correx.core.events.types.SessionId +import kotlinx.serialization.json.Json +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class ServerMessageSerializationTest { + + private val json = Json { + classDiscriminator = "type" + ignoreUnknownKeys = true + } + + @Test + fun `SessionStarted encodes sequence and sessionSequence`() { + val msg = ServerMessage.SessionStarted( + sessionId = SessionId("sess-1"), + workflowId = "wf-1", + sequence = 42, + sessionSequence = 7, + ) + val jsonStr = ProtocolSerializer.encodeServerMessage(msg) + assert(jsonStr.contains("\"type\":\"session.started\"")) { "expected type=session.started" } + assert(jsonStr.contains("\"sessionId\":\"sess-1\"")) { "expected sessionId" } + assert(jsonStr.contains("\"workflowId\":\"wf-1\"")) { "expected workflowId" } + assert(jsonStr.contains("\"sequence\":42")) { "expected sequence=42" } + assert(jsonStr.contains("\"sessionSequence\":7")) { "expected sessionSequence=7" } + } + + @Test + fun `SnapshotComplete encodes with only type field`() { + val msg = ServerMessage.SnapshotComplete + val jsonStr = ProtocolSerializer.encodeServerMessage(msg) + assert(jsonStr.contains("\"type\":\"snapshot_complete\"")) { "expected type=snapshot_complete" } + } + + @Test + fun `SessionSnapshot encodes lastSequence and lastSessionSequence`() { + val msg = ServerMessage.SessionSnapshot( + sessionId = SessionId("sess-1"), + workflowId = "wf-1", + status = "running", + currentStageId = "stage-1", + pauseReason = null, + pendingApproval = false, + approvalRequestId = null, + approvalTier = null, + approvalToolName = null, + approvalPreview = null, + state = SessionStateDto( + status = "running", + currentStageId = "stage-1", + pauseReason = null, + ), + pendingApprovals = emptyList(), + lastSequence = 100, + lastSessionSequence = 15, + ) + val jsonStr = ProtocolSerializer.encodeServerMessage(msg) + assert(jsonStr.contains("\"type\":\"session_snapshot\"")) { "expected type=session_snapshot" } + assert(jsonStr.contains("\"sessionId\":\"sess-1\"")) { "expected sessionId" } + assert(jsonStr.contains("\"lastSequence\":100")) { "expected lastSequence=100" } + assert(jsonStr.contains("\"lastSessionSequence\":15")) { "expected lastSessionSequence=15" } + } + + @Test + fun `SessionStarted round-trips through JSON`() { + val original = ServerMessage.SessionStarted( + sessionId = SessionId("sess-42"), + workflowId = "wf-abc", + sequence = 200, + sessionSequence = 33, + ) + val jsonStr = ProtocolSerializer.encodeServerMessage(original) + val decoded = json.decodeFromString(jsonStr) + assertEquals(original.sessionId, decoded.sessionId) + assertEquals(original.workflowId, decoded.workflowId) + assertEquals(200L, decoded.sequence) + assertEquals(33L, decoded.sessionSequence) + } +} diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/ApprovalReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/ApprovalReducerTest.kt index c4d2fe6d..c2faada2 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/ApprovalReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/ApprovalReducerTest.kt @@ -98,6 +98,8 @@ class ApprovalReducerTest { riskSummary = RiskSummaryDto(level = "HIGH", factors = emptyList(), recommendedAction = "review"), toolName = "bash", preview = "rm -rf /", + sequence = 1L, + sessionSequence = 1L, ) val (state, effects) = reduce(action = Action.ServerEventReceived(msg)) assertEquals("req-2", state.active?.requestId) diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt index 33d12119..8a35ef34 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt @@ -43,11 +43,15 @@ class RootReducerTest { val startMsg = com.correx.apps.server.protocol.ServerMessage.SessionStarted( sessionId = com.correx.core.events.types.SessionId("s1"), workflowId = "wf", + sequence = 1L, + sessionSequence = 1L, ) val (state1, _) = RootReducer.reduce(state0, Action.ServerEventReceived(startMsg), fixedClock) val startMsg2 = com.correx.apps.server.protocol.ServerMessage.SessionStarted( sessionId = com.correx.core.events.types.SessionId("s2"), workflowId = "wf", + sequence = 2L, + sessionSequence = 1L, ) val (state2, _) = RootReducer.reduce(state1, Action.ServerEventReceived(startMsg2), fixedClock) // selected is s1 (first); navigating down should move to s2 diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt index 0d6af391..6a2420e8 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/SessionsReducerTest.kt @@ -91,7 +91,7 @@ class SessionsReducerTest { @Test fun `ServerEventReceived SessionStarted appends session and sets selection`() { - val msg = ServerMessage.SessionStarted(sessionId = SessionId("s1"), workflowId = "wf1") + val msg = ServerMessage.SessionStarted(sessionId = SessionId("s1"), workflowId = "wf1", sequence = 1L, sessionSequence = 1L) val (state, effects) = reduce(action = Action.ServerEventReceived(msg)) assertEquals(1, state.sessions.size) assertEquals("s1", state.sessions[0].id) @@ -103,7 +103,7 @@ class SessionsReducerTest { @Test fun `ServerEventReceived SessionStarted preserves existing selection`() { val existing = SessionsState(sessions = listOf(session("existing")), selectedId = "existing") - val msg = ServerMessage.SessionStarted(sessionId = SessionId("new"), workflowId = "wf") + val msg = ServerMessage.SessionStarted(sessionId = SessionId("new"), workflowId = "wf", sequence = 1L, sessionSequence = 1L) val (state, _) = reduce(sessions = existing, action = Action.ServerEventReceived(msg)) assertEquals("existing", state.selectedId) assertEquals(2, state.sessions.size) @@ -112,7 +112,7 @@ class SessionsReducerTest { @Test fun `ServerEventReceived SessionCompleted updates status to COMPLETED`() { val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") - val msg = ServerMessage.SessionCompleted(sessionId = SessionId("s1")) + val msg = ServerMessage.SessionCompleted(sessionId = SessionId("s1"), sequence = 1L, sessionSequence = 1L) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("COMPLETED", state.sessions[0].status) assertEquals(1000L, state.sessions[0].lastEventAt) @@ -121,7 +121,7 @@ class SessionsReducerTest { @Test fun `ServerEventReceived SessionFailed updates status to FAILED`() { val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") - val msg = ServerMessage.SessionFailed(sessionId = SessionId("s1"), reason = "oops") + val msg = ServerMessage.SessionFailed(sessionId = SessionId("s1"), reason = "oops", sequence = 1L, sessionSequence = 1L) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("FAILED", state.sessions[0].status) } @@ -136,6 +136,8 @@ class SessionsReducerTest { sessionId = SessionId("s1"), stageId = StageId("stage-1"), occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals(null, state.sessions[0].currentStage) @@ -153,6 +155,8 @@ class SessionsReducerTest { stageId = StageId("stage-1"), reason = "error", occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals(null, state.sessions[0].currentStage) @@ -183,6 +187,8 @@ class SessionsReducerTest { sessionId = SessionId("s1"), stageId = StageId("stage-1"), outputSummary = "summary", responseText = "full response", occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("summary", state.sessions[0].lastOutput) @@ -197,6 +203,8 @@ class SessionsReducerTest { sessionId = SessionId("s1"), stageId = StageId("stage-1"), outputSummary = "", responseText = "", occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals(null, state.sessions[0].lastResponseText) @@ -209,6 +217,8 @@ class SessionsReducerTest { sessionId = SessionId("s1"), stageId = StageId("stage-1"), occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("stage-1", state.sessions[0].currentStage) @@ -223,6 +233,8 @@ class SessionsReducerTest { toolName = "file_write", outputSummary = "wrote 3 lines", occurredAt = fixedClock(), + sequence = 1L, + sessionSequence = 1L, ) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("file_write: wrote 3 lines", state.sessions[0].lastOutput) @@ -232,7 +244,7 @@ class SessionsReducerTest { @Test fun `ServerEventReceived SessionPaused with APPROVAL_PENDING sets status label`() { val s = SessionsState(sessions = listOf(session("s1")), selectedId = "s1") - val msg = ServerMessage.SessionPaused(sessionId = SessionId("s1"), reason = PauseReason.APPROVAL_PENDING) + val msg = ServerMessage.SessionPaused(sessionId = SessionId("s1"), reason = PauseReason.APPROVAL_PENDING, sequence = 1L, sessionSequence = 1L) val (state, _) = reduce(sessions = s, action = Action.ServerEventReceived(msg)) assertEquals("PAUSED awaiting approval", state.sessions[0].status) } diff --git a/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalReducerTest.kt b/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalReducerTest.kt index 5d65bc49..8a8ac110 100644 --- a/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalReducerTest.kt +++ b/core/approvals/src/test/kotlin/com/correx/core/approvals/DefaultApprovalReducerTest.kt @@ -38,6 +38,7 @@ class DefaultApprovalReducerTest { correlationId = null ), sequence = sequence, + sessionSequence = sequence, payload = payload ) } diff --git a/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt b/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt index 027ead40..fc381828 100644 --- a/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt +++ b/core/tools/src/test/kotlin/com/correx/core/tools/DefaultToolReducerTest.kt @@ -59,6 +59,7 @@ class DefaultToolReducerTest { correlationId = null ), sequence = 1L, + sessionSequence = 1L, payload = payload )