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 729e1598..60a3a7a6 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 @@ -6,6 +6,7 @@ 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.StatsCommand import com.correx.apps.cli.commands.StatusCommand import com.correx.apps.cli.commands.UndoCommand import com.github.ajalt.clikt.core.CliktCommand @@ -29,4 +30,5 @@ fun buildCli(): CorrexCli = CorrexCli().subcommands( UndoCommand(), EventsCommand(), ReplayCommand(), + StatsCommand(), ) diff --git a/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatsCommand.kt b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatsCommand.kt new file mode 100644 index 00000000..2683febf --- /dev/null +++ b/apps/cli/src/main/kotlin/com/correx/apps/cli/commands/StatsCommand.kt @@ -0,0 +1,182 @@ +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.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 FailureMetricsDto( + val inferenceFailures: Long = 0, + val inferenceTimeouts: Long = 0, + val toolFailures: Long = 0, + val toolRejections: Long = 0, + val stageFailures: Long = 0, + val workflowFailures: Long = 0, +) + +@Serializable +data class ProviderStatsDto( + val provider: String, + val completedCount: Long, + val totalLatencyMs: Long, + val promptTokens: Long, + val completionTokens: Long, + val tokensPerSecond: Double, +) + +@Serializable +data class ToolStatsDto( + val toolName: String, + val completedCount: Long, + val totalDurationMs: Long, +) + +@Serializable +data class TierApprovalStatsDto( + val tier: String, + val requestedCount: Long, + val resolvedCount: Long, + val totalWaitMs: Long, + val avgWaitMs: Long, +) + +@Serializable +data class StatsReportDto( + val sessionId: String, + val eventCount: Long = 0, + val sessionDurationMs: Long = 0, + val inferenceCount: Long = 0, + val inferenceMs: Long = 0, + val promptTokens: Long = 0, + val completionTokens: Long = 0, + val tokensPerSecond: Double = 0.0, + val perProvider: List = emptyList(), + val toolCount: Long = 0, + val toolMs: Long = 0, + val perTool: List = emptyList(), + val approvalsRequested: Long = 0, + val approvalsResolved: Long = 0, + val approvalsPending: Long = 0, + val approvalWaitMs: Long = 0, + val avgApprovalWaitMs: Long = 0, + val perTier: List = emptyList(), + val failures: FailureMetricsDto = FailureMetricsDto(), + val inferencePct: Double = 0.0, + val toolPct: Double = 0.0, + val approvalWaitPct: Double = 0.0, +) + +private const val MS_PER_SECOND = 1000L +private const val SECONDS_PER_MINUTE = 60L +private const val MINUTES_PER_HOUR = 60L + +private fun humanDuration(ms: Long): String { + if (ms <= 0L) return "0s" + val totalSeconds = ms / MS_PER_SECOND + val hours = totalSeconds / (SECONDS_PER_MINUTE * MINUTES_PER_HOUR) + val minutes = (totalSeconds / SECONDS_PER_MINUTE) % MINUTES_PER_HOUR + val seconds = totalSeconds % SECONDS_PER_MINUTE + return buildString { + if (hours > 0) append("${hours}h ") + if (hours > 0 || minutes > 0) append("${minutes}m ") + append("${seconds}s") + }.trim() +} + +private fun otherPct(report: StatsReportDto): Double = + (100.0 - report.inferencePct - report.toolPct - report.approvalWaitPct).coerceAtLeast(0.0) + +fun renderStats(report: StatsReportDto): String { + val lines = mutableListOf() + lines += "Session: ${report.sessionId}" + lines += "Events: ${report.eventCount}" + lines += "Duration: ${humanDuration(report.sessionDurationMs)}" + lines += "" + + lines += "Inference" + lines += " calls: ${report.inferenceCount}" + lines += " tokens: ${report.promptTokens} prompt + ${report.completionTokens} completion = " + + "${report.promptTokens + report.completionTokens}" + lines += " time: ${report.inferenceMs} ms (%.1f tok/s)".format(report.tokensPerSecond) + for (p in report.perProvider) { + lines += " %-20s %4d calls %6d tok %8d ms (%.1f tok/s)".format( + p.provider, p.completedCount, p.promptTokens + p.completionTokens, p.totalLatencyMs, p.tokensPerSecond, + ) + } + lines += "" + + lines += "Tools" + lines += " calls: ${report.toolCount} time: ${report.toolMs} ms" + for (t in report.perTool) { + lines += " %-24s %4d %8d ms".format(t.toolName, t.completedCount, t.totalDurationMs) + } + lines += "" + + lines += "Approvals" + lines += " requested: ${report.approvalsRequested} resolved: ${report.approvalsResolved} " + + "pending: ${report.approvalsPending}" + lines += " total wait: ${report.approvalWaitMs} ms (avg ${report.avgApprovalWaitMs} ms)" + for (tier in report.perTier) { + lines += " %-4s req %d res %d wait %d ms (avg %d ms)".format( + tier.tier, tier.requestedCount, tier.resolvedCount, tier.totalWaitMs, tier.avgWaitMs, + ) + } + lines += "" + + val f = report.failures + lines += "Failures" + lines += " inference: ${f.inferenceFailures} timeouts: ${f.inferenceTimeouts} " + + "tool: ${f.toolFailures} rejected: ${f.toolRejections} " + + "stage: ${f.stageFailures} workflow: ${f.workflowFailures}" + lines += "" + + lines += "Time accounting (% of session wall time)" + lines += " inference: %.1f%% tools: %.1f%% approval-wait: %.1f%% other: %.1f%%".format( + report.inferencePct, report.toolPct, report.approvalWaitPct, otherPct(report), + ) + + return lines.joinToString("\n") +} + +class StatsCommand : CliktCommand(name = "stats") { + 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 statsJson = 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/metrics" + + val client = HttpClient(CIO) { + install(ContentNegotiation) { json(statsJson) } + } + + runCatching { + if (outputJson) { + println(client.get(url).bodyAsText()) + } else { + println(renderStats(client.get(url).body())) + } + }.onFailure { e -> + System.err.println("Error fetching stats: ${e.message}") + } + + client.close() + } +} diff --git a/apps/cli/src/test/kotlin/com/correx/apps/cli/commands/StatsRenderTest.kt b/apps/cli/src/test/kotlin/com/correx/apps/cli/commands/StatsRenderTest.kt new file mode 100644 index 00000000..08ff99c8 --- /dev/null +++ b/apps/cli/src/test/kotlin/com/correx/apps/cli/commands/StatsRenderTest.kt @@ -0,0 +1,67 @@ +package com.correx.apps.cli.commands + +import org.junit.jupiter.api.Assertions.assertTrue +import org.junit.jupiter.api.Test + +class StatsRenderTest { + + private val report = StatsReportDto( + sessionId = "abc-123", + eventCount = 12, + sessionDurationMs = 90_000, + inferenceCount = 3, + inferenceMs = 6000, + promptTokens = 310, + completionTokens = 170, + tokensPerSecond = 28.3, + perProvider = listOf( + ProviderStatsDto("lfm:a", 2, 5000, 300, 150, 30.0), + ProviderStatsDto("lfm:b", 1, 1000, 10, 20, 20.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 = FailureMetricsDto(inferenceFailures = 1, toolFailures = 1, workflowFailures = 1), + inferencePct = 6.7, + toolPct = 1.7, + approvalWaitPct = 33.3, + ) + + @Test + fun `render includes session id and all section headers`() { + val out = renderStats(report) + assertTrue(out.contains("Session: abc-123")) + assertTrue(out.contains("Inference")) + assertTrue(out.contains("Tools")) + assertTrue(out.contains("Approvals")) + assertTrue(out.contains("Failures")) + assertTrue(out.contains("Time accounting")) + } + + @Test + fun `render shows per-provider and per-tier breakdown rows`() { + val out = renderStats(report) + assertTrue(out.contains("lfm:a"), "provider row missing") + assertTrue(out.contains("lfm:b"), "provider row missing") + assertTrue(out.contains("shell.read"), "tool row missing") + assertTrue(out.contains("T2"), "tier row missing") + } + + @Test + fun `render formats duration as human-readable`() { + // 90_000 ms -> 1m 30s + assertTrue(renderStats(report).contains("1m 30s"), "duration not humanized") + } + + @Test + fun `time accounting other is the remainder of the three tracked categories`() { + // 100 - 6.7 - 1.7 - 33.3 = 58.3 + assertTrue(renderStats(report).contains("other: 58.3%"), "remainder miscomputed") + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsInspectionService.kt b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsInspectionService.kt new file mode 100644 index 00000000..8f5c9aad --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsInspectionService.kt @@ -0,0 +1,148 @@ +package com.correx.apps.server.metrics + +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.SessionId +import com.correx.core.sessions.projections.replay.DefaultEventReplayer +import kotlinx.serialization.Serializable + +private const val MS_PER_SECOND = 1000.0 +private const val PERCENT = 100.0 + +@Serializable +data class ProviderStatsDto( + val provider: String, + val completedCount: Long, + val totalLatencyMs: Long, + val promptTokens: Long, + val completionTokens: Long, + val tokensPerSecond: Double, +) + +@Serializable +data class ToolStatsDto( + val toolName: String, + val completedCount: Long, + val totalDurationMs: Long, +) + +@Serializable +data class TierApprovalStatsDto( + val tier: String, + val requestedCount: Long, + val resolvedCount: Long, + val totalWaitMs: Long, + val avgWaitMs: Long, +) + +/** + * Render-friendly view of a session's metrics. Raw sums plus the derived ratios the operator + * actually reads; the time-accounting percentages split session wall time into inference / + * tool execution / awaiting-approval, with the remainder ("other": context build, orchestration, + * idle) recoverable as `100 − (inference + tool + approvalWait)`. + */ +@Serializable +data class MetricsReport( + val sessionId: String, + val eventCount: Long, + val sessionDurationMs: Long, + val inferenceCount: Long, + val inferenceMs: Long, + val promptTokens: Long, + val completionTokens: Long, + val tokensPerSecond: Double, + val perProvider: List, + val toolCount: Long, + val toolMs: Long, + val perTool: List, + val approvalsRequested: Long, + val approvalsResolved: Long, + val approvalsPending: Long, + val approvalWaitMs: Long, + val avgApprovalWaitMs: Long, + val perTier: List, + val failures: FailureMetrics, + val inferencePct: Double, + val toolPct: Double, + val approvalWaitPct: Double, +) + +/** + * Rebuilds [MetricsState] by replaying a session through [MetricsProjection], then derives the + * read-side ratios into a [MetricsReport]. Replay-only: no network, no live inference + * (Hard Invariant #8). + */ +class MetricsInspectionService(private val eventStore: EventStore) { + + private val replayer = DefaultEventReplayer(eventStore, MetricsProjection()) + + fun inspect(sessionId: SessionId): MetricsReport { + val state = replayer.rebuild(sessionId) + return toReport(sessionId, state) + } + + private fun toReport(sessionId: SessionId, s: MetricsState): MetricsReport { + val durationMs = sessionDurationMs(s) + return MetricsReport( + sessionId = sessionId.value, + eventCount = s.eventCount, + sessionDurationMs = durationMs, + inferenceCount = s.inference.completedCount, + inferenceMs = s.inference.totalLatencyMs, + promptTokens = s.inference.promptTokens, + completionTokens = s.inference.completionTokens, + tokensPerSecond = tokensPerSecond(s.inference.completionTokens, s.inference.totalLatencyMs), + perProvider = s.inference.perProvider.entries + .sortedByDescending { it.value.totalLatencyMs } + .map { (provider, m) -> + ProviderStatsDto( + provider = provider, + completedCount = m.completedCount, + totalLatencyMs = m.totalLatencyMs, + promptTokens = m.promptTokens, + completionTokens = m.completionTokens, + tokensPerSecond = tokensPerSecond(m.completionTokens, m.totalLatencyMs), + ) + }, + toolCount = s.tools.completedCount, + toolMs = s.tools.totalDurationMs, + perTool = s.tools.perTool.entries + .sortedByDescending { it.value.totalDurationMs } + .map { (name, m) -> ToolStatsDto(name, m.completedCount, m.totalDurationMs) }, + approvalsRequested = s.approvals.requestedCount, + approvalsResolved = s.approvals.resolvedCount, + approvalsPending = s.pendingApprovals.size.toLong(), + approvalWaitMs = s.approvals.totalWaitMs, + avgApprovalWaitMs = avg(s.approvals.totalWaitMs, s.approvals.resolvedCount), + perTier = s.approvals.perTier.entries + .sortedBy { it.key } + .map { (tier, m) -> + TierApprovalStatsDto( + tier = tier, + requestedCount = m.requestedCount, + resolvedCount = m.resolvedCount, + totalWaitMs = m.totalWaitMs, + avgWaitMs = avg(m.totalWaitMs, m.resolvedCount), + ) + }, + failures = s.failures, + inferencePct = pct(s.inference.totalLatencyMs, durationMs), + toolPct = pct(s.tools.totalDurationMs, durationMs), + approvalWaitPct = pct(s.approvals.totalWaitMs, durationMs), + ) + } + + private fun sessionDurationMs(s: MetricsState): Long { + val first = s.firstEventAt ?: return 0L + val last = s.lastEventAt ?: return 0L + return (last - first).inWholeMilliseconds.coerceAtLeast(0L) + } + + private fun tokensPerSecond(completionTokens: Long, latencyMs: Long): Double = + if (latencyMs <= 0L) 0.0 else completionTokens.toDouble() / (latencyMs / MS_PER_SECOND) + + private fun pct(part: Long, whole: Long): Double = + if (whole <= 0L) 0.0 else (part.toDouble() / whole.toDouble()) * PERCENT + + private fun avg(total: Long, count: Long): Long = + if (count <= 0L) 0L else total / count +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsProjection.kt b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsProjection.kt new file mode 100644 index 00000000..116a9fce --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsProjection.kt @@ -0,0 +1,123 @@ +package com.correx.apps.server.metrics + +import com.correx.core.events.events.ApprovalDecisionResolvedEvent +import com.correx.core.events.events.ApprovalRequestedEvent +import com.correx.core.events.events.InferenceCompletedEvent +import com.correx.core.events.events.InferenceFailedEvent +import com.correx.core.events.events.InferenceTimeoutEvent +import com.correx.core.events.events.StageFailedEvent +import com.correx.core.events.events.StoredEvent +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.WorkflowFailedEvent +import com.correx.core.sessions.projections.Projection +import kotlinx.datetime.Instant + +/** + * Pure fold deriving [MetricsState] from a session's event log + * (observability-spec §2: observability-as-projection — no separate trace store, no OTel). + * + * Every metric here is replayable: a metric added later can be recomputed against events + * recorded before it existed. This projection records no new events and reads only events + * other bounded contexts already emit; it is a disposable read-model (Hard Invariant #2). + */ +class MetricsProjection : Projection { + + override fun initial(): MetricsState = MetricsState() + + override fun apply(state: MetricsState, event: StoredEvent): MetricsState { + val at = event.metadata.timestamp + val spanned = state.copy( + firstEventAt = state.firstEventAt ?: at, + lastEventAt = at, + eventCount = state.eventCount + 1, + ) + return when (val payload = event.payload) { + is InferenceCompletedEvent -> spanned.applyInferenceCompleted(payload) + is InferenceFailedEvent -> spanned.bumpFailure { it.copy(inferenceFailures = it.inferenceFailures + 1) } + is InferenceTimeoutEvent -> spanned.bumpFailure { it.copy(inferenceTimeouts = it.inferenceTimeouts + 1) } + is ToolExecutionCompletedEvent -> spanned.applyToolCompleted(payload) + is ToolExecutionFailedEvent -> spanned.bumpFailure { it.copy(toolFailures = it.toolFailures + 1) } + is ToolExecutionRejectedEvent -> spanned.bumpFailure { it.copy(toolRejections = it.toolRejections + 1) } + is ApprovalRequestedEvent -> spanned.applyApprovalRequested(payload, at) + is ApprovalDecisionResolvedEvent -> spanned.applyApprovalResolved(payload) + is StageFailedEvent -> spanned.bumpFailure { it.copy(stageFailures = it.stageFailures + 1) } + is WorkflowFailedEvent -> spanned.bumpFailure { it.copy(workflowFailures = it.workflowFailures + 1) } + else -> spanned + } + } + + private fun MetricsState.bumpFailure(transform: (FailureMetrics) -> FailureMetrics): MetricsState = + copy(failures = transform(failures)) + + private fun MetricsState.applyInferenceCompleted(e: InferenceCompletedEvent): MetricsState { + val provider = e.providerId.value + val prev = inference.perProvider[provider] ?: ProviderInferenceMetrics() + val updated = prev.copy( + completedCount = prev.completedCount + 1, + totalLatencyMs = prev.totalLatencyMs + e.latencyMs, + promptTokens = prev.promptTokens + e.tokensUsed.promptTokens, + completionTokens = prev.completionTokens + e.tokensUsed.completionTokens, + ) + return copy( + inference = inference.copy( + completedCount = inference.completedCount + 1, + totalLatencyMs = inference.totalLatencyMs + e.latencyMs, + promptTokens = inference.promptTokens + e.tokensUsed.promptTokens, + completionTokens = inference.completionTokens + e.tokensUsed.completionTokens, + perProvider = inference.perProvider + (provider to updated), + ), + ) + } + + private fun MetricsState.applyToolCompleted(e: ToolExecutionCompletedEvent): MetricsState { + val durationMs = e.receipt.durationMs + val prev = tools.perTool[e.toolName] ?: ToolUsageMetrics() + val updated = prev.copy( + completedCount = prev.completedCount + 1, + totalDurationMs = prev.totalDurationMs + durationMs, + ) + return copy( + tools = tools.copy( + completedCount = tools.completedCount + 1, + totalDurationMs = tools.totalDurationMs + durationMs, + perTool = tools.perTool + (e.toolName to updated), + ), + ) + } + + private fun MetricsState.applyApprovalRequested(e: ApprovalRequestedEvent, at: Instant): MetricsState { + val tier = e.tier.name + val prev = approvals.perTier[tier] ?: TierApprovalMetrics() + return copy( + approvals = approvals.copy( + requestedCount = approvals.requestedCount + 1, + perTier = approvals.perTier + (tier to prev.copy(requestedCount = prev.requestedCount + 1)), + ), + pendingApprovals = pendingApprovals + (e.requestId.value to at), + ) + } + + private fun MetricsState.applyApprovalResolved(e: ApprovalDecisionResolvedEvent): MetricsState { + val requestedAt = pendingApprovals[e.requestId.value] + val waitMs = requestedAt + ?.let { (e.resolutionTimestamp - it).inWholeMilliseconds.coerceAtLeast(0L) } + ?: 0L + val tier = e.tier.name + val prev = approvals.perTier[tier] ?: TierApprovalMetrics() + return copy( + approvals = approvals.copy( + resolvedCount = approvals.resolvedCount + 1, + totalWaitMs = approvals.totalWaitMs + waitMs, + perTier = approvals.perTier + ( + tier to prev.copy( + resolvedCount = prev.resolvedCount + 1, + totalWaitMs = prev.totalWaitMs + waitMs, + ) + ), + ), + pendingApprovals = pendingApprovals - e.requestId.value, + ) + } +} diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsState.kt b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsState.kt new file mode 100644 index 00000000..23b7175a --- /dev/null +++ b/apps/server/src/main/kotlin/com/correx/apps/server/metrics/MetricsState.kt @@ -0,0 +1,85 @@ +package com.correx.apps.server.metrics + +import kotlinx.datetime.Instant +import kotlinx.serialization.Serializable + +/** + * Read-model accumulated by [MetricsProjection] — a pure fold over a single session's event log. + * + * Holds raw sums only. Derived ratios (token throughput, average wait, time-accounting + * percentages) are computed at render time by [MetricsInspectionService] so this stays a + * stable, replayable accumulator. Never persisted — rebuilt from events on demand + * (Hard Invariant #1: the event log is the only source of truth). + */ +@Serializable +data class MetricsState( + val firstEventAt: Instant? = null, + val lastEventAt: Instant? = null, + val eventCount: Long = 0, + val inference: InferenceMetrics = InferenceMetrics(), + val tools: ToolMetrics = ToolMetrics(), + val approvals: ApprovalMetrics = ApprovalMetrics(), + val failures: FailureMetrics = FailureMetrics(), + /** + * In-flight approval requests awaiting resolution: requestId → request-event timestamp. + * Transient correlation bookkeeping that lets request→decision latency be computed in a + * single forward fold. Resolved requests are removed; entries still present after the + * full replay are unresolved (their wait time is unknown and excluded from totals). + */ + val pendingApprovals: Map = emptyMap(), +) + +@Serializable +data class InferenceMetrics( + val completedCount: Long = 0, + val totalLatencyMs: Long = 0, + val promptTokens: Long = 0, + val completionTokens: Long = 0, + val perProvider: Map = emptyMap(), +) + +@Serializable +data class ProviderInferenceMetrics( + val completedCount: Long = 0, + val totalLatencyMs: Long = 0, + val promptTokens: Long = 0, + val completionTokens: Long = 0, +) + +@Serializable +data class ToolMetrics( + val completedCount: Long = 0, + val totalDurationMs: Long = 0, + val perTool: Map = emptyMap(), +) + +@Serializable +data class ToolUsageMetrics( + val completedCount: Long = 0, + val totalDurationMs: Long = 0, +) + +@Serializable +data class ApprovalMetrics( + val requestedCount: Long = 0, + val resolvedCount: Long = 0, + val totalWaitMs: Long = 0, + val perTier: Map = emptyMap(), +) + +@Serializable +data class TierApprovalMetrics( + val requestedCount: Long = 0, + val resolvedCount: Long = 0, + val totalWaitMs: Long = 0, +) + +@Serializable +data class FailureMetrics( + val inferenceFailures: Long = 0, + val inferenceTimeouts: Long = 0, + val toolFailures: Long = 0, + val toolRejections: Long = 0, + val stageFailures: Long = 0, + val workflowFailures: Long = 0, +) 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 b93ff63a..0e0c693b 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 @@ -1,6 +1,7 @@ package com.correx.apps.server.routes import com.correx.apps.server.ServerModule +import com.correx.apps.server.metrics.MetricsInspectionService import com.correx.apps.server.protocol.SessionConfigDto import com.correx.apps.server.replay.ReplayInspectionService import com.correx.apps.server.serialization.payloadDiscriminator @@ -82,6 +83,7 @@ fun Route.sessionRoutes(module: ServerModule) { resumeSessionRoute(module) getEventsRoute(module) getReplayRoute(module) + getMetricsRoute(module) webSocket("/stream") { val id = call.parameters["id"] ?: return@webSocket val lastEventId = call.request.queryParameters["lastEventId"]?.toLongOrNull() @@ -190,3 +192,13 @@ private fun Route.getReplayRoute(module: ServerModule) { call.respond(report) } } + +private fun Route.getMetricsRoute(module: ServerModule) { + val service = MetricsInspectionService(module.eventStore) + get("/metrics") { + 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/metrics/MetricsProjectionTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/metrics/MetricsProjectionTest.kt new file mode 100644 index 00000000..edc09837 --- /dev/null +++ b/apps/server/src/test/kotlin/com/correx/apps/server/metrics/MetricsProjectionTest.kt @@ -0,0 +1,257 @@ +package com.correx.apps.server.metrics + +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.InferenceCompletedEvent +import com.correx.core.events.events.InferenceFailedEvent +import com.correx.core.events.events.NewEvent +import com.correx.core.events.events.StoredEvent +import com.correx.core.events.events.ToolExecutionCompletedEvent +import com.correx.core.events.events.ToolExecutionFailedEvent +import com.correx.core.events.events.ToolReceipt +import com.correx.core.events.events.WorkflowFailedEvent +import com.correx.core.events.stores.EventStore +import com.correx.core.events.types.EventId +import com.correx.core.events.types.SessionId +import com.correx.core.inference.TokenUsage +import com.correx.core.utils.TypeId +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.datetime.Instant +import org.junit.jupiter.api.Assertions.assertEquals +import org.junit.jupiter.api.Test + +class MetricsProjectionTest { + + private val sessionId: SessionId = TypeId("metrics-session-1") + private val stageId = TypeId("stage-1") + private val t0 = Instant.parse("2026-01-01T00:00:00Z") + + private fun at(seconds: Long): Instant = t0.plus(kotlin.time.Duration.parse("${seconds}s")) + + private fun stored(payload: EventPayload, seq: Long, at: Instant): StoredEvent = + StoredEvent( + metadata = EventMetadata( + eventId = EventId("e$seq"), + sessionId = sessionId, + timestamp = at, + schemaVersion = 1, + causationId = null, + correlationId = null, + ), + sequence = seq, + sessionSequence = seq, + payload = payload, + ) + + private fun inferenceDone(provider: String, latencyMs: Long, prompt: Int, completion: Int) = + InferenceCompletedEvent( + requestId = TypeId("inf-$provider-$latencyMs"), + sessionId = sessionId, + stageId = stageId, + providerId = TypeId(provider), + tokensUsed = TokenUsage(promptTokens = prompt, completionTokens = completion), + latencyMs = latencyMs, + responseArtifactId = TypeId("art-$provider-$latencyMs"), + ) + + private fun toolDone(name: String, durationMs: Long) = + ToolExecutionCompletedEvent( + invocationId = TypeId("inv-$name-$durationMs"), + sessionId = sessionId, + toolName = name, + receipt = ToolReceipt( + invocationId = TypeId("inv-$name-$durationMs"), + toolName = name, + exitCode = 0, + outputSummary = "ok", + durationMs = durationMs, + tier = Tier.T1, + timestamp = t0, + ), + ) + + private val seed = listOf( + stored(inferenceDone("lfm:a", 2000, 100, 50), 1L, at(0)), + stored(inferenceDone("lfm:a", 3000, 200, 100), 2L, at(1)), + stored(inferenceDone("lfm:b", 1000, 10, 20), 3L, at(2)), + stored(toolDone("shell.read", 500), 4L, at(3)), + stored(toolDone("shell.read", 700), 5L, at(4)), + stored(toolDone("fs.write", 300), 6L, at(5)), + stored( + ApprovalRequestedEvent( + requestId = TypeId("req-1"), + tier = Tier.T2, + validationReportId = TypeId("vr-1"), + riskSummaryId = null, + sessionId = sessionId, + stageId = stageId, + projectId = null, + ), + 7L, at(10), + ), + stored( + ApprovalDecisionResolvedEvent( + decisionId = TypeId("dec-1"), + requestId = TypeId("req-1"), + outcome = ApprovalOutcome.APPROVED, + status = ApprovalStatus.COMPLETED, + tier = Tier.T2, + resolutionTimestamp = at(40), + reason = null, + ), + 8L, at(40), + ), + stored( + ApprovalRequestedEvent( + requestId = TypeId("req-2"), + tier = Tier.T2, + validationReportId = TypeId("vr-2"), + riskSummaryId = null, + sessionId = sessionId, + stageId = stageId, + projectId = null, + ), + 9L, at(50), + ), + stored( + InferenceFailedEvent( + requestId = TypeId("inf-fail"), + sessionId = sessionId, + stageId = stageId, + providerId = TypeId("lfm:a"), + reason = "boom", + ), + 10L, at(51), + ), + stored( + ToolExecutionFailedEvent( + invocationId = TypeId("inv-fail"), + sessionId = sessionId, + toolName = "shell.read", + reason = "nonzero exit", + ), + 11L, at(52), + ), + stored( + WorkflowFailedEvent( + sessionId = sessionId, + stageId = stageId, + reason = "aborted", + retryExhausted = true, + ), + 12L, at(60), + ), + ) + + 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 report() = MetricsInspectionService(SeededEventStore(seed)).inspect(sessionId) + + @Test + fun `session span and event count come from event metadata`() { + val r = report() + assertEquals(sessionId.value, r.sessionId) + assertEquals(12L, r.eventCount) + assertEquals(60_000L, r.sessionDurationMs) + } + + @Test + fun `inference totals and per-provider breakdown accumulate`() { + val r = report() + assertEquals(3L, r.inferenceCount) + assertEquals(6000L, r.inferenceMs) + assertEquals(310L, r.promptTokens) + assertEquals(170L, r.completionTokens) + + val a = r.perProvider.single { it.provider == "lfm:a" } + assertEquals(2L, a.completedCount) + assertEquals(5000L, a.totalLatencyMs) + assertEquals(300L, a.promptTokens) + assertEquals(150L, a.completionTokens) + + val b = r.perProvider.single { it.provider == "lfm:b" } + assertEquals(1L, b.completedCount) + assertEquals(1000L, b.totalLatencyMs) + } + + @Test + fun `tool totals and per-tool breakdown accumulate`() { + val r = report() + assertEquals(3L, r.toolCount) + assertEquals(1500L, r.toolMs) + + val read = r.perTool.single { it.toolName == "shell.read" } + assertEquals(2L, read.completedCount) + assertEquals(1200L, read.totalDurationMs) + + val write = r.perTool.single { it.toolName == "fs.write" } + assertEquals(1L, write.completedCount) + assertEquals(300L, write.totalDurationMs) + } + + @Test + fun `approval wait is correlated request-to-resolution and pending excludes wait`() { + val r = report() + assertEquals(2L, r.approvalsRequested) + assertEquals(1L, r.approvalsResolved) + assertEquals(1L, r.approvalsPending) + // req-1: requested at t+10s, resolved at t+40s -> 30s. req-2 unresolved -> contributes 0. + assertEquals(30_000L, r.approvalWaitMs) + assertEquals(30_000L, r.avgApprovalWaitMs) + + val t2 = r.perTier.single { it.tier == "T2" } + assertEquals(2L, t2.requestedCount) + assertEquals(1L, t2.resolvedCount) + assertEquals(30_000L, t2.totalWaitMs) + } + + @Test + fun `failure counts are classified by event type`() { + val f = report().failures + assertEquals(1L, f.inferenceFailures) + assertEquals(1L, f.toolFailures) + assertEquals(1L, f.workflowFailures) + assertEquals(0L, f.inferenceTimeouts) + assertEquals(0L, f.toolRejections) + assertEquals(0L, f.stageFailures) + } + + @Test + fun `time accounting splits session wall time`() { + val r = report() + // inference 6000/60000, tools 1500/60000, approval-wait 30000/60000 + assertEquals(10.0, r.inferencePct, 0.001) + assertEquals(2.5, r.toolPct, 0.001) + assertEquals(50.0, r.approvalWaitPct, 0.001) + } + + @Test + fun `empty session yields zeroed report`() { + val r = MetricsInspectionService(SeededEventStore(emptyList())).inspect(sessionId) + assertEquals(0L, r.eventCount) + assertEquals(0L, r.sessionDurationMs) + assertEquals(0L, r.inferenceCount) + assertEquals(0.0, r.tokensPerSecond, 0.001) + assertEquals(0.0, r.inferencePct, 0.001) + } +}