Изменения вводят live-канал OnlineOutbox и фоновую синхронизацию диалогов.

2026-09-30 11:43:49 +03:00
parent 509763cf36
commit bd7e079780
25 changed files with 491 additions and 102 deletions
@@ -11,6 +11,7 @@ import pw.binom.a2a.model.Role
 import pw.binom.a2a.model.TextPart
 import pw.binom.a2a.server.AgentHandler
 import pw.binom.agentik.outbox.Event
+import pw.binom.agentik.outbox.OnlineEvent
 import pw.binom.agentik.proto.Agent
 import pw.binom.agentik.proto.Content
 import pw.binom.agentik.proto.Conversation
@@ -27,9 +28,10 @@ private val log = KotlinLogging.logger {}
 *    -> новый диалог, контекст пересоздаётся.
 *  - id внутреннего диалога отдаётся клиенту в `metadata.agentikConversationId` ответа.
 *
- * Ответ A2A = склеенные [Event.AppendText] нашего хода. Подписку на [Conversation.events]
- * открываем ДО [Conversation.send] (иначе события начала хода могут быть упущены),
- * завершение хода ждём по [Event.End] / [Event.Interrupted] / [Event.Error].
+ * Ответ A2A = склеенные [OnlineEvent.AppendText] нашего хода. Подписку на онлайн-поток
+ * ([pw.binom.agentik.outbox.OnlineOutbox]) открываем ДО [Conversation.send] (live-only,
+ * без catchup — события начала хода иначе можно упустить), завершение хода ждём
+ * по durable-событиям [Event.End] / [Event.Interrupted] / [Event.Error].
 *
 * Ограничение v1: tool-события и картинки в A2A-ответ не транслируются;
 * при нескольких ходов в очереди за контекстом текст предыдущего хода
