feat(server,tui): session stats pane — metrics over the WS bus + Bubble Tea overlay

Surfaces the observability metrics (observability-spec §3-tier-2) in the Go TUI:
a session-summary pane showing duration, token throughput per provider, tool
time, approval latency per tier, failure counts, and the session wall-time
accounting split. Press `S` (or palette → "session stats").

Mirrors the artifacts/config fetch pattern exactly — a WS request/response, not a
new transport. Server-side reuses the already-tested MetricsInspectionService
(one MetricsReport definition, two wires: REST `correx stats` + WS).

Server:
- ClientMessage.GetSessionStats + ServerMessage.SessionStats(@SerialName
  session.stats), reusing MetricsReport as the nested payload.
- StreamQueries.sessionStats replays via MetricsInspectionService (pure read,
  replay-neutral); routed in GlobalStreamHandler.
- ServerMessageSerializationTest golden pins the session.stats wire format.

TUI (Go/Bubble Tea):
- protocol: TypeSessionStats + StatsDto/nested structs + GetSessionStats encoder
  + golden decode test (cross-language contract).
- OverlayStats pane (statsModal), `S` key + palette entry + footer hint,
  loading/cache-by-session state, bounded breakdown rows.
- demo `stats` preview case for serverless visual verification.

apps/server/src/main/kotlin/com/correx/apps/server/protocol/ClientMessage.kt

