diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt index 76ae7921..d1da14bb 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/Main.kt @@ -131,7 +131,9 @@ fun main() { routerFacade = routerFacade, orchestrationRepository = repositories.orchestrationRepository, approvalRepository = repositories.approvalRepository, + approvalConfig = correxConfig.approval, ) + module.start() log.info("==============================") embeddedServer(Netty, port = 8080) { configureServer(module) }.start(wait = true) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index e6447520..862ca1d5 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt @@ -1,17 +1,27 @@ package com.correx.apps.server +import com.correx.apps.server.approval.ApprovalCoordinator import com.correx.apps.server.registry.ProviderRegistry import com.correx.apps.server.registry.WorkflowRegistry import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.config.ApprovalConfig +import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.stores.EventStore import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator import com.correx.core.kernel.orchestration.OrchestrationConfig import com.correx.core.kernel.orchestration.OrchestrationRepository import com.correx.core.router.RouterFacade import com.correx.core.sessions.DefaultSessionRepository +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.flow.filter +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.flow.launchIn -data class ServerModule( +class ServerModule( val orchestrator: DefaultSessionOrchestrator, val eventStore: EventStore, val artifactStore: ArtifactStore, @@ -22,4 +32,30 @@ data class ServerModule( val routerFacade: RouterFacade, val orchestrationRepository: OrchestrationRepository, val approvalRepository: DefaultApprovalRepository, -) + approvalConfig: ApprovalConfig = ApprovalConfig(), + // Long-lived scope owned by the module — backs the ApprovalCoordinator timers + // and event-store subscription. SupervisorJob so one failure doesn't kill the + // whole module; Dispatchers.Default since the work is non-blocking and CPU-light. + val moduleScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default), + approvalCoordinator: ApprovalCoordinator? = null, +) { + val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator( + orchestrator = orchestrator, + config = approvalConfig, + scope = moduleScope, + ) + + private var subscriptionJob: Job? = null + + /** + * Subscribe to the event store and forward [ApprovalRequestedEvent] payloads to + * the [ApprovalCoordinator]. Idempotent: calling twice is a no-op for the second call. + */ + fun start() { + if (subscriptionJob != null) return + subscriptionJob = eventStore.subscribeAll() + .filter { it.payload is ApprovalRequestedEvent } + .onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) } + .launchIn(moduleScope) + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt deleted file mode 100644 index bc161ec6..00000000 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalConfig.kt +++ /dev/null @@ -1,5 +0,0 @@ -package com.correx.apps.server.approval - -data class ApprovalConfig( - val timeoutMs: Long = 300_000L, -) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt index dc9554a6..264db41c 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalCoordinator.kt @@ -1,6 +1,5 @@ package com.correx.apps.server.approval -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.RiskSummaryDto @@ -8,9 +7,9 @@ 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.config.ApprovalConfig import com.correx.core.events.events.ApprovalRequestedEvent import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId @@ -24,16 +23,17 @@ import kotlinx.coroutines.Job import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.datetime.Clock -import java.util.concurrent.* +import java.util.concurrent.ConcurrentHashMap import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision -class ApprovalCoordinator( +open class ApprovalCoordinator( private val orchestrator: ApprovalGateway, private val config: ApprovalConfig, - private val scope: CoroutineScope, + val scope: CoroutineScope, ) { private val sessionClients: ConcurrentHashMap> = ConcurrentHashMap() + private val globalClients: MutableSet = ConcurrentHashMap.newKeySet() private val resolved: ConcurrentHashMap = ConcurrentHashMap() private val timeoutJobs: ConcurrentHashMap = ConcurrentHashMap() @@ -45,6 +45,14 @@ class ApprovalCoordinator( sessionClients[sessionId]?.remove(session) } + fun registerGlobalClient(session: DefaultWebSocketServerSession) { + globalClients.add(session) + } + + fun unregisterGlobalClient(session: DefaultWebSocketServerSession) { + globalClients.remove(session) + } + suspend fun onApprovalRequested(event: ApprovalRequestedEvent) { val msg = ServerMessage.ApprovalRequired( sessionId = event.sessionId, @@ -64,7 +72,7 @@ class ApprovalCoordinator( scheduleTimeout(event.requestId, event.sessionId, event.stageId, event.tier) } - fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { + open fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { if (resolved.putIfAbsent(msg.requestId, true) != null) { return ServerMessage.ProtocolError( message = "Approval request ${msg.requestId.value} already resolved", @@ -73,7 +81,7 @@ class ApprovalCoordinator( ) } timeoutJobs.remove(msg.requestId)?.cancel() - val domain = msg.toDomainDecision(sessionId, null, Tier.T2) + val domain = msg.toDomain(sessionId, null, Tier.T2) return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) } .fold( onSuccess = { null }, @@ -89,7 +97,9 @@ class ApprovalCoordinator( private suspend fun broadcast(sessionId: SessionId, msg: ServerMessage) { val encoded = ProtocolSerializer.encodeServerMessage(msg) - sessionClients[sessionId]?.forEach { client -> + val sessionSubs = sessionClients[sessionId].orEmpty() + val recipients: Set = sessionSubs + globalClients + recipients.forEach { client -> runCatching { client.send(Frame.Text(encoded)) } } } @@ -119,34 +129,4 @@ class ApprovalCoordinator( } timeoutJobs[requestId] = job } - - private fun ClientMessage.ApprovalResponse.toDomainDecision( - sessionId: SessionId, - stageId: StageId?, - tier: Tier, - ): DomainApprovalDecision { - val outcome = when (decision) { - ApprovalDecision.APPROVE -> ApprovalOutcome.APPROVED - ApprovalDecision.REJECT -> ApprovalOutcome.REJECTED - ApprovalDecision.STEER -> ApprovalOutcome.APPROVED - } - val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = stageId, projectId = null) - val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) - val steering = if (decision == ApprovalDecision.STEER) { - steeringNote?.let { UserSteering(text = it, sessionId = sessionId, timestamp = Clock.System.now()) } - } else { - null - } - return DomainApprovalDecision( - id = null, - requestId = requestId, - outcome = outcome, - state = ApprovalStatus.COMPLETED, - tier = tier, - contextSnapshot = context, - resolutionTimestamp = Clock.System.now(), - reason = steeringNote, - userSteering = steering, - ) - } } diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalResponseMapper.kt b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalResponseMapper.kt new file mode 100644 index 00000000..e279cdf2 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/approval/ApprovalResponseMapper.kt @@ -0,0 +1,51 @@ +package com.correx.apps.server.approval + +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +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.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.sessions.ApprovalMode +import kotlinx.datetime.Clock +import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision + +/** + * Translates a wire-level [ClientMessage.ApprovalResponse] into a domain [DomainApprovalDecision]. + * + * Outcome is driven purely by [ClientMessage.ApprovalResponse.decision]: + * APPROVE -> APPROVED, REJECT -> REJECTED. + * + * A non-null [ClientMessage.ApprovalResponse.steeringNote] attaches a [UserSteering] regardless of + * the approve/reject decision — the user may steer alongside either choice. + */ +internal fun ClientMessage.ApprovalResponse.toDomain( + sessionId: SessionId, + stageId: StageId?, + tier: Tier, +): DomainApprovalDecision { + val outcome = when (decision) { + ApprovalDecision.APPROVE -> ApprovalOutcome.APPROVED + ApprovalDecision.REJECT -> ApprovalOutcome.REJECTED + } + val identity = ApprovalScopeIdentity(sessionId = sessionId, stageId = stageId, projectId = null) + val context = ApprovalContext(identity = identity, mode = ApprovalMode.PROMPT) + val steering = steeringNote?.let { + UserSteering(text = it, sessionId = sessionId, timestamp = Clock.System.now()) + } + return DomainApprovalDecision( + id = null, + requestId = requestId, + outcome = outcome, + state = ApprovalStatus.COMPLETED, + tier = tier, + contextSnapshot = context, + resolutionTimestamp = Clock.System.now(), + reason = steeringNote, + userSteering = steering, + ) +} 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 4ed64b62..b8d10f6f 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 @@ -9,7 +9,6 @@ import kotlinx.serialization.Serializable enum class ApprovalDecision { APPROVE, REJECT, - STEER, } @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 29e19b8f..55d5f212 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,17 +3,10 @@ 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 @@ -22,9 +15,7 @@ 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 @@ -62,6 +53,8 @@ class GlobalStreamHandler(private val module: ServerModule) { ) val mapper = DomainEventMapper(module.artifactStore) + module.approvalCoordinator.registerGlobalClient(session) + sendInitialSnapshot(session) // Launch global stream (snapshot + live-forward) concurrently with incoming frame loop @@ -95,6 +88,8 @@ class GlobalStreamHandler(private val module: ServerModule) { } } catch (_: ClosedReceiveChannelException) { log.info("client disconnected") + } finally { + module.approvalCoordinator.unregisterGlobalClient(session) } } @@ -130,46 +125,17 @@ 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 -> handleApprovalResponse(msg) + is ClientMessage.ApprovalResponse -> handleApprovalResponse(msg, sendFrame) 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 suspend fun handleApprovalResponse( + msg: ClientMessage.ApprovalResponse, + sendFrame: suspend (ServerMessage) -> Unit, + ) { + val scopeSessionId: SessionId = TypeId(msg.requestId.value) + module.approvalCoordinator.handleResponse(msg, scopeSessionId)?.let { sendFrame(it) } } 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 1f51767a..f3e0dc5e 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 @@ -2,27 +2,16 @@ package com.correx.apps.server.ws import com.correx.apps.server.ServerModule import com.correx.apps.server.bridge.DomainEventMapper -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.ServerMessage import com.correx.apps.server.protocol.ServerMessage.RouterResponseMessage -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.types.SessionId -import com.correx.core.sessions.ApprovalMode -import com.correx.core.utils.TypeId import io.ktor.server.websocket.DefaultWebSocketServerSession import io.ktor.websocket.Frame import io.ktor.websocket.readText import kotlinx.coroutines.channels.ClosedReceiveChannelException -import kotlinx.datetime.Clock import org.slf4j.LoggerFactory -import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision private val log = LoggerFactory.getLogger(SessionStreamHandler::class.java) @@ -46,6 +35,7 @@ class SessionStreamHandler(private val module: ServerModule) { } } + module.approvalCoordinator.registerClient(sessionId, session) try { for (frame in session.incoming) { if (frame is Frame.Text) { @@ -53,7 +43,7 @@ class SessionStreamHandler(private val module: ServerModule) { runCatching { ProtocolSerializer.decodeClientMessage(text) } .onSuccess { msg -> log.debug("recv: {}", msg::class.simpleName) - handleClientMessage(session, msg) + handleClientMessage(session, sessionId, msg) } .onFailure { log.warn("decode error: {}", it.message) @@ -64,10 +54,16 @@ class SessionStreamHandler(private val module: ServerModule) { } } catch (_: ClosedReceiveChannelException) { log.info("client disconnected session={}", sessionId.value) + } finally { + module.approvalCoordinator.unregisterClient(sessionId, session) } } - private suspend fun handleClientMessage(session: DefaultWebSocketServerSession, msg: ClientMessage) { + private suspend fun handleClientMessage( + session: DefaultWebSocketServerSession, + sessionId: SessionId, + msg: ClientMessage, + ) { when (msg) { is ClientMessage.Ping -> Unit is ClientMessage.CancelSession -> { @@ -76,15 +72,9 @@ class SessionStreamHandler(private val module: ServerModule) { } is 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, - ) - } + module.approvalCoordinator.handleResponse(msg, sessionId)?.let { reply -> + session.send(Frame.Text(ProtocolSerializer.encodeServerMessage(reply))) + } } is ClientMessage.ChatInput -> { @@ -121,28 +111,4 @@ class SessionStreamHandler(private val module: ServerModule) { } } - 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, - ) - } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt new file mode 100644 index 00000000..2bc67302 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/approval/ApprovalCoordinatorWiringTest.kt @@ -0,0 +1,178 @@ +package com.correx.apps.server.approval + +import com.correx.apps.server.protocol.ApprovalDecision +import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ServerMessage +import com.correx.core.approvals.Tier +import com.correx.core.config.ApprovalConfig +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.RiskSummaryId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.ValidationReportId +import com.correx.core.kernel.orchestration.ApprovalGateway +import com.correx.core.approvals.model.ApprovalDecision as DomainApprovalDecision +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.filter +import kotlinx.coroutines.flow.launchIn +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.yield +import kotlinx.coroutines.withTimeout +import kotlinx.datetime.Instant +import org.junit.jupiter.api.AfterEach +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertInstanceOf +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import java.util.concurrent.CopyOnWriteArrayList + +class ApprovalCoordinatorWiringTest { + + private val sessionId = SessionId("session-1") + private val requestId = ApprovalRequestId("req-1") + private val timestamp = Instant.parse("2026-01-01T00:00:00Z") + + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + + @AfterEach + fun tearDown() { + scope.cancel() + } + + private class RecordingGateway : ApprovalGateway { + val submissions = CopyOnWriteArrayList>() + override fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) { + submissions.add(requestId to decision) + } + } + + private fun storedEvent(payload: EventPayload, seq: Long): StoredEvent = StoredEvent( + metadata = EventMetadata( + eventId = EventId("evt-$seq"), + sessionId = sessionId, + timestamp = timestamp, + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = seq, + sessionSequence = seq, + payload = payload, + ) + + private fun fakeEventStore(liveFlow: MutableSharedFlow): EventStore = object : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = emptyList() + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = emptyList() + override fun lastSequence(sessionId: SessionId): Long = 0L + override fun subscribe(sessionId: SessionId): Flow = liveFlow + override fun allEvents(): Sequence = emptySequence() + override fun allSessionIds(): Set = emptySet() + override fun subscribeAll(): Flow = liveFlow + override suspend fun lastGlobalSequence(): Long = 0L + } + + @Test + fun `handleResponse forwards to gateway and returns null on success`() { + val gateway = RecordingGateway() + val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) + val msg = ClientMessage.ApprovalResponse( + requestId = requestId, + decision = ApprovalDecision.APPROVE, + steeringNote = null, + ) + + val reply = coord.handleResponse(msg, sessionId) + + assertNull(reply) + assertEquals(1, gateway.submissions.size) + assertEquals(requestId, gateway.submissions[0].first) + } + + @Test + fun `duplicate ApprovalResponse for same requestId returns ProtocolError`() { + val gateway = RecordingGateway() + val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) + val msg = ClientMessage.ApprovalResponse( + requestId = requestId, + decision = ApprovalDecision.APPROVE, + steeringNote = null, + ) + + val first = coord.handleResponse(msg, sessionId) + val second = coord.handleResponse(msg, sessionId) + + assertNull(first) + assertNotNull(second) + assertInstanceOf(ServerMessage.ProtocolError::class.java, second) + // Gateway only sees the first submission; dedupe blocks the second. + assertEquals(1, gateway.submissions.size) + } + + /** + * Proves the ServerModule.start() subscription wiring: an [ApprovalRequestedEvent] flowing + * through the event store reaches the coordinator's onApprovalRequested. We verify this by + * subscribing the same wiring the module uses and checking the coordinator's behavior + * (it schedules a timeout job and dedupes the requestId). + * + * After the event is observed, a subsequent handleResponse with a matching requestId should + * succeed once (the coordinator's internal `resolved` map was NOT pre-populated by the event). + * The behavioral assertion: the gateway sees the decision exactly once. + */ + /** + * Proves the ServerModule.start() subscription wiring shape: an [ApprovalRequestedEvent] + * flowing through the event store reaches the wired-up callback, and non-approval events + * are filtered out. + */ + @Test + fun `ApprovalRequestedEvent in store reaches coordinator via subscription`() = runBlocking { + val liveFlow = MutableSharedFlow(extraBufferCapacity = 16) + val store = fakeEventStore(liveFlow) + + val triggered = CompletableDeferred() + // Mirror the exact wiring used in ServerModule.start(). + val subscription = store.subscribeAll() + .filter { it.payload is ApprovalRequestedEvent } + .onEach { triggered.complete(it.payload as ApprovalRequestedEvent) } + .launchIn(scope) + + // Wait for subscription to attach so emit isn't dropped by the SharedFlow (replay=0). + while (liveFlow.subscriptionCount.value == 0) { + yield() + } + + val event = ApprovalRequestedEvent( + requestId = requestId, + tier = Tier.T2, + validationReportId = ValidationReportId("vr-1"), + riskSummaryId = RiskSummaryId("rs-1"), + sessionId = sessionId, + stageId = null, + projectId = null, + ) + // Non-approval event must be filtered out. + liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("stage-x")), seq = 1L)) + liveFlow.emit(storedEvent(event, seq = 2L)) + + val received = withTimeout(2_000L) { triggered.await() } + assertEquals(requestId, received.requestId) + subscription.cancel() + } +} diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt index 7b9781e1..4a14b340 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/InputReducer.kt @@ -69,7 +69,7 @@ object InputReducer { Effect.SendWs( ClientMessage.ApprovalResponse( requestId = ApprovalRequestId(active.requestId), - decision = ApprovalDecision.STEER, + decision = ApprovalDecision.APPROVE, steeringNote = text, ), ), diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/InputReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/InputReducerTest.kt index 2d1519d7..3f6df480 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/InputReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/InputReducerTest.kt @@ -132,7 +132,7 @@ class InputReducerTest { } @Test - fun `SubmitInput in STEER mode with active approval emits ApprovalResponse STEER`() { + fun `SubmitInput in STEER mode with active approval emits ApprovalResponse APPROVE with steering note`() { val (state, effects) = reduce( state = stateWithActiveApproval(inputMode = InputMode.STEER, inputBuffer = "steer this way"), action = Action.SubmitInput, @@ -141,7 +141,7 @@ class InputReducerTest { assertEquals(1, effects.size) val effect = effects[0] as Effect.SendWs val msg = effect.message as ClientMessage.ApprovalResponse - assertEquals(ApprovalDecision.STEER, msg.decision) + assertEquals(ApprovalDecision.APPROVE, msg.decision) assertEquals("steer this way", msg.steeringNote) assertEquals("req-1", msg.requestId.value) } @@ -155,7 +155,7 @@ class InputReducerTest { assertEquals(null, state.sessions.sessions[0].pendingApproval) assertEquals(1, effects.size) val msg = (effects[0] as Effect.SendWs).message as ClientMessage.ApprovalResponse - assertEquals(ApprovalDecision.STEER, msg.decision) + assertEquals(ApprovalDecision.APPROVE, msg.decision) assertEquals("go left", msg.steeringNote) assertEquals("req-1", msg.requestId.value) } 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 index 8bc2fc7d..5a0f2f91 100644 --- 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 @@ -46,14 +46,14 @@ class ApprovalResponseGlobalSocketRoundTripTest { fun `ApprovalResponse with steering note round-trips`() { val original: ClientMessage = ClientMessage.ApprovalResponse( requestId = ApprovalRequestId("req-steer"), - decision = ApprovalDecision.STEER, + decision = ApprovalDecision.APPROVE, 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(ApprovalDecision.APPROVE, decoded.decision) assertEquals("please retry with smaller batch", decoded.steeringNote) } }