kami/correx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: event-source router CHAT and STEERING conversation turns

Browse Source 
Adds ChatTurnEvent (role USER|ROUTER) so router conversation history
survives restart and is rebuilt from the event log via projection,
restoring Hard Invariant #1 for the CHAT path. Both CHAT and STEERING
now emit user + router turn events; STEERING additionally keeps its
existing SteeringNoteAddedEvent so the structured directive record is
preserved alongside the conversation turns.

RouterFacade drops the in-memory ConcurrentHashMap of histories and
rebuilds state through the repository after each emission. Reducer
appends turns to conversationHistory. ChatTurnEvent is registered in
eventModule.
This commit is contained in:
kami
2026-05-30 00:53:26 +04:00
parent 6feef150c9
commit a3f29e6eb9
5 changed files with 143 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/events/src/main/kotlin/com/correx/core/events/events/RouterEvents.kt
+21
View File
@@ -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,
}
core/events/src/main/kotlin/com/correx/core/events/serialization/Serialization.kt
+2
View File
@@ -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)
}
}
core/router/src/main/kotlin/com/correx/core/router/RouterFacade.kt
+56 -33
View File
@@ -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<SessionId, MutableList<RouterTurn>>()
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,
}
}
core/router/src/main/kotlin/com/correx/core/router/RouterReducer.kt
+21
View File
@@ -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
)
}
}
testing/deterministic/src/test/kotlin/RouterFacadeTest.kt
+43 -20
View File
@@ -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<RouterState>()
val eventStore = mockEventStore()
val replayer = DefaultEventReplayer<RouterState>(
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<RouterState>()
val eventStore = mockEventStore()
val replayer = DefaultEventReplayer<RouterState>(
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)
}
// --------------------------------------------------------------------------