fix(server): move stuck-approval-pause repair from WS connect to boot (S3)
SessionEventBridge.replaySnapshot ran per WS connection and appended a repair OrchestrationResumedEvent — a write side effect on a pure read path that two concurrent clients could double-append (and 'open TUI' mutated replay history). Moved to a one-shot ServerModule.repairStuckApprovalPauses() at boot; replaySnapshot is read-only again. The state it repairs comes from a fixed pre-Feb-13 bug, so no session newly enters it at runtime — boot coverage is sufficient. Note: :apps:server:test not run (core:kernel broken by concurrent work); edits are isolated and mirror the existing preRegisterPendingApprovals shape.
This commit is contained in:
@@ -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"
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user