fix(tui-02): route ApprovalResponse through global socket

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.
This commit is contained in:
2026-05-25 17:18:01 +04:00
parent 8f20f6aa24
commit c8a599c64f
3 changed files with 107 additions and 48 deletions
@@ -3,10 +3,17 @@ package com.correx.apps.server.ws
import com.correx.apps.server.ServerModule import com.correx.apps.server.ServerModule
import com.correx.apps.server.bridge.DomainEventMapper import com.correx.apps.server.bridge.DomainEventMapper
import com.correx.apps.server.bridge.SessionEventBridge 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.ClientMessage
import com.correx.apps.server.protocol.ProtocolSerializer import com.correx.apps.server.protocol.ProtocolSerializer
import com.correx.apps.server.protocol.ProviderHealthDto import com.correx.apps.server.protocol.ProviderHealthDto
import com.correx.apps.server.protocol.ServerMessage 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.EventMetadata
import com.correx.core.events.events.NewEvent import com.correx.core.events.events.NewEvent
import com.correx.core.events.events.StoredEvent 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.EventId
import com.correx.core.events.types.SessionId import com.correx.core.events.types.SessionId
import com.correx.core.inference.ProviderHealth import com.correx.core.inference.ProviderHealth
import com.correx.core.sessions.ApprovalMode
import com.correx.core.utils.TypeId import com.correx.core.utils.TypeId
import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision
import io.ktor.server.websocket.DefaultWebSocketServerSession import io.ktor.server.websocket.DefaultWebSocketServerSession
import io.ktor.websocket.Frame import io.ktor.websocket.Frame
import io.ktor.websocket.readText import io.ktor.websocket.readText
@@ -121,12 +130,48 @@ class GlobalStreamHandler(private val module: ServerModule) {
is ClientMessage.StartSession -> handleStartSession(session, msg, sendFrame) is ClientMessage.StartSession -> handleStartSession(session, msg, sendFrame)
is ClientMessage.CancelSession -> module.orchestrator.cancel(msg.sessionId) is ClientMessage.CancelSession -> module.orchestrator.cancel(msg.sessionId)
is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported")) is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported"))
is ClientMessage.ApprovalResponse -> is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg)
session.send(encodeError("ApprovalResponse must be sent to /sessions/{id}/stream"))
is ClientMessage.ChatInput -> Unit // handler deferred to subsequent task 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 = private fun encodeError(message: String): Frame.Text =
Frame.Text( Frame.Text(
ProtocolSerializer.encodeServerMessage( ProtocolSerializer.encodeServerMessage(
@@ -2,7 +2,6 @@ package com.correx.apps.tui.ws
import com.correx.apps.server.protocol.ClientMessage import com.correx.apps.server.protocol.ClientMessage
import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.server.protocol.ServerMessage
import com.correx.core.events.types.SessionId
import io.ktor.client.HttpClient import io.ktor.client.HttpClient
import io.ktor.client.engine.cio.CIO import io.ktor.client.engine.cio.CIO
import io.ktor.client.plugins.logging.LogLevel 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.WebSocketSession
import io.ktor.websocket.readText import io.ktor.websocket.readText
import io.ktor.websocket.send import io.ktor.websocket.send
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.delay import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.receiveAsFlow import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.launch
import kotlinx.serialization.encodeToString import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json 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 var session: WebSocketSession? = null
private val _messages = Channel<ServerMessage>(Channel.UNLIMITED) private val _messages = Channel<ServerMessage>(Channel.UNLIMITED)
@@ -55,16 +49,10 @@ class TuiWsClient(
suspend fun send(message: ClientMessage) { suspend fun send(message: ClientMessage) {
runCatching { runCatching {
// approval responses go to session stream, everything else to global session?.send(json.encodeToString(message))
val target = when (message) {
is ClientMessage.ApprovalResponse -> sessionWsSession ?: session
else -> session
}
target?.send(json.encodeToString(message))
} }
} }
suspend fun connect() { suspend fun connect() {
var delayMs = INITIAL_RETRY_DELAY_MS var delayMs = INITIAL_RETRY_DELAY_MS
var attempt = 0 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<ServerMessage>(frame.readText())
}.onSuccess { msg ->
_messages.send(msg)
}
}
}
}
}.also {
sessionWsSession = null
}
}
}
fun disconnectSession() {
sessionStreamJob?.cancel()
sessionStreamJob = null
sessionWsSession = null
}
fun close() { fun close() {
client.close() client.close()
} }
@@ -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)
}
}