fix: unblock sessions stuck in PAUSED state after approval resolved

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.
This commit is contained in:
2026-05-26 21:12:26 +04:00
parent e32bd121e5
commit 766b91081a
11 changed files with 140 additions and 19 deletions
@@ -3,6 +3,8 @@ package com.correx.apps.server
import com.correx.apps.server.approval.ApprovalCoordinator import com.correx.apps.server.approval.ApprovalCoordinator
import com.correx.apps.server.registry.ProviderRegistry import com.correx.apps.server.registry.ProviderRegistry
import com.correx.apps.server.registry.WorkflowRegistry 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.approvals.DefaultApprovalRepository
import com.correx.core.artifactstore.ArtifactStore import com.correx.core.artifactstore.ArtifactStore
import com.correx.core.config.ApprovalConfig import com.correx.core.config.ApprovalConfig
@@ -52,12 +54,33 @@ class ServerModule(
/** /**
* Subscribe to the event store and forward [ApprovalRequestedEvent] payloads to * Subscribe to the event store and forward [ApprovalRequestedEvent] payloads to
* the [ApprovalCoordinator]. Idempotent: calling twice is a no-op for the second call. * 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() { fun start() {
if (subscriptionJob != null) return if (subscriptionJob != null) return
preRegisterPendingApprovals()
subscriptionJob = eventStore.subscribeAll() subscriptionJob = eventStore.subscribeAll()
.filter { it.payload is ApprovalRequestedEvent } .filter { it.payload is ApprovalRequestedEvent }
.onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) } .onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) }
.launchIn(moduleScope) .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) }
}
}
} }
@@ -85,7 +85,7 @@ open class ApprovalCoordinator(
requestSessions[requestId] = sessionId 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) { if (resolved.putIfAbsent(msg.requestId, true) != null) {
return ServerMessage.ProtocolError( return ServerMessage.ProtocolError(
message = "Approval request ${msg.requestId.value} already resolved", message = "Approval request ${msg.requestId.value} already resolved",
@@ -10,6 +10,7 @@ import com.correx.core.events.events.WorkflowStartedEvent
import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.events.InferenceTimeoutEvent import com.correx.core.events.events.InferenceTimeoutEvent
import com.correx.core.events.events.OrchestrationPausedEvent 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.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.StoredEvent import com.correx.core.events.events.StoredEvent
@@ -90,6 +91,12 @@ suspend fun domainEventToServerMessage(
) )
is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence) is OrchestrationPausedEvent -> mapOrchestrationPaused(p, seq, sessionSequence)
is OrchestrationResumedEvent -> ServerMessage.SessionResumed(
sessionId = p.sessionId,
stageId = p.stageId,
sequence = seq,
sessionSequence = sessionSequence,
)
is InferenceStartedEvent -> ServerMessage.InferenceStarted( is InferenceStartedEvent -> ServerMessage.InferenceStarted(
sessionId = p.sessionId, sessionId = p.sessionId,
stageId = p.stageId, stageId = p.stageId,
@@ -12,10 +12,13 @@ import com.correx.core.approvals.ApprovalProjector
import com.correx.core.approvals.DefaultApprovalReducer import com.correx.core.approvals.DefaultApprovalReducer
import com.correx.core.approvals.model.ApprovalState import com.correx.core.approvals.model.ApprovalState
import com.correx.core.artifactstore.ArtifactStore 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.InferenceCompletedEvent
import com.correx.core.events.events.InferenceStartedEvent import com.correx.core.events.events.InferenceStartedEvent
import com.correx.core.events.events.InferenceTimeoutEvent 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.OrchestrationPausedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.events.StageCompletedEvent import com.correx.core.events.events.StageCompletedEvent
import com.correx.core.events.events.StageFailedEvent import com.correx.core.events.events.StageFailedEvent
import com.correx.core.events.events.StoredEvent 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.WorkflowCompletedEvent
import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.events.WorkflowFailedEvent
import com.correx.core.events.orchestration.OrchestrationStatus 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.stores.EventStore
import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ApprovalRequestId
import com.correx.core.events.types.SessionId import com.correx.core.events.types.SessionId
@@ -62,18 +67,46 @@ class SessionEventBridge(
} }
} else null } else null
val pendingApprovals = approvalState?.requests?.values val pendingApprovalRequests = approvalState?.let { state ->
?.filter { req -> approvalState.decisions.values.none { it.requestId == req.id } } state.requests.values
?.sortedWith(compareBy({ it.timestamp }, { it.id.value })) .filter { req -> state.decisions.values.none { it.requestId == req.id } }
?.map { .sortedWith(compareBy({ it.timestamp }, { it.id.value }))
ApprovalDto( } ?: emptyList()
requestId = it.id.value,
tier = it.tier.name, // Fix stuck sessions: orchestration says PAUSED + pendingApproval but there are no
toolName = it.toolName, // actual unresolved approval requests. This happens when approval was resolved but
preview = it.preview, // 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 val recentEvents = events
.mapNotNull { eventToEntry(it) } .mapNotNull { eventToEntry(it) }
@@ -53,6 +53,15 @@ sealed interface ServerMessage {
override val sessionSequence: Long, override val sessionSequence: Long,
) : ServerMessage, SessionMessage ) : 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 @Serializable
@SerialName("session.completed") @SerialName("session.completed")
data class SessionCompleted( data class SessionCompleted(
@@ -57,7 +57,7 @@ class ApprovalCoordinatorWiringTest {
private class RecordingGateway : ApprovalGateway { private class RecordingGateway : ApprovalGateway {
val submissions = CopyOnWriteArrayList<Pair<ApprovalRequestId, DomainApprovalDecision>>() val submissions = CopyOnWriteArrayList<Pair<ApprovalRequestId, DomainApprovalDecision>>()
override fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) { override suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: DomainApprovalDecision) {
submissions.add(requestId to decision) submissions.add(requestId to decision)
} }
} }
@@ -90,7 +90,7 @@ class ApprovalCoordinatorWiringTest {
} }
@Test @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 gateway = RecordingGateway()
val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
val msg = ClientMessage.ApprovalResponse( val msg = ClientMessage.ApprovalResponse(
@@ -107,7 +107,7 @@ class ApprovalCoordinatorWiringTest {
} }
@Test @Test
fun `duplicate ApprovalResponse for same requestId returns ProtocolError`() { fun `duplicate ApprovalResponse for same requestId returns ProtocolError`(): Unit = runBlocking {
val gateway = RecordingGateway() val gateway = RecordingGateway()
val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
val msg = ClientMessage.ApprovalResponse( val msg = ClientMessage.ApprovalResponse(
@@ -177,7 +177,7 @@ class ApprovalCoordinatorWiringTest {
} }
@Test @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 gateway = RecordingGateway()
val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope) val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
val otherSessionId = SessionId("other-session") val otherSessionId = SessionId("other-session")
@@ -165,6 +165,7 @@ object SessionsReducer {
val (result, effects) = when (msg) { val (result, effects) = when (msg) {
is ServerMessage.SessionStarted -> processSessionStartedMessage(msg, clock, sessions) is ServerMessage.SessionStarted -> processSessionStartedMessage(msg, clock, sessions)
is ServerMessage.SessionPaused -> processSessionPausedMessage(msg, sessions, clock) is ServerMessage.SessionPaused -> processSessionPausedMessage(msg, sessions, clock)
is ServerMessage.SessionResumed -> processSessionResumedMessage(msg, sessions, clock)
is ServerMessage.SessionCompleted -> processSessionCompletedMessage(sessions, msg, clock) is ServerMessage.SessionCompleted -> processSessionCompletedMessage(sessions, msg, clock)
is ServerMessage.SessionFailed -> processSessionFailedMessage(sessions, msg, clock) is ServerMessage.SessionFailed -> processSessionFailedMessage(sessions, msg, clock)
is ServerMessage.StageStarted -> processStageStartedMessage(msg, sessions) is ServerMessage.StageStarted -> processStageStartedMessage(msg, sessions)
@@ -189,6 +190,7 @@ object SessionsReducer {
private fun sessionIdFromMessage(msg: ServerMessage): String? = when (msg) { private fun sessionIdFromMessage(msg: ServerMessage): String? = when (msg) {
is ServerMessage.SessionStarted -> msg.sessionId.value is ServerMessage.SessionStarted -> msg.sessionId.value
is ServerMessage.SessionPaused -> msg.sessionId.value is ServerMessage.SessionPaused -> msg.sessionId.value
is ServerMessage.SessionResumed -> msg.sessionId.value
is ServerMessage.SessionCompleted -> msg.sessionId.value is ServerMessage.SessionCompleted -> msg.sessionId.value
is ServerMessage.SessionFailed -> msg.sessionId.value is ServerMessage.SessionFailed -> msg.sessionId.value
is ServerMessage.StageStarted -> msg.sessionId.value is ServerMessage.StageStarted -> msg.sessionId.value
@@ -507,6 +509,20 @@ object SessionsReducer {
) to emptyList() ) to emptyList()
} }
private fun processSessionResumedMessage(
msg: ServerMessage.SessionResumed,
sessions: SessionsState,
clock: () -> Long,
): Pair<SessionsState, List<Effect>> = 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( private fun processSessionStartedMessage(
msg: ServerMessage.SessionStarted, msg: ServerMessage.SessionStarted,
clock: () -> Long, clock: () -> Long,
@@ -64,6 +64,7 @@ object SnapshotPhaseReducer {
internal fun sessionIdOf(msg: ServerMessage): String? = when (msg) { internal fun sessionIdOf(msg: ServerMessage): String? = when (msg) {
is ServerMessage.SessionStarted -> msg.sessionId.value is ServerMessage.SessionStarted -> msg.sessionId.value
is ServerMessage.SessionPaused -> msg.sessionId.value is ServerMessage.SessionPaused -> msg.sessionId.value
is ServerMessage.SessionResumed -> msg.sessionId.value
is ServerMessage.SessionCompleted -> msg.sessionId.value is ServerMessage.SessionCompleted -> msg.sessionId.value
is ServerMessage.SessionFailed -> msg.sessionId.value is ServerMessage.SessionFailed -> msg.sessionId.value
is ServerMessage.StageStarted -> msg.sessionId.value is ServerMessage.StageStarted -> msg.sessionId.value
@@ -4,5 +4,5 @@ import com.correx.core.approvals.model.ApprovalDecision
import com.correx.core.events.types.ApprovalRequestId import com.correx.core.events.types.ApprovalRequestId
interface ApprovalGateway { interface ApprovalGateway {
fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision) suspend fun submitApprovalDecision(requestId: ApprovalRequestId, decision: ApprovalDecision)
} }
@@ -1,10 +1,15 @@
package com.correx.core.kernel.orchestration 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.approvals.model.ApprovalDecision
import com.correx.core.artifacts.ArtifactState import com.correx.core.artifacts.ArtifactState
import com.correx.core.artifactstore.ArtifactStore 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.ArtifactValidatedEvent
import com.correx.core.events.events.OrchestrationResumedEvent
import com.correx.core.events.orchestration.OrchestrationState 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.ApprovalRequestId
import com.correx.core.events.types.ArtifactId import com.correx.core.events.types.ArtifactId
import com.correx.core.events.types.ArtifactLifecyclePhase 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.execution.StageExecutionResult
import com.correx.core.transitions.graph.WorkflowGraph import com.correx.core.transitions.graph.WorkflowGraph
import com.correx.core.transitions.resolution.TransitionDecision import com.correx.core.transitions.resolution.TransitionDecision
import kotlinx.datetime.Clock
import org.slf4j.LoggerFactory import org.slf4j.LoggerFactory
import java.util.UUID
import java.util.concurrent.* import java.util.concurrent.*
import java.util.concurrent.atomic.* import java.util.concurrent.atomic.*
@@ -133,10 +140,33 @@ class DefaultSessionOrchestrator(
cancellations.getOrPut(sessionId) { AtomicBoolean(false) }.set(true) 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] val deferred = pendingApprovals[requestId]
?: error("No pending approval for requestId ${requestId.value}") if (deferred != null) {
deferred.complete(decision) 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( private suspend fun executeMove(
@@ -371,6 +371,7 @@ abstract class SessionOrchestrator(
} }
emitDecisionResolved(sessionId, domainRequest, decision) emitDecisionResolved(sessionId, domainRequest, decision)
if (!decision.isApproved) { if (!decision.isApproved) {
emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
val sourceId = toolCall.id ?: invocationId.value val sourceId = toolCall.id ?: invocationId.value
val assistantEntry = ContextEntry( val assistantEntry = ContextEntry(
id = ContextEntryId(UUID.randomUUID().toString()), id = ContextEntryId(UUID.randomUUID().toString()),
@@ -775,6 +776,7 @@ abstract class SessionOrchestrator(
emit(sessionId, OrchestrationResumedEvent(sessionId, stageId)) emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
StageExecutionResult.Success(emptyList()) StageExecutionResult.Success(emptyList())
} else { } else {
emit(sessionId, OrchestrationResumedEvent(sessionId, stageId))
StageExecutionResult.Failure(decision.reason ?: "approval rejected", retryable = false) StageExecutionResult.Failure(decision.reason ?: "approval rejected", retryable = false)
} }
} finally { } finally {