From eed557fe9c49c771ef36c59a815b3afa2988082d Mon Sep 17 00:00:00 2001 From: kami Date: Thu, 28 May 2026 23:10:29 +0400 Subject: [PATCH] fix: address P0-4 chat zombie, P0-5 double SessionStarted, P0-6 steering wrong content --- .../apps/server/bridge/DomainEventMapper.kt | 8 ++++ .../apps/server/ws/GlobalStreamHandler.kt | 38 +++++------------ .../core/events/events/SessionEvents.kt | 6 +++ .../orchestration/OrchestrationStatus.kt | 1 + .../events/serialization/Serialization.kt | 2 + .../DefaultOrchestrationReducer.kt | 7 ++++ .../com/correx/core/router/RouterFacade.kt | 27 +++--------- .../src/test/kotlin/RouterFacadeTest.kt | 41 ++++--------------- 8 files changed, 47 insertions(+), 83 deletions(-) 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 e6e984a8..45a05760 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 @@ -5,6 +5,7 @@ import com.correx.apps.server.protocol.RiskSummaryDto 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.ChatSessionStartedEvent import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.WorkflowStartedEvent import com.correx.core.events.events.InferenceStartedEvent @@ -45,6 +46,13 @@ suspend fun domainEventToServerMessage( ): ServerMessage? { val seq = event.sequence return when (val p = event.payload) { + is ChatSessionStartedEvent -> ServerMessage.SessionStarted( + sessionId = p.sessionId, + workflowId = "chat", + sequence = seq, + sessionSequence = sessionSequence, + ) + is WorkflowStartedEvent -> ServerMessage.SessionStarted( sessionId = p.sessionId, workflowId = p.workflowId, 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 df733034..27449e3e 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 @@ -14,10 +14,10 @@ import com.correx.apps.server.protocol.ToolDecl import com.correx.core.approvals.GrantScope import com.correx.core.approvals.Tier import com.correx.core.events.events.ApprovalGrantCreatedEvent +import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.StoredEvent -import com.correx.core.events.events.WorkflowStartedEvent import com.correx.core.events.types.GrantId import com.correx.core.events.stores.EventStore import com.correx.core.events.types.EventId @@ -275,8 +275,9 @@ class GlobalStreamHandler(private val module: ServerModule) { val sessionId = msg.sessionId log.info("starting chat session={}", sessionId.value) - // Emit WorkflowStartedEvent for the new session (no graph, no orchestrator). - val event = NewEvent( + // Emit ChatSessionStartedEvent for the new session (no graph, no orchestrator). + // streamGlobal picks this up and maps it to SessionStarted with the real sequence. + module.eventStore.append(NewEvent( metadata = EventMetadata( eventId = EventId(UUID.randomUUID().toString()), sessionId = sessionId, @@ -285,22 +286,8 @@ class GlobalStreamHandler(private val module: ServerModule) { causationId = null, correlationId = null, ), - payload = WorkflowStartedEvent( - sessionId = sessionId, - workflowId = "chat", - startStageId = StageId("none"), - ), - ) - module.eventStore.append(event) - - // Send SessionStarted so the TUI creates the session entry immediately. - val started = ServerMessage.SessionStarted( - sessionId = sessionId, - workflowId = "chat", - sequence = 0L, - sessionSequence = 0L, - ) - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) + payload = ChatSessionStartedEvent(sessionId = sessionId), + )) // Call router and send response. runCatching { @@ -340,17 +327,12 @@ class GlobalStreamHandler(private val module: ServerModule) { val sessionId: SessionId = TypeId(UUID.randomUUID().toString()) log.info("starting session={} workflow={}", sessionId.value, msg.workflowId) - // Send protocol messages FIRST. If the WS is already closed, these throw and the + // 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. - val started = ServerMessage.SessionStarted( - sessionId = sessionId, - workflowId = msg.workflowId, - sequence = 0L, - sessionSequence = 0L, - ) - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) - + // 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). val stages = graph.stages.map { (stageId, stageConfig) -> val tools = stageConfig.allowedTools.mapNotNull { toolName -> module.toolRegistry.resolve(toolName)?.let { tool -> diff --git a/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt b/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt index 8f253dc3..936cacc4 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/events/SessionEvents.kt @@ -38,3 +38,9 @@ data class SessionFailedEvent( val errorCode: String? = null, val errorMessage: String? = null ) : EventPayload + +@Serializable +@SerialName("ChatSessionStarted") +data class ChatSessionStartedEvent( + val sessionId: SessionId, +) : EventPayload diff --git a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationStatus.kt b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationStatus.kt index 395b1f32..9768adf5 100644 --- a/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationStatus.kt +++ b/core/events/src/main/kotlin/com/correx/core/events/orchestration/OrchestrationStatus.kt @@ -2,6 +2,7 @@ package com.correx.core.events.orchestration enum class OrchestrationStatus { IDLE, + CHAT, RUNNING, PAUSED, COMPLETED, 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 c64e1082..38a05e07 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 @@ -30,6 +30,7 @@ import com.correx.core.events.events.RetryAttemptedEvent import com.correx.core.events.events.RiskAssessedEvent import com.correx.core.events.events.SessionCompletedEvent import com.correx.core.events.events.SessionFailedEvent +import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.SessionPausedEvent import com.correx.core.events.events.SessionResumedEvent import com.correx.core.events.events.SessionStartedEvent @@ -101,6 +102,7 @@ val eventModule = SerializersModule { subclass(WorkflowCompletedEvent::class) subclass(RetryAttemptedEvent::class) subclass(RiskAssessedEvent::class) + subclass(ChatSessionStartedEvent::class) } } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt index 9372150e..c2633cac 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultOrchestrationReducer.kt @@ -1,5 +1,6 @@ package com.correx.core.kernel.orchestration +import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.OrchestrationPausedEvent import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.RetryAttemptedEvent @@ -16,6 +17,12 @@ class DefaultOrchestrationReducer : OrchestrationReducer { state: OrchestrationState, event: StoredEvent, ): OrchestrationState = when (val p = event.payload) { + is ChatSessionStartedEvent -> state.copy( + workflowId = "chat", + status = OrchestrationStatus.CHAT, + currentStageId = null, + ) + is WorkflowStartedEvent -> state.copy( workflowId = p.workflowId, status = OrchestrationStatus.RUNNING, 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 434e0b39..2e03cd6e 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,10 +1,6 @@ package com.correx.core.router -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.EventId import com.correx.core.events.types.InferenceRequestId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId @@ -71,24 +67,11 @@ class DefaultRouterFacade( history.add(RouterTurn(role = TurnRole.ROUTER, content = content, timestamp = Clock.System.now())) - if (mode == ChatMode.STEERING) { - val steeringEvent = SteeringNoteAddedEvent( - sessionId = sessionId, - content = input, - stageId = state.currentStageId, - ) - eventStore.append(NewEvent( - metadata = EventMetadata( - eventId = EventId(UUID.randomUUID().toString()), - sessionId = sessionId, - timestamp = Clock.System.now(), - schemaVersion = 1, - causationId = null, - correlationId = null, - ), - payload = steeringEvent, - )) - } + // NOTE: SteeringNoteAddedEvent emission is deferred to P4-1, which will wire + // proper validation (via ValidationPipeline) and consumption by the context builder. + // The current SteeringNoteAddedEvent had wrong content (user raw input instead of + // LLM-processed steering text) and was emitted without validation (P0-6). + // The steeringEmitted flag below still informs the TUI that steering mode was used. return RouterResponse(content = content, steeringEmitted = (mode == ChatMode.STEERING)) } diff --git a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt index 4432ade1..bbbb3774 100644 --- a/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt +++ b/testing/deterministic/src/test/kotlin/RouterFacadeTest.kt @@ -1,7 +1,6 @@ import com.correx.core.context.model.ContextPack import com.correx.core.context.model.TokenBudget import com.correx.core.events.events.NewEvent -import com.correx.core.events.events.SteeringNoteAddedEvent import com.correx.core.events.events.StoredEvent import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ContextPackId @@ -36,7 +35,6 @@ import kotlinx.coroutines.runBlocking import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertFalse 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 @@ -83,20 +81,17 @@ class RouterFacadeTest { } @Test - fun `STEERING mode appends SteeringNoteAddedEvent with correct session id and user input`(): Unit = runBlocking { + fun `STEERING mode does not append events to store`(): Unit = runBlocking { + // SteeringNoteAddedEvent emission is deferred to P4-1 (see P0-6). + // The steeringEmitted flag still informs the TUI that steering mode was used. 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 event = mockStore.appendedEvents[0] - assertTrue(event.payload is SteeringNoteAddedEvent) - val steeringEvent = event.payload as SteeringNoteAddedEvent - assertEquals(SessionId("session-xyz"), steeringEvent.sessionId) - assertEquals("steer this way", steeringEvent.content) + assertTrue(mockStore.appendedEvents.isEmpty()) } @Test - fun `STEERING mode appends SteeringNoteAddedEvent with stageId from state`(): Unit = runBlocking { + fun `STEERING mode does not append events regardless of state stageId`(): Unit = runBlocking { val mockStore = mockEventStore() val stageId = StageId("stage-A") val facade = DefaultRouterFacade( @@ -116,28 +111,7 @@ class RouterFacadeTest { config = RouterConfig(tokenBudget = TokenBudget(limit = 5000)), ) facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!", mode = ChatMode.STEERING) - val steeringEvent = mockStore.appendedEvents[0].payload as SteeringNoteAddedEvent - assertEquals(stageId, steeringEvent.stageId) - } - - @Test - fun `STEERING mode appends SteeringNoteAddedEvent with null stageId when state has none`(): Unit = runBlocking { - val mockStore = mockEventStore() - val facade = DefaultRouterFacade( - routerRepository = object : RouterRepository { - override suspend fun getRouterState(sessionId: SessionId): RouterState = - RouterState(sessionId = sessionId, workflowStatus = WorkflowStatus.IDLE, currentStageId = null) - }, - routerContextBuilder = object : RouterContextBuilder { - override suspend fun build(state: RouterState, budget: TokenBudget): ContextPack = emptyContextPack() - }, - inferenceRouter = mockInferenceRouter("response"), - eventStore = mockStore, - config = RouterConfig(tokenBudget = TokenBudget(limit = 5000)), - ) - facade.onUserInput(sessionId = SessionId("test-session"), input = "Hello!", mode = ChatMode.STEERING) - val steeringEvent = mockStore.appendedEvents[0].payload as SteeringNoteAddedEvent - assertNull(steeringEvent.stageId) + assertTrue(mockStore.appendedEvents.isEmpty()) } // -------------------------------------------------------------------------- @@ -475,7 +449,8 @@ class RouterFacadeTest { assertNotNull(response) assertEquals("inference response", response.content) assertTrue(response.steeringEmitted) - assertTrue(mockStore.appendedEvents.isNotEmpty()) + // Steering events are deferred to P4-1; no events appended yet. + assertTrue(mockStore.appendedEvents.isEmpty()) } // --------------------------------------------------------------------------