feat(server,cli): event stream inspector — /sessions/{id}/events + correx events

apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt

@@ -1,6 +1,7 @@
 package com.correx.apps.cli
 
 import com.correx.apps.cli.commands.ApproveCommand
+import com.correx.apps.cli.commands.EventsCommand
 import com.correx.apps.cli.commands.ProviderCommand
 import com.correx.apps.cli.commands.RunCommand
 import com.correx.apps.cli.commands.SessionCommand
@@ -25,4 +26,5 @@ fun buildCli(): CorrexCli = CorrexCli().subcommands(
     StatusCommand(),
     ProviderCommand(),
     UndoCommand(),
+    EventsCommand(),
 )

apps/cli/src/main/kotlin/com/correx/apps/cli/commands/EventsCommand.kt

@@ -0,0 +1,85 @@
+package com.correx.apps.cli.commands
+
+import com.correx.apps.cli.CorrexCli
+import com.correx.apps.cli.DEFAULT_PORT
+import com.github.ajalt.clikt.core.CliktCommand
+import com.github.ajalt.clikt.parameters.arguments.argument
+import com.github.ajalt.clikt.parameters.options.default
+import com.github.ajalt.clikt.parameters.options.option
+import io.ktor.client.HttpClient
+import io.ktor.client.call.body
+import io.ktor.client.engine.cio.CIO
+import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
+import io.ktor.client.request.get
+import io.ktor.serialization.kotlinx.json.json
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.builtins.ListSerializer
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonElement
+
+@Serializable
+data class EventRow(
+    val sequence: Long,
+    val eventId: String,
+    val type: String,
+    val timestamp: String,
+    val causationId: String?,
+    val correlationId: String?,
+    val payload: JsonElement,
+)
+
+fun renderEventRows(rows: List<EventRow>): String {
+    if (rows.isEmpty()) return "(no events)"
+    val header = "%-6s  %-40s  %-30s  %s".format("seq", "type", "timestamp", "causation→")
+    val lines = rows.map { r ->
+        "%-6d  %-40s  %-30s  %s".format(
+            r.sequence,
+            r.type,
+            r.timestamp,
+            r.causationId ?: "-",
+        )
+    }
+    return (listOf(header) + lines).joinToString("\n")
+}
+
+class EventsCommand : CliktCommand(name = "events") {
+    private val sessionId by argument("sessionId")
+    private val type by option("--type")
+    private val fromSeq by option("--from-seq")
+    private val limit by option("--limit")
+    private val host by option("--host").default("localhost")
+    private val port by option("--port").default("$DEFAULT_PORT")
+
+    private val eventsJson = Json { ignoreUnknownKeys = true }
+
+    override fun run(): Unit = runBlocking {
+        val portInt = port.toIntOrNull() ?: DEFAULT_PORT
+        val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false
+
+        val params = buildList {
+            type?.let { add("type=$it") }
+            fromSeq?.let { add("fromSeq=$it") }
+            limit?.let { add("limit=$it") }
+        }.joinToString("&")
+        val query = if (params.isNotEmpty()) "?$params" else ""
+        val url = "http://$host:$portInt/sessions/$sessionId/events$query"
+
+        val client = HttpClient(CIO) {
+            install(ContentNegotiation) { json(eventsJson) }
+        }
+
+        runCatching {
+            val rows = client.get(url).body<List<EventRow>>()
+            if (outputJson) {
+                println(eventsJson.encodeToString(ListSerializer(EventRow.serializer()), rows))
+            } else {
+                println(renderEventRows(rows))
+            }
+        }.getOrElse { e ->
+            System.err.println("Error fetching events: ${e.message}")
+        }
+
+        client.close()
+    }
+}

apps/cli/src/test/kotlin/com/correx/apps/cli/commands/EventRenderTest.kt

