From d6bada6f1051575c1b2f5292c6fd3860020094ba Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 7 Jul 2026 18:30:09 +0400 Subject: [PATCH] feat(clarification): rehydrate pending questions on reconnect + restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stage parked on operator clarification lost its modal whenever the client disconnected or the server restarted — the questions were stranded in the event log with no way to re-surface them (unlike approvals, which already rehydrate). Two composing mechanisms, mirroring the approval path: - On reconnect: SessionEventBridge.replaySnapshot() now re-pushes the last unanswered ClarificationRequestedEvent for a PAUSED/CLARIFICATION_PENDING session as a ClarificationRequired message. Gated on the orchestrator still holding a live deferred (SessionOrchestrator.liveClarificationRequestIds) so a client never sees a modal whose answer could not be delivered. - On restart: ServerModule.resumeAbandonedSessions() eagerly relaunches clarification-parked sessions (approval-parked stay lazy). rehydrate+resume re-runs the parked stage, which re-emits a fresh, live ClarificationRequestedEvent (round budget MAX=3, 1 prior round recorded, so it re-parks rather than silently proceeding), re-arming the deferred that the reconnect path gates on. pauseReason == "CLARIFICATION_PENDING" is the discriminator because the reducer sets pendingApproval=true for every pause. --- .../com/correx/apps/server/ServerModule.kt | 23 ++++- .../apps/server/bridge/SessionEventBridge.kt | 40 ++++++++ .../apps/server/ws/GlobalStreamHandler.kt | 1 + .../server/bridge/SessionEventBridgeTest.kt | 99 +++++++++++++++++++ .../orchestration/SessionOrchestrator.kt | 9 ++ 5 files changed, 167 insertions(+), 5 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 b5131788..c0bacf8b 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 @@ -520,11 +520,21 @@ class ServerModule( val cutoff = Clock.System.now().minus(maxAgeMinutes, DateTimeUnit.MINUTE) eventStore.allSessionIds().forEach { sessionId -> val orchState = orchestrationRepository.getState(sessionId) - // Skip PAUSED sessions — they are waiting for user approval, not for computation. - // preRegisterPendingApprovals() already routes their responses. When the user - // approves, submitApprovalDecision emits OrchestrationResumedEvent which triggers - // the subscription above to continue the session. - if (orchState.status != OrchestrationStatus.RUNNING) return@forEach + // Skip approval-PAUSED sessions — they wait for user approval, not for computation. + // preRegisterPendingApprovals() already routes their responses. When the user approves, + // submitApprovalDecision emits OrchestrationResumedEvent which triggers the subscription + // above to continue the session. + // + // Clarification-PAUSED sessions are the exception: there is no lazy trigger that + // re-surfaces the modal after a restart (the parked stage's coroutine died with the + // deferred), so we eagerly relaunch them. rehydrate+resume re-runs the parked stage, + // which re-emits a fresh, live ClarificationRequestedEvent (round budget MAX=3, only 1 + // prior round recorded, so it re-parks rather than silently proceeding). The re-emitted + // event live-pushes the modal to connected clients, and SessionEventBridge.replaySnapshot + // re-pushes it to clients that connect afterwards. + val clarificationParked = orchState.status == OrchestrationStatus.PAUSED && + orchState.pauseReason == CLARIFICATION_PENDING_REASON + if (orchState.status != OrchestrationStatus.RUNNING && !clarificationParked) return@forEach val lastEventAt = eventStore.read(sessionId).lastOrNull()?.metadata?.timestamp ?: return@forEach if (lastEventAt < cutoff) { log.info( @@ -593,6 +603,9 @@ class ServerModule( /** Workflow stage id that produces the `design` artifact (examples/workflows/role_pipeline.toml). */ const val ARCHITECT_STAGE_ID = "architect" + /** Pause reason emitted when a stage parks on operator clarification (see SessionOrchestrator). */ + private const val CLARIFICATION_PENDING_REASON = "CLARIFICATION_PENDING" + /** Cap on the architect decision text embedded when no `approach` field is present. */ private const val DECISION_TEXT_CAP = 4000 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 5f7bd8e1..925b4bd3 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 @@ -20,6 +20,8 @@ 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.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 @@ -48,6 +50,11 @@ import com.correx.core.tools.registry.ToolRegistry // list isn't truncated on reopen (was 7, which dropped most of a 100+ event run). private const val SNAPSHOT_EVENT_LIMIT = 200 +// Pause reason emitted by SessionOrchestrator.requestClarificationIfNeeded. The reducer sets +// pendingApproval=true for every pause, so this reason string is the only reliable way to tell a +// clarification pause from an approval pause. +private const val CLARIFICATION_PENDING_REASON = "CLARIFICATION_PENDING" + class SessionEventBridge( private val eventStore: EventStore, private val artifactStore: ArtifactStore, @@ -55,9 +62,15 @@ class SessionEventBridge( private val workflowRegistry: WorkflowRegistry, private val toolRegistry: ToolRegistry, private val approvalCoordinator: ApprovalCoordinator? = null, + // Request-ids of clarifications the orchestrator is still parked on with a live deferred. A + // reconnecting client only re-opens a clarification modal that can actually be answered; after a + // restart the deferred is gone until the parked stage is relaunched (see + // ServerModule.resumeAbandonedSessions), which re-arms it. + private val livePendingClarificationIds: () -> Set = { emptySet() }, private val send: suspend (ServerMessage) -> Unit, ) { private val approvalProjector = ApprovalProjector(DefaultApprovalReducer()) + private val mapper = DomainEventMapper(artifactStore) /** * Resolves the last stage's inference output from CAS so a reopened session restores its output @@ -189,10 +202,37 @@ class SessionEventBridge( stages = stages, )) } + + rePushPendingClarification(orchState, events) } send(ServerMessage.SnapshotComplete) } + /** + * Re-surfaces a clarification the given session is parked on so a reconnecting client re-opens + * the modal (parallel to the pendingApprovals re-registration above). Fires only for a + * PAUSED/CLARIFICATION_PENDING session whose last unanswered ClarificationRequestedEvent still + * has a *live* deferred in the orchestrator — a request the running process can actually resolve. + * After a bare restart the deferred is gone until ServerModule.resumeAbandonedSessions relaunches + * the parked stage (which re-emits a fresh, live ClarificationRequestedEvent); until then we do + * not show a modal that would resume nothing. + */ + private suspend fun rePushPendingClarification( + orchState: com.correx.core.events.orchestration.OrchestrationState, + events: List, + ) { + if (orchState.status != OrchestrationStatus.PAUSED) return + if (orchState.pauseReason != CLARIFICATION_PENDING_REASON) return + val answered = events.mapNotNullTo(mutableSetOf()) { + (it.payload as? ClarificationAnsweredEvent)?.requestId?.value + } + val live = livePendingClarificationIds() + events.lastOrNull { + val p = it.payload + p is ClarificationRequestedEvent && p.requestId.value !in answered && p.requestId.value in live + }?.let { stored -> mapper.map(stored, stored.sessionSequence)?.let { send(it) } } + } + private fun eventToEntry(event: StoredEvent): EventEntryDto? { val ts = event.metadata.timestamp.toEpochMilliseconds() return when (val p = event.payload) { 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 63fa9577..ca08b3b8 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 @@ -105,6 +105,7 @@ class GlobalStreamHandler(private val module: ServerModule) { toolRegistry = module.toolRegistry, send = sendFrame, approvalCoordinator = module.approvalCoordinator, + livePendingClarificationIds = { module.orchestrator.liveClarificationRequestIds() }, ) val mapper = DomainEventMapper(module.artifactStore) diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt index c4dd5d82..c6467745 100644 --- a/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt +++ b/apps/server/src/test/kotlin/com/correx/apps/server/bridge/SessionEventBridgeTest.kt @@ -5,8 +5,13 @@ import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.server.registry.WorkflowRegistry import com.correx.core.artifactstore.ArtifactStore import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.ClarificationAnsweredEvent +import com.correx.core.events.events.ClarificationAnswer +import com.correx.core.events.events.ClarificationQuestion +import com.correx.core.events.events.ClarificationRequestedEvent import com.correx.core.events.events.EventPayload 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.StoredEvent @@ -17,6 +22,7 @@ import com.correx.core.events.orchestration.OrchestrationState import com.correx.core.events.orchestration.OrchestrationStatus import com.correx.core.events.stores.EventStore import com.correx.core.events.types.ArtifactId +import com.correx.core.events.types.ClarificationRequestId import com.correx.core.events.types.EventId import com.correx.core.events.types.SessionId import com.correx.core.events.types.StageId @@ -92,6 +98,99 @@ class SessionEventBridgeTest { }, ) + private fun clarificationPausedRepository(): OrchestrationRepository = OrchestrationRepository( + object : EventReplayer { + override fun rebuild(sessionId: SessionId): OrchestrationState = OrchestrationState( + workflowId = workflowId, + status = OrchestrationStatus.PAUSED, + currentStageId = stageId, + // The reducer sets pendingApproval=true for every pause; the reason discriminates. + pendingApproval = true, + pauseReason = "CLARIFICATION_PENDING", + ) + }, + ) + + private val clarRequestId = ClarificationRequestId("req-1") + + // A stage parking on clarification records the pause then the request (see + // SessionOrchestrator.requestClarificationIfNeeded) — the pause event is what stops the + // stuck-session fixer from mistaking the clarification pause for a resolved approval. + private fun clarParkEvents(answered: Boolean = false): List = buildList { + add(storedEvent(OrchestrationPausedEvent(sessionId, stageId, "CLARIFICATION_PENDING"), seq = 1L)) + add( + storedEvent( + ClarificationRequestedEvent( + requestId = clarRequestId, + sessionId = sessionId, + stageId = stageId, + questions = listOf(ClarificationQuestion(id = "stack", prompt = "Which stack?")), + ), + seq = 2L, + ), + ) + if (answered) { + add( + storedEvent( + ClarificationAnsweredEvent( + requestId = clarRequestId, + sessionId = sessionId, + stageId = stageId, + answers = listOf(ClarificationAnswer(questionId = "stack", value = "React")), + ), + seq = 3L, + ), + ) + } + } + + @Test + fun `replaySnapshot re-pushes a pending clarification when the deferred is live`() = runTest { + val store = fakeEventStore(allEventsList = clarParkEvents()) + val sent = mutableListOf() + val bridge = SessionEventBridge( + store, noopArtifactStore, clarificationPausedRepository(), noopWorkflowRegistry, noopToolRegistry, + livePendingClarificationIds = { setOf(clarRequestId.value) }, + ) { sent.add(it) } + + bridge.replaySnapshot() + + val clar = sent.filterIsInstance() + assertEquals(1, clar.size) + assertEquals(clarRequestId, clar[0].requestId) + assertEquals(stageId, clar[0].stageId) + } + + @Test + fun `replaySnapshot does not re-push a clarification whose deferred died with a restart`() = runTest { + val store = fakeEventStore(allEventsList = clarParkEvents()) + val sent = mutableListOf() + // Empty live set = no in-process deferred (bare restart) — showing the modal would trap the + // operator since the answer could not be delivered. + val bridge = SessionEventBridge( + store, noopArtifactStore, clarificationPausedRepository(), noopWorkflowRegistry, noopToolRegistry, + livePendingClarificationIds = { emptySet() }, + ) { sent.add(it) } + + bridge.replaySnapshot() + + assertEquals(0, sent.filterIsInstance().size) + } + + @Test + fun `replaySnapshot does not re-push an already-answered clarification`() = runTest { + val store = fakeEventStore(allEventsList = clarParkEvents(answered = true)) + val sent = mutableListOf() + val bridge = SessionEventBridge( + store, noopArtifactStore, clarificationPausedRepository(), noopWorkflowRegistry, noopToolRegistry, + livePendingClarificationIds = { setOf(clarRequestId.value) }, + ) { sent.add(it) } + + bridge.replaySnapshot() + + assertEquals(0, sent.filterIsInstance().size) + } + private val noopWorkflowRegistry = object : WorkflowRegistry { override fun listAll(): List = emptyList() override fun find(workflowId: String): com.correx.core.transitions.graph.WorkflowGraph? = null 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 d159f443..3936e104 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 @@ -312,6 +312,15 @@ abstract class SessionOrchestrator( ConcurrentHashMap>> = ConcurrentHashMap() + /** + * Request-ids of clarifications this process is still parked on (the [CompletableDeferred] is + * live and awaitable). A reconnecting client uses this to re-push only clarifications that can + * actually be answered — a request whose deferred died with a server restart is excluded, so the + * client never shows a modal that would resume nothing. + */ + fun liveClarificationRequestIds(): Set = + pendingClarifications.keys.mapTo(mutableSetOf()) { it.value } + /** Cache mapping "sessionId:artifactId" to JSON-serialized artifact payload content. * Populated when ProcessResult artifacts are stored on tool failure. * Used by DefaultSessionOrchestrator.step() to populate EvaluationContext.artifactContent. */