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/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,

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<ClarificationAnswer>,
+    ) : ClientMessage()
+
     @Serializable
     data class CreateGrant(
         val sessionId: SessionId,

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<ClarificationQuestion>,
+        override val sequence: Long,
+        override val sessionSequence: Long,
+    ) : ServerMessage, SessionMessage
+
     // -- System / infra --
 
     @Serializable

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 =

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

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(