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 ed7e7397..ba6e2936 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 @@ -9,7 +9,9 @@ import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore import com.correx.core.config.ApprovalConfig import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.OrchestrationResumedEvent import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.SessionId import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator import com.correx.core.kernel.orchestration.OrchestrationConfig import com.correx.core.kernel.orchestration.OrchestrationRepository @@ -17,6 +19,7 @@ import com.correx.core.router.RouterFacade import com.correx.core.sessions.DefaultSessionRepository import com.correx.core.tools.registry.ToolRegistry import com.correx.core.events.orchestration.OrchestrationStatus +import java.util.concurrent.ConcurrentHashMap import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job @@ -55,6 +58,7 @@ class ServerModule( ) private var subscriptionJob: Job? = null + private val activeSessionJobs = ConcurrentHashMap() /** * Subscribe to the event store and forward [ApprovalRequestedEvent] payloads to @@ -73,26 +77,50 @@ class ServerModule( .filter { it.payload is ApprovalRequestedEvent } .onEach { approvalCoordinator.onApprovalRequested(it.payload as ApprovalRequestedEvent) } .launchIn(moduleScope) + + // When a PAUSED session's approval is resolved without a live orchestrator coroutine + // (server was restarted while approval was pending), emit OrchestrationResumedEvent + // triggers this subscription to continue the session. + eventStore.subscribeAll() + .filter { it.payload is OrchestrationResumedEvent } + .onEach { event -> + val sessionId = (event.payload as OrchestrationResumedEvent).sessionId + if (activeSessionJobs.containsKey(sessionId)) return@onEach + val orchState = orchestrationRepository.getState(sessionId) + if (orchState.status != OrchestrationStatus.RUNNING) return@onEach + val graph = workflowRegistry.find(orchState.workflowId) ?: return@onEach + launchSessionResume(sessionId, graph) + } + .launchIn(moduleScope) + } + + private fun launchSessionResume(sessionId: SessionId, graph: com.correx.core.transitions.graph.WorkflowGraph) { + if (activeSessionJobs.containsKey(sessionId)) return + val job = moduleScope.launch { + runCatching { + orchestrator.resume(sessionId, graph, defaultOrchestrationConfig) + }.onFailure { e -> + log.error("resumeSession: session={} failed", sessionId.value, e) + } + activeSessionJobs.remove(sessionId) + } + activeSessionJobs[sessionId] = job } private fun resumeAbandonedSessions() { eventStore.allSessionIds().forEach { sessionId -> val orchState = orchestrationRepository.getState(sessionId) - if (orchState.status != OrchestrationStatus.RUNNING && - orchState.status != OrchestrationStatus.PAUSED - ) return@forEach + // 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 val graph = workflowRegistry.find(orchState.workflowId) ?: run { log.warn("resumeAbandoned: no graph for session={} workflowId={}", sessionId.value, orchState.workflowId) return@forEach } log.info("resumeAbandoned: launching session={} workflow={}", sessionId.value, orchState.workflowId) - moduleScope.launch { - runCatching { - orchestrator.resume(sessionId, graph, defaultOrchestrationConfig) - }.onFailure { e -> - log.error("resumeAbandoned: session={} failed", sessionId.value, e) - } - } + launchSessionResume(sessionId, graph) } } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt index 80c8c8c3..88c62e1a 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/TuiApp.kt @@ -163,9 +163,9 @@ private fun renderIdleLayout(frame: Frame, state: TuiState) { private fun renderInSessionLayout(frame: Frame, state: TuiState) { val area = frame.area() val selectedSession = state.sessions.sessions.firstOrNull { it.id == state.sessions.selectedId } - val hasDiff = selectedSession?.tools?.any { - it.status == ToolDisplayStatus.COMPLETED && it.diff != null - } == true + val hasDiff = selectedSession?.tools + ?.lastOrNull { it.status == ToolDisplayStatus.COMPLETED } + ?.diff != null val constraints = mutableListOf( Constraint.length(STATUS_HEIGHT), @@ -197,12 +197,20 @@ private fun renderInSessionLayout(frame: Frame, state: TuiState) { private fun renderApprovalLayout(frame: Frame, state: TuiState) { val area = frame.area() + val selectedSession = state.sessions.sessions.firstOrNull { it.id == state.sessions.selectedId } + val hasDiff = selectedSession?.tools + ?.lastOrNull { it.status == ToolDisplayStatus.COMPLETED } + ?.diff != null + val constraints = mutableListOf( Constraint.length(STATUS_HEIGHT), ) if (state.eventStripVisible) { constraints.add(Constraint.length(EVENT_STRIP_HEIGHT)) } + if (hasDiff) { + constraints.add(Constraint.length(DIFF_HEIGHT)) + } constraints.add(Constraint.fill()) constraints.add(Constraint.length(INPUT_HEIGHT)) @@ -215,6 +223,9 @@ private fun renderApprovalLayout(frame: Frame, state: TuiState) { if (state.eventStripVisible) { frame.renderWidget(eventHistoryStripWidget(state), vertRects[rectIdx++]) } + if (hasDiff) { + frame.renderWidget(diffViewerWidget(state), vertRects[rectIdx++]) + } frame.renderWidget(approvalSurfaceWidget(state), vertRects[rectIdx++]) frame.renderWidget(inputBarWidget(state, DisplayState.APPROVAL), vertRects[rectIdx]) } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/DiffViewer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/DiffViewer.kt index 400aee44..faa9faec 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/components/DiffViewer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/components/DiffViewer.kt @@ -23,8 +23,11 @@ private val yellowStyle = Style.create().yellow() fun diffViewerWidget(state: TuiState): Paragraph { val selectedSession = state.sessions.sessions.firstOrNull { it.id == state.sessions.selectedId } + // Only show the diff for the most recently completed tool. If the last tool has no diff + // (e.g. a shell command), the viewer is hidden — stale diffs from earlier tools are noise. val toolWithDiff = selectedSession?.tools - ?.lastOrNull { it.status == ToolDisplayStatus.COMPLETED && it.diff != null } + ?.lastOrNull { it.status == ToolDisplayStatus.COMPLETED } + ?.takeIf { it.diff != null } val diffText = toolWithDiff?.diff ?: return Paragraph.builder().build()