From 766b91081a3404f97f3728a2af63ef9eb5de0799 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 26 May 2026 21:12:26 +0400 Subject: [PATCH] fix: unblock sessions stuck in PAUSED state after approval resolved MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sessions that had an approval resolved (or never needed one) but never received OrchestrationResumedEvent would show the last tool as RUNNING forever with no subsequent events. Three coordinated fixes: 1. SessionOrchestrator now emits OrchestrationResumedEvent in both the approved and rejected approval branches, so the paused flag clears correctly going forward. 2. SessionEventBridge.replaySnapshot() detects the stuck pattern (pendingApproval=true with no unresolved requests) and appends a synthetic OrchestrationResumedEvent so historical sessions self-heal on next TUI connect. 3. DomainEventMapper maps OrchestrationResumedEvent → SessionResumed; TUI SessionsReducer/SnapshotPhaseReducer handle the new message, clearing pendingApproval and setting status back to ACTIVE. Also adds server-restart recovery path in DefaultSessionOrchestrator (emit decision + resume events directly when no live deferred exists) and pre-registers pending approvals in ServerModule.start() to close the race between reconnect and ApprovalResponse routing. --- .../com/correx/apps/server/ServerModule.kt | 23 ++++++++ .../server/approval/ApprovalCoordinator.kt | 2 +- .../apps/server/bridge/DomainEventMapper.kt | 7 +++ .../apps/server/bridge/SessionEventBridge.kt | 53 +++++++++++++++---- .../apps/server/protocol/ServerMessage.kt | 9 ++++ .../approval/ApprovalCoordinatorWiringTest.kt | 8 +-- .../apps/tui/reducer/SessionsReducer.kt | 16 ++++++ .../apps/tui/reducer/SnapshotPhaseReducer.kt | 1 + .../kernel/orchestration/ApprovalGateway.kt | 2 +- .../DefaultSessionOrchestrator.kt | 36 +++++++++++-- .../orchestration/SessionOrchestrator.kt | 2 + 11 files changed, 140 insertions(+), 19 deletions(-) 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 cd971900..ac742a09 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 @@ -3,6 +3,8 @@ 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.ApprovalProjector +import com.correx.core.approvals.DefaultApprovalReducer import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.config.ApprovalConfig @@ -52,12 +54,33 @@ class ServerModule( /** * Subscribe to the event store and forward [ApprovalRequestedEvent] payloads to * the [ApprovalCoordinator]. Idempotent: calling twice is a no-op for the second call. + * + * Also pre-populates the coordinator's session lookup map from any approvals that were + * pending when the server last stopped. This prevents a race where a reconnecting TUI + * sends an ApprovalResponse before the per-connection replaySnapshot() has registered + * the session mapping. */ fun start() { if (subscriptionJob != null) return + preRegisterPendingApprovals() subscriptionJob = eventStore.subscribeAll() .filter { it.payload is ApprovalRequestedEvent } .onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) } .launchIn(moduleScope) } + + private fun preRegisterPendingApprovals() { + val projector = ApprovalProjector(DefaultApprovalReducer()) + eventStore.allSessionIds().forEach { sessionId -> + val orchState = orchestrationRepository.getState(sessionId) + if (!orchState.pendingApproval) return@forEach + val events = eventStore.read(sessionId) + val approvalState = events.fold(projector.initial()) { state, event -> + projector.apply(state, event) + } + approvalState.requests.values + .filter { req -> approvalState.decisions.values.none { it.requestId == req.id } } + .forEach { req -> approvalCoordinator.registerPendingRequest(req.id, sessionId) } + } + } } 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 3eec7fa6..e305b5c7 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 @@ -85,7 +85,7 @@ open class ApprovalCoordinator( requestSessions[requestId] = sessionId } - open fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? { + open suspend 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", 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 4f113b79..e6e984a8 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 @@ -10,6 +10,7 @@ import com.correx.core.events.events.WorkflowStartedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceTimeoutEvent import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StoredEvent @@ -90,6 +91,12 @@ suspend fun domainEventToServerMessage( ) is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence) + is OrchestrationResumedEvent -> ServerMessage.SessionResumed( + sessionId = p.sessionId, + stageId = p.stageId, + sequence = seq, + sessionSequence = sessionSequence, + ) is InferenceStartedEvent -> ServerMessage.InferenceStarted( sessionId = p.sessionId, stageId = p.stageId, diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt index 93e359af..850b8101 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/bridge/SessionEventBridge.kt @@ -12,10 +12,13 @@ import com.correx.core.approvals.ApprovalProjector import com.correx.core.approvals.DefaultApprovalReducer import com.correx.core.approvals.model.ApprovalState import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.EventMetadata import com.correx.core.events.events.InferenceCompletedEvent import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceTimeoutEvent +import com.correx.core.events.events.NewEvent import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StoredEvent @@ -27,6 +30,8 @@ import com.correx.core.events.events.TransitionExecutedEvent import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.orchestration.OrchestrationStatus +import com.correx.core.events.types.EventId +import kotlinx.datetime.Clock import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.SessionId @@ -62,18 +67,46 @@ class SessionEventBridge( } } else null - val pendingApprovals = approvalState?.requests?.values - ?.filter { req -> approvalState.decisions.values.none { it.requestId == req.id } } - ?.sortedWith(compareBy({ it.timestamp }, { it.id.value })) - ?.map { - ApprovalDto( - requestId = it.id.value, - tier = it.tier.name, - toolName = it.toolName, - preview = it.preview, + val pendingApprovalRequests = approvalState?.let { state -> + state.requests.values + .filter { req -> state.decisions.values.none { it.requestId == req.id } } + .sortedWith(compareBy({ it.timestamp }, { it.id.value })) + } ?: emptyList() + + // Fix stuck sessions: orchestration says PAUSED + pendingApproval but there are no + // actual unresolved approval requests. This happens when approval was resolved but + // OrchestrationResumedEvent was not emitted (pre-Feb-13 orchestrator bug). Append the + // missing event permanently so the session state corrects on all future replays. + if (orchState.pendingApproval && pendingApprovalRequests.isEmpty()) { + val alreadyResumed = events.any { it.payload is OrchestrationResumedEvent } + val stageId = orchState.currentStageId + if (!alreadyResumed && stageId != null) { + val resumeEvent = NewEvent( + metadata = EventMetadata( + eventId = EventId(java.util.UUID.randomUUID().toString()), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + payload = OrchestrationResumedEvent( + sessionId = sessionId, + stageId = stageId, + ), ) + eventStore.append(resumeEvent) } - ?: emptyList() + } + + val pendingApprovals = pendingApprovalRequests.map { + ApprovalDto( + requestId = it.id.value, + tier = it.tier.name, + toolName = it.toolName, + preview = it.preview, + ) + } val recentEvents = events .mapNotNull { eventToEntry(it) } 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 585fd2c5..7ae50111 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 @@ -53,6 +53,15 @@ sealed interface ServerMessage { override val sessionSequence: Long, ) : ServerMessage, SessionMessage + @Serializable + @SerialName("session.resumed") + data class SessionResumed( + val sessionId: SessionId, + val stageId: StageId? = null, + override val sequence: Long, + override val sessionSequence: Long, + ) : ServerMessage, SessionMessage + @Serializable @SerialName("session.completed") data class SessionCompleted( 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 index 50e70c1f..38ecb0dd 100644 --- 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 @@ -57,7 +57,7 @@ class ApprovalCoordinatorWiringTest { private class RecordingGateway : ApprovalGateway { val submissions = CopyOnWriteArrayList>() - override fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) { + override suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) { submissions.add(requestId to decision) } } @@ -90,7 +90,7 @@ class ApprovalCoordinatorWiringTest { } @Test - fun `handleResponse forwards to gateway and returns null on success`() { + fun `handleResponse forwards to gateway and returns null on success`(): Unit = runBlocking { val gateway = RecordingGateway() val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val msg = ClientMessage.ApprovalResponse( @@ -107,7 +107,7 @@ class ApprovalCoordinatorWiringTest { } @Test - fun `duplicate ApprovalResponse for same requestId returns ProtocolError`() { + fun `duplicate ApprovalResponse for same requestId returns ProtocolError`(): Unit = runBlocking { val gateway = RecordingGateway() val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val msg = ClientMessage.ApprovalResponse( @@ -177,7 +177,7 @@ class ApprovalCoordinatorWiringTest { } @Test - fun `lookupSession returns mapping registered by onApprovalRequested and cleared by handleResponse`() = runBlocking { + fun `lookupSession returns mapping registered by onApprovalRequested and cleared by handleResponse`(): Unit = runBlocking { val gateway = RecordingGateway() val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val otherSessionId = SessionId("other-session") diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt index e846f004..606e4974 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt @@ -165,6 +165,7 @@ object SessionsReducer { val (result, effects) = when (msg) { is ServerMessage.SessionStarted -> processSessionStartedMessage(msg, clock, sessions) is ServerMessage.SessionPaused -> processSessionPausedMessage(msg, sessions, clock) + is ServerMessage.SessionResumed -> processSessionResumedMessage(msg, sessions, clock) is ServerMessage.SessionCompleted -> processSessionCompletedMessage(sessions, msg, clock) is ServerMessage.SessionFailed -> processSessionFailedMessage(sessions, msg, clock) is ServerMessage.StageStarted -> processStageStartedMessage(msg, sessions) @@ -189,6 +190,7 @@ object SessionsReducer { private fun sessionIdFromMessage(msg: ServerMessage): String? = when (msg) { is ServerMessage.SessionStarted -> msg.sessionId.value is ServerMessage.SessionPaused -> msg.sessionId.value + is ServerMessage.SessionResumed -> msg.sessionId.value is ServerMessage.SessionCompleted -> msg.sessionId.value is ServerMessage.SessionFailed -> msg.sessionId.value is ServerMessage.StageStarted -> msg.sessionId.value @@ -507,6 +509,20 @@ object SessionsReducer { ) to emptyList() } + private fun processSessionResumedMessage( + msg: ServerMessage.SessionResumed, + sessions: SessionsState, + clock: () -> Long, + ): Pair> = sessions.copy( + sessions = sessions.sessions.map { s -> + if (s.id == msg.sessionId.value) s.copy( + status = "ACTIVE", + pendingApproval = null, + lastEventAt = clock(), + ) else s + }, + ) to emptyList() + private fun processSessionStartedMessage( msg: ServerMessage.SessionStarted, clock: () -> Long, diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt index 1ba264d7..145ea6d5 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SnapshotPhaseReducer.kt @@ -64,6 +64,7 @@ object SnapshotPhaseReducer { internal fun sessionIdOf(msg: ServerMessage): String? = when (msg) { is ServerMessage.SessionStarted -> msg.sessionId.value is ServerMessage.SessionPaused -> msg.sessionId.value + is ServerMessage.SessionResumed -> msg.sessionId.value is ServerMessage.SessionCompleted -> msg.sessionId.value is ServerMessage.SessionFailed -> msg.sessionId.value is ServerMessage.StageStarted -> msg.sessionId.value diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt index e9c13840..ff4b81b3 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/ApprovalGateway.kt @@ -4,5 +4,5 @@ import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.events.types.ApprovalRequestId interface ApprovalGateway { - fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) + suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) } diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt index 1f02e276..062e996d 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/DefaultSessionOrchestrator.kt @@ -1,10 +1,15 @@ package com.correx.core.kernel.orchestration +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus import com.correx.core.approvals.model.ApprovalDecision import com.correx.core.artifacts.ArtifactState import com.correx.core.artifactstore.ArtifactStore +import com.correx.core.events.events.ApprovalDecisionResolvedEvent import com.correx.core.events.events.ArtifactValidatedEvent +import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.orchestration.OrchestrationState +import com.correx.core.events.types.ApprovalDecisionId import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.ArtifactLifecyclePhase @@ -16,7 +21,9 @@ import com.correx.core.sessions.Session import com.correx.core.transitions.execution.StageExecutionResult import com.correx.core.transitions.graph.WorkflowGraph import com.correx.core.transitions.resolution.TransitionDecision +import kotlinx.datetime.Clock import org.slf4j.LoggerFactory +import java.util.UUID import java.util.concurrent.* import java.util.concurrent.atomic.* @@ -133,10 +140,33 @@ class DefaultSessionOrchestrator( cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true) } - override fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) { + override suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) { val deferred = pendingApprovals[requestId] - ?: error("No pending approval for requestId ${requestId.value}") - deferred.complete(decision) + if (deferred != null) { + deferred.complete(decision) + return + } + // No live orchestrator coroutine for this request — server restarted while approval was pending. + // Record the decision and emit resume to unblock the session in the event store. + log.warn( + "submitApprovalDecision: no live deferred for requestId={} (server restart?), emitting events directly", + requestId.value, + ) + val sessionId = decision.contextSnapshot.identity.sessionId + val stageId = decision.contextSnapshot.identity.stageId + ?: orchestrationRepository.getState(sessionId).currentStageId + ?: return + emit(sessionId, ApprovalDecisionResolvedEvent( + decisionId = ApprovalDecisionId(UUID.randomUUID().toString()), + requestId = requestId, + outcome = decision.outcome ?: ApprovalOutcome.REJECTED, + status = ApprovalStatus.COMPLETED, + tier = decision.tier, + resolutionTimestamp = Clock.System.now(), + reason = decision.reason, + userSteering = decision.userSteering, + )) + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) } private suspend fun executeMove( diff --git a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt index 755a3c48..2c2d3c4f 100644 --- a/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt +++ b/core/kernel/src/main/kotlin/com/correx/core/kernel/orchestration/SessionOrchestrator.kt @@ -371,6 +371,7 @@ abstract class SessionOrchestrator( } emitDecisionResolved(sessionId, domainRequest, decision) if (!decision.isApproved) { + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) val sourceId = toolCall.id ?: invocationId.value val assistantEntry = ContextEntry( id = ContextEntryId(UUID.randomUUID().toString()), @@ -775,6 +776,7 @@ abstract class SessionOrchestrator( emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) StageExecutionResult.Success(emptyList()) } else { + emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) StageExecutionResult.Failure(decision.reason ?: "approval rejected", retryable = false) } } finally {