From 6242b590e47f6711ac3bc8d11b20f5f277919e5e Mon Sep 17 00:00:00 2001 From: kami Date: Sun, 14 Jun 2026 03:35:59 +0400 Subject: [PATCH] feat(server): clarification WS protocol + wiring Surfaces the clarification loop over the wire: ServerMessage.ClarificationRequired (clarification.required, carrying the structured questions) is mapped from ClarificationRequestedEvent; ClientMessage.ClarificationResponse carries the operator's answers and is routed to orchestrator.submitClarification on both the global and session streams. Reuses the core ClarificationQuestion/Answer types as the wire shape. Round-trip tests for both directions. --- .../apps/server/bridge/DomainEventMapper.kt | 9 +++++ .../apps/server/protocol/ClientMessage.kt | 11 ++++++ .../apps/server/protocol/ServerMessage.kt | 18 +++++++++ .../apps/server/ws/GlobalStreamHandler.kt | 11 ++++++ .../apps/server/ws/SessionStreamHandler.kt | 8 ++++ .../ServerMessageSerializationTest.kt | 39 +++++++++++++++++++ 6 files changed, 96 insertions(+) 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 e34640b4..bebb835d 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 @@ -15,6 +15,7 @@ import com.correx.core.events.events.ArtifactValidatingEvent import com.correx.core.events.events.ExecutionPlanLockedEvent import com.correx.core.events.events.ChatSessionStartedEvent import com.correx.core.events.events.ChatTurnEvent +import com.correx.core.events.events.ClarificationRequestedEvent import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.SessionWorkspaceBoundEvent import com.correx.core.events.events.WorkflowStartedEvent @@ -221,6 +222,14 @@ suspend fun domainEventToServerMessage( ) is ApprovalRequestedEvent -> mapApprovalRequested(p, seq, sessionSequence) + is ClarificationRequestedEvent -> ServerMessage.ClarificationRequired( + sessionId = p.sessionId, + requestId = p.requestId, + stageId = p.stageId, + questions = p.questions, + sequence = seq, + sessionSequence = sessionSequence, + ) is ApprovalDecisionResolvedEvent -> ServerMessage.ApprovalResolved( // ApprovalDecisionResolvedEvent has no sessionId on its payload — read it from the event envelope sessionId = event.metadata.sessionId, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt index 7aba33e5..f4b9c001 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt @@ -2,7 +2,9 @@ package com.correx.apps.server.protocol import com.correx.core.approvals.GrantScope import com.correx.core.approvals.Tier +import com.correx.core.events.events.ClarificationAnswer import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.ClarificationRequestId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import com.correx.core.router.ChatMode @@ -61,6 +63,15 @@ sealed class ClientMessage { val steeringNote: String?, ) : ClientMessage() + /** The operator's answers to a stage's open questions (see ServerMessage.ClarificationRequired). */ + @Serializable + data class ClarificationResponse( + val sessionId: SessionId, + val stageId: StageId, + val requestId: ClarificationRequestId, + val answers: List, + ) : ClientMessage() + @Serializable data class CreateGrant( val sessionId: SessionId, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 9a9b8aed..f41de858 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -2,8 +2,10 @@ package com.correx.apps.server.protocol import com.correx.apps.server.metrics.MetricsReport import com.correx.core.approvals.Tier +import com.correx.core.events.events.ClarificationQuestion import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ClarificationRequestId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId import kotlinx.serialization.SerialName @@ -305,6 +307,22 @@ sealed interface ServerMessage { override val sessionSequence: Long, ) : ServerMessage, SessionMessage + /** + * A stage raised open questions (from a ClarificationRequestedEvent) and the run is parked until + * the operator answers. The client renders [questions] as an interactive form and replies with a + * ClientMessage.ClarificationResponse carrying [requestId] and [stageId]. + */ + @Serializable + @SerialName("clarification.required") + data class ClarificationRequired( + val sessionId: SessionId, + val requestId: ClarificationRequestId, + val stageId: StageId, + val questions: List, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + // -- System / infra -- @Serializable 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 521f9397..f29ed987 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 @@ -234,6 +234,7 @@ class GlobalStreamHandler(private val module: ServerModule) { is ClientMessage.UpdateConfig -> sendFrame(queries.updateConfig(msg.patch)) is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported")) is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg, sendFrame) + is ClientMessage.ClarificationResponse -> handleClarificationResponse(msg) is ClientMessage.CreateGrant -> handleCreateGrant(msg, sendFrame) is ClientMessage.ChatInput -> { // The USER and ROUTER turns are emitted as ChatTurnEvents inside onUserInput @@ -293,6 +294,16 @@ class GlobalStreamHandler(private val module: ServerModule) { module.approvalCoordinator.handleResponse(msg, scopeSessionId)?.let { sendFrame(it) } } + // The message carries its own sessionId/stageId/requestId, so no lookup is needed — the parked + // stage completes and re-runs with the answers in context (DefaultSessionOrchestrator). + private suspend fun handleClarificationResponse(msg: ClientMessage.ClarificationResponse) { + runCatching { + module.orchestrator.submitClarification(msg.sessionId, msg.stageId, msg.requestId, msg.answers) + }.onFailure { + log.error("submitClarification failed for session={}: {}", msg.sessionId.value, it.message, it) + } + } + private fun errorResponse(message: String) = ServerMessage.ProtocolError(message) private fun encodeError(message: String): Frame.Text = diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt index 0805be8c..5e3a8abd 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ws/SessionStreamHandler.kt @@ -106,6 +106,14 @@ class SessionStreamHandler(private val module: ServerModule) { } } + is ClientMessage.ClarificationResponse -> { + runCatching { + module.orchestrator.submitClarification(msg.sessionId, msg.stageId, msg.requestId, msg.answers) + }.onFailure { + log.error("submitClarification failed for session={}: {}", msg.sessionId.value, it.message, it) + } + } + else -> { log.warn("unexpected message type={} in session stream", msg::class.simpleName) val error = ServerMessage.ProtocolError("Unexpected message type in session stream") diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt index d0041aea..328f91af 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt @@ -193,6 +193,45 @@ class ServerMessageSerializationTest { assertEquals(50.0, decoded.stats.approvalWaitPct) } + @Test + fun `ClarificationRequired encodes type, stageId and structured questions`() { + val msg = ServerMessage.ClarificationRequired( + sessionId = SessionId("sess-c"), + requestId = SessionId("req-1"), + stageId = SessionId("analyst"), + questions = listOf( + com.correx.core.events.events.ClarificationQuestion( + id = "stack", + prompt = "Which stack?", + options = listOf("React", "Vue"), + header = "Stack", + ), + ), + sequence = 9, + sessionSequence = 4, + ) + val jsonStr = ProtocolSerializer.encodeServerMessage(msg) + assert(jsonStr.contains("\"type\":\"clarification.required\"")) { "expected type=clarification.required" } + assert(jsonStr.contains("\"prompt\":\"Which stack?\"")) { "expected question prompt" } + assert(jsonStr.contains("\"React\"")) { "expected options" } + } + + @Test + fun `ClarificationResponse decodes from the client wire format`() { + val wire = """ + {"type":"com.correx.apps.server.protocol.ClientMessage.ClarificationResponse", + "sessionId":"sess-c","stageId":"analyst","requestId":"req-1", + "answers":[{"questionId":"stack","value":"React"}]} + """.trimIndent() + val decoded = ProtocolSerializer.decodeClientMessage(wire) + assert(decoded is ClientMessage.ClarificationResponse) { "expected ClarificationResponse, got $decoded" } + val response = decoded as ClientMessage.ClarificationResponse + assertEquals("analyst", response.stageId.value) + assertEquals(1, response.answers.size) + assertEquals("stack", response.answers.single().questionId) + assertEquals("React", response.answers.single().value) + } + @Test fun `SessionAnnounced round-trips through JSON`() { val original = ServerMessage.SessionAnnounced(