@@ -48,17 +50,21 @@ class A2aBridge(private val agent: Agent) : AgentHandler {
        val since = conv.updatedAt
        val reply = StringBuilder()
        val turnDone = CompletableDeferred<Unit>()
-        val subscription = async {
+        // Онлайн-поток — дельты ответа (live-only, без catchup).
+        val onlineJob = async {
+            agent.onlineOutbox.onlineEvents(conv.id).collect { e ->
+                if (e is OnlineEvent.AppendText) reply.append(e.body)
+            }
+        }
+        // Durable-поток — терминатор хода (catchup + live).
+        val turnJob = async {
            agent.outbox.conversationEvents(since, conv.id).collect { ce ->
-                val e = ce.event
-                when (e) {
-                    is Event.AppendText -> reply.append(e.body)
+                when (val e = ce.event) {
                    is Event.End, is Event.Interrupted -> turnDone.complete(Unit)
                    is Event.Error ->
-                        if (!turnDone.completeExceptionally(
-                                IllegalStateException("agent turn failed: ${e.message}")
-                            )
-                        ) {}
+                        turnDone.completeExceptionally(
+                            IllegalStateException("agent turn failed: ${e.message}")
+                        )
                    else -> {}
                }
            }
@@ -67,7 +73,8 @@ class A2aBridge(private val agent: Agent) : AgentHandler {
        try {
            turnDone.await()
        } finally {
-            subscription.cancel()
+            onlineJob.cancel()
+            turnJob.cancel()
        }
        log.info { "a2a context=$contextId conv=${conv.id} reply=${reply.length} chars" }
        Message(
@@ -29,6 +29,8 @@ import pw.binom.agentik.outbox.AgentEvent
 import pw.binom.agentik.outbox.Event as OutboxEvent
 import pw.binom.agentik.outbox.CommonEvent
 import pw.binom.agentik.outbox.MutableOutboxStore
+import pw.binom.agentik.outbox.MutableOnlineOutbox
+import pw.binom.agentik.outbox.OnlineOutbox
 import pw.binom.agentik.journal.JournalStore
 import pw.binom.agentik.outbox.OutboxStore
 import pw.binom.agentik.proto.Conversation as ProtoConversation
@@ -213,6 +215,13 @@ private val eventStore: MutableOutboxStore = pw.binom.agentik.outbox.inmemory.In
    ttl = null,
 )

+/**
+ * Live-канал стриминга ответа (дельты текста/картинок и фазовые маркеры).
+ * Онлайн-события никогда не сохраняются и не реплеятся — см. [OnlineOutbox].
+ * Durable-события по-прежнему идут в [eventStore].
+ */
+private val onlineEventStore: MutableOnlineOutbox = pw.binom.agentik.outbox.inmemory.InMemoryOnlineOutbox()
+
    /**
     * Собирает **актуальный** список тулов для диспетчеризации:
     * внешние из [toolProviders] + testTools + встроенные (read_skill,
@@ -413,6 +422,15 @@ private val eventStore: MutableOutboxStore = pw.binom.agentik.outbox.inmemory.In
    override val outbox: OutboxStore
        get() = eventStore

+    /**
+     * Read-only view of [onlineEventStore] для HTTP-фасада в `:server`
+     * (`Route.agentikAgent` → `/conversations/{id}/online` SSE).
+     *
+     * Live-only: без catchup/курсора, события не сохраняются.
+     */
+    override val onlineOutbox: OnlineOutbox
+        get() = onlineEventStore
+
    /**
     * Read-only view of [mutableConversationStore] для HTTP-фасада в `:server`
     * (`GET /conversations` → список [ConversationRecord] для UI).
@@ -456,6 +474,7 @@ private val eventStore: MutableOutboxStore = pw.binom.agentik.outbox.inmemory.In
            workingMemoryStore = workingMemoryStore,
            reflectionStore = reflectionStore,
            eventStore = eventStore,
+            onlineEventStore = onlineEventStore,
            llm = llm,
            systemPrompt = systemPrompt,
            systemPromptResolver = { buildRuntimeSystemPrompt() },
@@ -528,6 +547,7 @@ private val eventStore: MutableOutboxStore = pw.binom.agentik.outbox.inmemory.In
        workingMemoryStore = workingMemoryStore,
        reflectionStore = reflectionStore,
        eventStore = eventStore,
+            onlineEventStore = onlineEventStore,
        llm = llm,
        systemPrompt = systemPrompt,
        systemPromptResolver = { buildRuntimeSystemPrompt() },
@@ -5,10 +5,18 @@ import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
 import pw.binom.agentik.outbox.CommonEvent
 import pw.binom.agentik.outbox.Event
+import pw.binom.agentik.outbox.MutableOnlineOutbox
 import pw.binom.agentik.outbox.MutableOutboxStore
+import pw.binom.agentik.outbox.OnlineEvent

+/**
+ * Фасад эмиссии и чтения событий одного диалога. Разводит два канала:
+ *  - durable ([Event]) → [globalEventStore] (`:outbox`), с catchup по `after`;
+ *  - online ([OnlineEvent]) → [onlineStore], live-only (без catchup).
+ */
 internal class ConversationEvents(
    private val globalEventStore: MutableOutboxStore,
+    private val onlineStore: MutableOnlineOutbox,
    private val conversationId: String,
 ) {
    fun tryEmit(event: Event): Boolean {
@@ -27,4 +35,11 @@ internal class ConversationEvents(
    fun events(after: kotlin.time.Instant?): Flow<Event> =
        globalEventStore.conversationEvents(after = after, conversationId = conversationId)
            .map { it.event }
+
+    /** Best-effort эмиссия онлайн-события — без блокировки продюсера и без хранения. */
+    fun tryEmitOnline(event: OnlineEvent): Boolean =
+        onlineStore.tryAppendOnline(conversationId, event)
+
+    /** Live-поток онлайн-событий диалога (без catchup — см. KDoc [pw.binom.agentik.outbox.OnlineOutbox]). */
+    fun onlineEvents(): Flow<OnlineEvent> = onlineStore.onlineEvents(conversationId)
 }
@@ -22,6 +22,8 @@ import pw.binom.agentik.memory.MemoryStore
 import pw.binom.agentik.proto.Content as ProtoContent
 import pw.binom.agentik.proto.Conversation as ProtoConversation
 import pw.binom.agentik.outbox.Event as ProtoEvent
+import pw.binom.agentik.outbox.OnlineEvent
+import pw.binom.agentik.outbox.MutableOnlineOutbox
 import pw.binom.agentik.reflection.ReflectionStore
 import pw.binom.agentik.proto.Message as ProtoMessage
 import pw.binom.agentik.proto.MessageContext as ProtoMessageContext
@@ -53,6 +55,11 @@ class ConversationLoop(
    private val workingMemoryStore: ContextStore,
    private val reflectionStore: ReflectionStore?,
    private val eventStore: pw.binom.agentik.outbox.MutableOutboxStore,
+    /**
+     * Live-канал онлайн-событий (дельты ответа). Не сохраняется; подписка
+     * возможна только «онлайн». Durable-события по-прежнему в [eventStore].
+     */
+    private val onlineEventStore: MutableOnlineOutbox,
    private val llm: LiteLlm,
    /**
     * Базовый системный промпт, который задаётся беседе при создании.
@@ -102,6 +109,7 @@ class ConversationLoop(

    private val events = ConversationEvents(
        globalEventStore = eventStore,
+        onlineStore = onlineEventStore,
        conversationId = state.id,
    )

@@ -259,8 +267,8 @@ class ConversationLoop(
            compactor.compactPreTurnIfNeeded()
        }

-        emitEvent(ProtoEvent.StartReasoning(date = turnStarted))
-        emitEvent(ProtoEvent.StartResponse(date = now(), responseType = ProtoEvent.ResponseType.TEXT))
+        emitOnline(OnlineEvent.StartReasoning(date = turnStarted))
+        emitOnline(OnlineEvent.StartResponse(date = now(), responseType = OnlineEvent.ResponseType.TEXT))

        val parts = userRecord.content.mapNotNull { c ->
            when (c) {
@@ -327,7 +335,7 @@ class ConversationLoop(
                        lc.sendStreamContents(pendingParts).collect { delta ->
                            if (delta.text.isNotEmpty()) {
                                reply.append(delta.text)
-                                emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text))
+                                emitOnline(OnlineEvent.AppendText(date = now(), body = delta.text))
                            }
                            if (delta.toolCalls.isNotEmpty()) {
                                collectedCalls.addAll(delta.toolCalls)
@@ -365,7 +373,7 @@ class ConversationLoop(
                        }
                        if (delta.text.isNotEmpty()) {
                            reply.append(delta.text)
-                            emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text))
+                            emitOnline(OnlineEvent.AppendText(date = now(), body = delta.text))
                        }
                        if (delta.toolCalls.isNotEmpty()) {
                            nextCalls.addAll(delta.toolCalls)
@@ -382,7 +390,7 @@ class ConversationLoop(
                                lc.sendStreamContents(listOf(LiteContentPart.Text(" "))).collect { followUp ->
                                    if (followUp.text.isNotEmpty()) {
                                        reply.append(followUp.text)
-                                        emitEvent(ProtoEvent.AppendText(date = now(), body = followUp.text))
+                                        emitOnline(OnlineEvent.AppendText(date = now(), body = followUp.text))
                                    }
                                    if (followUp.toolCalls.isNotEmpty()) {
                                        collectedPostTool.addAll(followUp.toolCalls)
@@ -491,6 +499,10 @@ class ConversationLoop(
        events.tryEmit(event)
    }

+    private fun emitOnline(event: OnlineEvent) {
+        events.tryEmitOnline(event)
+    }
+
    private suspend fun failTurn(message: String, code: String? = null) {
        val ts = now()
        if (!state.isTemporal) {
@@ -8,6 +8,7 @@ import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import pw.binom.agentik.outbox.AgentEvent
+import pw.binom.agentik.outbox.OnlineEvent
 import pw.binom.agentik.proto.Content
 import pw.binom.agentik.outbox.Event as ProtoEvent
 import pw.binom.agentik.skill.mining.SkillReadTool
@@ -362,33 +363,42 @@ class ChatAgentTest {
    }

    @Test
-    fun `Working event is the first event of a turn`() = runTest {
-        // Working-маркер обязан прийти самым первым событием хода, до
-        // StartReasoning/StartResponse/AppendText/End. Это позволяет UI
-        // показать спиннер сразу же при отправке, не дожидаясь первого
-        // токена от LLM.
+    fun `Working is first durable event, streaming goes to online channel`() = runTest {
+        // Working-маркер обязан прийти самым первым durable-событием хода, до
+        // End. Это позволяет UI показать спиннер сразу же при отправке, не
+        // дожидаясь первого токена от LLM.
+        //
+        // Стриминг (StartReasoning / StartResponse / AppendText) — теперь
+        // онлайн-события (live-only, не сохраняются) и приходят из
+        // [OnlineOutbox], а НЕ из durable-потока.
        val agent = newAgent()
        fakeLlm.reply = "ok"
        val conv = agent.createConversation(temp = false)

-        val events = mutableListOf<ProtoEvent>()
-        val job = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
-            agent.outbox.conversationEvents(Instant.DISTANT_PAST, conv.id).collect { events.add(it.event) }
+        val durable = mutableListOf<ProtoEvent>()
+        val online = mutableListOf<OnlineEvent>()
+        val durableJob = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
+            agent.outbox.conversationEvents(Instant.DISTANT_PAST, conv.id).collect { durable.add(it.event) }
+        }
+        val onlineJob = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
+            agent.onlineOutbox.onlineEvents(conv.id).collect { online.add(it) }
        }
        conv.send(listOf(Content.Text("hi")))
        delay(50)
-        job.cancel()
+        durableJob.cancel()
+        onlineJob.cancel()

-        // Working — первый event хода (индекс 0).
-        assertTrue(events.isNotEmpty(), "no events captured: $events")
-        val first = events.first()
+        // Working — первый durable event хода (индекс 0).
+        assertTrue(durable.isNotEmpty(), "no events captured: $durable")
+        val first = durable.first()
        assertIs<ProtoEvent.Working>(first)
-        assertTrue(events.any { it is ProtoEvent.StartReasoning }, "no StartReasoning: $events")
-        assertTrue(events.any { it is ProtoEvent.StartResponse }, "no StartResponse: $events")
-        assertTrue(events.any { it is ProtoEvent.End }, "no End: $events")
+        assertTrue(durable.any { it is ProtoEvent.End }, "no End: $durable")
+        // Ранее стриминговые маркеры — теперь в онлайн-канале.
+        assertTrue(online.any { it is OnlineEvent.StartReasoning }, "no StartReasoning: $online")
+        assertTrue(online.any { it is OnlineEvent.StartResponse }, "no StartResponse: $online")

        // Working не дублируется (emit'ится один раз на send()).
-        assertEquals(1, events.count { it is ProtoEvent.Working }, "Working emitted >1 times: $events")
+        assertEquals(1, durable.count { it is ProtoEvent.Working }, "Working emitted >1 times: $durable")
    }

    @Test