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 ce6fff73..916d139a 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 @@ -186,6 +186,7 @@ class ServerModule( fun start() { if (subscriptionJob != null) return preRegisterPendingApprovals() + repairStuckApprovalPauses() resumeAbandonedSessions() subscriptionJob = eventStore.subscribeAll() .filter { it.payload is ApprovalRequestedEvent } @@ -646,6 +647,53 @@ class ServerModule( } } + /** + * One-shot boot repair for sessions stuck PAUSED + pendingApproval with no actual unresolved + * approval request — the symptom of a pre-Feb-13 orchestrator bug where approval was resolved + * but OrchestrationResumedEvent was never emitted. Appends the missing resume so the projection + * corrects permanently. Previously this ran per WS connection inside SessionEventBridge. + * replaySnapshot, which put a racy write on a pure read path (two concurrent clients could + * double-append). Running it once at boot removes that hazard; the bug that creates the state is + * fixed, so no session newly enters it at runtime. + * + * Guard: skip when there are more OrchestrationPausedEvents than ApprovalRequestedEvents — the + * session just entered the gate and its ApprovalRequestedEvent isn't stored yet; a spurious + * resume would hide the pending approval. + */ + private fun repairStuckApprovalPauses() { + 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) } + val hasUnresolved = approvalState.requests.values + .any { req -> approvalState.decisions.values.none { it.requestId == req.id } } + val pauseCount = events.count { it.payload is OrchestrationPausedEvent } + val approvalRequestCount = events.count { it.payload is ApprovalRequestedEvent } + val hasUnpairedPause = pauseCount > approvalRequestCount + val alreadyResumed = events.any { it.payload is OrchestrationResumedEvent } + val stageId = orchState.currentStageId + if (hasUnresolved || hasUnpairedPause || alreadyResumed || stageId == null) return@forEach + log.info("repairStuckApprovalPauses: emitting missing resume for stuck session={}", sessionId.value) + moduleScope.launch { + eventStore.append( + 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), + ), + ) + } + } + } + companion object { /** Workflow stage id that produces the `design` artifact (examples/workflows/role_pipeline.toml). */ const val ARCHITECT_STAGE_ID = "architect" 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 c0abb96b..cca340c2 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 @@ -14,18 +14,15 @@ 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.ApprovalRequestedEvent import com.correx.core.events.events.ArtifactCreatedEvent import com.correx.core.events.events.ChatTurnEvent import com.correx.core.events.events.ClarificationAnsweredEvent import com.correx.core.events.events.ClarificationRequestedEvent 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.SessionWorkspaceBoundEvent import com.correx.core.events.events.StageFailedEvent @@ -40,8 +37,6 @@ import com.correx.core.events.events.WorkflowCompletedEvent import com.correx.core.events.events.WorkflowStartedEvent 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.SessionId import com.correx.core.kernel.orchestration.OrchestrationRepository @@ -108,39 +103,11 @@ class SessionEventBridge( .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. - // - // Guard: if there are more OrchestrationPausedEvents than ApprovalRequestedEvents the - // session just entered the approval gate and ApprovalRequestedEvent has not been stored - // yet (tiny race window between two sequential emits). Do NOT fire in that case — - // appending a spurious resume would hide the pending approval from the TUI. - val pauseCount = events.count { it.payload is OrchestrationPausedEvent } - val approvalRequestCount = events.count { it.payload is ApprovalRequestedEvent } - val hasUnpairedPause = pauseCount > approvalRequestCount - if (orchState.pendingApproval && pendingApprovalRequests.isEmpty() && !hasUnpairedPause) { - 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) - } - } + // NOTE: repairing stuck approval-pauses (PAUSED + pendingApproval but no unresolved + // request, from a pre-Feb-13 orchestrator bug) used to happen HERE, per connection — + // a write side effect on a pure read path that two concurrent clients could double-fire. + // It now runs once at boot in ServerModule.repairStuckApprovalPauses(). replaySnapshot is + // read-only again. val pendingApprovals = pendingApprovalRequests.map { ApprovalDto(