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 6fc2a38d..f686708d 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 @@ -192,6 +192,35 @@ 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 + // 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))) + + val stages = graph.stages.map { (stageId, stageConfig) -> + val tools = stageConfig.allowedTools.mapNotNull { toolName -> + module.toolRegistry.resolve(toolName)?.let { tool -> + ToolDecl(name = tool.name, tier = tool.tier.level) + } + } + StageToolDecl(stageId = stageId.value, tools = tools) + } + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage( + ServerMessage.StageToolManifest( + sessionId = sessionId, + workflowId = msg.workflowId, + stages = stages, + ), + ))) + + // Only launch orchestrator after protocol messages are delivered successfully module.moduleScope.launch { runCatching { module.orchestrator.run(sessionId, graph, module.defaultOrchestrationConfig) } .onFailure { ex -> @@ -216,35 +245,6 @@ class GlobalStreamHandler(private val module: ServerModule) { ) } } - - runCatching { - val started = ServerMessage.SessionStarted( - sessionId = sessionId, - workflowId = msg.workflowId, - sequence = 0L, - sessionSequence = 0L, - ) - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(started))) - - // Emit StageToolManifest for the newly created session (ARCH-TOOL-1) - val stages = graph.stages.map { (stageId, stageConfig) -> - val tools = stageConfig.allowedTools.mapNotNull { toolName -> - module.toolRegistry.resolve(toolName)?.let { tool -> - ToolDecl(name = tool.name, tier = tool.tier.level) - } - } - StageToolDecl(stageId = stageId.value, tools = tools) - } - session.send(Frame.Text(ProtocolSerializer.encodeServerMessage( - ServerMessage.StageToolManifest( - sessionId = sessionId, - workflowId = msg.workflowId, - stages = stages, - ), - ))) - }.onFailure { ex -> - log.warn("failed to send session start messages for session={}: {}", sessionId.value, ex.message) - } } }