diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/RouterEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/RouterEvents.kt new file mode 100644 index 00000000..b085b737 --- /dev/null +++ b/core/events/src/main/kotlin/com/correx/core/events/events/RouterEvents.kt @@ -0,0 +1,21 @@ +package com.correx.core.events.events + +import com.correx.core.events.types.SessionId +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +@Serializable +@SerialName("ChatTurn") +data class ChatTurnEvent( + val sessionId: SessionId, + val turnId: String, + val role: ChatTurnRole, + val content: String, + val timestampMs: Long, +) : EventPayload + +@Serializable +enum class ChatTurnRole { + USER, + ROUTER, +} diff --git a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt index bbe1bb4e..ffdf4997 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt @@ -8,6 +8,7 @@ import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.ArtifactValidatedEvent import com.correx.core.events.events.ArtifactValidatingEvent import com.correx.core.events.events.ChatSessionStartedEvent +import com.correx.core.events.events.ChatTurnEvent import com.correx.core.events.events.EventPayload import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.InferenceFailedEvent @@ -71,6 +72,7 @@ val eventModule = SerializersModule { subclass(RetryAttemptedEvent::class) subclass(RiskAssessedEvent::class) subclass(ChatSessionStartedEvent::class) + subclass(ChatTurnEvent::class) } } diff --git a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt index 59fea505..43865a0a 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt @@ -1,11 +1,11 @@ package com.correx.core.router +import com.correx.core.events.events.ChatTurnEvent +import com.correx.core.events.events.ChatTurnRole import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.stores.EventStore -import com.correx.core.events.types.CausationId -import com.correx.core.events.types.CorrelationId import com.correx.core.events.types.EventId import com.correx.core.events.types.InferenceRequestId import com.correx.core.events.types.SessionId @@ -16,11 +16,8 @@ import com.correx.core.inference.ModelCapability import com.correx.core.inference.ResponseFormat import com.correx.core.router.model.RouterConfig import com.correx.core.router.model.RouterResponse -import com.correx.core.router.model.RouterTurn -import com.correx.core.router.model.TurnRole import kotlinx.datetime.Clock import java.util.* -import java.util.concurrent.* interface RouterFacade { suspend fun onUserInput( @@ -39,22 +36,19 @@ class DefaultRouterFacade( private val validateSteering: (suspend (String) -> String?)? = null, ) : RouterFacade { - private val histories = ConcurrentHashMap>() - override suspend fun onUserInput( sessionId: SessionId, input: String, mode: ChatMode, ): RouterResponse { - val state = routerRepository.getRouterState(sessionId) + // Emit USER turn event + emitChatTurn(sessionId, input, ChatTurnRole.USER) - val history = histories.getOrPut(sessionId) { mutableListOf() } - history.add(RouterTurn(role = TurnRole.USER, content = input, timestamp = Clock.System.now())) + // Rebuild state with user turn appended + val stateWithUserTurn = routerRepository.getRouterState(sessionId) + val effectiveStageId = stateWithUserTurn.currentStageId ?: StageId.NONE - val stateWithHistory = state.copy(conversationHistory = history.toList()) - val effectiveStageId = state.currentStageId ?: StageId.NONE - - val contextPack = routerContextBuilder.build(stateWithHistory, config.tokenBudget) + val contextPack = routerContextBuilder.build(stateWithUserTurn, config.tokenBudget) val provider = inferenceRouter.route(effectiveStageId, setOf(ModelCapability.General)) val inferenceRequest = InferenceRequest( requestId = InferenceRequestId(UUID.randomUUID().toString()), @@ -67,36 +61,65 @@ class DefaultRouterFacade( val inferenceResponse = provider.infer(inferenceRequest) val content = inferenceResponse.text - history.add(RouterTurn(role = TurnRole.ROUTER, content = content, timestamp = Clock.System.now())) + // Emit ROUTER turn event + emitChatTurn(sessionId, content, ChatTurnRole.ROUTER) if (mode == ChatMode.STEERING) { val validationError = validateSteering?.invoke(content) if (validationError == null) { - eventStore.append( - NewEvent( - metadata = EventMetadata( - eventId = EventId(UUID.randomUUID().toString()), - sessionId = sessionId, - timestamp = Clock.System.now(), - schemaVersion = 1, - causationId = null, - correlationId = null, - ), - payload = SteeringNoteAddedEvent( - sessionId = sessionId, - content = content, - stageId = effectiveStageId.takeIf { it != StageId.NONE }, - ), - ), - ) + emitSteeringNote(sessionId, content, effectiveStageId) } } return RouterResponse(content = content, steeringEmitted = (mode == ChatMode.STEERING)) } + + private suspend fun emitChatTurn(sessionId: SessionId, content: String, role: ChatTurnRole) { + val turnId = UUID.randomUUID().toString() + val nowMs = Clock.System.now().toEpochMilliseconds() + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = ChatTurnEvent( + sessionId = sessionId, + turnId = turnId, + role = role, + content = content, + timestampMs = nowMs, + ), + ), + ) + } + + private suspend fun emitSteeringNote(sessionId: SessionId, content: String, effectiveStageId: StageId) { + eventStore.append( + NewEvent( + metadata = EventMetadata( + eventId = EventId(UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = SteeringNoteAddedEvent( + sessionId = sessionId, + content = content, + stageId = effectiveStageId.takeIf { it != StageId.NONE }, + ), + ), + ) + } } enum class ChatMode { CHAT, STEERING, -} \ No newline at end of file +} diff --git a/core/router/src/main/kotlin/com/correx/core/router/RouterReducer.kt b/core/router/src/main/kotlin/com/correx/core/router/RouterReducer.kt index 751860d9..b146ecb1 100644 --- a/core/router/src/main/kotlin/com/correx/core/router/RouterReducer.kt +++ b/core/router/src/main/kotlin/com/correx/core/router/RouterReducer.kt @@ -1,5 +1,6 @@ package com.correx.core.router +import com.correx.core.events.events.ChatTurnEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.StageCompletedEvent @@ -10,8 +11,11 @@ import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.events.WorkflowStartedEvent import com.correx.core.router.model.RouterL2Entry import com.correx.core.router.model.RouterState +import com.correx.core.router.model.RouterTurn import com.correx.core.router.model.StageOutcomeKind +import com.correx.core.router.model.TurnRole import com.correx.core.router.model.WorkflowStatus +import kotlinx.datetime.Instant interface RouterReducer { val initial: RouterState @@ -37,6 +41,7 @@ class DefaultRouterReducer : RouterReducer { is OrchestrationResumedEvent -> handleOrchestrationResumed(state) is StageCompletedEvent -> handleStageCompleted(state, event) is StageFailedEvent -> handleStageFailed(state, event) + is ChatTurnEvent -> handleChatTurn(state, event) else -> state } } @@ -102,4 +107,20 @@ class DefaultRouterReducer : RouterReducer { currentStageId = null ) } + + private fun handleChatTurn(state: RouterState, event: StoredEvent): RouterState { + val payload = event.payload as ChatTurnEvent + val turnRole = when (payload.role) { + com.correx.core.events.events.ChatTurnRole.USER -> TurnRole.USER + com.correx.core.events.events.ChatTurnRole.ROUTER -> TurnRole.ROUTER + } + val turn = RouterTurn( + role = turnRole, + content = payload.content, + timestamp = Instant.fromEpochMilliseconds(payload.timestampMs), + ) + return state.copy( + conversationHistory = state.conversationHistory + turn + ) + } } diff --git a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt index 3ee4ab93..1760d252 100644 --- a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt @@ -21,14 +21,17 @@ import com.correx.core.inference.ResponseFormat import com.correx.core.inference.TokenUsage import com.correx.core.router.ChatMode import com.correx.core.router.DefaultRouterFacade +import com.correx.core.router.DefaultRouterReducer import com.correx.core.router.RouterContextBuilder import com.correx.core.router.RouterFacade +import com.correx.core.router.RouterProjector import com.correx.core.router.RouterRepository import com.correx.core.router.model.RouterConfig import com.correx.core.router.model.RouterResponse import com.correx.core.router.model.RouterState import com.correx.core.router.model.TurnRole import com.correx.core.router.model.WorkflowStatus +import com.correx.core.sessions.projections.replay.DefaultEventReplayer import com.correx.testing.fixtures.inference.MockTokenizer import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.runBlocking @@ -61,11 +64,19 @@ class RouterFacadeTest { } @Test - fun `CHAT mode does not append to EventStore`(): Unit = runBlocking { + fun `CHAT mode appends USER and ROUTER ChatTurnEvent to EventStore`(): Unit = runBlocking { val mockStore = mockEventStore() val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.CHAT) facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!") - assertTrue(mockStore.appendedEvents.isEmpty()) + assertEquals(2, mockStore.appendedEvents.size) + val userEvent = mockStore.appendedEvents[0].payload + val routerEvent = mockStore.appendedEvents[1].payload + assertTrue(userEvent is com.correx.core.events.events.ChatTurnEvent) + assertTrue(routerEvent is com.correx.core.events.events.ChatTurnEvent) + assertEquals("Hello!", (userEvent as com.correx.core.events.events.ChatTurnEvent).content) + assertEquals(com.correx.core.events.events.ChatTurnRole.USER, userEvent.role) + assertEquals("inference response", (routerEvent as com.correx.core.events.events.ChatTurnEvent).content) + assertEquals(com.correx.core.events.events.ChatTurnRole.ROUTER, routerEvent.role) } // -------------------------------------------------------------------------- @@ -81,14 +92,20 @@ class RouterFacadeTest { } @Test - fun `STEERING mode appends SteeringNoteAddedEvent to store`(): Unit = runBlocking { + fun `STEERING mode appends USER ChatTurnEvent, ROUTER ChatTurnEvent, and SteeringNoteAddedEvent to store`(): Unit = runBlocking { val mockStore = mockEventStore() val facade = facadeWithMocks(eventStore = mockStore, chatMode = ChatMode.STEERING) facade.onUserInput(sessionId = SessionId("session-xyz"), input = "steer this way") - assertEquals(1, mockStore.appendedEvents.size) - val payload = mockStore.appendedEvents.single().payload - assertTrue(payload is com.correx.core.events.events.SteeringNoteAddedEvent) - assertEquals("inference response", (payload as com.correx.core.events.events.SteeringNoteAddedEvent).content) + assertEquals(3, mockStore.appendedEvents.size) + val userEvent = mockStore.appendedEvents[0].payload + val routerEvent = mockStore.appendedEvents[1].payload + val steeringEvent = mockStore.appendedEvents[2].payload + assertTrue(userEvent is com.correx.core.events.events.ChatTurnEvent) + assertTrue(routerEvent is com.correx.core.events.events.ChatTurnEvent) + assertTrue(steeringEvent is com.correx.core.events.events.SteeringNoteAddedEvent) + assertEquals("steer this way", (userEvent as com.correx.core.events.events.ChatTurnEvent).content) + assertEquals("inference response", (routerEvent as com.correx.core.events.events.ChatTurnEvent).content) + assertEquals("inference response", (steeringEvent as com.correx.core.events.events.SteeringNoteAddedEvent).content) } @Test @@ -112,10 +129,10 @@ class RouterFacadeTest { config = RouterConfig(tokenBudget = TokenBudget(limit = 5000)), ) facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!", mode = ChatMode.STEERING) - assertEquals(1, mockStore.appendedEvents.size) - val payload = mockStore.appendedEvents.single().payload - assertTrue(payload is com.correx.core.events.events.SteeringNoteAddedEvent) - assertEquals("steering response", (payload as com.correx.core.events.events.SteeringNoteAddedEvent).content) + assertEquals(3, mockStore.appendedEvents.size) + val steeringEvent = mockStore.appendedEvents[2].payload + assertTrue(steeringEvent is com.correx.core.events.events.SteeringNoteAddedEvent) + assertEquals("steering response", (steeringEvent as com.correx.core.events.events.SteeringNoteAddedEvent).content) } // -------------------------------------------------------------------------- @@ -125,14 +142,15 @@ class RouterFacadeTest { @Test fun `conversation history grows per call - user and router turns appended`(): Unit = runBlocking { val capturedStates = mutableListOf() + val eventStore = mockEventStore() + val replayer = DefaultEventReplayer( + store = eventStore, + projection = RouterProjector(DefaultRouterReducer()), + ) val facade = DefaultRouterFacade( routerRepository = object : RouterRepository { override suspend fun getRouterState(sessionId: SessionId): RouterState = - RouterState( - sessionId = sessionId, - workflowStatus = WorkflowStatus.RUNNING, - currentStageId = StageId("s1"), - ) + replayer.rebuild(sessionId) }, routerContextBuilder = object : RouterContextBuilder { override suspend fun build(state: RouterState, budget: TokenBudget): ContextPack { @@ -141,7 +159,7 @@ class RouterFacadeTest { } }, inferenceRouter = mockInferenceRouter("router reply"), - eventStore = mockEventStore(), + eventStore = eventStore, config = RouterConfig(tokenBudget = TokenBudget(limit = 5000)), ) val sessionId = SessionId("history-session") @@ -162,10 +180,15 @@ class RouterFacadeTest { @Test fun `conversation history is session-scoped - different sessions do not share history`(): Unit = runBlocking { val capturedStates = mutableListOf() + val eventStore = mockEventStore() + val replayer = DefaultEventReplayer( + store = eventStore, + projection = RouterProjector(DefaultRouterReducer()), + ) val facade = DefaultRouterFacade( routerRepository = object : RouterRepository { override suspend fun getRouterState(sessionId: SessionId): RouterState = - RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.RUNNING) + replayer.rebuild(sessionId) }, routerContextBuilder = object : RouterContextBuilder { override suspend fun build(state: RouterState, budget: TokenBudget): ContextPack { @@ -174,7 +197,7 @@ class RouterFacadeTest { } }, inferenceRouter = mockInferenceRouter("response"), - eventStore = mockEventStore(), + eventStore = eventStore, config = RouterConfig(tokenBudget = TokenBudget(limit = 5000)), ) facade.onUserInput(sessionId = SessionId("session-A"), input = "message A") @@ -453,7 +476,7 @@ class RouterFacadeTest { assertNotNull(response) assertEquals("inference response", response.content) assertTrue(response.steeringEmitted) - assertEquals(1, mockStore.appendedEvents.size) + assertEquals(3, mockStore.appendedEvents.size) } // --------------------------------------------------------------------------