From a04cd1fa2405c183ef2a094757c37918885bac32 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 12 Jun 2026 14:25:31 +0400 Subject: [PATCH] =?UTF-8?q?feat(server,cli):=20event=20stream=20inspector?= =?UTF-8?q?=20=E2=80=94=20/sessions/{id}/events=20+=20correx=20events?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../kotlin/com/correx/apps/cli/CorrexCli.kt | 2 + .../correx/apps/cli/commands/EventsCommand.kt | 85 +++++ .../apps/cli/commands/EventRenderTest.kt | 64 ++++ .../apps/server/routes/SessionRoutes.kt | 50 ++- .../server/routes/EventInspectRoutesTest.kt | 298 ++++++++++++++++++ 5 files changed, 487 insertions(+), 12 deletions(-) create mode 100644 apps/cli/src/main/kotlin/com/correx/apps/cli/commands/EventsCommand.kt create mode 100644 apps/cli/src/test/kotlin/com/correx/apps/cli/commands/EventRenderTest.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/routes/EventInspectRoutesTest.kt diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt index 4dd24b73..a5530a86 100644 --- a/apps/cli/src/main/kotlin/com/correx/apps/cli/CorrexCli.kt +++ b/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(), ) diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/EventsCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/EventsCommand.kt new file mode 100644 index 00000000..b072f927 --- /dev/null +++ b/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): 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>() + 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() + } +} diff --git a/apps/cli/src/test/kotlin/com/correx/apps/cli/commands/EventRenderTest.kt b/apps/cli/src/test/kotlin/com/correx/apps/cli/commands/EventRenderTest.kt new file mode 100644 index 00000000..2351a5de --- /dev/null +++ b/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")) + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt b/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt index e7164b5f..796f403b 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/routes/SessionRoutes.kt +++ b/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 events = module.eventStore.readFrom(TypeId(id), from) - call.respond( - events.map { e -> - EventResponse( - eventId = e.metadata.eventId.value, - sequence = e.sequence, - sessionId = e.metadata.sessionId.value, + val fromSeq = call.request.queryParameters["fromSeq"]?.toLongOrNull() ?: 0L + val limit = call.request.queryParameters["limit"]?.toIntOrNull() ?: DEFAULT_EVENT_LIMIT + val typeFilter = call.request.queryParameters["type"] + val rows = module.eventStore.readFrom(TypeId(id), fromSeq) + .asSequence() + .map { stored -> + val payloadJson = eventJson.encodeToJsonElement(EventPayload.serializer(), stored.payload) + 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) } } diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/routes/EventInspectRoutesTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/routes/EventInspectRoutesTest.kt new file mode 100644 index 00000000..16291ab5 --- /dev/null +++ b/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) : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = events.filter { it.metadata.sessionId == sessionId } + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = + 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 = emptyFlow() + override fun subscribeAll(): Flow = emptyFlow() + override suspend fun lastGlobalSequence(): Long = 0L + override fun allEvents(): Sequence = emptySequence() + override fun allSessionIds(): Set = 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() + override suspend fun healthCheckAll() = emptyMap() + } + val noopWorkflowRegistry = object : WorkflowRegistry { + override fun listAll(): List = 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) + } + } +}