fix(server): send SessionStarted/StageToolManifest before launching orchestrator
Reordered handleStartSession so protocol messages are sent before the orchestrator launches in module.moduleScope. If the WS is already closed, session.send() throws and the orchestrator never launches — avoiding silent data loss where the session runs but the client never learns it exists.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user