From 450b492937932038d63c1f0e7cecbb67e1f053c8 Mon Sep 17 00:00:00 2001 From: kami Date: Tue, 26 May 2026 13:22:48 +0400 Subject: [PATCH] fix(tui): status bar model, snapshot events, session-start selection Three fixes: 1. Status bar now shows model/provider from ProviderStatusChanged: - RootReducer sets currentModel and providerType when the message arrives - providerType is LOCAL for llama/local providers, REMOTE otherwise - Removes the stale '(no model) (local)' display 2. SessionSnapshot now carries recent event history: - New EventEntryDto in the protocol with timestamp/type/detail - SessionSnapshot.recentEvents populated by bridge from event store - Bridge reads last 7 display-relevant events per session - TUI processSessionSnapshotMessage maps them to TuiEventEntry - Completed sessions finally show their event history 3. SessionStarted now switches selectedId to the new session: - RootReducer cross-field weave updates selectedId alongside sessionEntered - Previously kept the old selectedId, navigating user into the wrong session - Works whether starting a session from the list or during replay --- .../apps/server/bridge/SessionEventBridge.kt | 40 ++++++++++++++++ .../com/correx/apps/server/protocol/Dtos.kt | 7 +++ .../apps/server/protocol/ServerMessage.kt | 1 + .../correx/apps/tui/reducer/RootReducer.kt | 23 ++++++++-- .../apps/tui/reducer/SessionsReducer.kt | 8 ++++ .../apps/tui/reducer/RootReducerTest.kt | 46 +++++++++++++++++++ 6 files changed, 121 insertions(+), 4 deletions(-) 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 d9f7d430..8f30b5c3 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 @@ -1,6 +1,7 @@ package com.correx.apps.server.bridge import com.correx.apps.server.protocol.ApprovalDto +import com.correx.apps.server.protocol.EventEntryDto import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.server.protocol.SessionStateDto import com.correx.apps.server.protocol.StageToolDecl @@ -8,6 +9,20 @@ import com.correx.apps.server.protocol.ToolDecl import com.correx.apps.server.registry.WorkflowRegistry import com.correx.core.approvals.DefaultApprovalRepository import com.correx.core.artifactstore.ArtifactStore +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.OrchestrationPausedEvent +import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.events.StageFailedEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.ToolExecutionCompletedEvent +import com.correx.core.events.events.ToolExecutionFailedEvent +import com.correx.core.events.events.ToolExecutionRejectedEvent +import com.correx.core.events.events.ToolInvocationRequestedEvent +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowFailedEvent import com.correx.core.events.orchestration.OrchestrationStatus import com.correx.core.events.stores.EventStore import com.correx.core.events.types.SessionId @@ -48,6 +63,10 @@ class SessionEventBridge( } ?: emptyList() + val recentEvents = eventStore.read(sessionId) + .mapNotNull { eventToEntry(it) } + .takeLast(7) + send(ServerMessage.SessionSnapshot( sessionId = sessionId, workflowId = orchState.workflowId, @@ -57,6 +76,7 @@ class SessionEventBridge( pauseReason = orchState.pauseReason, ), pendingApprovals = pendingApprovals, + recentEvents = recentEvents, lastSequence = lastGlobal, lastSessionSequence = lastSession, )) @@ -82,6 +102,26 @@ class SessionEventBridge( send(ServerMessage.SnapshotComplete) } + private fun eventToEntry(event: StoredEvent): EventEntryDto? { + val ts = event.metadata.timestamp.toEpochMilliseconds() + return when (val p = event.payload) { + is TransitionExecutedEvent -> EventEntryDto(ts, "StageStarted", p.to.value) + is StageCompletedEvent -> EventEntryDto(ts, "StageCompleted", p.stageId.value) + is StageFailedEvent -> EventEntryDto(ts, "StageFailed", p.stageId.value) + is InferenceStartedEvent -> EventEntryDto(ts, "InferenceStarted", p.stageId.value) + is InferenceCompletedEvent -> EventEntryDto(ts, "InferenceCompleted", p.stageId.value) + is InferenceTimeoutEvent -> EventEntryDto(ts, "InferenceTimeout", p.stageId.value) + is ToolInvocationRequestedEvent -> EventEntryDto(ts, "ToolStarted", p.toolName) + is ToolExecutionCompletedEvent -> EventEntryDto(ts, "ToolCompleted", p.toolName) + is ToolExecutionFailedEvent -> EventEntryDto(ts, "ToolFailed", p.toolName) + is ToolExecutionRejectedEvent -> EventEntryDto(ts, "ToolRejected", p.toolName) + is OrchestrationPausedEvent -> EventEntryDto(ts, "SessionPaused", p.reason) + is WorkflowCompletedEvent -> EventEntryDto(ts, "SessionCompleted", "") + is WorkflowFailedEvent -> EventEntryDto(ts, "SessionFailed", p.reason) + else -> null + } + } + suspend fun streamLive(sessionId: SessionId) { var sessionSequence = 0L eventStore.subscribe(sessionId).collect { event -> diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt index 3527a05b..86c59a90 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/Dtos.kt @@ -54,3 +54,10 @@ data class ToolDecl( val name: String, val tier: Int, ) + +@Serializable +data class EventEntryDto( + val timestamp: Long, + val type: String, + val detail: String, +) diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt index 9cc6d76b..aa21062f 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt +++ b/apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt @@ -79,6 +79,7 @@ sealed interface ServerMessage { val workflowId: String, val state: SessionStateDto, val pendingApprovals: List, + val recentEvents: List = emptyList(), val lastSequence: Long, val lastSessionSequence: Long, override val sequence: Long? = null, diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt index 21e21a2e..cbd60c89 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/RootReducer.kt @@ -5,6 +5,7 @@ import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.tui.input.Action import com.correx.apps.tui.state.DisplayState import com.correx.apps.tui.state.InputMode +import com.correx.apps.tui.state.ProviderType import com.correx.apps.tui.state.TuiState import com.correx.apps.tui.state.displayState import org.slf4j.LoggerFactory @@ -138,11 +139,25 @@ object RootReducer { } // Auto-enter a newly started session when SessionStarted arrives. + // When a ProviderStatusChanged arrives, update model/provider labels. is Action.ServerEventReceived -> { - if (action.message is ServerMessage.SessionStarted) { - withBgReset.copy(sessionEntered = true) - } else { - withBgReset + val msg = action.message + when { + msg is ServerMessage.SessionStarted -> { + withBgReset.copy( + sessionEntered = true, + sessions = withBgReset.sessions.copy(selectedId = msg.sessionId.value), + ) + } + msg is ServerMessage.ProviderStatusChanged -> { + withBgReset.copy( + currentModel = msg.providerId, + providerType = if (msg.providerId.contains("llama", ignoreCase = true) || + msg.providerId.contains("local", ignoreCase = true) + ) ProviderType.LOCAL else ProviderType.REMOTE, + ) + } + else -> withBgReset } } diff --git a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt index 01ec440e..b353688f 100644 --- a/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt +++ b/apps/tui/src/main/kotlin/com/correx/apps/tui/reducer/SessionsReducer.kt @@ -217,6 +217,13 @@ object SessionsReducer { ) } val status = if (firstPending != null) "PAUSED awaiting approval" else msg.state.status + val eventEntries = msg.recentEvents.map { entry -> + TuiEventEntry( + timestamp = formatTime(entry.timestamp), + type = entry.type, + detail = entry.detail, + ) + } val summary = SessionSummary( id = msg.sessionId.value, status = status, @@ -226,6 +233,7 @@ object SessionsReducer { currentStage = msg.state.currentStageId, currentStageId = msg.state.currentStageId, pendingApproval = approvalInfo, + recentEvents = eventEntries, ) val selected = sessionState.selectedId ?: msg.sessionId.value return sessionState.copy( diff --git a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt index 53644cde..a26783c1 100644 --- a/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt +++ b/apps/tui/src/test/kotlin/com/correx/apps/tui/reducer/RootReducerTest.kt @@ -1,12 +1,14 @@ package com.correx.apps.tui.reducer import com.correx.apps.server.protocol.ClientMessage +import com.correx.apps.server.protocol.ProviderHealthDto import com.correx.apps.server.protocol.RiskSummaryDto import com.correx.apps.server.protocol.ServerMessage import com.correx.apps.tui.input.Action import com.correx.apps.tui.state.ConnectionState import com.correx.apps.tui.state.DisplayState import com.correx.apps.tui.state.InputMode +import com.correx.apps.tui.state.ProviderType import com.correx.apps.tui.state.SessionSummary import com.correx.apps.tui.state.SessionsState import com.correx.apps.tui.state.TuiState @@ -341,4 +343,48 @@ class RootReducerTest { assertEquals(null, state.sessions.sessions.find { it.id == "s1" }?.pendingApproval) assertEquals("r1", state.sessions.sessions.find { it.id == "s2" }?.pendingApproval?.requestId) } + + @Test + fun `ProviderStatusChanged updates currentModel and providerType`() { + val initial = TuiState(snapshotPhase = false, currentModel = null, providerType = ProviderType.LOCAL) + val msg = ServerMessage.ProviderStatusChanged( + providerId = "llama.cpp", + status = ProviderHealthDto("llama.cpp", "healthy", null), + ) + val (state, _) = RootReducer.reduce(initial, Action.ServerEventReceived(msg), fixedClock) + assertEquals("llama.cpp", state.currentModel) + assertEquals(ProviderType.LOCAL, state.providerType) + } + + @Test + fun `ProviderStatusChanged for remote provider sets providerType to REMOTE`() { + val initial = TuiState(snapshotPhase = false, currentModel = null, providerType = ProviderType.LOCAL) + val msg = ServerMessage.ProviderStatusChanged( + providerId = "openai", + status = ProviderHealthDto("openai", "healthy", null), + ) + val (state, _) = RootReducer.reduce(initial, Action.ServerEventReceived(msg), fixedClock) + assertEquals("openai", state.currentModel) + assertEquals(ProviderType.REMOTE, state.providerType) + } + + @Test + fun `SessionStarted switches selectedId to new session when another session is selected`() { + val initial = TuiState( + snapshotPhase = false, + sessions = SessionsState( + sessions = listOf(session("s1"), session("s2")), + selectedId = "s1", + ), + ) + val startedMsg = ServerMessage.SessionStarted( + sessionId = SessionId("s2"), + workflowId = "wf", + sequence = 1L, + sessionSequence = 1L, + ) + val (state, _) = RootReducer.reduce(initial, Action.ServerEventReceived(startedMsg), fixedClock) + assertEquals("s2", state.sessions.selectedId) + assertTrue(state.sessionEntered) + } }