fix(approval): resolve requestId→sessionId lookup in global socket handler

Fix the remaining TODO from 2e3b8e5: replace the TypeId(requestId.value) hack with a proper requestId→sessionId mapping stored in ApprovalCoordinator and populated when onApprovalRequested fires from the event store subscription. Changes across the full bugfix batch (all 9 bugs from TUI rework): - Bug 1: ApprovalCoordinator broadcasts non-null toolName/preview - Bug 2: ApprovalDto carries toolName/preview for late-connected clients - Bug 3: SessionEventBridge.replaySnapshot() preserves toolName/preview - Bug 4: TUI SessionsReducer processes enriched ApprovalDto fields - Bug 5: DomainEventMapper passes sessionSequence; GlobalStreamHandler tracks per-session counters so live events aren't dropped - Bug 6: KeyResolver restores Ctrl+L/E/C/N bindings - Bug 7: SessionSnapshot carries workflowId for display names - Bug 8: ↑↓ navigates selection only; Enter/sessionEntered gates IN_SESSION - Bug 9: EventHistoryStrip capped at 4 lines to prevent overflow - Bonus: ApprovalCoordinator.lookupSession() with proper requestId→sessionId mapping
2026-05-26 13:09:10 +04:00
parent f6ef028883
commit c142f146d4
17 changed files with 132 additions and 56 deletions
@@ -36,6 +36,7 @@ open class ApprovalCoordinator(
    private val globalClients: MutableSet<DefaultWebSocketServerSession> = ConcurrentHashMap.newKeySet()
    private val resolved: ConcurrentHashMap<ApprovalRequestId, Boolean> = ConcurrentHashMap()
    private val timeoutJobs: ConcurrentHashMap<ApprovalRequestId, Job> = ConcurrentHashMap()
+    private val requestSessions: ConcurrentHashMap<ApprovalRequestId, SessionId> = ConcurrentHashMap()

    fun registerClient(sessionId: SessionId, session: DefaultWebSocketServerSession) {
        sessionClients.getOrPut(sessionId) { ConcurrentHashMap.newKeySet() }.add(session)
@@ -54,6 +55,7 @@ open class ApprovalCoordinator(
    }

    suspend fun onApprovalRequested(event: ApprovalRequestedEvent) {
+        requestSessions[event.requestId] = event.sessionId
        val msg = ServerMessage.ApprovalRequired(
            sessionId = event.sessionId,
            requestId = event.requestId,
@@ -63,8 +65,8 @@ open class ApprovalCoordinator(
                factors = emptyList(),
                recommendedAction = "Review and approve or reject",
            ),
-            toolName = null,
-            preview = null,
+            toolName = event.toolName,
+            preview = event.preview,
            sequence = 0L,
            sessionSequence = 0L,
        )
@@ -72,6 +74,8 @@ open class ApprovalCoordinator(
        scheduleTimeout(event.requestId, event.sessionId, event.stageId, event.tier)
    }

+    fun lookupSession(requestId: ApprovalRequestId): SessionId? = requestSessions[requestId]
+
    open fun handleResponse(msg: ClientMessage.ApprovalResponse, sessionId: SessionId): ServerMessage? {
        if (resolved.putIfAbsent(msg.requestId, true) != null) {
            return ServerMessage.ProtocolError(
@@ -81,6 +85,7 @@ open class ApprovalCoordinator(
            )
        }
        timeoutJobs.remove(msg.requestId)?.cancel()
+        requestSessions.remove(msg.requestId)
        val domain = msg.toDomain(sessionId, null, Tier.T2)
        return runCatching { orchestrator.submitApprovalDecision(msg.requestId, domain) }
            .fold(
@@ -26,8 +26,8 @@ import org.slf4j.LoggerFactory
 private val log = LoggerFactory.getLogger("DomainEventMapper")

 class DomainEventMapper(private val artifactStore: ArtifactStore = NoopArtifactStore) {
-    suspend fun map(event: StoredEvent): ServerMessage? =
-        domainEventToServerMessage(event, artifactStore)
+    suspend fun map(event: StoredEvent, sessionSequence: Long = 0L): ServerMessage? =
+        domainEventToServerMessage(event, artifactStore, sessionSequence = sessionSequence)
 }

 private object NoopArtifactStore : ArtifactStore {
@@ -38,11 +38,19 @@ class SessionEventBridge(
            val pendingApprovals = approvalState?.requests?.values
                ?.filter { req -> approvalState.decisions.values.none { it.requestId == req.id } }
                ?.sortedWith(compareBy({ it.timestamp }, { it.id.value }))
-                ?.map { ApprovalDto(requestId = it.id.value, tier = it.tier.name) }
+                ?.map {
+                    ApprovalDto(
+                        requestId = it.id.value,
+                        tier = it.tier.name,
+                        toolName = it.toolName,
+                        preview = it.preview,
+                    )
+                }
                ?: emptyList()

            send(ServerMessage.SessionSnapshot(
                sessionId = sessionId,
+                workflowId = orchState.workflowId,
                state = SessionStateDto(
                    status = orchState.status.name,
                    currentStageId = orchState.currentStageId?.value,
@@ -33,6 +33,8 @@ data class SessionStateDto(
 data class ApprovalDto(
    val requestId: String,
    val tier: String,
+    val toolName: String? = null,
+    val preview: String? = null,
 )

@Serializable
@@ -76,6 +76,7 @@ sealed interface ServerMessage {
    @SerialName("session_snapshot")
    data class SessionSnapshot(
        val sessionId: SessionId,
+        val workflowId: String,
        val state: SessionStateDto,
        val pendingApprovals: List<ApprovalDto>,
        val lastSequence: Long,
@@ -138,10 +138,18 @@ class GlobalStreamHandler(private val module: ServerModule) {
        msg: ClientMessage.ApprovalResponse,
        sendFrame: suspend (ServerMessage) -> Unit,
    ) {
-        // TODO: scopeSessionId is a placeholder derived from requestId — the global socket
-        // is not bound to a single session. Switch to a proper requestId → sessionId lookup
-        // (via ApprovalCoordinator or the approval repository) once that mapping is exposed.
-        val scopeSessionId: SessionId = TypeId(msg.requestId.value)
+        val scopeSessionId = module.approvalCoordinator.lookupSession(msg.requestId)
+        if (scopeSessionId == null) {
+            log.warn("handleApprovalResponse: no session found for requestId={}", msg.requestId.value)
+            sendFrame(
+                ServerMessage.ProtocolError(
+                    message = "Unknown approval request",
+                    sequence = null,
+                    sessionSequence = null,
+                ),
+            )
+            return
+        }
        module.approvalCoordinator.handleResponse(msg, scopeSessionId)?.let { sendFrame(it) }
    }

@@ -248,11 +256,18 @@ internal suspend fun streamGlobal(
        signaled.collect { buffer.send(it) }
    }

+    // Per-session sequence counters so live events carry the correct sessionSequence
+    // and are not silently dropped by the TUI's SnapshotPhaseReducer dedup filter.
+    val sessionSequences = mutableMapOf<String, Long>()
+
    try {
        subscribed.await()
        bridge.replaySnapshot()
        for (event in buffer) {
-            val msg = mapper.map(event) ?: continue
+            val sid = event.metadata.sessionId.value
+            val seq = sessionSequences.getOrDefault(sid, 0L) + 1
+            sessionSequences[sid] = seq
+            val msg = mapper.map(event, sessionSequence = seq) ?: continue
            sendFrame(msg)
        }
    } finally {
@@ -175,4 +175,36 @@ class ApprovalCoordinatorWiringTest {
        assertEquals(requestId, received.requestId)
        subscription.cancel()
    }
+
+    @Test
+    fun `lookupSession returns mapping registered by onApprovalRequested and cleared by handleResponse`() = runBlocking {
+        val gateway = RecordingGateway()
+        val coord = ApprovalCoordinator(gateway, ApprovalConfig(timeoutMs = 300_000L), scope)
+        val otherSessionId = SessionId("other-session")
+
+        // Initially no mapping
+        assertNull(coord.lookupSession(requestId))
+
+        // After onApprovalRequested, lookupSession returns the registered sessionId
+        val event = ApprovalRequestedEvent(
+            requestId = requestId,
+            tier = Tier.T2,
+            validationReportId = ValidationReportId("vr-1"),
+            riskSummaryId = null,
+            sessionId = otherSessionId,
+            stageId = null,
+            projectId = null,
+        )
+        coord.onApprovalRequested(event)
+        assertEquals(otherSessionId, coord.lookupSession(requestId))
+
+        // After handleResponse resolves the request, the mapping is cleaned up
+        val msg = ClientMessage.ApprovalResponse(
+            requestId = requestId,
+            decision = ApprovalDecision.APPROVE,
+            steeringNote = null,
+        )
+        coord.handleResponse(msg, otherSessionId)
+        assertNull(coord.lookupSession(requestId))
+    }
 }
@@ -39,6 +39,7 @@ class ServerMessageSerializationTest {
    fun `SessionSnapshot encodes lastSequence and lastSessionSequence`() {
        val msg = ServerMessage.SessionSnapshot(
            sessionId = SessionId("sess-1"),
+            workflowId = "wf-abc",
            state = SessionStateDto(
                status = "running",
                currentStageId = "stage-1",
@@ -34,8 +34,9 @@ fun eventHistoryStripWidget(state: TuiState): Paragraph {
            }
            add(Line.from(Span.styled(stageInfo, dimStyle)))

-            // Recent events (plain text, compact format)
-            val events = selectedSession.recentEvents
+            // Recent events (plain text, compact format) — capped to fit allocated height.
+            // Line 0 = stage info, lines 1-3 = events (max 3).
+            val events = selectedSession.recentEvents.take(3)
            if (events.isEmpty()) {
                add(Line.from(Span.styled("no events", dimStyle)))
            } else {
@@ -50,33 +51,6 @@ fun eventHistoryStripWidget(state: TuiState): Paragraph {
                    add(Line.from(Span.raw(row)))
                }
            }
-
-            // Tool manifest for current stage (from toolsByStage)
-            val stageId = selectedSession.currentStageId
-            if (stageId != null) {
-                val tools = selectedSession.toolsByStage[stageId]
-                if (!tools.isNullOrEmpty()) {
-                    add(Line.from(Span.styled("declared tools:", dimStyle)))
-                    for (tool in tools) {
-                        add(Line.from(Span.raw("  ${tool.name} T${tool.tier}")))
-                    }
-                }
-            }
-
-            // Runtime tool records (TuiToolRecord from SessionSummary.tools)
-            if (selectedSession.tools.isNotEmpty()) {
-                add(Line.from(Span.styled("tool records:", dimStyle)))
-                for (tool in selectedSession.tools) {
-                    val statusSymbol = when (tool.status) {
-                        ToolDisplayStatus.REQUESTED -> "?"
-                        ToolDisplayStatus.STARTED -> "\u25B6"
-                        ToolDisplayStatus.COMPLETED -> "\u2713"
-                        ToolDisplayStatus.FAILED -> "\u2717"
-                        ToolDisplayStatus.REJECTED -> "\u2298"
-                    }
-                    add(Line.from(Span.raw("  $statusSymbol ${tool.name} T${tool.tier}")))
-                }
-            }
        }
    }

@@ -43,6 +43,7 @@ object KeyResolver {
                KeyEvent.Enter -> if (inputText.isBlank()) null else Action.SubmitInput
                KeyEvent.Tab -> Action.CycleMode
                KeyEvent.Backspace -> Action.Backspace
+                KeyEvent.NewSession -> Action.OpenNewSessionPrompt
                is KeyEvent.CharInput -> Action.AppendChar(key.ch)
                else -> null
            }
@@ -54,6 +55,9 @@ object KeyResolver {
                KeyEvent.Tab -> Action.CycleMode
                KeyEvent.Backspace -> Action.Backspace
                is KeyEvent.CharInput -> Action.AppendChar(key.ch)
+                KeyEvent.Cancel -> Action.CancelSelectedSession
+                KeyEvent.ReturnToSessionList -> Action.ReturnToSessionList
+                KeyEvent.ToggleEventStrip -> Action.ToggleEventStrip
                KeyEvent.ShowPendingApproval -> if (hasPendingApproval) Action.ShowPendingApproval else null
                else -> null
            }
@@ -61,6 +65,9 @@ object KeyResolver {
            DisplayState.APPROVAL -> when (key) {
                KeyEvent.Enter -> Action.SubmitApprovalDecision
                KeyEvent.Escape -> Action.CancelInput
+                KeyEvent.Cancel -> Action.CancelInput
+                KeyEvent.ReturnToSessionList -> Action.ReturnToSessionList
+                KeyEvent.ToggleEventStrip -> Action.ToggleEventStrip
                KeyEvent.Tab -> Action.CycleMode
                KeyEvent.Backspace -> Action.Backspace
                is KeyEvent.CharInput -> Action.AppendChar(key.ch)
@@ -104,27 +104,43 @@ object RootReducer {

            is Action.CancelInput -> when {
                prevInputMode == InputMode.FILTER -> withBgReset
-                else -> withBgReset.copy(approvalDismissed = true)
+                else -> withBgReset.copy(approvalDismissed = true, sessionEntered = false)
            }

+            is Action.ReturnToSessionList -> withBgReset.copy(sessionEntered = false)
+
            is Action.ToggleEventStrip -> {
                if (withBgReset.displayState == DisplayState.IDLE) withBgReset
                else withBgReset.copy(eventStripVisible = !withBgReset.eventStripVisible)
            }

-            // Input history append on SubmitInput in IN_SESSION + ROUTER (ARCH-HISTORY-1).
+            // Input history append on SubmitInput in IN_SESSION + ROUTER (ARCH-HISTORY-1),
+            // or enter a selected session from the list in IDLE.
            is Action.SubmitInput -> {
                val text = prevInputBuffer.trim()
-                if (text.isNotBlank() && sessions.selectedId != null &&
-                    prevInputMode == InputMode.ROUTER &&
-                    withBgReset.displayState == com.correx.apps.tui.state.DisplayState.IN_SESSION
-                ) {
-                    val selectedId = sessions.selectedId!!
-                    val currentHistory = withBgReset.inputHistory[selectedId] ?: emptyList()
-                    val updatedHistory = (currentHistory + text).takeLast(50)
-                    withBgReset.copy(
-                        inputHistory = withBgReset.inputHistory + (selectedId to updatedHistory),
-                    )
+                when {
+                    text.isBlank() && sessions.selectedId != null && prevInputMode != InputMode.FILTER ->
+                        withBgReset.copy(sessionEntered = true)
+
+                    text.isNotBlank() && sessions.selectedId != null &&
+                        prevInputMode == InputMode.ROUTER &&
+                        withBgReset.displayState == com.correx.apps.tui.state.DisplayState.IN_SESSION -> {
+                        val selectedId = sessions.selectedId!!
+                        val currentHistory = withBgReset.inputHistory[selectedId] ?: emptyList()
+                        val updatedHistory = (currentHistory + text).takeLast(50)
+                        withBgReset.copy(
+                            inputHistory = withBgReset.inputHistory + (selectedId to updatedHistory),
+                        )
+                    }
+
+                    else -> withBgReset
+                }
+            }
+
+            // Auto-enter a newly started session when SessionStarted arrives.
+            is Action.ServerEventReceived -> {
+                if (action.message is ServerMessage.SessionStarted) {
+                    withBgReset.copy(sessionEntered = true)
                } else {
                    withBgReset
                }
@@ -212,16 +212,16 @@ object SessionsReducer {
                sessionId = msg.sessionId.value,
                tier = it.tier,
                riskSummary = "unknown",
-                toolName = null,
-                preview = null,
+                toolName = it.toolName,
+                preview = it.preview,
            )
        }
        val status = if (firstPending != null) "PAUSED awaiting approval" else msg.state.status
        val summary = SessionSummary(
            id = msg.sessionId.value,
            status = status,
-            workflowId = msg.sessionId.value,
-            name = msg.sessionId.value,
+            workflowId = msg.workflowId,
+            name = msg.workflowId,
            lastEventAt = clock(),
            currentStage = msg.state.currentStageId,
            currentStageId = msg.state.currentStageId,
@@ -10,6 +10,7 @@ val TuiState.displayState: DisplayState
        return when {
            approvalDismissed -> DisplayState.IN_SESSION
            hasApproval -> DisplayState.APPROVAL
-            else -> DisplayState.IN_SESSION
+            sessionEntered -> DisplayState.IN_SESSION
+            else -> DisplayState.IDLE
        }
    }
@@ -35,6 +35,12 @@ data class TuiState(
    val savedInputBuffer: String = "",
    val pendingDecision: ApprovalDecision? = null,
    val approvalDismissed: Boolean = false,
+    /**
+     * True when the user has explicitly "entered" a session (pressed Enter on it or
+     * just started a new one). False when simply navigating the session list with ↑↓.
+     * Drives [displayState]: IN_SESSION requires both [selectedId] and [sessionEntered].
+     */
+    val sessionEntered: Boolean = false,
    val currentModel: String? = null,
    val providerType: ProviderType = ProviderType.LOCAL,
    val routerConnected: Boolean = false,
@@ -146,6 +146,7 @@ class RootReducerTest {
    fun `NavigateUp in IN_SESSION at index -1 saves buffer and loads last history entry`() {
        val state = TuiState(
            snapshotPhase = false,
+            sessionEntered = true,
            sessions = SessionsState(
                sessions = listOf(session("s1")),
                selectedId = "s1",
@@ -165,6 +166,7 @@ class RootReducerTest {
    fun `NavigateUp in IN_SESSION from index above 0 decrements index and loads entry`() {
        val state = TuiState(
            snapshotPhase = false,
+            sessionEntered = true,
            sessions = SessionsState(
                sessions = listOf(session("s1")),
                selectedId = "s1",
@@ -235,6 +237,7 @@ class RootReducerTest {
    fun `NavigateDown in IN_SESSION at last entry restores savedInputBuffer and resets index`() {
        val state = TuiState(
            snapshotPhase = false,
+            sessionEntered = true,
            sessions = SessionsState(
                sessions = listOf(session("s1")),
                selectedId = "s1",
@@ -253,6 +256,7 @@ class RootReducerTest {
    fun `NavigateDown in IN_SESSION from below last entry increments index`() {
        val state = TuiState(
            snapshotPhase = false,
+            sessionEntered = true,
            sessions = SessionsState(
                sessions = listOf(session("s1")),
                selectedId = "s1",
@@ -324,6 +328,7 @@ class RootReducerTest {
    fun `ApprovalRequired on non-selected session does not transition to APPROVAL`() {
        val initial = TuiState(
            snapshotPhase = false,
+            sessionEntered = true,
            sessions = SessionsState(
                sessions = listOf(session("s1"), session("s2")),
                selectedId = "s1",
@@ -300,6 +300,7 @@ class SessionsReducerTest {
    fun `SessionSnapshot with pendingApprovals populates pendingApproval on summary`() {
        val msg = ServerMessage.SessionSnapshot(
            sessionId = SessionId("s1"),
+            workflowId = "wf-1",
            state = SessionStateDto("PAUSED", null, null),
            pendingApprovals = listOf(ApprovalDto(requestId = "r9", tier = "T2")),
            lastSequence = 1L,
@@ -353,6 +354,7 @@ class SessionsReducerTest {
    fun `SessionSnapshot without pendingApprovals leaves pendingApproval null`() {
        val msg = ServerMessage.SessionSnapshot(
            sessionId = SessionId("s1"),
+            workflowId = "wf-1",
            state = SessionStateDto("ACTIVE", null, null),
            pendingApprovals = emptyList(),
            lastSequence = 1L,
@@ -108,6 +108,7 @@ class SnapshotPhaseReducerTest {
        val state = TuiState(snapshotPhase = true)
        val snapshot = ServerMessage.SessionSnapshot(
            sessionId = SessionId("s1"),
+            workflowId = "wf-1",
            state = SessionStateDto(status = "ACTIVE", currentStageId = null, pauseReason = null),
            pendingApprovals = emptyList(),
            lastSequence = 1L,