fix(server,persistence): stop slow WS client stalling kernel; log dropped emits (#53)

Two event-delivery back-pressure faults from the core audit (#7).

(a) streamGlobal forwarded globalFlow → a per-connection Channel with
buffer.send (SUSPEND). globalFlow is also SUSPEND-overflow, so one wedged
WebSocket client back-pressured globalFlow → append() suspended → the whole
kernel stalled. Now the collector uses trySend; on overflow it closes the
buffer with SlowClientException, tearing down that one connection (client
reconnects and re-syncs via replaySnapshot) instead of blocking append().

(b) Per-session subscriptions (extraBufferCapacity 64) used tryEmit with the
return ignored: a lagging in-process collector (e.g. LiveArtifactRepository)
silently diverged from the durable log. Both SqliteEventStore and
InMemoryEventStore now warnIfDropped() — the event is still persisted; only the
live SharedFlow drops, and we surface it. ponytail: log-only; bounded
rebuild-on-lag if divergence ever bites.

Added slf4j-api to :infrastructure:persistence (already the standard logging
dep across modules) for the warn.

Green: :infrastructure:persistence + :apps:server compile+detekt+test (the one
pre-existing ModelLifecycleWiringTest failure needs a real llama-server, fails
identically on clean checkout).

apps/server/src/main/kotlin/com/correx/apps/server/ws/GlobalStreamHandler.kt

@@ -65,6 +65,9 @@ import java.util.UUID
 
 private val log = LoggerFactory.getLogger(GlobalStreamHandler::class.java)
 private const val BUFFER_CAPACITY = 1024
+
+/** Raised to tear down a WS connection whose forward buffer overflowed, so the client reconnects. */
+private class SlowClientException : RuntimeException("global WS stream buffer overflow")
 private const val RESOURCE_PUSH_INTERVAL_MS = 2500L
 private const val BYTES_PER_MB = 1024L * 1024L
 
@@ -788,8 +791,17 @@ internal suspend fun streamGlobal(
         is SharedFlow<StoredEvent> -> source.onSubscription { subscribed.complete(Unit) }
         else -> source.onStart { subscribed.complete(Unit) }
     }
+    // trySend, never send: a blocking send here would back-pressure globalFlow (SUSPEND overflow),
+    // suspending append() and stalling the whole kernel on one wedged client. On overflow we instead
+    // drop the connection (close the buffer); the client reconnects and re-syncs via replaySnapshot.
     val subscription = launch {
-        signaled.collect { buffer.send(it) }
+        signaled.collect {
+            val result = buffer.trySend(it)
+            if (result.isFailure && !result.isClosed) {
+                log.warn("global WS stream buffer overflow (slow client); dropping connection to re-sync")
+                buffer.close(SlowClientException())
+            }
+        }
     }
 
     try {

infrastructure/persistence/build.gradle

@@ -11,6 +11,7 @@ dependencies {
     implementation(project(":core:artifacts"))
     implementation(project(":core:artifacts-store"))
     implementation "org.xerial:sqlite-jdbc"
+    implementation "org.slf4j:slf4j-api:2.0.16"
     testImplementation(testFixtures(project(":testing:contracts")))
     testImplementation(project(":testing:fixtures"))
     testImplementation "org.junit.jupiter:junit-jupiter"

infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/InMemoryEventStore.kt

@@ -9,9 +9,12 @@ import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.asSharedFlow
+import org.slf4j.LoggerFactory
 import java.util.concurrent.*
 import java.util.concurrent.atomic.*
 
+private val log = LoggerFactory.getLogger(InMemoryEventStore::class.java)
+
 class InMemoryEventStore : EventStore {
     private val streams = ConcurrentHashMap<SessionId, MutableList<StoredEvent>>()
     private val sequences = ConcurrentHashMap<SessionId, AtomicLong>()
@@ -33,7 +36,7 @@ class InMemoryEventStore : EventStore {
             }
             stored = doAppend(event, stream)
         }
-        subscriptions[event.metadata.sessionId]?.tryEmit(stored)
+        subscriptions[event.metadata.sessionId]?.let { warnIfDropped(it.tryEmit(stored), stored) }
         globalFlow.emit(stored)
         return stored
     }
@@ -47,7 +50,7 @@ class InMemoryEventStore : EventStore {
             stored = events.map { doAppend(it, stream) }
         }
         val flow = subscriptions[sessionId]
-        if (flow != null) stored.forEach { flow.tryEmit(it) }
+        if (flow != null) stored.forEach { warnIfDropped(flow.tryEmit(it), it) }
         stored.forEach { globalFlow.emit(it) }
         return stored
     }
@@ -77,6 +80,19 @@ class InMemoryEventStore : EventStore {
 
     override fun allSessionIds(): Set<SessionId> = sequences.keys
 
+    /** A lagging per-session collector's buffer (extraBufferCapacity 64) is full; the event is durably
+     *  stored but dropped from the live SharedFlow. Surface it rather than swallow the false return. */
+    private fun warnIfDropped(emitted: Boolean, event: StoredEvent) {
+        if (!emitted) {
+            log.warn(
+                "dropped live subscription emit for session {} seq {} ({}): buffer full, collector lagging",
+                event.metadata.sessionId.value,
+                event.sequence,
+                event.payload::class.simpleName,
+            )
+        }
+    }
+
     private fun doAppend(event: NewEvent, stream: MutableList<StoredEvent>): StoredEvent {
         val sessionSeq = sequences.computeIfAbsent(event.metadata.sessionId) { AtomicLong(0) }
             .incrementAndGet()

infrastructure/persistence/src/main/kotlin/com/correx/infrastructure/persistence/SqliteEventStore.kt

@@ -21,10 +21,13 @@ import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import kotlinx.coroutines.withContext
 import kotlinx.datetime.Instant
+import org.slf4j.LoggerFactory
 import java.sql.Connection
 import java.sql.ResultSet
 import java.util.concurrent.*
 
+private val log = LoggerFactory.getLogger(SqliteEventStore::class.java)
+
 class SqliteEventStore(
     private val connection: Connection,
     private val jsonSerializer: JsonEventSerializer = JsonEventSerializer(eventJson),
@@ -89,7 +92,7 @@ class SqliteEventStore(
             }
         }
         val result = checkNotNull(stored)
-        subscriptions[event.metadata.sessionId]?.tryEmit(result)
+        subscriptions[event.metadata.sessionId]?.let { warnIfDropped(it.tryEmit(result), result) }
         globalFlow.emit(result)
         return result
     }
@@ -120,7 +123,7 @@ class SqliteEventStore(
             }
         }
         val flow = subscriptions[sessionId]
-        if (flow != null) stored.forEach { flow.tryEmit(it) }
+        if (flow != null) stored.forEach { warnIfDropped(flow.tryEmit(it), it) }
         stored.forEach { globalFlow.emit(it) }
         return stored
     }
@@ -234,6 +237,24 @@ class SqliteEventStore(
      */
     private inline fun <T> withConnection(block: () -> T): T = synchronized(connection) { block() }
 
+    /**
+     * A per-session subscription's buffer (extraBufferCapacity 64) is full and this event was dropped.
+     * The event is durably persisted (the drop is only on the live SharedFlow), but a lagging in-process
+     * collector — e.g. LiveArtifactRepository — silently diverges from the log. Surface it instead of
+     * swallowing the false return. ponytail: log-only; give that collector a bounded rebuild-on-lag if
+     * divergence ever bites in practice.
+     */
+    private fun warnIfDropped(emitted: Boolean, event: StoredEvent) {
+        if (!emitted) {
+            log.warn(
+                "dropped live subscription emit for session {} seq {} ({}): buffer full, collector lagging",
+                event.metadata.sessionId.value,
+                event.sequence,
+                event.payload::class.simpleName,
+            )
+        }
+    }
+
     /**
      * Inserts [event], letting SQLite assign both the global `sequence` and the per-session
      * `session_sequence` as `MAX+1` subqueries evaluated inside the INSERT itself, and RETURNs the