feat(clarification): rehydrate pending questions on reconnect + restart

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.
2026-07-07 18:30:09 +04:00
parent 8cfc590c7e
commit d6bada6f10
5 changed files with 167 additions and 5 deletions
@@ -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

@@ -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<String> = { 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<StoredEvent>,
+    ) {
+        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) {
@@ -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)

@@ -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<OrchestrationState> {
+            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<StoredEvent> = 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<ServerMessage>()
+        val bridge = SessionEventBridge(
+            store, noopArtifactStore, clarificationPausedRepository(), noopWorkflowRegistry, noopToolRegistry,
+            livePendingClarificationIds = { setOf(clarRequestId.value) },
+        ) { sent.add(it) }
+
+        bridge.replaySnapshot()
+
+        val clar = sent.filterIsInstance<ServerMessage.ClarificationRequired>()
+        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<ServerMessage>()
+        // 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<ServerMessage.ClarificationRequired>().size)
+    }
+
+    @Test
+    fun `replaySnapshot does not re-push an already-answered clarification`() = runTest {
+        val store = fakeEventStore(allEventsList = clarParkEvents(answered = true))
+        val sent = mutableListOf<ServerMessage>()
+        val bridge = SessionEventBridge(
+            store, noopArtifactStore, clarificationPausedRepository(), noopWorkflowRegistry, noopToolRegistry,
+            livePendingClarificationIds = { setOf(clarRequestId.value) },
+        ) { sent.add(it) }
+
+        bridge.replaySnapshot()
+
+        assertEquals(0, sent.filterIsInstance<ServerMessage.ClarificationRequired>().size)
+    }
+
    private val noopWorkflowRegistry = object : WorkflowRegistry {
        override fun listAll(): List<com.correx.apps.server.registry.WorkflowSummary> = emptyList()
        override fun find(workflowId: String): com.correx.core.transitions.graph.WorkflowGraph? = null
@@ -312,6 +312,15 @@ abstract class SessionOrchestrator(
        ConcurrentHashMap<ClarificationRequestId, CompletableDeferred<List<ClarificationAnswer>>> =
        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<String> =
+        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. */