@@ -0,0 +1,64 @@
+package com.correx.apps.cli.commands
+
+import kotlinx.serialization.json.JsonObject
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class EventRenderTest {
+
+    private val rows = listOf(
+        EventRow(
+            sequence = 1L,
+            eventId = "e1",
+            type = "WorkspaceStateObserved",
+            timestamp = "2026-06-12T10:00:00Z",
+            causationId = null,
+            correlationId = null,
+            payload = JsonObject(emptyMap()),
+        ),
+        EventRow(
+            sequence = 2L,
+            eventId = "e2",
+            type = "InitialIntent",
+            timestamp = "2026-06-12T10:01:00Z",
+            causationId = "c1",
+            correlationId = null,
+            payload = JsonObject(emptyMap()),
+        ),
+    )
+
+    @Test
+    fun `renderEventRows includes sequence numbers in order`() {
+        val output = renderEventRows(rows)
+        val lines = output.lines().filter { it.isNotBlank() }
+        // header + 2 data rows
+        assertTrue(lines.size >= 3)
+        assertTrue(lines[1].trimStart().startsWith("1"))
+        assertTrue(lines[2].trimStart().startsWith("2"))
+    }
+
+    @Test
+    fun `renderEventRows includes type column content`() {
+        val output = renderEventRows(rows)
+        assertTrue(output.contains("WorkspaceStateObserved"))
+        assertTrue(output.contains("InitialIntent"))
+    }
+
+    @Test
+    fun `renderEventRows shows causation id when present`() {
+        val output = renderEventRows(rows)
+        assertTrue(output.contains("c1"))
+    }
+
+    @Test
+    fun `renderEventRows shows dash when causation id absent`() {
+        val output = renderEventRows(rows)
+        assertTrue(output.contains("-"))
+    }
+
+    @Test
+    fun `renderEventRows returns no-events message for empty list`() {
+        val output = renderEventRows(emptyList())
+        assertTrue(output.contains("no events"))
+    }
+}

apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt

@@ -3,6 +3,8 @@ package com.correx.apps.server.routes
 import com.correx.apps.server.ServerModule
 import com.correx.apps.server.protocol.SessionConfigDto
 import com.correx.apps.server.ws.SessionStreamHandler
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.serialization.eventJson
 import com.correx.core.events.types.SessionId
 import com.correx.core.utils.TypeId
 import io.ktor.http.HttpStatusCode
@@ -14,9 +16,25 @@ import io.ktor.server.routing.post
 import io.ktor.server.routing.route
 import io.ktor.server.websocket.webSocket
 import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonPrimitive
 import org.slf4j.LoggerFactory
 import java.util.*
 
+internal const val DEFAULT_EVENT_LIMIT = 200
+
+@Serializable
+data class EventRow(
+    val sequence: Long,
+    val eventId: String,
+    val type: String,
+    val timestamp: String,
+    val causationId: String?,
+    val correlationId: String?,
+    val payload: JsonElement,
+)
+
 @Serializable
 data class StartSessionRequest(val workflowId: String, val config: SessionConfigDto? = null)
 
