fix: address P0-4 chat zombie, P0-5 double SessionStarted, P0-6 steering wrong content

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,

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).
+        // Emit ChatSessionStartedEvent for the new session (no graph, no orchestrator).
-        val event = NewEvent(
+        // 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(
+            payload = ChatSessionStartedEvent(sessionId = sessionId),
-                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)))
 
         // 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(
+        // SessionStarted is not sent here — it is derived from WorkflowStartedEvent
-            sessionId = sessionId,
+        // by streamGlobal via DomainEventMapper, ensuring exactly one SessionStarted
-            workflowId = msg.workflowId,
+        // per session with the correct sequence number (P0-5).
-            sequence = 0L,
-            sessionSequence = 0L,
-        )
-        session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started)))
-
         val stages = graph.stages.map { (stageId, stageConfig) ->
             val tools = stageConfig.allowedTools.mapNotNull { toolName ->
                 module.toolRegistry.resolve(toolName)?.let { tool ->

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

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,

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)
     }
 }
 

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,

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) {
+        // NOTE: SteeringNoteAddedEvent emission is deferred to P4-1, which will wire
-            val steeringEvent = SteeringNoteAddedEvent(
+        // proper validation (via ValidationPipeline) and consumption by the context builder.
-                sessionId = sessionId,
+        // The current SteeringNoteAddedEvent had wrong content (user raw input instead of
-                content = input,
+        // LLM-processed steering text) and was emitted without validation (P0-6).
-                stageId = state.currentStageId,
+        // The steeringEmitted flag below still informs the TUI that steering mode was used.
-            )
-            eventStore.append(NewEvent(
-                metadata = EventMetadata(
-                    eventId = EventId(UUID.randomUUID().toString()),
-                    sessionId = sessionId,
-                    timestamp = Clock.System.now(),
-                    schemaVersion = 1,
-                    causationId = null,
-                    correlationId = null,
-                ),
-                payload = steeringEvent,
-            ))
-        }
 
         return RouterResponse(content = content, steeringEmitted = (mode == ChatMode.STEERING))
     }

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)
+        assertTrue(mockStore.appendedEvents.isEmpty())
-        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)
     }
 
     @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
+        assertTrue(mockStore.appendedEvents.isEmpty())
-        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)
     }
 
     // --------------------------------------------------------------------------
@@ -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())
     }
 
     // --------------------------------------------------------------------------