feat(server,cli): session replay inspection — timeline + determinism digest

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(),
 )

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<TimelineEntryDto>,
+    val digest: String,
+    val deterministic: Boolean,
+)
+
+private const val SEPARATOR_WIDTH = 110
+
+private fun renderReport(report: ReplayReportDto): String {
+    val lines = mutableListOf<String>()
+    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<ReplayReportDto>(body)
+                !parsed.deterministic
+            } else {
+                val report = client.get(url).body<ReplayReportDto>()
+                println(renderReport(report))
+                !report.deterministic
+            }
+        }.getOrElse { e ->
+            System.err.println("Error fetching replay: ${e.message}")
+            false
+        }
+
+        client.close()
+        if (nonDeterministic) throw ProgramResult(1)
+    }
+}

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<TimelineEntry>,
+    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) }
+    }
+}

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)
+    }
+}

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<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()
+    }
+
+    @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<StoredEvent> {
+                reads += 1
+                return if (reads % 2 == 0) seedEvents.dropLast(1) else seedEvents
+            }
+        }
+        val service = ReplayInspectionService(unstableStore)
+        val report = service.inspect(sessionId)
+
+        assertFalse(report.deterministic)
+    }
+}