From 64b58e3b05db7f2447a2282bb54635ab0c82ef9b Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 12 Jun 2026 14:32:24 +0400 Subject: [PATCH] =?UTF-8?q?feat(server,cli):=20session=20replay=20inspecti?= =?UTF-8?q?on=20=E2=80=94=20timeline=20+=20determinism=20digest?= 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/ReplayCommand.kt | 95 +++++++++ .../server/replay/ReplayInspectionService.kt | 94 +++++++++ .../apps/server/routes/SessionRoutes.kt | 12 ++ .../replay/ReplayInspectionServiceTest.kt | 194 ++++++++++++++++++ 5 files changed, 397 insertions(+) create mode 100644 apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ReplayCommand.kt create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/replay/ReplayInspectionService.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/replay/ReplayInspectionServiceTest.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 a5530a86..729e1598 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 @@ -3,6 +3,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.ReplayCommand import com.correx.apps.cli.commands.RunCommand import com.correx.apps.cli.commands.SessionCommand import com.correx.apps.cli.commands.StatusCommand @@ -27,4 +28,5 @@ fun buildCli(): CorrexCli = CorrexCli().subcommands( ProviderCommand(), UndoCommand(), EventsCommand(), + ReplayCommand(), ) diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ReplayCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ReplayCommand.kt new file mode 100644 index 00000000..ae51c750 --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/ReplayCommand.kt @@ -0,0 +1,95 @@ +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.core.ProgramResult +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.client.statement.bodyAsText +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json + +@Serializable +data class TimelineEntryDto( + val sequence: Long, + val type: String, + val stageId: String?, + val summary: String, +) + +@Serializable +data class ReplayReportDto( + val sessionId: String, + val eventCount: Long, + val timeline: List, + val digest: String, + val deterministic: Boolean, +) + +private const val SEPARATOR_WIDTH = 110 + +private fun renderReport(report: ReplayReportDto): String { + val lines = mutableListOf() + lines += "Session: ${report.sessionId}" + lines += "Events: ${report.eventCount}" + lines += "Digest: ${report.digest}" + lines += "Deterministic: ${report.deterministic}" + lines += "" + lines += "%-6s %-35s %-20s %s".format("seq", "type", "stageId", "summary") + lines += "-".repeat(SEPARATOR_WIDTH) + for (entry in report.timeline) { + lines += "%-6d %-35s %-20s %s".format( + entry.sequence, + entry.type, + entry.stageId ?: "-", + entry.summary, + ) + } + return lines.joinToString("\n") +} + +class ReplayCommand : CliktCommand(name = "replay") { + private val sessionId by argument("sessionId") + private val host by option("--host").default("localhost") + private val port by option("--port").default("$DEFAULT_PORT") + + private val replayJson = Json { ignoreUnknownKeys = true } + + override fun run(): Unit = runBlocking { + val portInt = port.toIntOrNull() ?: DEFAULT_PORT + val outputJson = (currentContext.findRoot().command as? CorrexCli)?.json ?: false + val url = "http://$host:$portInt/sessions/$sessionId/replay" + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(replayJson) } + } + + val nonDeterministic = runCatching { + if (outputJson) { + val body = client.get(url).bodyAsText() + println(body) + val parsed = replayJson.decodeFromString(body) + !parsed.deterministic + } else { + val report = client.get(url).body() + println(renderReport(report)) + !report.deterministic + } + }.getOrElse { e -> + System.err.println("Error fetching replay: ${e.message}") + false + } + + client.close() + if (nonDeterministic) throw ProgramResult(1) + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/replay/ReplayInspectionService.kt b/apps/server/src/main/kotlin/com/correx/apps/server/replay/ReplayInspectionService.kt new file mode 100644 index 00000000..28245e46 --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/replay/ReplayInspectionService.kt @@ -0,0 +1,94 @@ +package com.correx.apps.server.replay + +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.OrchestrationPausedEvent +import com.correx.core.events.events.OrchestrationResumedEvent +import com.correx.core.events.events.RetryAttemptedEvent +import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.events.StageFailedEvent +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.TransitionExecutedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.serialization.eventJson +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.SessionId +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import java.security.MessageDigest + +@Serializable +data class TimelineEntry( + val sequence: Long, + val type: String, + val stageId: String?, + val summary: String, +) + +@Serializable +data class ReplayReport( + val sessionId: String, + val eventCount: Long, + val timeline: List, + val digest: String, + val deterministic: Boolean, +) + +class ReplayInspectionService(private val eventStore: EventStore) { + + fun inspect(sessionId: SessionId): ReplayReport { + val events = eventStore.read(sessionId) + val timeline = events.map { stored -> + val element = eventJson.encodeToJsonElement(EventPayload.serializer(), stored.payload) + val type = element.jsonObject["type"]?.jsonPrimitive?.content ?: "unknown" + toTimelineEntry(stored.sequence, stored.payload, type) + } + val digest = computeDigest(sessionId) + val digest2 = computeDigest(sessionId) + return ReplayReport( + sessionId = sessionId.value, + eventCount = events.size.toLong(), + timeline = timeline, + digest = digest, + deterministic = digest == digest2, + ) + } + + private fun toTimelineEntry(sequence: Long, payload: EventPayload, type: String): TimelineEntry = + when (payload) { + is WorkflowStartedEvent -> TimelineEntry(sequence, type, payload.startStageId.value, "Workflow started: ${payload.workflowId}") + is StageCompletedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Stage completed: ${payload.stageId.value}") + is StageFailedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Stage failed: ${payload.stageId.value} — ${payload.reason}") + is ToolExecutionCompletedEvent -> TimelineEntry(sequence, type, null, "Tool executed: ${payload.toolName}") + is ToolExecutionFailedEvent -> TimelineEntry(sequence, type, null, "Tool failed: ${payload.toolName} — ${payload.reason}") + is ToolExecutionRejectedEvent -> TimelineEntry(sequence, type, null, "Tool rejected: ${payload.toolName} — ${payload.reason}") + is ApprovalRequestedEvent -> TimelineEntry(sequence, type, payload.stageId?.value, "Approval requested: tier ${payload.tier}") + is ApprovalDecisionResolvedEvent -> TimelineEntry(sequence, type, null, "Approval resolved: ${payload.outcome}") + is OrchestrationPausedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Orchestration paused at ${payload.stageId.value}: ${payload.reason}") + is OrchestrationResumedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Orchestration resumed at ${payload.stageId.value}") + is RetryAttemptedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Retry attempt ${payload.attemptNumber}/${payload.maxAttempts} at ${payload.stageId.value}: ${payload.failureReason}") + is TransitionExecutedEvent -> TimelineEntry(sequence, type, payload.from.value, "Transition ${payload.from.value} → ${payload.to.value}") + is WorkflowCompletedEvent -> TimelineEntry(sequence, type, payload.terminalStageId.value, "Workflow completed after ${payload.totalStages} stages") + is WorkflowFailedEvent -> TimelineEntry(sequence, type, payload.stageId.value, "Workflow failed: ${payload.reason}") + else -> TimelineEntry(sequence, type, null, type) + } + + private fun computeDigest(sessionId: SessionId): String { + val events = eventStore.read(sessionId) + val digest = MessageDigest.getInstance("SHA-256") + for (event in events) { + val element = eventJson.encodeToJsonElement(EventPayload.serializer(), event.payload) + val type = element.jsonObject["type"]?.jsonPrimitive?.content ?: "unknown" + val payloadJsonString = eventJson.encodeToString(EventPayload.serializer(), event.payload) + val line = "${event.sequence}|$type|$payloadJsonString" + digest.update(line.toByteArray(Charsets.UTF_8)) + } + return digest.digest().joinToString("") { "%02x".format(it) } + } +} 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 796f403b..c120b65b 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 @@ -2,6 +2,7 @@ package com.correx.apps.server.routes import com.correx.apps.server.ServerModule import com.correx.apps.server.protocol.SessionConfigDto +import com.correx.apps.server.replay.ReplayInspectionService import com.correx.apps.server.ws.SessionStreamHandler import com.correx.core.events.events.EventPayload import com.correx.core.events.serialization.eventJson @@ -81,6 +82,7 @@ fun Route.sessionRoutes(module: ServerModule) { undoSessionRoute(module) resumeSessionRoute(module) getEventsRoute(module) + getReplayRoute(module) webSocket("/stream") { val id = call.parameters["id"] ?: return@webSocket val lastEventId = call.request.queryParameters["lastEventId"]?.toLongOrNull() @@ -179,3 +181,13 @@ private fun Route.getEventsRoute(module: ServerModule) { call.respond(rows) } } + +private fun Route.getReplayRoute(module: ServerModule) { + val service = ReplayInspectionService(module.eventStore) + get("/replay") { + val id = call.parameters["id"] + ?: return@get call.respond(HttpStatusCode.BadRequest, "Missing session id") + val report = service.inspect(TypeId(id)) + call.respond(report) + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/replay/ReplayInspectionServiceTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/replay/ReplayInspectionServiceTest.kt new file mode 100644 index 00000000..abff67f5 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/replay/ReplayInspectionServiceTest.kt @@ -0,0 +1,194 @@ +package com.correx.apps.server.replay + +import com.correx.core.approvals.ApprovalOutcome +import com.correx.core.approvals.ApprovalStatus +import com.correx.core.approvals.Tier +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.EventMetadata +import com.correx.core.events.events.EventPayload +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.TransitionExecutedEvent +import com.correx.core.events.events.WorkflowCompletedEvent +import com.correx.core.events.events.WorkflowStartedEvent +import com.correx.core.events.events.StageCompletedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.ApprovalDecisionId +import com.correx.core.events.types.ApprovalRequestId +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.events.types.StageId +import com.correx.core.events.types.TransitionId +import com.correx.core.utils.TypeId +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.datetime.Clock +import kotlinx.datetime.Instant +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Assertions.assertFalse +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class ReplayInspectionServiceTest { + + private val sessionId: SessionId = TypeId("test-session-1") + private val stageId: StageId = TypeId("stage-1") + private val toStageId: StageId = TypeId("stage-2") + private val transitionId: TransitionId = TypeId("t-1") + private val requestId: ApprovalRequestId = TypeId("req-1") + private val decisionId: ApprovalDecisionId = TypeId("dec-1") + + private fun storedEvent(payload: EventPayload, seq: Long): StoredEvent = + StoredEvent( + metadata = EventMetadata( + eventId = EventId("e$seq"), + sessionId = sessionId, + timestamp = Clock.System.now(), + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = seq, + sessionSequence = seq, + payload = payload, + ) + + private val seedEvents = listOf( + storedEvent(WorkflowStartedEvent(sessionId = sessionId, workflowId = "wf-1", startStageId = stageId), 1L), + storedEvent(StageCompletedEvent(sessionId = sessionId, stageId = stageId, transitionId = transitionId), 2L), + storedEvent(ApprovalRequestedEvent( + requestId = requestId, + tier = Tier.T2, + validationReportId = TypeId("vr-1"), + riskSummaryId = null, + sessionId = sessionId, + stageId = stageId, + projectId = null, + ), 3L), + storedEvent(ApprovalDecisionResolvedEvent( + decisionId = decisionId, + requestId = requestId, + outcome = ApprovalOutcome.APPROVED, + status = ApprovalStatus.COMPLETED, + tier = Tier.T2, + resolutionTimestamp = Instant.parse("2026-01-01T00:00:00Z"), + reason = null, + ), 4L), + storedEvent(WorkflowCompletedEvent(sessionId = sessionId, terminalStageId = stageId, totalStages = 1), 5L), + storedEvent(TransitionExecutedEvent( + sessionId = sessionId, + from = stageId, + to = toStageId, + transitionId = transitionId, + ), 6L), + ) + + 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() + } + + @Test + fun `inspect returns correct event count and timeline`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + assertEquals(sessionId.value, report.sessionId) + assertEquals(6L, report.eventCount) + assertEquals(6, report.timeline.size) + } + + @Test + fun `timeline type uses SerialName discriminator not simpleName`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + assertEquals("WorkflowStarted", report.timeline[0].type) + assertEquals("StageCompleted", report.timeline[1].type) + assertEquals("WorkflowCompleted", report.timeline[4].type) + } + + @Test + fun `timeline entries carry stageId where applicable`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + val workflowEntry = report.timeline[0] + assertEquals(stageId.value, workflowEntry.stageId) + + val stageEntry = report.timeline[1] + assertEquals(stageId.value, stageEntry.stageId) + + val approvalEntry = report.timeline[2] + assertEquals(stageId.value, approvalEntry.stageId) + + val decisionEntry = report.timeline[3] + assertEquals(null, decisionEntry.stageId) + + val completedEntry = report.timeline[4] + assertEquals(stageId.value, completedEntry.stageId) + } + + @Test + fun `TransitionExecutedEvent summary includes from and to stages`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + val transitionEntry = report.timeline[5] + assertEquals("TransitionExecuted", transitionEntry.type) + assertTrue(transitionEntry.summary.contains(stageId.value), "summary should mention from-stage") + assertTrue(transitionEntry.summary.contains(toStageId.value), "summary should mention to-stage") + assertFalse(transitionEntry.summary == "TransitionExecuted", "summary should not be the bare catch-all type") + } + + @Test + fun `digest is 64 hex characters`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + assertEquals(64, report.digest.length) + assertTrue(report.digest.matches(Regex("[0-9a-f]+"))) + } + + @Test + fun `digest is stable across two inspect calls`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report1 = service.inspect(sessionId) + val report2 = service.inspect(sessionId) + + assertEquals(report1.digest, report2.digest) + } + + @Test + fun `deterministic is true for stable store`() { + val service = ReplayInspectionService(SeededEventStore(seedEvents)) + val report = service.inspect(sessionId) + + assertTrue(report.deterministic) + } + + @Test + fun `deterministic is false when store returns different events per read`() { + val unstableStore = object : EventStore by SeededEventStore(seedEvents) { + private var reads = 0 + override fun read(sessionId: SessionId): List { + reads += 1 + return if (reads % 2 == 0) seedEvents.dropLast(1) else seedEvents + } + } + val service = ReplayInspectionService(unstableStore) + val report = service.inspect(sessionId) + + assertFalse(report.deterministic) + } +}