feat(server): llama-server health probe

Second HealthProbe (after disk-watermark): liveness via GET /health on the
configured llama base URL + tokens/sec degradation watch derived from recorded
InferenceCompletedEvents (windowed average vs configurable threshold, opt-in via
llamaTpsWarnBelow). Plugs into the existing HealthMonitor edge-detection loop;
no new events. Observability spec §4.

apps/server/src/main/kotlin/com/correx/apps/server/Main.kt

@@ -125,6 +125,8 @@ fun main() {
     // it stays null on the static path. System RAM (overlaid below) is owner-agnostic, so the gauge
     // renders on both paths regardless of whether correx manages the model process.
     var managedPidSupplier: () -> Long? = { null }
+    // Base URL used by the llama-server health probe; derived from whichever path is active.
+    var llamaBaseUrl: String = "http://127.0.0.1:10000"
 
     if (correxConfig.models.isNotEmpty()) {
         val settings = correxConfig.modelsSettings
@@ -151,6 +153,7 @@ fun main() {
         inferenceRouter = managedRouter
         modelSwapper = managedRouter
         managedPidSupplier = { modelManager.currentPid() }
+        llamaBaseUrl = "http://${settings.host}:${settings.port}"
         // Shutdown hook to kill the managed llama-server
         Runtime.getRuntime().addShutdownHook(Thread {
             log.info("Shutdown: unloading managed model '{}'", targetModelConfig.id)
@@ -165,6 +168,9 @@ fun main() {
         firstProvider = staticProviders.first()
         infraRegistry = InfrastructureModule.createProviderRegistry(staticProviders)
         inferenceRouter = DefaultInferenceRouter(infraRegistry, FirstAvailableRoutingStrategy())
+        llamaBaseUrl = correxConfig.providers.firstOrNull()?.url
+            ?: System.getenv("CORREX_LLAMA_URL")
+            ?: "http://127.0.0.1:10000"
     }
 
     // Build the resource gauge for whichever GPU vendor is present (pid-less GPU/VRAM reads work on
@@ -436,6 +442,13 @@ fun main() {
                         warnBytes = hc.diskWarnBytes,
                         warnGrowthBytesPerMin = hc.diskWarnGrowthBytesPerMin,
                     ),
+                    com.correx.apps.server.health.LlamaServerHealthProbe(
+                        llamaBaseUrl = llamaBaseUrl,
+                        httpClient = io.ktor.client.HttpClient(io.ktor.client.engine.cio.CIO),
+                        eventStore = eventStore,
+                        livenessTimeoutMs = hc.llamaLivenessTimeoutMs,
+                        tpsWarnBelow = hc.llamaTpsWarnBelow,
+                    ),
                 ),
                 intervalMs = hc.intervalMs,
                 initialStatuses = seeded,

apps/server/src/main/kotlin/com/correx/apps/server/health/LlamaServerHealthProbe.kt

@@ -0,0 +1,159 @@
+package com.correx.apps.server.health
+
+import com.correx.core.events.events.HealthSubject
+import com.correx.core.events.events.InferenceCompletedEvent
+import com.correx.core.events.stores.EventStore
+import io.ktor.client.HttpClient
+import io.ktor.client.request.get
+import io.ktor.http.isSuccess
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+
+private const val MS_PER_SECOND = 1000.0
+private const val TPS_WINDOW = 10
+
+/**
+ * Health probe for the local llama-server (observability-spec §4: LLAMA_SERVER subject).
+ *
+ * **Liveness** — every tick issues a bounded `GET [llamaBaseUrl]/health` via [httpClient].
+ * A 200-family response is live; any non-2xx, connection error, or [livenessTimeoutMs]
+ * expiry is DEGRADED. The [httpClient] is injected so tests can supply a [MockEngine].
+ *
+ * **Degradation (tokens/sec)** — on each tick the probe scans new [InferenceCompletedEvent]s
+ * from the event store (events appended since the previous tick, tracked by
+ * [lastScannedGlobalSequence]), accumulates the per-call tokens/sec into a sliding window
+ * of size [TPS_WINDOW], and reports DEGRADED when the window average falls below
+ * [tpsWarnBelow]. A threshold of 0.0 disables the degradation dimension; a window that
+ * has not yet collected any samples reports HEALTHY (no baseline).
+ *
+ * The probe only observes and returns a [HealthObservation]; it never emits events.
+ * [HealthMonitor] owns edge detection and event emission (Invariants #8/#9).
+ */
+class LlamaServerHealthProbe(
+    private val llamaBaseUrl: String,
+    private val httpClient: HttpClient,
+    private val eventStore: EventStore,
+    private val livenessTimeoutMs: Long = 5_000,
+    private val tpsWarnBelow: Double = 0.0,
+) : HealthProbe {
+
+    override val subject: HealthSubject = HealthSubject.LLAMA_SERVER
+
+    /**
+     * Monotonic global sequence up to which we have already scanned for inference events.
+     * Assumes single-coroutine sequential invocation: [HealthMonitor] calls probes sequentially
+     * within one coroutine tick, so no concurrent mutation of this field or [tpsWindow] occurs.
+     */
+    private var lastScannedGlobalSequence: Long = -1L
+
+    /**
+     * Sliding window of recent per-call tokens/sec values (newest at the end).
+     * See [lastScannedGlobalSequence] for the single-coroutine assumption.
+     */
+    private val tpsWindow: ArrayDeque<Double> = ArrayDeque(TPS_WINDOW)
+
+    override suspend fun observe(): HealthObservation {
+        val livenessObservation = checkLiveness()
+        if (livenessObservation.status == HealthStatus.DEGRADED) return livenessObservation
+        return checkThroughput() ?: livenessObservation
+    }
+
+    private suspend fun checkLiveness(): HealthObservation {
+        val result = runCatching {
+            withTimeout(livenessTimeoutMs) {
+                httpClient.get("$llamaBaseUrl/health")
+            }
+        }
+        val response = result.getOrNull()
+        return when {
+            result.isFailure -> degraded(
+                metric = "liveness",
+                value = 0L,
+                threshold = 1L,
+                detail = "GET $llamaBaseUrl/health failed: ${result.exceptionOrNull()?.message}",
+            )
+            response != null && !response.status.isSuccess() -> degraded(
+                metric = "liveness",
+                value = response.status.value.toLong(),
+                threshold = 200L,
+                detail = "GET $llamaBaseUrl/health returned HTTP ${response.status.value}",
+            )
+            else -> healthy(
+                metric = "liveness",
+                value = 1L,
+                threshold = 1L,
+                detail = "llama-server live at $llamaBaseUrl",
+            )
+        }
+    }
+
+    private suspend fun checkThroughput(): HealthObservation? {
+        if (tpsWarnBelow <= 0.0) {
+            // Dimension disabled: still advance the cursor so a later live-enable doesn't rescan
+            // the entire history from sequence 0.
+            lastScannedGlobalSequence = withContext(Dispatchers.IO) { eventStore.lastGlobalSequence() }
+            return null
+        }
+        withContext(Dispatchers.IO) { scanNewInferenceEvents() }
+        if (tpsWindow.isEmpty()) return null
+        val avgTps = tpsWindow.average()
+        val avgTpsLong = (avgTps * TPS_SCALE).toLong()
+        val thresholdLong = (tpsWarnBelow * TPS_SCALE).toLong()
+        return if (avgTps < tpsWarnBelow) {
+            degraded(
+                metric = "tokens_per_second",
+                value = avgTpsLong,
+                threshold = thresholdLong,
+                detail = "avg tokens/sec %.2f below warn threshold %.2f (window=%d)".format(
+                    avgTps, tpsWarnBelow, tpsWindow.size,
+                ),
+            )
+        } else {
+            healthy(
+                metric = "tokens_per_second",
+                value = avgTpsLong,
+                threshold = thresholdLong,
+                detail = "avg tokens/sec %.2f (window=%d)".format(avgTps, tpsWindow.size),
+            )
+        }
+    }
+
+    /**
+     * Scans [EventStore.allEvents] for [InferenceCompletedEvent]s with a global sequence
+     * number beyond [lastScannedGlobalSequence], appending each call's tokens/sec to
+     * [tpsWindow] (capped at [TPS_WINDOW]). Updates [lastScannedGlobalSequence] so the
+     * next tick only scans new events — O(new events), not O(all events).
+     */
+    private fun scanNewInferenceEvents() {
+        var maxSeen = lastScannedGlobalSequence
+        eventStore.allEvents()
+            .filter { it.sequence > lastScannedGlobalSequence }
+            .forEach { event ->
+                if (event.sequence > maxSeen) maxSeen = event.sequence
+                val payload = event.payload as? InferenceCompletedEvent ?: return@forEach
+                // Skip events with no completion tokens (e.g. stop-token-only responses) to
+                // avoid dragging the windowed average to zero on a healthy model.
+                if (payload.tokensUsed.completionTokens <= 0) return@forEach
+                val tps = if (payload.latencyMs > 0L) {
+                    payload.tokensUsed.completionTokens.toDouble() / (payload.latencyMs / MS_PER_SECOND)
+                } else {
+                    0.0
+                }
+                if (tpsWindow.size >= TPS_WINDOW) tpsWindow.removeFirst()
+                tpsWindow.addLast(tps)
+            }
+        lastScannedGlobalSequence = maxSeen
+    }
+
+    private fun degraded(metric: String, value: Long, threshold: Long, detail: String) =
+        HealthObservation(subject, HealthStatus.DEGRADED, metric, value, threshold, detail)
+
+    private fun healthy(metric: String, value: Long, threshold: Long, detail: String) =
+        HealthObservation(subject, HealthStatus.HEALTHY, metric, value, threshold, detail)
+
+    companion object {
+        /** Scale factor: store tps×100 as Long to fit [HealthObservation.observedValue]. */
+        private const val TPS_SCALE = 100L
+    }
+}

apps/server/src/test/kotlin/com/correx/apps/server/health/LlamaServerHealthProbeTest.kt

@@ -0,0 +1,202 @@
+package com.correx.apps.server.health
+
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.InferenceCompletedEvent
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.StoredEvent
+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 io.ktor.client.HttpClient
+import io.ktor.client.engine.mock.MockEngine
+import io.ktor.client.engine.mock.respond
+import io.ktor.http.HttpStatusCode
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.runBlocking
+import kotlinx.datetime.Instant
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+
+class LlamaServerHealthProbeTest {
+
+    private val sessionId: SessionId = TypeId("s-1")
+    private val stageId = TypeId("stage-1")
+
+    // ---------------------------------------------------------------------------
+    // Helpers
+    // ---------------------------------------------------------------------------
+
+    private fun httpClient(status: HttpStatusCode): HttpClient =
+        HttpClient(MockEngine { respond("", status) })
+
+    private fun httpClientThrowing(): HttpClient =
+        HttpClient(MockEngine { error("connection refused") })
+
+    private fun inference(seq: Long, completionTokens: Int, latencyMs: Long): StoredEvent =
+        StoredEvent(
+            metadata = EventMetadata(
+                eventId = EventId("e$seq"),
+                sessionId = sessionId,
+                timestamp = Instant.parse("2026-01-01T00:00:00Z"),
+                schemaVersion = 1,
+                causationId = null,
+                correlationId = null,
+            ),
+            sequence = seq,
+            sessionSequence = seq,
+            payload = InferenceCompletedEvent(
+                requestId = TypeId("req-$seq"),
+                sessionId = sessionId,
+                stageId = stageId,
+                providerId = TypeId("lfm"),
+                tokensUsed = TokenUsage(promptTokens = 10, completionTokens = completionTokens),
+                latencyMs = latencyMs,
+                responseArtifactId = TypeId("art-$seq"),
+            ),
+        )
+
+    /** Minimal EventStore backed by a mutable list; allEvents() returns all stored events. */
+    private class ListEventStore(private val events: MutableList<StoredEvent> = mutableListOf()) : 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 = events.lastOrNull()?.sequence ?: 0L
+        override fun allEvents(): Sequence<StoredEvent> = events.asSequence()
+        override fun allSessionIds(): Set<SessionId> = events.map { it.metadata.sessionId }.toSet()
+
+        fun add(event: StoredEvent) { events.add(event) }
+    }
+
+    private fun probe(
+        status: HttpStatusCode = HttpStatusCode.OK,
+        store: ListEventStore = ListEventStore(),
+        tpsWarnBelow: Double = 0.0,
+    ) = LlamaServerHealthProbe(
+        llamaBaseUrl = "http://localhost:11434",
+        httpClient = httpClient(status),
+        eventStore = store,
+        livenessTimeoutMs = 5_000,
+        tpsWarnBelow = tpsWarnBelow,
+    )
+
+    // ---------------------------------------------------------------------------
+    // Liveness tests
+    // ---------------------------------------------------------------------------
+
+    @Test
+    fun `healthy when llama-server returns 200`(): Unit = runBlocking {
+        val obs = probe(status = HttpStatusCode.OK).observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("liveness", obs.metric)
+        assertEquals(1L, obs.observedValue)
+    }
+
+    @Test
+    fun `degraded when llama-server returns non-200`(): Unit = runBlocking {
+        val obs = probe(status = HttpStatusCode.ServiceUnavailable).observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("liveness", obs.metric)
+        assertTrue(obs.detail.contains("503"), "expected 503 in detail: ${obs.detail}")
+    }
+
+    @Test
+    fun `degraded on connection error`(): Unit = runBlocking {
+        val p = LlamaServerHealthProbe(
+            llamaBaseUrl = "http://localhost:11434",
+            httpClient = httpClientThrowing(),
+            eventStore = ListEventStore(),
+            livenessTimeoutMs = 5_000,
+            tpsWarnBelow = 0.0,
+        )
+        val obs = p.observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("liveness", obs.metric)
+    }
+
+    // ---------------------------------------------------------------------------
+    // Tokens/sec degradation tests
+    // ---------------------------------------------------------------------------
+
+    @Test
+    fun `healthy when tps above threshold`(): Unit = runBlocking {
+        // 100 tokens in 1000 ms = 100 tps, threshold is 50 tps
+        val store = ListEventStore()
+        store.add(inference(seq = 1, completionTokens = 100, latencyMs = 1_000))
+        val obs = probe(store = store, tpsWarnBelow = 50.0).observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("tokens_per_second", obs.metric)
+    }
+
+    @Test
+    fun `degraded when tps below threshold`(): Unit = runBlocking {
+        // 10 tokens in 1000 ms = 10 tps, threshold is 50 tps
+        val store = ListEventStore()
+        store.add(inference(seq = 1, completionTokens = 10, latencyMs = 1_000))
+        val obs = probe(store = store, tpsWarnBelow = 50.0).observe()
+        assertEquals(HealthStatus.DEGRADED, obs.status)
+        assertEquals("tokens_per_second", obs.metric)
+        assertTrue(obs.detail.contains("below"), "detail should mention 'below': ${obs.detail}")
+    }
+
+    @Test
+    fun `healthy when no inference events and tpsWarnBelow configured`(): Unit = runBlocking {
+        // No baseline — can't degrade without samples
+        val obs = probe(tpsWarnBelow = 50.0).observe()
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        // No inference events means the liveness path dominates
+        assertEquals("liveness", obs.metric)
+    }
+
+    @Test
+    fun `tpsWarnBelow zero disables degradation check`(): Unit = runBlocking {
+        val store = ListEventStore()
+        store.add(inference(seq = 1, completionTokens = 1, latencyMs = 100_000)) // very slow
+        val obs = probe(store = store, tpsWarnBelow = 0.0).observe()
+        // Degradation disabled; liveness is healthy -> overall HEALTHY
+        assertEquals(HealthStatus.HEALTHY, obs.status)
+        assertEquals("liveness", obs.metric)
+    }
+
+    @Test
+    fun `only scans new events on subsequent ticks`(): Unit = runBlocking {
+        val store = ListEventStore()
+        store.add(inference(seq = 1, completionTokens = 100, latencyMs = 1_000)) // 100 tps
+
+        val p = LlamaServerHealthProbe(
+            llamaBaseUrl = "http://localhost:11434",
+            httpClient = httpClient(HttpStatusCode.OK),
+            eventStore = store,
+            livenessTimeoutMs = 5_000,
+            tpsWarnBelow = 50.0,
+        )
+
+        // First tick: sees inference at seq=1, 100 tps -> HEALTHY
+        assertEquals(HealthStatus.HEALTHY, p.observe().status)
+
+        // Add a very slow inference at seq=2
+        store.add(inference(seq = 2, completionTokens = 1, latencyMs = 10_000)) // 0.1 tps
+
+        // Second tick: sees only seq=2 (new), window = [100, 0.1], avg ~50.05 -> still HEALTHY
+        val obs2 = p.observe()
+        // avg(100, 0.1) = 50.05 which is above 50 threshold -> HEALTHY
+        assertEquals(HealthStatus.HEALTHY, obs2.status)
+
+        // Add another very slow inference
+        store.add(inference(seq = 3, completionTokens = 1, latencyMs = 10_000)) // 0.1 tps
+
+        // Third tick: window = [100, 0.1, 0.1], avg ~33.4 -> DEGRADED
+        val obs3 = p.observe()
+        assertEquals(HealthStatus.DEGRADED, obs3.status)
+        assertEquals("tokens_per_second", obs3.metric)
+    }
+}

core/config/src/main/kotlin/com/correx/core/config/CorrexConfig.kt

@@ -24,6 +24,11 @@ data class CorrexConfig(
  * the on-disk footprint every [intervalMs] and records a degraded/restored event on the edge.
  * [diskWarnBytes] is the absolute event-log + CAS size that trips a warning; [diskWarnGrowthBytesPerMin]
  * is the growth rate that warns log compaction is becoming non-theoretical. Defaults suit a homelab.
+ *
+ * [llamaLivenessTimeoutMs] bounds the GET /health ping to the llama-server; a connection error or
+ * timeout within this window is reported as DEGRADED (liveness). [llamaTpsWarnBelow] is the
+ * tokens/sec floor below which the probe reports degradation (KV-cache pressure, thermal throttle);
+ * 0.0 disables the degradation dimension (no baseline to compare against on a fresh install).
  */
 @Serializable
 data class HealthConfig(
@@ -31,6 +36,8 @@ data class HealthConfig(
     val intervalMs: Long = 30_000,
     val diskWarnBytes: Long = 5_000_000_000,
     val diskWarnGrowthBytesPerMin: Long = 50_000_000,
+    val llamaLivenessTimeoutMs: Long = 5_000,
+    val llamaTpsWarnBelow: Double = 0.0,
 )
 
 /**