feat(server,cli): observability metrics as a replayable projection + correx stats
Implements observability-spec §2 + §3-tier-1: metrics as a pure fold over the
event log, no third pillar, no OTel. Every metric is replayable over the full
historical log because it derives from events other contexts already emit — no
new EventPayload subclasses, so no serialization registry change.
- MetricsProjection: Projection<MetricsState> folding inference/tool/approval/
stage/workflow events into raw sums; approval request→decision latency is
correlated in-fold via a transient pending-request map.
- MetricsInspectionService: rebuilds via DefaultEventReplayer, derives read-side
ratios (token throughput, avg approval wait, session wall-time accounting split
into inference / tool / approval-wait / other) into MetricsReport.
- GET /sessions/{id}/metrics route, mirroring the replay endpoint.
- `correx stats <sessionId>` CLI client (--json passthrough), mirroring replay.
- Tests: seeded-event fold correctness with controlled timestamps (7) + CLI
render smoke tests (4).
This commit is contained in:
@@ -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(),
|
||||
)
|
||||
|
||||
@@ -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<ProviderStatsDto> = emptyList(),
|
||||
val toolCount: Long = 0,
|
||||
val toolMs: Long = 0,
|
||||
val perTool: List<ToolStatsDto> = emptyList(),
|
||||
val approvalsRequested: Long = 0,
|
||||
val approvalsResolved: Long = 0,
|
||||
val approvalsPending: Long = 0,
|
||||
val approvalWaitMs: Long = 0,
|
||||
val avgApprovalWaitMs: Long = 0,
|
||||
val perTier: List<TierApprovalStatsDto> = 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<String>()
|
||||
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()
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user