refactor: rename the router subsystem to Talkie
The 'router' name was misleading — it's the always-on conversational front-end
(CHAT triage + STEERING into a running workflow), not a routing layer, and it's
distinct from InferenceRouter. Rename core:router -> core:talkie, package
com.correx.core.router -> com.correx.core.talkie, and RouterFacade/Config/State/
Repository/ContextBuilder/Projector/Reducer/Response -> Talkie*. Config section
[router] -> [talkie] (legacy [router] still read as fallback).
Persisted wire formats are preserved: ChatTurnRole.ROUTER and the
RouterNarrationEvent @SerialName("RouterNarration") stay, so existing event
logs still replay. RouterNarrationEvent the class is now TalkieNarrationEvent.
This commit is contained in:
@@ -0,0 +1,267 @@
|
||||
import com.correx.core.events.events.OrchestrationPausedEvent
|
||||
import com.correx.core.events.events.OrchestrationResumedEvent
|
||||
import com.correx.core.events.events.StageCompletedEvent
|
||||
import com.correx.core.events.events.StageFailedEvent
|
||||
import com.correx.core.events.events.SteeringNoteAddedEvent
|
||||
import com.correx.core.events.events.ToolInvokedEvent
|
||||
import com.correx.core.events.events.WorkflowCompletedEvent
|
||||
import com.correx.core.events.events.WorkflowFailedEvent
|
||||
import com.correx.core.events.events.WorkflowStartedEvent
|
||||
import com.correx.core.events.types.SessionId
|
||||
import com.correx.core.events.types.StageId
|
||||
import com.correx.core.talkie.DefaultTalkieReducer
|
||||
import com.correx.core.talkie.model.StageOutcomeKind
|
||||
import com.correx.core.talkie.model.WorkflowStatus
|
||||
import com.correx.testing.fixtures.EventFixtures.stored
|
||||
import kotlinx.datetime.Instant
|
||||
import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.Assertions.assertNotNull
|
||||
import org.junit.jupiter.api.Assertions.assertNull
|
||||
import org.junit.jupiter.api.Assertions.assertTrue
|
||||
import org.junit.jupiter.api.Test
|
||||
|
||||
class TalkieReducerTest {
|
||||
|
||||
private val reducer = DefaultTalkieReducer()
|
||||
private val sessionId = SessionId("s1")
|
||||
private val stageId = StageId("stage-1")
|
||||
private val workflowId = "workflow-test"
|
||||
|
||||
@Test
|
||||
fun `initial state has IDLE status and empty collections`() {
|
||||
val state = reducer.initial
|
||||
assertEquals(WorkflowStatus.IDLE, state.workflowStatus)
|
||||
assertNull(state.sessionId)
|
||||
assertNull(state.currentStageId)
|
||||
assertTrue(state.l2Memory.isEmpty())
|
||||
assertTrue(state.conversationHistory.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `WorkflowStartedEvent sets sessionId, RUNNING status and startStageId`() {
|
||||
val state = reducer.initial
|
||||
val updated = reducer.reduce(
|
||||
state,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
assertEquals(sessionId, updated.sessionId)
|
||||
assertEquals(WorkflowStatus.RUNNING, updated.workflowStatus)
|
||||
assertEquals(stageId, updated.currentStageId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `WorkflowCompletedEvent sets COMPLETED and clears currentStageId`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val completed = reducer.reduce(
|
||||
started,
|
||||
stored(payload = WorkflowCompletedEvent(sessionId, stageId, 3)),
|
||||
)
|
||||
assertEquals(WorkflowStatus.COMPLETED, completed.workflowStatus)
|
||||
assertNull(completed.currentStageId)
|
||||
assertEquals(sessionId, completed.sessionId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `WorkflowFailedEvent sets FAILED and clears currentStageId`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val failed = reducer.reduce(
|
||||
started,
|
||||
stored(payload = WorkflowFailedEvent(sessionId, stageId, "timeout", false)),
|
||||
)
|
||||
assertEquals(WorkflowStatus.FAILED, failed.workflowStatus)
|
||||
assertNull(failed.currentStageId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `OrchestrationPausedEvent sets PAUSED status`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val paused = reducer.reduce(
|
||||
started,
|
||||
stored(payload = OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")),
|
||||
)
|
||||
assertEquals(WorkflowStatus.PAUSED, paused.workflowStatus)
|
||||
assertEquals(sessionId, paused.sessionId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `OrchestrationResumedEvent sets RUNNING status from PAUSED`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val paused = reducer.reduce(
|
||||
started,
|
||||
stored(payload = OrchestrationPausedEvent(sessionId, stageId, "APPROVAL_PENDING")),
|
||||
)
|
||||
val resumed = reducer.reduce(
|
||||
paused,
|
||||
stored(payload = OrchestrationResumedEvent(sessionId, stageId)),
|
||||
)
|
||||
assertEquals(WorkflowStatus.RUNNING, resumed.workflowStatus)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `StageCompletedEvent appends RouterL2Entry with SUCCESS to l2Memory`() {
|
||||
val state = reducer.initial
|
||||
val storedEvent = stored(payload = StageCompletedEvent(sessionId, stageId, StageId("t1")))
|
||||
val updated = reducer.reduce(state, storedEvent)
|
||||
assertEquals(1, updated.l2Memory.size)
|
||||
val entry = updated.l2Memory[0]
|
||||
assertEquals(stageId, entry.stageId)
|
||||
assertEquals(StageOutcomeKind.SUCCESS, entry.outcome)
|
||||
assertTrue(entry.summary.contains("completed"))
|
||||
assertNotNull(entry.timestamp)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `StageCompletedEvent preserves other state fields`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val updated = reducer.reduce(
|
||||
started,
|
||||
stored(payload = StageCompletedEvent(sessionId, StageId("stage-2"), StageId("t1"))),
|
||||
)
|
||||
assertEquals(sessionId, updated.sessionId)
|
||||
assertEquals(WorkflowStatus.RUNNING, updated.workflowStatus)
|
||||
assertEquals(stageId, updated.currentStageId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `StageFailedEvent appends RouterL2Entry with FAILURE and clears currentStageId`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val storedEvent = stored(payload = StageFailedEvent(sessionId, stageId, StageId("t1"), "timeout"))
|
||||
val updated = reducer.reduce(started, storedEvent)
|
||||
assertEquals(1, updated.l2Memory.size)
|
||||
val entry = updated.l2Memory[0]
|
||||
assertEquals(stageId, entry.stageId)
|
||||
assertEquals(StageOutcomeKind.FAILURE, entry.outcome)
|
||||
assertEquals("timeout", entry.reason)
|
||||
assertNull(updated.currentStageId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `multiple StageCompletedEvents append to l2Memory sequentially`() {
|
||||
val state = reducer.initial
|
||||
val s1 = reducer.reduce(state, stored(payload = StageCompletedEvent(sessionId, StageId("a"), StageId("t1"))))
|
||||
val s2 = reducer.reduce(s1, stored(payload = StageCompletedEvent(sessionId, StageId("b"), StageId("t2"))))
|
||||
val s3 = reducer.reduce(s2, stored(payload = StageCompletedEvent(sessionId, StageId("c"), StageId("t3"))))
|
||||
assertEquals(3, s3.l2Memory.size)
|
||||
assertEquals(StageId("a"), s3.l2Memory[0].stageId)
|
||||
assertEquals(StageId("b"), s3.l2Memory[1].stageId)
|
||||
assertEquals(StageId("c"), s3.l2Memory[2].stageId)
|
||||
s3.l2Memory.forEach {
|
||||
assertEquals(StageOutcomeKind.SUCCESS, it.outcome)
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `SteeringNoteAddedEvent accumulates into pendingSteeringNotes without touching conversation`() {
|
||||
val state = reducer.initial
|
||||
val updated = reducer.reduce(
|
||||
state,
|
||||
stored(payload = SteeringNoteAddedEvent(sessionId, "do something different")),
|
||||
)
|
||||
assertEquals(listOf("do something different"), updated.pendingSteeringNotes)
|
||||
assertTrue(updated.conversationHistory.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `SteeringNoteAddedEvent preserves workflow status and current stage`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(
|
||||
initial,
|
||||
stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)),
|
||||
)
|
||||
val withNote = reducer.reduce(
|
||||
started,
|
||||
stored(payload = SteeringNoteAddedEvent(sessionId, "shift focus")),
|
||||
)
|
||||
assertEquals(listOf("shift focus"), withNote.pendingSteeringNotes)
|
||||
assertEquals(WorkflowStatus.RUNNING, withNote.workflowStatus)
|
||||
assertEquals(stageId, withNote.currentStageId)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `steering notes are harvested into the completing stage's L2 entry and then cleared`() {
|
||||
val initial = reducer.initial
|
||||
val started = reducer.reduce(initial, stored(payload = WorkflowStartedEvent(sessionId, workflowId, stageId)))
|
||||
val steered = reducer.reduce(started, stored(payload = SteeringNoteAddedEvent(sessionId, "call it the runtime")))
|
||||
val completed = reducer.reduce(steered, stored(payload = StageCompletedEvent(sessionId, stageId, StageId("t1"))))
|
||||
|
||||
assertEquals(listOf("call it the runtime"), completed.l2Memory[0].steeringNotes)
|
||||
assertTrue(completed.pendingSteeringNotes.isEmpty())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `unknown event type returns unchanged state`() {
|
||||
val state = reducer.initial
|
||||
val updated = reducer.reduce(state, stored(payload = ToolInvokedEvent("test")))
|
||||
assertEquals(state, updated)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `l2Memory entry preserves timestamp from event metadata`() {
|
||||
val storedEvent = stored(
|
||||
payload = StageCompletedEvent(sessionId, stageId, StageId("t1")),
|
||||
timestamp = Instant.parse("2026-06-15T12:00:00Z"),
|
||||
)
|
||||
val updated = reducer.reduce(reducer.initial, storedEvent)
|
||||
assertEquals(1, updated.l2Memory.size)
|
||||
assertNotNull(updated.l2Memory[0].timestamp)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `full lifecycle start complete pause resume start another stage fail`() {
|
||||
val s0 = reducer.initial
|
||||
assertEquals(WorkflowStatus.IDLE, s0.workflowStatus)
|
||||
|
||||
val s1 = reducer.reduce(s0, stored(payload = WorkflowStartedEvent(sessionId, workflowId, StageId("stage-A"))))
|
||||
assertEquals(WorkflowStatus.RUNNING, s1.workflowStatus)
|
||||
assertEquals(StageId("stage-A"), s1.currentStageId)
|
||||
|
||||
val s2 = reducer.reduce(s1, stored(payload = StageCompletedEvent(sessionId, StageId("stage-A"), StageId("t1"))))
|
||||
assertEquals(1, s2.l2Memory.size)
|
||||
assertEquals(StageOutcomeKind.SUCCESS, s2.l2Memory[0].outcome)
|
||||
|
||||
val s3 =
|
||||
reducer.reduce(s2, stored(payload = WorkflowCompletedEvent(sessionId, StageId("stage-A"), 1)))
|
||||
assertEquals(WorkflowStatus.COMPLETED, s3.workflowStatus)
|
||||
assertNull(s3.currentStageId)
|
||||
|
||||
val s4 = reducer.reduce(
|
||||
s3,
|
||||
stored(payload = OrchestrationPausedEvent(sessionId, StageId("stage-A"), "manual pause")),
|
||||
)
|
||||
assertEquals(WorkflowStatus.PAUSED, s4.workflowStatus)
|
||||
|
||||
val s5 = reducer.reduce(s4, stored(payload = OrchestrationResumedEvent(sessionId, StageId("stage-A"))))
|
||||
assertEquals(WorkflowStatus.RUNNING, s5.workflowStatus)
|
||||
|
||||
val s6 = reducer.reduce(
|
||||
s5,
|
||||
stored(payload = StageFailedEvent(sessionId, StageId("stage-B"), StageId("t2"), "llm error")),
|
||||
)
|
||||
assertEquals(StageOutcomeKind.FAILURE, s6.l2Memory[s6.l2Memory.size - 1].outcome)
|
||||
assertNull(s6.currentStageId)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user