feat(server): Task 3.3 — NarrationSubscriber + wiring

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

@@ -1,6 +1,7 @@
 package com.correx.apps.server
 
 import com.correx.apps.server.approval.ApprovalCoordinator
+import com.correx.apps.server.narration.NarrationSubscriber
 import com.correx.apps.server.registry.ProviderRegistry
 import com.correx.apps.server.registry.WorkflowRegistry
 import com.correx.apps.server.workspace.WorkspaceResolver
@@ -65,6 +66,7 @@ class ServerModule(
     // Dispatchers.Default since the work is non-blocking and CPU-light.
     val moduleScope: CoroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default),
     approvalCoordinator: ApprovalCoordinator? = null,
+    val narrationMaxPerRun: Int = 100,
 ) {
     val approvalCoordinator: ApprovalCoordinator = approvalCoordinator ?: ApprovalCoordinator(
         orchestrator = orchestrator,
@@ -105,6 +107,10 @@ class ServerModule(
                 launchSessionResume(sessionId, graph)
             }
             .launchIn(moduleScope)
+
+        // Live-only: subscribeAll() replays nothing and ServerModule is never built under
+        // ReplayOrchestrator, so narration never re-fires on restart/replay (invariant #8).
+        NarrationSubscriber(eventStore = eventStore, routerFacade = routerFacade, scope = moduleScope, maxPerRun = narrationMaxPerRun).start()
     }
 
     /**

apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt

@@ -0,0 +1,98 @@
+package com.correx.apps.server.narration
+
+import com.correx.core.events.events.OrchestrationPausedEvent
+import com.correx.core.events.events.StageCompletedEvent
+import com.correx.core.events.events.StageFailedEvent
+import com.correx.core.events.events.StoredEvent
+import com.correx.core.events.events.WorkflowCompletedEvent
+import com.correx.core.events.events.WorkflowFailedEvent
+import com.correx.core.events.events.WorkflowStartedEvent
+import com.correx.core.events.stores.EventStore
+import com.correx.core.events.types.SessionId
+import com.correx.core.router.RouterFacade
+import com.correx.core.router.model.NarrationTrigger
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.launch
+import org.slf4j.LoggerFactory
+import java.util.concurrent.ConcurrentHashMap
+
+class NarrationSubscriber(
+    private val eventStore: EventStore,
+    private val routerFacade: RouterFacade,
+    private val scope: CoroutineScope,
+    private val maxPerRun: Int = DEFAULT_MAX_PER_RUN,
+) {
+    private data class SessionLane(val channel: Channel<NarrationTrigger>, var used: Int)
+
+    private val lanes = ConcurrentHashMap<String, SessionLane>()
+
+    fun start() {
+        eventStore.subscribeAll().onEach { handle(it) }.launchIn(scope)
+    }
+
+    private fun handle(event: StoredEvent) {
+        val sid = event.metadata.sessionId
+        when (val p = event.payload) {
+            is WorkflowStartedEvent -> lanes[sid.value]?.let { it.used = 0 }
+            is StageCompletedEvent -> enqueue(
+                sid,
+                NarrationTrigger(kind = "stage_completed", instruction = "Stage ${p.stageId.value} completed. Summarise what was accomplished."),
+            )
+            is StageFailedEvent -> enqueue(
+                sid,
+                NarrationTrigger(kind = "stage_failed", instruction = "Stage ${p.stageId.value} failed: ${p.reason}. Explain what went wrong."),
+            )
+            is WorkflowCompletedEvent -> enqueue(
+                sid,
+                NarrationTrigger(kind = "workflow_completed", instruction = "The workflow finished successfully. Provide a brief summary."),
+            )
+            is WorkflowFailedEvent -> enqueue(
+                sid,
+                NarrationTrigger(kind = "workflow_failed", instruction = "The workflow failed: ${p.reason}. Explain the failure to the user."),
+            )
+            is OrchestrationPausedEvent -> enqueue(
+                sid,
+                NarrationTrigger(kind = "paused", instruction = "Orchestration paused: ${p.reason}. Inform the user."),
+            )
+            else -> Unit
+        }
+    }
+
+    private fun enqueue(sessionId: SessionId, trigger: NarrationTrigger) {
+        val lane = lanes.computeIfAbsent(sessionId.value) { startLane(sessionId) }
+        if (lane.used >= maxPerRun) {
+            log.debug("narration budget reached for session {}; skipping {}", sessionId.value, trigger.kind)
+            return
+        }
+        lane.used += 1
+        lane.channel.trySend(trigger)
+    }
+
+    private fun startLane(sessionId: SessionId): SessionLane {
+        val channel = Channel<NarrationTrigger>(capacity = Channel.UNLIMITED)
+        scope.launch {
+            for (trigger in channel) {
+                runCatching { routerFacade.narrate(sessionId, trigger) }
+                    .onFailure { e ->
+                        if (e is CancellationException) throw e
+                        log.warn(
+                            "narration failed for session {} trigger {}: {}",
+                            sessionId.value,
+                            trigger.kind,
+                            e.message,
+                        )
+                    }
+            }
+        }
+        return SessionLane(channel, used = 0)
+    }
+
+    companion object {
+        private const val DEFAULT_MAX_PER_RUN = 100
+        private val log = LoggerFactory.getLogger(NarrationSubscriber::class.java)
+    }
+}

apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt

@@ -0,0 +1,160 @@
+package com.correx.apps.server.narration
+
+import com.correx.core.events.events.EventMetadata
+import com.correx.core.events.events.EventPayload
+import com.correx.core.events.events.NewEvent
+import com.correx.core.events.events.StageCompletedEvent
+import com.correx.core.events.events.StageFailedEvent
+import com.correx.core.events.events.StoredEvent
+import com.correx.core.events.events.WorkflowCompletedEvent
+import com.correx.core.events.events.WorkflowStartedEvent
+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.events.types.StageId
+import com.correx.core.events.types.TransitionId
+import com.correx.core.router.ChatMode
+import com.correx.core.router.RouterFacade
+import com.correx.core.router.model.NarrationTrigger
+import com.correx.core.router.model.RouterResponse
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlinx.coroutines.yield
+import kotlinx.datetime.Instant
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import java.util.concurrent.CopyOnWriteArrayList
+
+class NarrationSubscriberTest {
+
+    private val sessionId = SessionId("session-1")
+    private val timestamp = Instant.parse("2026-01-01T00:00:00Z")
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+
+    @AfterEach
+    fun tearDown() {
+        scope.cancel()
+    }
+
+    private val liveFlow = MutableSharedFlow<StoredEvent>(extraBufferCapacity = 64)
+
+    private val fakeStore = 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 = 0L
+    }
+
+    private class RecordingRouterFacade : RouterFacade {
+        val calls = CopyOnWriteArrayList<Pair<SessionId, NarrationTrigger>>()
+
+        override suspend fun onUserInput(sessionId: SessionId, input: String, mode: ChatMode): RouterResponse =
+            error("unused in narration tests")
+
+        override suspend fun narrate(sessionId: SessionId, trigger: NarrationTrigger) {
+            calls.add(sessionId to trigger)
+        }
+    }
+
+    private fun storedEvent(payload: EventPayload, seq: Long): StoredEvent = StoredEvent(
+        metadata = EventMetadata(
+            eventId = EventId("evt-$seq"),
+            sessionId = sessionId,
+            timestamp = timestamp,
+            schemaVersion = 1,
+            causationId = null,
+            correlationId = null,
+        ),
+        sequence = seq,
+        sessionSequence = seq,
+        payload = payload,
+    )
+
+    @Test
+    fun `triggers are narrated in order for matching events`(): Unit = runBlocking {
+        val facade = RecordingRouterFacade()
+        NarrationSubscriber(fakeStore, facade, scope).start()
+
+        // Wait for subscription to be attached
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        liveFlow.emit(storedEvent(StageFailedEvent(sessionId, StageId("b"), TransitionId("t2"), reason = "x"), seq = 3L))
+        liveFlow.emit(storedEvent(WorkflowCompletedEvent(sessionId, StageId("c"), totalStages = 2), seq = 4L))
+
+        withTimeout(2_000L) {
+            while (facade.calls.size < 3) yield()
+        }
+
+        assertEquals(3, facade.calls.size)
+        assertEquals("stage_completed", facade.calls[0].second.kind)
+        assertEquals("stage_failed", facade.calls[1].second.kind)
+        assertEquals("workflow_completed", facade.calls[2].second.kind)
+    }
+
+    @Test
+    fun `maxPerRun=1 skips second trigger after first narration`(): Unit = runBlocking {
+        val facade = RecordingRouterFacade()
+        NarrationSubscriber(fakeStore, facade, scope, maxPerRun = 1).start()
+
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("b"), TransitionId("t2")), seq = 3L))
+
+        withTimeout(2_000L) {
+            while (facade.calls.size < 1) yield()
+        }
+        // Give a moment for a second call to arrive (it shouldn't)
+        kotlinx.coroutines.delay(100L)
+
+        assertEquals(1, facade.calls.size)
+        assertEquals("stage_completed", facade.calls[0].second.kind)
+    }
+
+    @Test
+    fun `a throwing narrate does not prevent subsequent narrations`(): Unit = runBlocking {
+        var callCount = 0
+        val facade = object : RouterFacade {
+            val calls = CopyOnWriteArrayList<String>()
+
+            override suspend fun onUserInput(sessionId: SessionId, input: String, mode: ChatMode): RouterResponse =
+                error("unused")
+
+            override suspend fun narrate(sessionId: SessionId, trigger: NarrationTrigger) {
+                callCount++
+                if (callCount == 1) throw RuntimeException("simulated failure")
+                calls.add(trigger.kind)
+            }
+        }
+        NarrationSubscriber(fakeStore, facade, scope).start()
+
+        while (liveFlow.subscriptionCount.value == 0) yield()
+
+        liveFlow.emit(storedEvent(WorkflowStartedEvent(sessionId, "wf", StageId("s0")), seq = 1L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("a"), TransitionId("t1")), seq = 2L))
+        liveFlow.emit(storedEvent(StageCompletedEvent(sessionId, StageId("b"), TransitionId("t2")), seq = 3L))
+
+        withTimeout(2_000L) {
+            while (facade.calls.size < 1) yield()
+        }
+
+        assertEquals(1, facade.calls.size)
+        assertEquals("stage_completed", facade.calls[0])
+    }
+}