@@ -42,6 +42,10 @@ sealed class ClientMessage {
     @Serializable
     data class ListArtifacts(val sessionId: SessionId) : ClientMessage()
 
+    /** Operator request for a session's derived metrics (replied to with a SessionStats). */
+    @Serializable
+    data class GetSessionStats(val sessionId: SessionId) : ClientMessage()
+
     /** Operator request for the current editable config (replied to with a ConfigSnapshot). */
     @Serializable
     data object GetConfig : ClientMessage()

apps/server/src/main/kotlin/com/correx/apps/server/protocol/ServerMessage.kt

@@ -1,5 +1,6 @@
 package com.correx.apps.server.protocol
 
+import com.correx.apps.server.metrics.MetricsReport
 import com.correx.core.approvals.Tier
 import com.correx.core.events.types.ApprovalRequestId
 import com.correx.core.events.types.ArtifactId
@@ -425,6 +426,20 @@ sealed interface ServerMessage {
         override val sessionSequence: Long? = null,
     ) : ServerMessage, NonEventMessage
 
+    /**
+     * A session's derived metrics, returned in response to a GetSessionStats request. Not
+     * event-derived (it is a projection-replay snapshot), so cursors are null. [stats] is the same
+     * [MetricsReport] the REST `/sessions/{id}/metrics` endpoint serves — one definition, two wires.
+     */
+    @Serializable
+    @SerialName("session.stats")
+    data class SessionStats(
+        val sessionId: SessionId,
+        val stats: MetricsReport,
+        override val sequence: Long? = null,
+        override val sessionSequence: Long? = null,
+    ) : ServerMessage, NonEventMessage
+
     /**
      * The current editable config (reply to GetConfig / UpdateConfig). [restartRequired] lists the
      * keys in a just-applied patch that need a server restart to take effect; [error] is set (and

apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt

@@ -229,6 +229,7 @@ class GlobalStreamHandler(private val module: ServerModule) {
                     .onFailure { log.error("cancel failed for session={}: {}", msg.sessionId.value, it.message, it) }
             }
             is ClientMessage.ListArtifacts -> sendFrame(queries.listArtifacts(msg.sessionId))
+            is ClientMessage.GetSessionStats -> sendFrame(queries.sessionStats(msg.sessionId))
             is ClientMessage.GetConfig -> sendFrame(queries.configSnapshot(error = null, restartRequired = emptyList()))
             is ClientMessage.UpdateConfig -> sendFrame(queries.updateConfig(msg.patch))
             is ClientMessage.ResumeSession -> session.send(encodeError("ResumeSession not supported"))

apps/server/src/main/kotlin/com/correx/apps/server/ws/StreamQueries.kt

@@ -2,6 +2,7 @@ package com.correx.apps.server.ws
 
 import com.correx.apps.server.ServerModule
 import com.correx.apps.server.config.ConfigUpdateResult
+import com.correx.apps.server.metrics.MetricsInspectionService
 import com.correx.apps.server.protocol.ArtifactSummaryDto
 import com.correx.apps.server.protocol.ConfigFieldDto
 import com.correx.apps.server.protocol.ServerMessage
@@ -33,6 +34,18 @@ private val lenientJson = Json { ignoreUnknownKeys = true; isLenient = true }
  */
 class StreamQueries(private val module: ServerModule) {
 
+    private val metricsService = MetricsInspectionService(module.eventStore)
+
+    /**
+     * A session's derived metrics as a snapshot frame. Replays the event log through
+     * [MetricsInspectionService] (the same projection the REST metrics endpoint uses) — pure read,
+     * no events appended, so it is replay-neutral (invariant #1/#8).
+     */
+    suspend fun sessionStats(sessionId: SessionId): ServerMessage.SessionStats {
+        val report = withContext(Dispatchers.IO) { metricsService.inspect(sessionId) }
+        return ServerMessage.SessionStats(sessionId = sessionId, stats = report)
+    }
+
     /**
      * Reads a session's events to assemble its artifact listing (creation order preserved),
      * resolving each artifact's stored bytes via the CAS artifact store. Pure snapshot read —

apps/server/src/test/kotlin/com/correx/apps/server/protocol/ServerMessageSerializationTest.kt

@@ -1,5 +1,10 @@
 package com.correx.apps.server.protocol
 
+import com.correx.apps.server.metrics.FailureMetrics
+import com.correx.apps.server.metrics.MetricsReport
+import com.correx.apps.server.metrics.ProviderStatsDto
+import com.correx.apps.server.metrics.TierApprovalStatsDto
+import com.correx.apps.server.metrics.ToolStatsDto
 import com.correx.core.events.types.SessionId
 import kotlinx.serialization.json.Json
 import org.junit.jupiter.api.Assertions.assertEquals
@@ -146,6 +151,48 @@ class ServerMessageSerializationTest {
         assertEquals(listOf("server.port"), decoded.restartRequired)
     }
 
+    @Test
+    fun `SessionStats encodes type, sessionId and nested metrics`() {
+        val msg = ServerMessage.SessionStats(
+            sessionId = SessionId("sess-stats"),
+            stats = MetricsReport(
+                sessionId = "sess-stats",
+                eventCount = 12,
+                sessionDurationMs = 60_000,
+                inferenceCount = 3,
+                inferenceMs = 6000,
+                promptTokens = 310,
+                completionTokens = 170,
+                tokensPerSecond = 28.3,
+                perProvider = listOf(ProviderStatsDto("lfm:a", 2, 5000, 300, 150, 30.0)),
+                toolCount = 3,
+                toolMs = 1500,
+                perTool = listOf(ToolStatsDto("shell.read", 2, 1200)),
+                approvalsRequested = 2,
+                approvalsResolved = 1,
+                approvalsPending = 1,
+                approvalWaitMs = 30_000,
+                avgApprovalWaitMs = 30_000,
+                perTier = listOf(TierApprovalStatsDto("T2", 2, 1, 30_000, 30_000)),
+                failures = FailureMetrics(inferenceFailures = 1, toolFailures = 1, workflowFailures = 1),
+                inferencePct = 10.0,
+                toolPct = 2.5,
+                approvalWaitPct = 50.0,
+            ),
+        )
+        val jsonStr = ProtocolSerializer.encodeServerMessage(msg)
+        assert(jsonStr.contains("\"type\":\"session.stats\"")) { "expected type=session.stats" }
+        assert(jsonStr.contains("\"sessionId\":\"sess-stats\"")) { "expected sessionId" }
+        assert(jsonStr.contains("\"completionTokens\":170")) { "expected nested metric field" }
+        assert(jsonStr.contains("\"provider\":\"lfm:a\"")) { "expected per-provider row" }
+
+        val decoded = json.decodeFromString<ServerMessage.SessionStats>(jsonStr)
+        assertEquals("sess-stats", decoded.stats.sessionId)
+        assertEquals(3, decoded.stats.inferenceCount)
+        assertEquals(1, decoded.stats.perProvider.size)
+        assertEquals(50.0, decoded.stats.approvalWaitPct)
+    }
+
     @Test
     fun `SessionAnnounced round-trips through JSON`() {
         val original = ServerMessage.SessionAnnounced(