@@ -34,9 +52,6 @@ data class SessionSummaryResponse(
 @Serializable
 data class SessionStateResponse(val sessionId: String, val status: String, val createdAt: String?)
 
-@Serializable
-data class EventResponse(val eventId: String, val sequence: Long, val sessionId: String)
-
 @Serializable
 data class StartSessionResponse(val sessionId: String)
 
@@ -140,16 +155,27 @@ private fun Route.getEventsRoute(module: ServerModule) {
     get("/events") {
         val id = call.parameters["id"]
             ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id")
-        val from = call.request.queryParameters["from"]?.toLongOrNull() ?: 0L
+        val fromSeq = call.request.queryParameters["fromSeq"]?.toLongOrNull() ?: 0L
-        val events = module.eventStore.readFrom(TypeId(id), from)
+        val limit = call.request.queryParameters["limit"]?.toIntOrNull() ?: DEFAULT_EVENT_LIMIT
-        call.respond(
+        val typeFilter = call.request.queryParameters["type"]
-            events.map { e ->
+        val rows = module.eventStore.readFrom(TypeId(id), fromSeq)
-                EventResponse(
+            .asSequence()
-                    eventId = e.metadata.eventId.value,
+            .map { stored ->
-                    sequence = e.sequence,
+                val payloadJson = eventJson.encodeToJsonElement(EventPayload.serializer(), stored.payload)
-                    sessionId = e.metadata.sessionId.value,
+                val type = (payloadJson as? JsonObject)?.get("type")?.jsonPrimitive?.content ?: "unknown"
+                EventRow(
+                    sequence = stored.sequence,
+                    eventId = stored.metadata.eventId.value,
+                    type = type,
+                    timestamp = stored.metadata.timestamp.toString(),
+                    causationId = stored.metadata.causationId?.value,
+                    correlationId = stored.metadata.correlationId?.value,
+                    payload = payloadJson,
                 )
             }
-        )
+            .filter { typeFilter == null || it.type == typeFilter }
+            .take(limit)
+            .toList()
+        call.respond(rows)
     }
 }

apps/server/src/test/kotlin/com/correx/apps/server/routes/EventInspectRoutesTest.kt

@@ -0,0 +1,298 @@
+package com.correx.apps.server.routes
+
+import com.correx.apps.server.ServerModule
+import com.correx.apps.server.configureServer
+import com.correx.apps.server.registry.ProviderRegistry
+import com.correx.apps.server.registry.WorkflowRegistry
+import com.correx.apps.server.registry.WorkflowSummary
+import com.correx.apps.server.undo.SessionUndoService
+import com.correx.core.artifactstore.ArtifactStore
+import com.correx.core.context.builder.DefaultContextPackBuilder
+import com.correx.core.context.compression.DefaultContextCompressor
+import com.correx.core.events.EventDispatcher
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.InitialIntentEvent
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.StoredEvent
+import com.correx.core.events.events.WorkspaceStateObservedEvent
+import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.ArtifactId
+import com.correx.core.events.types.EventId
+import com.correx.core.events.types.SessionId
+import com.correx.core.inference.InferenceProjector
+import com.correx.core.inference.InferenceProvider
+import com.correx.core.inference.InferenceRepository
+import com.correx.core.inference.ProviderHealth
+import com.correx.core.kernel.orchestration.DefaultOrchestrationReducer
+import com.correx.core.kernel.orchestration.DefaultSessionOrchestrator
+import com.correx.core.kernel.orchestration.OrchestrationConfig
+import com.correx.core.kernel.orchestration.OrchestrationProjector
+import com.correx.core.kernel.orchestration.OrchestrationRepository
+import com.correx.core.kernel.orchestration.OrchestratorEngines
+import com.correx.core.kernel.orchestration.OrchestratorRepositories
+import com.correx.core.kernel.retry.DefaultRetryCoordinator
+import com.correx.core.events.types.ProviderId
+import com.correx.core.risk.DefaultRiskAssessor
+import com.correx.core.router.model.RouterConfig
+import com.correx.core.sessions.DefaultSessionReducer
+import com.correx.core.sessions.DefaultSessionRepository
+import com.correx.core.sessions.SessionProjector
+import com.correx.core.sessions.projections.replay.DefaultEventReplayer
+import com.correx.core.transitions.graph.WorkflowGraph
+import com.correx.core.transitions.resolution.DefaultTransitionResolver
+import com.correx.core.validation.pipeline.ValidationPipeline
+import com.correx.infrastructure.InfrastructureModule
+import com.correx.infrastructure.inference.commons.UnavailableProbe
+import com.correx.infrastructure.tools.FileEditConfig
+import com.correx.infrastructure.tools.FileReadConfig
+import com.correx.infrastructure.tools.FileWriteConfig
+import com.correx.infrastructure.tools.ShellConfig
+import com.correx.infrastructure.tools.ToolConfig
+import io.ktor.client.request.get
+import io.ktor.client.statement.bodyAsText
+import io.ktor.http.HttpStatusCode
+import io.ktor.server.testing.testApplication
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.datetime.Clock
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonArray
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.jsonPrimitive
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.io.TempDir
+import java.nio.file.Path
+
+class EventInspectRoutesTest {
+
+    private val testJson = Json { ignoreUnknownKeys = true }
+
+    private val s1 = SessionId("s1")
+
+    private fun storedEvent(payload: com.correx.core.events.events.EventPayload, seq: Long): StoredEvent =
+        StoredEvent(
+            metadata = EventMetadata(
+                eventId = EventId("e$seq"),
+                sessionId = s1,
+                timestamp = Clock.System.now(),
+                schemaVersion = 1,
+                causationId = null,
+                correlationId = null,
+            ),
+            sequence = seq,
+            sessionSequence = seq,
+            payload = payload,
+        )
+
+    private val seedEvents = listOf(
+        storedEvent(WorkspaceStateObservedEvent(sessionId = s1, workspaceRoot = "/repo", stateKey = "git:abc", source = "git"), 1L),
+        storedEvent(WorkspaceStateObservedEvent(sessionId = s1, workspaceRoot = "/repo", stateKey = "git:def", source = "git"), 2L),
+        storedEvent(InitialIntentEvent(sessionId = s1, intent = "build the thing"), 3L),
+    )
+
+    private class SeededEventStore(private val events: List<StoredEvent>) : EventStore {
+        override suspend fun append(event: NewEvent): StoredEvent = error("unused")
+        override suspend fun appendAll(events: List<NewEvent>): List<StoredEvent> = error("unused")
+        override fun read(sessionId: SessionId): List<StoredEvent> = events.filter { it.metadata.sessionId == sessionId }
+        override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> =
+            events.filter { it.metadata.sessionId == sessionId && it.sequence >= fromSequence }
+        override fun lastSequence(sessionId: SessionId): Long? = events.filter { it.metadata.sessionId == sessionId }.lastOrNull()?.sequence
+        override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = emptyFlow()
+        override fun subscribeAll(): Flow<StoredEvent> = emptyFlow()
+        override suspend fun lastGlobalSequence(): Long = 0L
+        override fun allEvents(): Sequence<StoredEvent> = emptySequence()
+        override fun allSessionIds(): Set<SessionId> = events.map { it.metadata.sessionId }.toSet()
+    }
+
+    private fun buildModule(eventStore: EventStore, tempDir: Path): ServerModule {
+        val artifactStore = object : ArtifactStore {
+            override suspend fun put(bytes: ByteArray): ArtifactId = ArtifactId("noop")
+            override suspend fun get(id: ArtifactId): ByteArray? = null
+            override suspend fun flushBefore(commit: suspend () -> Unit) = commit()
+        }
+        val provider = InfrastructureModule.createLlamaCppProvider(
+            modelId = "test-model",
+            modelPath = "/dev/null",
+            baseUrl = "http://127.0.0.1:1",
+        )
+        val providerRegistry = InfrastructureModule.createProviderRegistry(listOf(provider))
+        val inferenceRouter = com.correx.core.inference.DefaultInferenceRouter(
+            providerRegistry,
+            com.correx.infrastructure.inference.FirstAvailableRoutingStrategy(),
+        )
+        val toolConfig = ToolConfig(
+            shell = ShellConfig(enabled = false, allowedExecutables = emptySet(), workingDir = tempDir),
+            fileRead = FileReadConfig(enabled = false, allowedPaths = setOf(tempDir)),
+            fileWrite = FileWriteConfig(enabled = false, allowedPaths = setOf(tempDir), workingDir = tempDir),
+            fileEdit = FileEditConfig(enabled = false, allowedPaths = setOf(tempDir), workingDir = tempDir),
+        )
+        val toolRegistry = InfrastructureModule.createToolRegistry(toolConfig)
+        val eventDispatcher = EventDispatcher(eventStore)
+        val toolExecutor = InfrastructureModule.createToolExecutor(
+            registry = toolRegistry,
+            eventDispatcher = eventDispatcher,
+            workDir = tempDir,
+            artifactStore = null,
+        )
+        val engines = OrchestratorEngines(
+            transitionResolver = DefaultTransitionResolver { condition, ctx -> condition.evaluate(ctx) },
+            contextPackBuilder = DefaultContextPackBuilder(DefaultContextCompressor()),
+            inferenceRouter = inferenceRouter,
+            validationPipeline = ValidationPipeline(validators = emptyList()),
+            approvalEngine = com.correx.core.approvals.domain.DefaultApprovalEngine(),
+            riskAssessor = DefaultRiskAssessor(),
+            toolRegistry = toolRegistry,
+            toolExecutor = toolExecutor,
+        )
+        val repositories = OrchestratorRepositories(
+            eventStore = eventStore,
+            inferenceRepository = InferenceRepository(DefaultEventReplayer(eventStore, InferenceProjector())),
+            orchestrationRepository = OrchestrationRepository(
+                DefaultEventReplayer(eventStore, OrchestrationProjector(DefaultOrchestrationReducer())),
+            ),
+            sessionRepository = DefaultSessionRepository(
+                DefaultEventReplayer(eventStore, SessionProjector(DefaultSessionReducer())),
+            ),
+            artifactRepository = InfrastructureModule.createArtifactRepository(eventStore),
+            approvalRepository = InfrastructureModule.createApprovalRepository(eventStore),
+        )
+        val orchestrator = DefaultSessionOrchestrator(
+            repositories = repositories,
+            engines = engines,
+            retryCoordinator = DefaultRetryCoordinator(eventStore),
+            artifactStore = artifactStore,
+            tokenizer = provider.tokenizer,
+            decisionJournalRepository = InfrastructureModule.createDecisionJournalRepository(eventStore),
+        )
+        val routerFacade = InfrastructureModule.createRouterFacade(
+            eventStore = eventStore,
+            inferenceRouter = inferenceRouter,
+            config = RouterConfig(),
+            tokenizer = provider.tokenizer,
+        )
+        val noopProviderRegistry = object : ProviderRegistry {
+            override fun listAll() = emptyList<InferenceProvider>()
+            override suspend fun healthCheckAll() = emptyMap<ProviderId, ProviderHealth>()
+        }
+        val noopWorkflowRegistry = object : WorkflowRegistry {
+            override fun listAll(): List<WorkflowSummary> = emptyList()
+            override fun find(workflowId: String): WorkflowGraph? = null
+        }
+        val sessionUndoService = SessionUndoService(
+            eventStore = eventStore,
+            artifactStore = artifactStore,
+            bootRoots = setOf(tempDir),
+        )
+        return ServerModule(
+            orchestrator = orchestrator,
+            eventStore = eventStore,
+            artifactStore = artifactStore,
+            sessionRepository = repositories.sessionRepository,
+            workflowRegistry = noopWorkflowRegistry,
+            providerRegistry = noopProviderRegistry,
+            defaultOrchestrationConfig = OrchestrationConfig(sandboxRoot = tempDir),
+            routerFacade = routerFacade,
+            orchestrationRepository = repositories.orchestrationRepository,
+            approvalRepository = repositories.approvalRepository,
+            toolRegistry = toolRegistry,
+            sessionUndoService = sessionUndoService,
+            resourceProbe = UnavailableProbe,
+        )
+    }
+
+    @Test
+    fun `GET sessions id events returns all events ascending sequence`(@TempDir tempDir: Path) {
+        val eventStore = SeededEventStore(seedEvents)
+        val module = buildModule(eventStore, tempDir)
+
+        testApplication {
+            application { configureServer(module) }
+            val client = createClient {}
+
+            val response = client.get("/sessions/s1/events")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val array = testJson.parseToJsonElement(response.bodyAsText()) as JsonArray
+            assertEquals(3, array.size)
+            assertEquals(1L, (array[0] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+            assertEquals(2L, (array[1] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+            assertEquals(3L, (array[2] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+        }
+    }
+
+    @Test
+    fun `GET sessions id events with type filter returns matching events only`(@TempDir tempDir: Path) {
+        val eventStore = SeededEventStore(seedEvents)
+        val module = buildModule(eventStore, tempDir)
+
+        testApplication {
+            application { configureServer(module) }
+            val client = createClient {}
+
+            val response = client.get("/sessions/s1/events?type=InitialIntent")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val array = testJson.parseToJsonElement(response.bodyAsText()) as JsonArray
+            assertEquals(1, array.size)
+            assertEquals("InitialIntent", (array[0] as JsonObject)["type"]!!.jsonPrimitive.content)
+        }
+    }
+
+    @Test
+    fun `GET sessions id events with fromSeq skips earlier sequences`(@TempDir tempDir: Path) {
+        val eventStore = SeededEventStore(seedEvents)
+        val module = buildModule(eventStore, tempDir)
+
+        testApplication {
+            application { configureServer(module) }
+            val client = createClient {}
+
+            val response = client.get("/sessions/s1/events?fromSeq=2")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val array = testJson.parseToJsonElement(response.bodyAsText()) as JsonArray
+            assertEquals(2, array.size)
+            assertEquals(2L, (array[0] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+            assertEquals(3L, (array[1] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+        }
+    }
+
+    @Test
+    fun `GET sessions id events with limit truncates after filtering`(@TempDir tempDir: Path) {
+        val eventStore = SeededEventStore(seedEvents)
+        val module = buildModule(eventStore, tempDir)
+
+        testApplication {
+            application { configureServer(module) }
+            val client = createClient {}
+
+            val response = client.get("/sessions/s1/events?limit=1")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val array = testJson.parseToJsonElement(response.bodyAsText()) as JsonArray
+            assertEquals(1, array.size)
+            assertEquals(1L, (array[0] as JsonObject)["sequence"]!!.jsonPrimitive.content.toLong())
+
+            val filtered = client.get("/sessions/s1/events?type=WorkspaceStateObserved&limit=1")
+            val filteredArray = testJson.parseToJsonElement(filtered.bodyAsText()) as JsonArray
+            assertEquals(1, filteredArray.size)
+            assertEquals(
+                "WorkspaceStateObserved",
+                (filteredArray[0] as JsonObject)["type"]!!.jsonPrimitive.content,
+            )
+        }
+    }
+
+    @Test
+    fun `GET sessions id events for unknown session returns empty array`(@TempDir tempDir: Path) {
+        val eventStore = SeededEventStore(seedEvents)
+        val module = buildModule(eventStore, tempDir)
+
+        testApplication {
+            application { configureServer(module) }
+            val client = createClient {}
+
+            val response = client.get("/sessions/no-such-session/events")
+            assertEquals(HttpStatusCode.OK, response.status)
+            val array = testJson.parseToJsonElement(response.bodyAsText()) as JsonArray
+            assertEquals(0, array.size)
+        }
+    }
+}