fix(server-05): synchronize subscription before snapshot read

Close the narrow race where `streamGlobal` launched the subscribeAll
collector and immediately read `lastGlobalSequence()` — events appended
in that window were dropped by the replay=0 SharedFlow before the
collector actually subscribed. Now use a CompletableDeferred barrier
signaled via `onSubscription` (SharedFlow) or `onStart` (cold fallback),
awaited before `bridge.replaySnapshot()`.

Closes the remaining gap on finding #1 from
docs/specs/2026-05-24-review-fixes.

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

@@ -19,10 +19,14 @@ import com.correx.core.utils.TypeId
 import io.ktor.server.websocket.DefaultWebSocketServerSession
 import io.ktor.websocket.Frame
 import io.ktor.websocket.readText
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.channels.BufferOverflow
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ClosedReceiveChannelException
 import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.onSubscription
 import kotlinx.coroutines.launch
 import kotlinx.datetime.Clock
 import org.slf4j.LoggerFactory
@@ -197,11 +201,20 @@ internal suspend fun streamGlobal(
         onBufferOverflow = BufferOverflow.SUSPEND,
     )
 
+    // Subscription must be active BEFORE replaySnapshot reads lastGlobalSequence,
+    // otherwise events appended in that window are lost (replay=0 SharedFlow).
+    val subscribed = CompletableDeferred<Unit>()
+    val source = eventStore.subscribeAll()
+    val signaled = when (source) {
+        is SharedFlow<StoredEvent> -> source.onSubscription { subscribed.complete(Unit) }
+        else -> source.onStart { subscribed.complete(Unit) }
+    }
     val subscription = launch {
-        eventStore.subscribeAll().collect { buffer.send(it) }
+        signaled.collect { buffer.send(it) }
     }
 
     try {
+        subscribed.await()
         bridge.replaySnapshot()
         for (event in buffer) {
             val msg = mapper.map(event) ?: continue

apps/server/src/test/kotlin/com/correx/apps/server/ws/GlobalStreamHandlerTest.kt

@@ -22,6 +22,7 @@ import com.correx.core.events.types.StageId
 import com.correx.core.kernel.orchestration.OrchestrationRepository
 import com.correx.core.sessions.projections.replay.DefaultEventReplayer
 import com.correx.core.sessions.projections.replay.EventReplayer
+import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
@@ -236,6 +237,42 @@ class GlobalStreamHandlerTest {
         assertInstanceOf(ServerMessage.SessionStarted::class.java, messages[1])
     }
 
+    @Test
+    fun `barrier - subscription is registered before lastGlobalSequence is read`() = runTest {
+        val liveFlow = MutableSharedFlow<StoredEvent>(extraBufferCapacity = 16)
+        val observedAtSnapshotRead = CompletableDeferred<Int>()
+
+        val store = object : 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> = emptyList()
+            override fun readFrom(sessionId: SessionId, fromSequence: Long): List<StoredEvent> = emptyList()
+            override fun lastSequence(sessionId: SessionId): Long = 0L
+            override fun subscribe(sessionId: SessionId): Flow<StoredEvent> = liveFlow
+            override fun allEvents(): Sequence<StoredEvent> = emptySequence()
+            override fun allSessionIds(): Set<SessionId> = emptySet()
+            override fun subscribeAll(): Flow<StoredEvent> = liveFlow
+            override suspend fun lastGlobalSequence(): Long {
+                // Capture the subscription count at the moment replaySnapshot reads the offset.
+                // The barrier in streamGlobal must guarantee count > 0 here.
+                observedAtSnapshotRead.complete(liveFlow.subscriptionCount.value)
+                return 0L
+            }
+        }
+        val approvals = fakeApprovalRepository(store)
+        val bridge = SessionEventBridge(store, noopArtifactStore, idleOrchestrationRepository(), approvals) { }
+        val mapper = DomainEventMapper(noopArtifactStore)
+
+        val job = launch {
+            streamGlobal(store, bridge, mapper) { }
+        }
+
+        val count = observedAtSnapshotRead.await()
+        job.cancelAndJoin()
+
+        assertEquals(1, count, "subscribeAll() must have an active subscriber before lastGlobalSequence() is read")
+    }
+
     @Test
     fun `shutdown - cancel outer scope ends subscription with no leaks`() = runTest {
         val liveFlow = MutableSharedFlow<StoredEvent>()