From c8a599c64fa1af0a05d0ba88aa0538a665da13c2 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 25 May 2026 17:18:01 +0400 Subject: [PATCH] fix(tui-02): route ApprovalResponse through global socket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Complete the tui-02 spec acceptance criteria: delete the orphan `connectSession`, `disconnectSession`, `sessionStreamJob`, and `sessionWsSession` members from TuiWsClient, and remove the `send()` fallback that previously routed ApprovalResponse over a per-session socket. All ClientMessage types now travel the single global socket. Server side: GlobalStreamHandler.handleClientMessage now actually handles ApprovalResponse (was returning a protocol error), converting to a domain ApprovalDecision and dispatching to the orchestrator — mirrors SessionStreamHandler's prior logic. Adds ApprovalResponseGlobalSocketRoundTripTest covering APPROVE and STEER encode/decode through ProtocolSerializer. --- .../apps/server/ws/GlobalStreamHandler.kt | 49 ++++++++++++++- .../com/correx/apps/tui/ws/TuiWsClient.kt | 47 +-------------- ...provalResponseGlobalSocketRoundTripTest.kt | 59 +++++++++++++++++++ 3 files changed, 107 insertions(+), 48 deletions(-) create mode 100644 apps/tui/src/test/kotlin/com/correx/apps/tui/ws/ApprovalResponseGlobalSocketRoundTripTest.kt 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 4d30d973..29e19b8f 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 @@ -3,10 +3,17 @@ package com.correx.apps.server.ws import com.correx.apps.server.ServerModule import com.correx.apps.server.bridge.DomainEventMapper import com.correx.apps.server.bridge.SessionEventBridge +import com.correx.apps.server.protocol.ApprovalDecision import com.correx.apps.server.protocol.ClientMessage import com.correx.apps.server.protocol.ProtocolSerializer import com.correx.apps.server.protocol.ProviderHealthDto import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.Tier +import com.correx.core.approvals.UserSteering +import com.correx.core.approvals.model.ApprovalContext +import com.correx.core.approvals.model.ApprovalScopeIdentity import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.NewEvent import com.correx.core.events.events.StoredEvent @@ -15,7 +22,9 @@ import com.correx.core.events.stores.EventStore import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.inference.ProviderHealth +import com.correx.core.sessions.ApprovalMode import com.correx.core.utils.TypeId +import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision import io.ktor.server.websocket.DefaultWebSocketServerSession import io.ktor.websocket.Frame import io.ktor.websocket.readText @@ -121,12 +130,48 @@ class GlobalStreamHandler(private val module: ServerModule) { is ClientMessage.StartSession -> handleStartSession(session, msg, sendFrame) is ClientMessage.CancelSession -> module.orchestrator.cancel(msg.sessionId) is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported")) - is ClientMessage.ApprovalResponse -> - session.send(encodeError("ApprovalResponse must be sent to /sessions/{id}/stream")) + is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg) is ClientMessage.ChatInput -> Unit // handler deferred to subsequent task } } + private fun handleApprovalResponse(msg: ClientMessage.ApprovalResponse) { + val domainDecision = msg.toDomainDecision(msg.requestId.value) + runCatching { module.orchestrator.submitApprovalDecision(msg.requestId, domainDecision) } + .onFailure { + log.warn( + "submitApprovalDecision failed for request={}: {}", + msg.requestId.value, + it.message, + ) + } + } + + private fun ClientMessage.ApprovalResponse.toDomainDecision(sessionIdValue: String): DomainApprovalDecision { + val outcome = when (decision) { + ApprovalDecision.APPROVE -> ApprovalOutcome.APPROVED + ApprovalDecision.REJECT -> ApprovalOutcome.REJECTED + ApprovalDecision.STEER -> ApprovalOutcome.REJECTED + } + val scopeSessionId: SessionId = TypeId(sessionIdValue) + val identity = ApprovalScopeIdentity(sessionId = scopeSessionId, stageId = null, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val steering = steeringNote?.let { + UserSteering(text = it, sessionId = scopeSessionId, timestamp = Clock.System.now()) + } + return DomainApprovalDecision( + id = null, + requestId = requestId, + outcome = outcome, + state = ApprovalStatus.COMPLETED, + tier = Tier.T2, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = steeringNote, + userSteering = steering, + ) + } + private fun encodeError(message: String): Frame.Text = Frame.Text( ProtocolSerializer.encodeServerMessage( diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt index 3fdb1ec5..370c2bd6 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/ws/TuiWsClient.kt @@ -2,7 +2,6 @@ package com.correx.apps.tui.ws import com.correx.apps.server.protocol.ClientMessage import com.correx.apps.server.protocol.ServerMessage -import com.correx.core.events.types.SessionId import io.ktor.client.HttpClient import io.ktor.client.engine.cio.CIO import io.ktor.client.plugins.logging.LogLevel @@ -14,13 +13,10 @@ import io.ktor.websocket.Frame import io.ktor.websocket.WebSocketSession import io.ktor.websocket.readText import io.ktor.websocket.send -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Job import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.receiveAsFlow -import kotlinx.coroutines.launch import kotlinx.serialization.encodeToString import kotlinx.serialization.json.Json @@ -43,8 +39,6 @@ class TuiWsClient( } } - private var sessionStreamJob: Job? = null - private var sessionWsSession: WebSocketSession? = null private var session: WebSocketSession? = null private val _messages = Channel(Channel.UNLIMITED) @@ -55,16 +49,10 @@ class TuiWsClient( suspend fun send(message: ClientMessage) { runCatching { - // approval responses go to session stream, everything else to global - val target = when (message) { - is ClientMessage.ApprovalResponse -> sessionWsSession ?: session - else -> session - } - target?.send(json.encodeToString(message)) + session?.send(json.encodeToString(message)) } } - suspend fun connect() { var delayMs = INITIAL_RETRY_DELAY_MS var attempt = 0 @@ -96,39 +84,6 @@ class TuiWsClient( } } - fun connectSession(sessionId: SessionId, scope: CoroutineScope) { - sessionStreamJob?.cancel() - sessionStreamJob = scope.launch { - runCatching { - client.webSocket( - method = HttpMethod.Get, - host = host, - port = port, - path = "/sessions/${sessionId.value}/stream", - ) { - sessionWsSession = this - for (frame in incoming) { - if (frame is Frame.Text) { - runCatching { - json.decodeFromString(frame.readText()) - }.onSuccess { msg -> - _messages.send(msg) - } - } - } - } - }.also { - sessionWsSession = null - } - } - } - - fun disconnectSession() { - sessionStreamJob?.cancel() - sessionStreamJob = null - sessionWsSession = null - } - fun close() { client.close() } diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/ApprovalResponseGlobalSocketRoundTripTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/ApprovalResponseGlobalSocketRoundTripTest.kt new file mode 100644 index 00000000..8bc2fc7d --- /dev/null +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/ws/ApprovalResponseGlobalSocketRoundTripTest.kt @@ -0,0 +1,59 @@ +package com.correx.apps.tui.ws + +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ProtocolSerializer +import com.correx.core.events.types.ApprovalRequestId +import kotlinx.serialization.encodeToString +import kotlinx.serialization.json.Json +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertInstanceOf +import org.junit.jupiter.api.Test + +/** + * Per tui-02 invariant 5: all ClientMessage types (including ApprovalResponse) are sent over + * the single global socket. This test asserts that the JSON encoding used by TuiWsClient.send() + * round-trips through the server-side global socket decoder (ProtocolSerializer.decodeClientMessage) + * as an ApprovalResponse. + */ +class ApprovalResponseGlobalSocketRoundTripTest { + + // Mirrors the Json config in TuiWsClient + private val clientJson = Json { + classDiscriminator = "type" + ignoreUnknownKeys = true + } + + @Test + fun `ApprovalResponse encoded by TUI client decodes on the global socket`() { + val original: ClientMessage = ClientMessage.ApprovalResponse( + requestId = ApprovalRequestId("req-123"), + decision = ApprovalDecision.APPROVE, + steeringNote = null, + ) + + val wire = clientJson.encodeToString(original) + val decoded = ProtocolSerializer.decodeClientMessage(wire) + + assertInstanceOf(ClientMessage.ApprovalResponse::class.java, decoded) + val approval = decoded as ClientMessage.ApprovalResponse + assertEquals("req-123", approval.requestId.value) + assertEquals(ApprovalDecision.APPROVE, approval.decision) + assertEquals(null, approval.steeringNote) + } + + @Test + fun `ApprovalResponse with steering note round-trips`() { + val original: ClientMessage = ClientMessage.ApprovalResponse( + requestId = ApprovalRequestId("req-steer"), + decision = ApprovalDecision.STEER, + steeringNote = "please retry with smaller batch", + ) + + val wire = clientJson.encodeToString(original) + val decoded = ProtocolSerializer.decodeClientMessage(wire) as ClientMessage.ApprovalResponse + + assertEquals(ApprovalDecision.STEER, decoded.decision) + assertEquals("please retry with smaller batch", decoded.steeringNote) + } +}