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

feat(tui): event-derived ordering + retire Kotlin TUI

Browse Source 
Rebuild the Go TUI transcript from the event stream instead of
optimistic local echoes. ChatTurnEvent now maps to ServerMessage.ChatTurn
(chat.turn); the TUI auto-vivifies sessions and renders user+router turns
from events. SessionStarted is de-special-cased into SessionAnnounced
(still carries workflowId), dropping the router.response shadow and
optimistic echoes.

Delete the unused Kotlin TUI (apps/tui) and add a kotlinx<->Go
golden-fixture test to lock the wire protocol. Gitignore server runtime
logs.
This commit is contained in:
kami
2026-06-02 23:38:20 +04:00
parent b2f60d09bb
commit 1017bfffef
60 changed files with 231 additions and 4813 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/main/kotlin/com/correx/apps/server/bridge/DomainEventMapper.kt
+12 -2
View File
@@ -10,6 +10,7 @@ import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.ApprovalRequestedEvent
import com.correx.core.events.events.ArtifactCreatedEvent
import com.correx.core.events.events.ChatSessionStartedEvent
import com.correx.core.events.events.ChatTurnEvent
import com.correx.core.events.events.InferenceCompletedEvent
import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.events.InferenceStartedEvent
@@ -54,20 +55,29 @@ suspend fun domainEventToServerMessage(
): ServerMessage? {
val seq = event.sequence
return when (val p = event.payload) {
is ChatSessionStartedEvent -> ServerMessage.SessionStarted(
is ChatSessionStartedEvent -> ServerMessage.SessionAnnounced(
sessionId = p.sessionId,
workflowId = "chat",
sequence = seq,
sessionSequence = sessionSequence,
)
is WorkflowStartedEvent -> ServerMessage.SessionStarted(
is WorkflowStartedEvent -> ServerMessage.SessionAnnounced(
sessionId = p.sessionId,
workflowId = p.workflowId,
sequence = seq,
sessionSequence = sessionSequence,
)
is ChatTurnEvent -> ServerMessage.ChatTurn(
sessionId = p.sessionId,
turnId = p.turnId,
role = p.role.name,
content = p.content,
sequence = seq,
sessionSequence = sessionSequence,
)
is WorkflowCompletedEvent -> ServerMessage.SessionCompleted(
sessionId = p.sessionId,
sequence = seq,
apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt
+20 -4
View File
@@ -33,18 +33,34 @@ sealed interface ServerMessage {
// -- Session lifecycle --
/**
* @see SessionStartedEvent — emitted when a session begins.
* TODO(cleanup-NN): retire SessionStarted once TUI migrates to event-derived ordering.
* Event-derived session announcement: carries the session's [workflowId] so the client
* can name it. Derived from WorkflowStartedEvent / ChatSessionStartedEvent. The client
* applies it through the same auto-vivify path as any other event (no special-casing).
*/
@Serializable
@SerialName("session.started")
data class SessionStarted(
@SerialName("session.announced")
data class SessionAnnounced(
val sessionId: SessionId,
val workflowId: String,
override val sequence: Long,
override val sessionSequence: Long,
) : ServerMessage, SessionMessage
/**
* Event-derived conversation turn (USER or ROUTER), from ChatTurnEvent. The client
* rebuilds the chat transcript from these, ordered by [sessionSequence].
*/
@Serializable
@SerialName("chat.turn")
data class ChatTurn(
val sessionId: SessionId,
val turnId: String,
val role: String,
val content: String,
override val sequence: Long,
override val sessionSequence: Long,
) : ServerMessage, SessionMessage
@Serializable
@SerialName("session.paused")
data class SessionPaused(
apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt
+8 -17
View File
@@ -193,14 +193,11 @@ class GlobalStreamHandler(private val module: ServerModule) {
is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg, sendFrame)
is ClientMessage.CreateGrant -> handleCreateGrant(msg, sendFrame)
is ClientMessage.ChatInput -> {
// The USER and ROUTER turns are emitted as ChatTurnEvents inside onUserInput
// and reach the client as chat.turn frames via streamGlobal — no direct
// RouterResponseMessage (the conversation is event-derived).
runCatching {
module.routerFacade.onUserInput(msg.sessionId, msg.text, msg.mode)
}.onSuccess { response ->
sendFrame(ServerMessage.RouterResponseMessage(
sessionId = msg.sessionId,
content = response.content,
steeringEmitted = response.steeringEmitted,
))
}.onFailure {
log.error("routerFacade.onUserInput failed: {}", it.message)
sendFrame(ServerMessage.ProtocolError("Router error: ${it.message}"))
@@ -327,7 +324,8 @@ class GlobalStreamHandler(private val module: ServerModule) {
log.info("starting chat session={}", sessionId.value)
// Emit ChatSessionStartedEvent for the new session (no graph, no orchestrator).
// streamGlobal picks this up and maps it to SessionStarted with the real sequence.
// streamGlobal maps it to a session.announced frame with the real sequence, and the
// USER/ROUTER turns from onUserInput stream as chat.turn frames — all event-derived.
module.eventStore.append(NewEvent(
metadata = EventMetadata(
eventId = EventId(UUID.randomUUID().toString()),
@@ -340,15 +338,8 @@ class GlobalStreamHandler(private val module: ServerModule) {
payload = ChatSessionStartedEvent(sessionId = sessionId),
))
// Call router and send response.
runCatching {
module.routerFacade.onUserInput(sessionId, msg.text)
}.onSuccess { response ->
sendFrame(ServerMessage.RouterResponseMessage(
sessionId = sessionId,
content = response.content,
steeringEmitted = response.steeringEmitted,
))
}.onFailure {
log.error("routerFacade.onUserInput failed: {}", it.message)
sendFrame(ServerMessage.ProtocolError(
@@ -381,9 +372,9 @@ class GlobalStreamHandler(private val module: ServerModule) {
// Send StageToolManifest FIRST. If the WS is already closed, this throws and the
// orchestrator never launches — no silent data loss. The exception propagates to the
// runCatching in handleClientMessage (caller) and is logged there.
// SessionStarted is not sent here — it is derived from WorkflowStartedEvent
// by streamGlobal via DomainEventMapper, ensuring exactly one SessionStarted
// per session with the correct sequence number (P0-5).
// No session announce is sent here — it is derived from WorkflowStartedEvent by
// streamGlobal via DomainEventMapper (one session.announced per session with the
// correct sequence number).
val stages = graph.stages.map { (stageId, stageConfig) ->
val tools = stageConfig.allowedTools.mapNotNull { toolName ->
module.toolRegistry.resolve(toolName)?.let { tool ->
apps/server/src/test/kotlin/com/correx/apps/server/bridge/DomainEventMapperTest.kt
+22 -2
View File
@@ -9,6 +9,8 @@ import com.correx.core.approvals.Tier
import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.events.events.ApprovalDecisionResolvedEvent
import com.correx.core.events.events.ApprovalRequestedEvent
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.InferenceCompletedEvent
import com.correx.core.events.events.InferenceStartedEvent
@@ -88,12 +90,30 @@ class DomainEventMapperTest {
)
@Test
fun `WorkflowStartedEvent maps to SessionStarted`(): Unit = runTest {
fun `WorkflowStartedEvent maps to SessionAnnounced`(): Unit = runTest {
val event =
storedEvent(WorkflowStartedEvent(sessionId = sessionId, startStageId = stageId, workflowId = workflowId))
val result = domainEventToServerMessage(event, noopStore, sessionSequence = 0L)
assertEquals(
ServerMessage.SessionStarted(sessionId, workflowId, event.sequence, 0L),
ServerMessage.SessionAnnounced(sessionId, workflowId, event.sequence, 0L),
result,
)
}
@Test
fun `ChatTurnEvent maps to ChatTurn`(): Unit = runTest {
val event = storedEvent(
ChatTurnEvent(
sessionId = sessionId,
turnId = "turn-1",
role = ChatTurnRole.ROUTER,
content = "hello",
timestampMs = 1L,
),
)
val result = domainEventToServerMessage(event, noopStore, sessionSequence = 3L)
assertEquals(
ServerMessage.ChatTurn(sessionId, "turn-1", "ROUTER", "hello", event.sequence, 3L),
result,
)
}
apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt
+23 -6
View File
@@ -13,21 +13,38 @@ class ServerMessageSerializationTest {
}
@Test
fun `SessionStarted encodes sequence and sessionSequence`() {
val msg = ServerMessage.SessionStarted(
fun `SessionAnnounced encodes sequence and sessionSequence`() {
val msg = ServerMessage.SessionAnnounced(
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("\"type\":\"session.announced\"")) { "expected type=session.announced" }
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 `ChatTurn encodes role content and cursors`() {
val msg = ServerMessage.ChatTurn(
sessionId = SessionId("sess-1"),
turnId = "turn-9",
role = "ROUTER",
content = "hi there",
sequence = 5,
sessionSequence = 2,
)
val jsonStr = ProtocolSerializer.encodeServerMessage(msg)
assert(jsonStr.contains("\"type\":\"chat.turn\"")) { "expected type=chat.turn" }
assert(jsonStr.contains("\"turnId\":\"turn-9\"")) { "expected turnId" }
assert(jsonStr.contains("\"role\":\"ROUTER\"")) { "expected role" }
assert(jsonStr.contains("\"content\":\"hi there\"")) { "expected content" }
}
@Test
fun `SnapshotComplete encodes with only type field`() {
val msg = ServerMessage.SnapshotComplete
@@ -57,15 +74,15 @@ class ServerMessageSerializationTest {
}
@Test
fun `SessionStarted round-trips through JSON`() {
val original = ServerMessage.SessionStarted(
fun `SessionAnnounced round-trips through JSON`() {
val original = ServerMessage.SessionAnnounced(
sessionId = SessionId("sess-42"),
workflowId = "wf-abc",
sequence = 200,
sessionSequence = 33,
)
val jsonStr = ProtocolSerializer.encodeServerMessage(original)
val decoded = json.decodeFromString<ServerMessage.SessionStarted>(jsonStr)
val decoded = json.decodeFromString<ServerMessage.SessionAnnounced>(jsonStr)
assertEquals(original.sessionId, decoded.sessionId)
assertEquals(original.workflowId, decoded.workflowId)
assertEquals(200L, decoded.sequence)
apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt
+3 -3
View File
@@ -158,7 +158,7 @@ class GlobalStreamHandlerTest {
// Then 3 WorkflowStarted messages (seq 6, 7, 8)
val live = messages.drop(2)
assertEquals(3, live.size)
live.forEach { assertInstanceOf(ServerMessage.SessionStarted::class.java, it) }
live.forEach { assertInstanceOf(ServerMessage.SessionAnnounced::class.java, it) }
}
@Test
@@ -198,7 +198,7 @@ class GlobalStreamHandlerTest {
}
val completeIdx = messages.indexOf(ServerMessage.SnapshotComplete)
val afterComplete = messages.drop(completeIdx + 1)
assert(afterComplete.any { msg -> msg is ServerMessage.SessionStarted }) {
assert(afterComplete.any { msg -> msg is ServerMessage.SessionAnnounced }) {
"Race event (seq=6) was lost in iteration $it! Messages: $messages"
}
}
@@ -232,7 +232,7 @@ class GlobalStreamHandlerTest {
assertEquals(ServerMessage.SnapshotComplete, messages[0])
assertEquals(2, messages.size)
assertInstanceOf(ServerMessage.SessionStarted::class.java, messages[1])
assertInstanceOf(ServerMessage.SessionAnnounced::class.java, messages[1])
}
@Test