From 689020e16105bbc964d13f3110f031de6a811af9 Mon Sep 17 00:00:00 2001 From: kami Date: Wed, 3 Jun 2026 18:41:58 +0400 Subject: [PATCH] =?UTF-8?q?feat(server):=20Task=203.3=20=E2=80=94=20Narrat?= =?UTF-8?q?ionSubscriber=20+=20wiring?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/correx/apps/server/ServerModule.kt | 6 + .../server/narration/NarrationSubscriber.kt | 98 +++++++++++ .../narration/NarrationSubscriberTest.kt | 160 ++++++++++++++++++ 3 files changed, 264 insertions(+) create mode 100644 apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt create mode 100644 apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt b/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt index 1d48baeb..6c69d6d6 100644 --- a/apps/server/src/main/kotlin/com/correx/apps/server/ServerModule.kt +++ b/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() } /** diff --git a/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt b/apps/server/src/main/kotlin/com/correx/apps/server/narration/NarrationSubscriber.kt new file mode 100644 index 00000000..42b64435 --- /dev/null +++ b/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, var used: Int) + + private val lanes = ConcurrentHashMap() + + 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(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) + } +} diff --git a/apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt b/apps/server/src/test/kotlin/com/correx/apps/server/narration/NarrationSubscriberTest.kt new file mode 100644 index 00000000..db607dab --- /dev/null +++ b/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(extraBufferCapacity = 64) + + private val fakeStore = object : EventStore { + override suspend fun append(event: NewEvent): StoredEvent = error("unused") + override suspend fun appendAll(events: List): List = error("unused") + override fun read(sessionId: SessionId): List = emptyList() + override fun readFrom(sessionId: SessionId, fromSequence: Long): List = emptyList() + override fun lastSequence(sessionId: SessionId): Long = 0L + override fun subscribe(sessionId: SessionId): Flow = liveFlow + override fun allEvents(): Sequence = emptySequence() + override fun allSessionIds(): Set = emptySet() + override fun subscribeAll(): Flow = liveFlow + override suspend fun lastGlobalSequence(): Long = 0L + } + + private class RecordingRouterFacade : RouterFacade { + val calls = CopyOnWriteArrayList>() + + 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() + + 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]) + } +}