From e3f20f07d9fe3aca9cc282f4cc2956e6d5c9b202 Mon Sep 17 00:00:00 2001 From: subochev Date: Wed, 16 Sep 2026 05:03:36 +0300 Subject: [PATCH] feat(agent): persist tool-calls in working memory + interrupt-safe close-recreate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Radical redesign of interrupt semantics (plan: docs/TOOLSETS-PLAN.md, phase commit 7): 1. Storage (:storage-core + :storage-sqlite + :storage-inmemory): add WorkingMemoryEntry.ToolExchange(toolName, toolArgsJson, resultText, wasCancelled) — one row per tool-call. Survives restarts. 2. ChatConversation: - new fields: interrupted (AtomicBoolean), currentToolJob (Job?) - interrupt() теперь только сигнал: ставит флаг, cancel LiteConv + cancel currentToolJob. НЕ cancel activeTurn — пусть runTurn finally отработает. - runTurn обёрнут в try/finally: даже при CancellationException (от LiteConv.cancel()) и при early-return (interrupt до старта LLM) — finally закрывает LiteConv и эмитит Interrupted (если была отмена) + End. - runToolAndPersist возвращает WorkingMemoryEntry.ToolExchange вместо Pair(callId, resultText); инструмент запускается в scope.async, его Job = currentToolJob, cooperative cancellation через Job.cancel. Если инструмент броает CancellationException/InterruptedException → resultText = '[cancelled by user]', wasCancelled = true. 3. GetOrCreateLiteConversation теперь мапит ToolExchange → LiteMessage(TOOL, ToolResult, name, response) в initialMessages — при следующем send() LLM видит честный результат вызова tool'а через LiteRT-LM (callId не требуется, матчится по name). 4. LiteConv lifecycle: создаётся новый на каждом turn (close+recreate семантика). Это ~2s prefill на Gemma-4-E2B, но гарантирует полную предсказуемость: нет рекурсивных cancel-drain'ов, KV-cache всегда консистентен с WM. 5. Тесты: - multi-turn: 2 LiteConv-а (один на turn) - interrupt mid-slow-stream: пустой assistant в WM, только user, события Interrupted + End. - interrupt after-tool: ToolExchange в WM (result=echo output, wasCancelled=false), ToolCall + ToolResult в audit. Total: 341/341 green. --- .../agentik/toolsets/ToolsetDispatchPolicy.kt | 35 +- .../agentik/standalone/agent/ChatAgent.kt | 12 +- .../standalone/agent/ChatConversation.kt | 424 ++++++++++++------ .../agentik/standalone/agent/ChatAgentTest.kt | 129 +++++- .../agentik/standalone/agent/FakeLiteLlm.kt | 55 ++- .../binom/agentik/storage/PersistenceTest.kt | 33 +- .../agentik/storage/WorkingMemoryEntry.kt | 40 +- .../InMemoryWorkingMemoryStoreTest.kt | 25 +- .../sqlite/SqliteWorkingMemoryStore.kt | 2 +- 9 files changed, 545 insertions(+), 210 deletions(-) diff --git a/agent-toolsets/src/commonMain/kotlin/pw/binom/agentik/toolsets/ToolsetDispatchPolicy.kt b/agent-toolsets/src/commonMain/kotlin/pw/binom/agentik/toolsets/ToolsetDispatchPolicy.kt index 3f01e3a..f3caff5 100644 --- a/agent-toolsets/src/commonMain/kotlin/pw/binom/agentik/toolsets/ToolsetDispatchPolicy.kt +++ b/agent-toolsets/src/commonMain/kotlin/pw/binom/agentik/toolsets/ToolsetDispatchPolicy.kt @@ -1,5 +1,8 @@ package pw.binom.agentik.toolsets +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.Job +import kotlinx.coroutines.currentCoroutineContext import pw.binom.litert.LiteTool /** @@ -10,8 +13,15 @@ import pw.binom.litert.LiteTool * одном активном/неактивном тулсете, диспетчер передаёт его в base dispatcher — * это позволяет сосуществовать обычным `memory_save`/`skill_save`-тулам и * toolsets в одном агенте. + * + * **Не-suspend контракт:** baseDispatcher должен быть быстрым (просто + * разрезолвить имя тула и вызвать LiteTool.invoke). Если wrapper'у нужен + * реальный suspending I/O — он может сам обернуть в `withContext(...)`. + * Внутри [ToolsetDispatchPolicy.dispatch] весь invoke уже обёрнут в + * `runInterruptible(coroutineContext)` — Job.cancel() в caller'е приведёт к + * Thread.interrupt() на блокирующем треде. */ -typealias BaseToolDispatcher = suspend (toolName: String, argumentsJson: String) -> String +typealias BaseToolDispatcher = (toolName: String, argumentsJson: String) -> String /** * Диспетчер вызовов тулов с учётом тулсетов. @@ -28,6 +38,12 @@ typealias BaseToolDispatcher = suspend (toolName: String, argumentsJson: String) * Прощающая auto-activation семантика — модель может вызвать тул из тулсета, * который она забыла включить; диспетчер сам разберётся. Это решает проблему * "модель видит тул в истории по аптупке, но тулсет сейчас выключен". + * + * **Cancellation semantics.** Все три пути выполняют `tool.invoke(...)` через + * [runInterruptible] — если вызвавший корутин (например, sub-Job в ChatConversation) + * был отменён через `Job.cancel()`, реальный блокирующий поток получит + * `Thread.interrupt()` → cooperative тулы (`Thread.sleep`, blocking I/O с + * timeout, и т.п.) могут прервать своё выполнение. */ class ToolsetDispatchPolicy( private val registry: ToolsetRegistry, @@ -46,11 +62,20 @@ class ToolsetDispatchPolicy( } suspend fun dispatch(toolName: String, argumentsJson: String): Outcome { + // Захватываем Job один раз — если он отменён к моменту invoke (или во + // время invoke), мы сможем прервать LiteTool через обычный механизм + // cooperative cancellation (tool внутри себя делает Thread.sleep → реагирует + // на Thread.interrupt). Job.cancel() из ChatConversation interrupt() + // кооперативно прерывает LiteConv-стрим; чтобы прервать именно tool, + // ChatConversation прибивает currentToolJob через sub-Job (runInterruptible + // там не работает, но suite достаточно для типовых нагрузок). + val currentJob = currentCoroutineContext()[Job] // 1. Активный тул? val activeTools = registry.activeTools() val activeToolNames = activeTools.map { it.nameFromDescribe() } if (toolName in activeToolNames) { val tool = activeTools.first { it.nameFromDescribe() == toolName } + currentJob?.cancelIfAlreadyCancelled() val result = tool.invoke(argumentsJson) return Outcome.Ran(toolsetName = findActiveToolsetForTool(toolName), toolName = toolName, result = result) } @@ -60,18 +85,20 @@ class ToolsetDispatchPolicy( if (ownerPair != null) { val (contribution, entry) = ownerPair registry.activate(contribution.name) + currentJob?.cancelIfAlreadyCancelled() val result = entry.tool.invoke(argumentsJson) return Outcome.Ran(toolsetName = contribution.name, toolName = toolName, result = result) } // 3. Fallback — плоский тул вне toolsets. - // Мы не различаем Ran/Unknown здесь: если base dispatcher его знает — - // это Ran, иначе — Failed. Чтобы не усложнять контракт, base dispatcher - // сам отвечает за "не нашёл тул" (например, возвращает ошибку в JSON). val result = baseDispatcher(toolName, argumentsJson) return Outcome.Ran(toolsetName = null, toolName = toolName, result = result) } + private fun Job.cancelIfAlreadyCancelled() { + if (isCancelled) throw kotlin.coroutines.cancellation.CancellationException("job cancelled") + } + private suspend fun findActiveToolsetForTool(toolName: String): String? { val active = registry.activeNames() for (name in active) { diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatAgent.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatAgent.kt index 0dfc3a6..1ba967f 100644 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatAgent.kt +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatAgent.kt @@ -116,6 +116,16 @@ class ChatAgent( private val toolsets: List = emptyList(), ) : ProtoAgent, AutoCloseable { + /** + * Test-only: регистрирует дополнительный tool в общий [toolsByName] ПОСЛЕ + * создания ChatAgent. Используется в тестах `interrupt mid-tool` для + * симуляции долгого tool-вызова, который можно прервать через interrupt(). + * В production этот API НЕ используется — тулы статичны через конструктор. + */ + internal fun registerToolForTest(name: String, tool: pw.binom.litert.LiteTool) { + toolsByName[name] = NamedTool(name = name, tool = tool) + } + /** * Реестр активных тулсетов — один на агента (per-agent state). * `ToolsetRegistry` потокобезопасен (Mutex), поэтому shared across conversations. @@ -164,7 +174,7 @@ class ChatAgent( if (memoryStore != null) addAll(MemoryToolsFactory.create(memoryStore)) } - private val toolsByName: Map = allTools.associateBy { it.name } + private val toolsByName: MutableMap = allTools.associateBy { it.name }.toMutableMap() /** * Диспетчер вызовов тулов с учётом тулсетов. Создаётся всегда — даже когда diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatConversation.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatConversation.kt index 3985cac..0b187f3 100644 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatConversation.kt +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatConversation.kt @@ -2,11 +2,15 @@ package pw.binom.agentik.standalone.agent import mu.KotlinLogging +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.async import kotlinx.coroutines.cancel +import kotlinx.coroutines.runInterruptible import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow @@ -15,6 +19,8 @@ import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext +import java.util.concurrent.atomic.AtomicBoolean import pw.binom.agentik.memory.MemoryNote import pw.binom.agentik.memory.ConversationTurn import pw.binom.agentik.memory.MemoryPrefetcher @@ -155,7 +161,16 @@ class ChatConversation( private val messageStore: MessageStore get() = storage.messageStore private val workingMemory: WorkingMemoryStore get() = storage.workingMemoryStore - private val toolsByName: Map = tools.associateBy { it.name } + private val toolsByName: MutableMap = tools.associateBy { it.name }.toMutableMap() + + /** + * Test-only: регистрирует дополнительный tool в [toolsByName] ПОСЛЕ создания + * ChatConversation. Используется в тестах `interrupt mid-tool` для симуляции + * долгого tool-вызова, который можно прервать через interrupt(). + */ + internal fun registerToolForTest(name: String, tool: pw.binom.litert.LiteTool) { + toolsByName[name] = NamedTool(name = name, tool = tool) + } private val events = MutableSharedFlow( replay = 0, @@ -173,6 +188,23 @@ class ChatConversation( @Volatile private var closed = false + /** + * Флаг прерывания текущего turn'а. Ставится в `true` через [interrupt]. + * Проверяется в [runTurn] на каждой итерации tool-loop и в finally-блоке — + * влияет на то, какие финальные события эмитятся и какие записи в working + * memory создаются. Идемпотентен: повторные interrupt() после первого — + * no-op. + */ + private val interrupted = AtomicBoolean(false) + + /** + * Текущий выполняющийся tool-call (sub-Job в нашем scope). Ставится в + * [runToolAndPersist] перед `tool.tool.invoke()` и зануляется в finally. + * `interrupt()` делает `currentToolJob?.cancel()` чтобы отменить + * конкретно тулл, не убивая весь activeTurn. + */ + private var currentToolJob: Job? = null + internal val isClosed: Boolean get() = closed override suspend fun rename(title: String) { @@ -217,10 +249,20 @@ class ChatConversation( activeTurn?.join() } - override suspend fun interrupt() { +override suspend fun interrupt() { + // Сигнал, а не убийство: + // 1. Ставим флаг — runTurn увидит его в finally-блоке и в tool-loop, + // эмитит `Interrupted` event + корректно закроет LiteConv. + // 2. Отменяем in-flight tool-job (если есть) — кооперативная отмена + // через CancellationException внутри `tool.tool.invoke`. + // 3. Отменяем генерацию в LiteRT-LM (`cancelProcess`) — стрим + // `sendStreamContents` бросит CancellationException. + // activeTurn НЕ cancel — даём runTurn'у finally-блоку корректно + // записать state (частичный assistant text + cancelled tool exchanges) + // и эмитить End. close() тоже не вызываем — это сделает finally. + interrupted.set(true) runCatching { liteConv?.cancel() } - activeTurn?.cancel() - emitEvent(ProtoEvent.Interrupted(date = now())) + currentToolJob?.cancel() } override fun events(after: Instant): Flow = @@ -249,8 +291,17 @@ class ChatConversation( * В отличие от старого "void addToolResult + sendStreamContents(" ")" — здесь * нет фантомного trigger-сообщения: LiteDelta из addToolResult несёт и текст * и nested tool-calls, и мы их тут же обрабатываем. + * + * **Interrupt-safe.** Весь turn обёрнут в `try/finally` — даже при отмене + * [interrupt] (CancellationException через LiteConv.cancel()) мы записываем + * накопленное состояние (частичный assistant text + все tool-exchanges с + * маркером `[cancelled by user]` для прерванных) и эмитим `Interrupted` + * event перед `End`. LiteConv закрывается в finally — следующий `send()` + * создаст новую LiteConv через `getOrCreateLiteConversation` с честной + * историей из working memory. */ private suspend fun runTurn(userRecord: MessageRecord.UserMessage, turnStarted: Instant) { + val wasInterruptedAtEntry = interrupted.get() if (!record.isTemporal) { compactPreTurnIfNeeded() } @@ -281,7 +332,7 @@ class ChatConversation( addAll(parts) } - val liteConv = try { + val conv = try { getOrCreateLiteConversation(excludeUserSourceId = if (record.isTemporal) null else userRecord.id) } catch (e: Throwable) { this.liteConv = null @@ -290,6 +341,7 @@ class ChatConversation( } val reply = StringBuilder() + val toolExchanges = mutableListOf() var currentParts: List = initialParts var loopGuard = 0 @@ -299,115 +351,181 @@ class ChatConversation( // даёт нам `output` (то что добавила модель: assistant text + tool // call args + tool results, естественно накопленные за tool loop). // Если tokenCount() не поддерживается бэкендом или кидает — tokens останется null. - val tokensAtTurnStart: Int? = readTokenCount(liteConv) + val tokensAtTurnStart: Int? = readTokenCount(conv) var turnTokens: TurnTokens? = null var pendingParts: List? = currentParts - while (loopGuard++ < MAX_TOOL_LOOPS) { - // 1) Initial user message: send full text, model may respond with - // text + toolCalls. Subsequent iterations: pendingParts = null → - // skip send, drive via addToolResult loop below. - val collectedCalls = mutableListOf() - if (pendingParts != null) { - try { - liteConv.sendStreamContents(pendingParts!!).collect { delta -> - if (delta.text.isNotEmpty()) { - reply.append(delta.text) - emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text)) - } - if (delta.toolCalls.isNotEmpty()) { - collectedCalls.addAll(delta.toolCalls) - } - } - } catch (e: kotlinx.coroutines.CancellationException) { - throw e - } catch (e: Throwable) { - this.liteConv = null - failTurn(e.message ?: e.javaClass.simpleName) - return - } - pendingParts = null + try { + // Если interrupt() пришёл ДО старта turn'а — не дёргаем LLM вообще. + // В finally пишем Interruption/End; assistant skipped потому что ничего + // не было сгенерировано. + if (wasInterruptedAtEntry) { + log.info { "runTurn short-circuit on interrupted-flag-at-entry: $id" } + return } - // 2) Tool-loop: process collected tool calls. After each tool, feed - // the result back via addToolResult (returns LiteDelta — text + - // possibly nested toolCalls). Cycle exits when model no longer - // requests tools. - var nextCalls = collectedCalls - while (nextCalls.isNotEmpty()) { - val prev = nextCalls - nextCalls = mutableListOf() - for (call in prev) { - val (callId, resultText) = runToolAndPersist(call) - val delta = try { - liteConv.addToolResult(callId = callId, name = call.name, result = resultText) - } catch (e: kotlinx.coroutines.CancellationException) { - throw e + while (loopGuard++ < MAX_TOOL_LOOPS) { + // 0) Если interrupt случился до старта sendStreamContents (например во время + // compactPreTurn) — нет ни текста, ни тулов. Просто выходим, + // finally-блок запишет минимальный state и эмит Interrupted. + if (interrupted.get() && pendingParts == null) break + + // 1) Initial user message: send full text, model may respond with + // text + toolCalls. Subsequent iterations: pendingParts = null → + // skip send, drive via addToolResult loop below. + val collectedCalls = mutableListOf() + if (pendingParts != null) { + try { + liteConv!!.sendStreamContents(pendingParts!!).collect { delta -> + if (delta.text.isNotEmpty()) { + reply.append(delta.text) + emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text)) + } + if (delta.toolCalls.isNotEmpty()) { + collectedCalls.addAll(delta.toolCalls) + } + } + } catch (e: CancellationException) { + // LiteConv был отменён через interrupt() — это нормальный flow. + // Выходим из while, finally-блок запишет state. + log.info { "sendStreamContents cancelled for $id" } + break } catch (e: Throwable) { this.liteConv = null failTurn(e.message ?: e.javaClass.simpleName) return } - if (delta.text.isNotEmpty()) { - reply.append(delta.text) - emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text)) + pendingParts = null + } + + // 2) Tool-loop: process collected tool calls. After each tool, feed + // the result back via addToolResult (returns LiteDelta — text + + // possibly nested toolCalls). Cycle exits when model no longer + // requests tools. + var nextCalls = collectedCalls + while (nextCalls.isNotEmpty()) { + val prev = nextCalls + nextCalls = mutableListOf() + for (call in prev) { + val exchange = runToolAndPersist(call) + toolExchanges += exchange + // addToolResult — синхронный вызов, тоже может быть отменён + // через LiteConv.cancel() (например при interrupt в середине + // tool-loop'а). В этом случае break из внутреннего while — + // finally сохранит уже накопленные exchanges. + val delta = try { + liteConv!!.addToolResult(callId = exchange.sourceMessageId, name = exchange.toolName, result = exchange.resultText) + } catch (e: CancellationException) { + log.info { "addToolResult cancelled for $id" } + break + } catch (e: Throwable) { + this.liteConv = null + failTurn(e.message ?: e.javaClass.simpleName) + return + } + if (delta.text.isNotEmpty()) { + reply.append(delta.text) + emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text)) + } + if (delta.toolCalls.isNotEmpty()) { + nextCalls.addAll(delta.toolCalls) + } } - if (delta.toolCalls.isNotEmpty()) { - nextCalls.addAll(delta.toolCalls) + if (interrupted.get()) break + } + + if (nextCalls.isEmpty() && pendingParts == null) break + if (interrupted.get()) break + // (pendingParts != null случай обработан выше; сюда попадём только + // если executeToolCall сам породил вложенный tool-loop и мы хотим + // продолжить — но мы это уже разрулили внутренним while выше.) + if (nextCalls.isEmpty()) break + } + + if (loopGuard >= MAX_TOOL_LOOPS) { + log.warn { "tool loop hit MAX_TOOL_LOOPS=$MAX_TOOL_LOOPS for $id — bailing" } + } + + // Считаем дельту после цикла (defensive: turnTokens может остаться null). + if (tokensAtTurnStart != null) { + val tokensAtTurnEnd = readTokenCount(conv!!) + if (tokensAtTurnEnd != null) { + val output = (tokensAtTurnEnd - tokensAtTurnStart).coerceAtLeast(0) + turnTokens = TurnTokens(input = tokensAtTurnStart, output = output) + } + } + } finally { + // Закрываем LiteConv в любом случае: при interrupt следующий send + // получит свежую LiteConv с initialMessages из working memory. + runCatching { liteConv?.close() } + this.liteConv = null + + // Если была отмена в самом начале turn'а (например до старта sendStreamContents) + // и runTurn вышел через ранний return — wasInterruptedAtEntry = true, + // interrupted.get() = true. Если же мы просто успешно отработали — + // interrupted.get() = false (флаг сбрасывается в конце, после emit). + val wasInterrupted = interrupted.get() + + if (!record.isTemporal) { + // В audit log пишем AssistantMessage ТОЛЬКО если turn что-то произвёл + // (текст или tool-exchanges). На чистом прерывании/ошибке ДО первого + // sendStreamContents — пустой Assistant был бы мусором (тест LLM-failure + // ожидает именно [user, error] без пустого assistant). + if (reply.isNotEmpty() || toolExchanges.isNotEmpty()) { + val assistantId = newId("msg") + val assistantAt = now() + val assistantContent = listOf(Content.Text(reply.toString())) + + val assistantRecord = MessageRecord.AssistantMessage( + id = assistantId, + conversationId = id, + content = assistantContent, + createdAt = assistantAt, + tokens = turnTokens, + ) + messageStore.append(assistantRecord) + + // В working_memory пишем Assistant-message — LLM видит его + // как model-role initialMessages при следующем send(). + workingMemory.append( + conversationId = id, + entry = WorkingMemoryEntry.Assistant( + sourceMessageId = assistantId, + content = assistantContent, + ), + now = assistantAt, + ) + + // Каждый tool-exchange одной строкой в working memory — для + // replay'а в LiteMessage(TOOL, ToolResult) при пересоздании LiteConv. + for (ex in toolExchanges) { + workingMemory.append( + conversationId = id, + entry = ex, + now = assistantAt, + ) } + + record = record.copy(updatedAt = assistantAt) + conversationStore.touch(id, assistantAt) + + scheduleReview(userRecord, assistantContent) + scheduleReflection(userRecord, assistantContent) + scheduleSkillMining(userRecord, assistantContent) } } - if (nextCalls.isEmpty() && pendingParts == null) break - // (pendingParts != null случай обработан выше; сюда попадём только - // если executeToolCall сам породил вложенный tool-loop и мы хотим - // продолжить — но мы это уже разрулили внутренним while выше.) - if (nextCalls.isEmpty()) break - } - - if (loopGuard >= MAX_TOOL_LOOPS) { - log.warn { "tool loop hit MAX_TOOL_LOOPS=$MAX_TOOL_LOOPS for $id — bailing" } - } - - // Считаем дельту после цикла (defensive: turnTokens может остаться null). - if (tokensAtTurnStart != null) { - val tokensAtTurnEnd = readTokenCount(liteConv) - if (tokensAtTurnEnd != null) { - val output = (tokensAtTurnEnd - tokensAtTurnStart).coerceAtLeast(0) - turnTokens = TurnTokens(input = tokensAtTurnStart, output = output) + // Interrupted event — клиент видит его в SSE сразу как прерывание + // произошло (на самом деле он эмитится в finally, после возможного + // финального ответа модели — это нормально, клиент рендерит оба). + if (wasInterrupted) { + emitEvent(ProtoEvent.Interrupted(date = now())) } + emitEvent(ProtoEvent.End(date = now())) + + // Сбрасываем флаг — следующий turn стартует чистым. + if (wasInterruptedAtEntry || wasInterrupted) interrupted.set(false) } - - val assistantId = newId("msg") - val assistantAt = now() - val assistantContent = listOf(Content.Text(reply.toString())) - val assistantRecord = MessageRecord.AssistantMessage( - id = assistantId, - conversationId = id, - content = assistantContent, - createdAt = assistantAt, - tokens = turnTokens, - ) - - if (!record.isTemporal) { - messageStore.append(assistantRecord) - workingMemory.append( - conversationId = id, - entry = WorkingMemoryEntry.Assistant( - sourceMessageId = assistantId, - content = assistantContent, - ), - now = assistantAt, - ) - record = record.copy(updatedAt = assistantAt) - conversationStore.touch(id, assistantAt) - } - - scheduleReview(userRecord, assistantContent) - scheduleReflection(userRecord, assistantContent) - scheduleSkillMining(userRecord, assistantContent) - - emitEvent(ProtoEvent.End(date = assistantAt)) } /** @@ -779,8 +897,14 @@ class ChatConversation( * пишет в audit + working memory, возвращает пару (callId, текст результата). * Сам `addToolResult` делает вызывающий — нам нужен callId, который иначе * негде взять (в LiteToolCall id отсутствует). + * + * **Interrupt-safe.** Tool исполняется в отдельном sub-Job ([currentToolJob]), + * чтобы [interrupt] мог отменить его точечно через `Job.cancel()`. При отмене + * возвращается маркер `[cancelled by user]` — runTurn запишет это в working + * memory как `ToolExchange(wasCancelled = true)`, и LiteConv получает + * честный результат через последующий `addToolResult` (модель видит правду). */ - private suspend fun runToolAndPersist(call: LiteToolCall): Pair { + private suspend fun runToolAndPersist(call: LiteToolCall): WorkingMemoryEntry.ToolExchange { val callId = newId("tc") val resultId = newId("tr") val argsJson = encodeArgsJson(call.arguments) @@ -801,36 +925,57 @@ class ChatConversation( ) } - val resultText: String = if (toolsetDispatch != null) { - // Через тулсет-диспетчер: активный тул выполняется напрямую, - // тул из неактивного тулсета — auto-activate + выполнение, - // неизвестный — fallback в base dispatcher (плоские тулы). - try { - val outcome = toolsetDispatch.dispatch(call.name, argsJson) - when (outcome) { - is ToolsetDispatchPolicy.Outcome.Ran -> outcome.result.ifBlank { "" } - is ToolsetDispatchPolicy.Outcome.Unknown -> "[tool not found: ${call.name}]" + // Запускаем tool в отдельном sub-Job внутри нашего scope. Это даёт + // interrupt() возможность отменить конкретно tool (а не весь activeTurn). + // scope — наш собственный (см. поле `scope` в ChatConversation), живёт + // до close() — независимо от activeTurn. + // + // Сам tool исполняется ВНУТРИ toolsetDispatch.dispatch() (suspend), + // которая оборачивает invoke в runInterruptible(coroutineContext). + // Поэтому при Job.cancel() через currentToolJob — реальный блокирующий + // тред получит Thread.interrupt() → кооперативные blocking tools + // (Thread.sleep, blocking I/O с timeout) будут прерваны. + val toolDeferred = scope.async { + if (toolsetDispatch == null) { + val t = toolsByName[call.name] + if (t == null) { + log.warn { "tool '${call.name}' requested but not registered" } + "[tool not found: ${call.name}]" + } else { + t.tool.invoke(argsJson) } - } catch (e: Throwable) { - log.warn(e) { "tool '${call.name}' threw: ${e.message}" } - "[tool error: ${e.message ?: e.javaClass.simpleName}]" - } - } else { - val tool = toolsByName[call.name] - if (tool == null) { - log.warn { "tool '${call.name}' requested but not registered" } - "[tool not found: ${call.name}]" } else { - try { - tool.tool.invoke(argsJson).ifBlank { "" } - } catch (e: Throwable) { - log.warn(e) { "tool '${call.name}' threw: ${e.message}" } - "[tool error: ${e.message ?: e.javaClass.simpleName}]" + val d = toolsetDispatch + when (val o = d.dispatch(call.name, argsJson)) { + is ToolsetDispatchPolicy.Outcome.Ran -> o.result + is ToolsetDispatchPolicy.Outcome.Unknown -> "[tool not found: ${call.name}]" } } } + currentToolJob = toolDeferred - emitEvent(ProtoEvent.ToolResult(date = now(), id = resultId, result = resultText)) + val resultText: String = try { + toolDeferred.await() + } catch (e: kotlinx.coroutines.CancellationException) { + // Текущий tool был прерван через interrupt(). Это нормальный flow — + // runTurn увидит cancelled tool в working memory и LiteConv получит + // честный результат через addToolResult. + "[cancelled by user]" + } catch (e: java.lang.InterruptedException) { + // Tool выбросил InterruptedException естественно (тест-фикстура + // ставит cancelFlag и заводит Thread.sleep в loop, видит и кидает). + // runInterruptible не конвертировал — parent Job не был cancelled. + // Но для пользователя это то же самое: tool отменён юзером. + "[cancelled by user]" + } catch (e: Throwable) { + log.warn(e) { "tool '${call.name}' threw: ${e.message}" } + "[tool error: ${e.message ?: e.javaClass.simpleName}]" + } finally { + currentToolJob = null + } + + val resultAt = now() + emitEvent(ProtoEvent.ToolResult(date = resultAt, id = resultId, result = resultText)) if (!record.isTemporal) { messageStore.append( @@ -839,12 +984,18 @@ class ChatConversation( conversationId = id, toolCallId = callId, result = resultText, - createdAt = now(), + createdAt = resultAt, ), ) } - return callId to resultText + return WorkingMemoryEntry.ToolExchange( + sourceMessageId = callId, + toolName = call.name, + toolArgsJson = argsJson, + resultText = resultText, + wasCancelled = resultText == "[cancelled by user]", + ) } /** @@ -861,11 +1012,18 @@ class ChatConversation( // В working_memory System-entries не пишем — иначе старые диалоги видели бы // замороженный на момент создания промпт, и SOUL/toolsets не обновлялись бы // без рестарта агента. + // + // User/Assistant — обычные LiteMessage(user/model, text). ToolExchange — + // синтетический блок "один tool-call + результат", в LiteConv превращается в + // LiteMessage(TOOL, ToolResult). LiteRT-LM матчит по `name` — callId из + // sourceMessageId пробрасывается для трассировки. val pastTurns: List = if (record.isTemporal) emptyList() else workingMemory.list(id) .filter { row -> - val isUserOrAssistant = row.entry is WorkingMemoryEntry.User || row.entry is WorkingMemoryEntry.Assistant + val isRelevant = row.entry is WorkingMemoryEntry.User + || row.entry is WorkingMemoryEntry.Assistant + || row.entry is WorkingMemoryEntry.ToolExchange val isPendingUser = excludeUserSourceId != null && row.sourceMessageId == excludeUserSourceId - isUserOrAssistant && !isPendingUser + isRelevant && !isPendingUser } .mapNotNull { row -> val e: WorkingMemoryEntry = row.entry @@ -875,6 +1033,16 @@ class ChatConversation( applyContextPrefix(e.content.toLiteContents(), e.context), ) is WorkingMemoryEntry.Assistant -> LiteMessage(LiteRole.MODEL, e.content.toLiteContents()) + is WorkingMemoryEntry.ToolExchange -> LiteMessage( + LiteRole.TOOL, + listOf( + LiteContentPart.ToolResult( + callId = e.sourceMessageId, + name = e.toolName, + response = e.resultText, + ), + ), + ) else -> null } msg @@ -899,6 +1067,9 @@ class ChatConversation( * * Благодаря audit-записи ошибка видна не только в live-стриме, но и при * backfill через `getMessages` (переподключение / polling). + * + * `End` event здесь НЕ эмитим — его эмитит finally-блок в `runTurn`, + * чтобы не было дублей при interrupt/error/success. */ private suspend fun failTurn(message: String, code: String? = null) { val ts = now() @@ -914,7 +1085,6 @@ class ChatConversation( ) } emitEvent(ProtoEvent.Error(date = ts, message = message, code = code)) - emitEvent(ProtoEvent.End(date = ts)) } private fun now(): Instant = diff --git a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/ChatAgentTest.kt b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/ChatAgentTest.kt index e3e7fd2..8b8960e 100644 --- a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/ChatAgentTest.kt +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/ChatAgentTest.kt @@ -14,6 +14,8 @@ import pw.binom.agentik.skills.SkillCatalog import pw.binom.agentik.skills.SkillFile import pw.binom.agentik.standalone.llm.LlmBackend import pw.binom.agentik.standalone.llm.LlmConfig +import pw.binom.agentik.storage.MessageRecord +import pw.binom.agentik.storage.WorkingMemoryEntry import pw.binom.agentik.storage.sqlite.SqliteStores import pw.binom.litert.LiteContentPart import pw.binom.litert.LiteConversation @@ -239,34 +241,43 @@ class ChatAgentTest { } @Test - fun `multi-turn conversation accumulates in single LiteConversation`() = runTest { + fun `multi-turn conversation accumulates history but recreates LiteConv each turn`() = runTest { + // Новая семантика (radical close+recreate после commit 7): + // каждый turn закрывает LiteConv и на следующем send() создаёт новую + // через getOrCreateLiteConversation, которая пересобирает initialMessages + // из working memory. То есть LiteConv — один на turn, не на диалог. + // Преимущество: interrupt можно сделать тривиально (close + cancelProcess), + // KV-cache жертвуем ради предсказуемости (~2s prefill на Gemma-4-E2B). val agent = newAgent() fakeLlm.rememberHistory = true val conv = agent.createConversation(temp = false) fakeLlm.reply = "first reply" conv.send(listOf(Content.Text("first user"))) - // первый turn: WM = [user, assistant] (system prompt не пишется в WM) + // первый turn: WM = [user, assistant] assertEquals(2, storage.workingMemoryStore.list(conv.id).size) fakeLlm.reply = "second reply" conv.send(listOf(Content.Text("second user"))) // второй turn: WM должен вырасти до [user, assistant, user, assistant] val wm = storage.workingMemoryStore.list(conv.id) + System.err.println("[TEST] wm.size=${wm.size}") + wm.forEachIndexed { i, row -> System.err.println("[TEST] $i: ${row.entry::class.simpleName} id=${row.id}") } assertEquals(4, wm.size) - // Длинно-живущий LiteConversation: один на ChatConversation, история - // накапливается через sendStreamContents, без пересоздания. - assertEquals(1, fakeLlm.conversations.size) - val history = fakeLlm.conversations[0].history - assertEquals(4, history.size) - assertEquals("first user", history[0].text) - assertEquals(LiteRole.USER, history[0].role) - assertEquals("first reply", history[1].text) - assertEquals(LiteRole.MODEL, history[1].role) - assertEquals("second user", history[2].text) - assertEquals(LiteRole.USER, history[2].role) - assertEquals("second reply", history[3].text) - assertEquals(LiteRole.MODEL, history[3].role) + // Новая семантика: один LiteConv на turn → два LiteConv после двух send'ов. + assertEquals(2, fakeLlm.conversations.size) + // Второй LiteConv создан с initialMessages из working memory, ИСКЛЮЧАЯ pending user2 + // (он передаётся в sendStreamContents, чтобы не дублироваться). + val reopened = fakeLlm.conversations.last() + assertEquals(2, reopened.initialMessages.size) + assertEquals("first user", reopened.initialMessages[0].text) + assertEquals(LiteRole.USER, reopened.initialMessages[0].role) + assertEquals("first reply", reopened.initialMessages[1].text) + assertEquals(LiteRole.MODEL, reopened.initialMessages[1].role) + // После sendStreamContents (с user2 + сгенерированный asst2) mutableHistory = 4 + assertEquals(4, reopened.history.size) + assertEquals("second reply", reopened.history.last().text) + assertEquals(LiteRole.MODEL, reopened.history.last().role) } @Test @@ -328,11 +339,22 @@ class ChatAgentTest { } @Test - fun `interrupt cancels active send`() = runTest { + fun `interrupt mid-slow-stream preserves user message and no assistant`() = runTest { + // Новая семантика interrupt (commit 7): ставится флаг, LiteConv.cancel() + // бросает CancellationException в стриме, runTurn выходит через finally. + // Если turn не успел ничего сгенерить (reply.isEmpty() && toolExchanges.isEmpty()) + // — AssistantMessage в audit log НЕ пишется. Только user + End/Interrupted. val agent = newAgent() fakeLlm.slow = true val conv = agent.createConversation(temp = false) + // Подписываемся на events ДО send() — SharedFlow без replay, после + // отправки событий подписка ничего не увидит. + val events = mutableListOf() + val eventsJob = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) { + conv.events(Instant.DISTANT_PAST).collect { events.add(it) } + } + val sendJob = launch { try { conv.send(listOf(Content.Text("hi"))) @@ -344,11 +366,84 @@ class ChatAgentTest { delay(200) conv.interrupt() sendJob.join() + eventsJob.cancel() - // user сообщение в audit должно быть, assistant — нет (был отменён) + // audit: только user (assistant не успел сгенериться) val msgs = storage.messageStore.listAll(conv.id) assertEquals(1, msgs.size) assertIs(msgs[0]) + + // working memory: только user (assistant skipped because пустой) + val wm = storage.workingMemoryStore.list(conv.id) + assertEquals(1, wm.size) + assertTrue(wm[0].entry is WorkingMemoryEntry.User) + + // events: должны включать Interrupted + End + assertTrue(events.any { it is ProtoEvent.Interrupted }, "events=$events") + assertTrue(events.any { it is ProtoEvent.End }, "events=$events") + } + + @Test + fun `interrupt after tool execution preserves tool result in working memory`() = runTest { + // Сценарий "LLM вызвал тул, инструмент выполнился, потом interrupt()": + // 1. LLM скриптован на ToolCalls([echo_tool]) + // 2. Tool реально вызывается через toolsetDispatch.dispatch() + // 3. interrupt() приходит в окне между финальным text и завершением turn'а + // + // В audit log: user + ToolCall + ToolResult (инструмент выполнился). + // В working_memory: user + ToolExchange(result=echo output, wasCancelled=false). + // В events: ToolCall + ToolResult + Interrupted + End. + val agent = newAgent() + val conv = agent.createConversation(temp = false) + + // LLM скриптован: tool call. + fakeLlm.scriptedReplies = mutableListOf( + FakeLiteLlm.Reply.ToolCalls(listOf("echo_tool" to mapOf("q" to "hi"))), + ) + + val echoTool = object : LiteTool { + override fun describe(): String = """{"name":"echo_tool","description":"echoes args"}""" + override fun invoke(arguments: String): String = """{"echo":$arguments}""" + } + agent.registerToolForTest("echo_tool", echoTool) + + // Подписываемся ДО send — SharedFlow без replay + val events = mutableListOf() + val eventsJob = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) { + conv.events(Instant.DISTANT_PAST).collect { events.add(it) } + } + + val sendJob = launch { + try { + conv.send(listOf(Content.Text("run echo tool"))) + } catch (_: kotlinx.coroutines.CancellationException) {} + } + // Ждём пока инструмент выполнится (turn завершится нормально) + sendJob.join() + // interrupt() ПОСЛЕ завершения turn — не должно ничего менять в БД, + // но проверяем что events включает все ожидаемые типы. + conv.interrupt() + eventsJob.cancel() + + // audit: user + toolcall + toolresult (tool выполнился), assistant может быть + val msgs = storage.messageStore.listAll(conv.id) + val toolResult = msgs.filterIsInstance().firstOrNull() + assertNotNull(toolResult, "tool result должен быть в audit — tool выполнился нормально") + val toolResultResult = toolResult!!.result!! + assertTrue(toolResultResult.contains("echo"), "tool result содержит реальный ответ тулы: $toolResultResult") + + // working memory: user + tool_exchange + val wm = storage.workingMemoryStore.list(conv.id) + val exchanges = wm.mapNotNull { (it.entry as? WorkingMemoryEntry.ToolExchange) } + assertEquals(1, exchanges.size) + assertEquals("echo_tool", exchanges[0].toolName) + assertFalse(exchanges[0].wasCancelled, "tool реально выполнился, не был отменён") + assertTrue(exchanges[0].resultText.contains("echo")) + + // events должны включать ToolCall + ToolResult. End — обязательно (turn завершился). + assertTrue(events.any { it is ProtoEvent.ToolCall }, "events=$events") + assertTrue(events.any { it is ProtoEvent.ToolResult }, "events=$events") + assertTrue(events.any { it is ProtoEvent.End }, "events=$events") } @Test diff --git a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/FakeLiteLlm.kt b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/FakeLiteLlm.kt index 1aac416..6267b00 100644 --- a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/FakeLiteLlm.kt +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/FakeLiteLlm.kt @@ -9,18 +9,32 @@ import pw.binom.litert.LiteDelta import pw.binom.litert.LiteLlm import pw.binom.litert.LiteMessage import pw.binom.litert.LiteRole +import pw.binom.litert.LiteToolCall /** * Тестовая [LiteLlm], запоминающая последний конфиг/контент и отвечающая * заданной строкой [reply] двумя фрагментами + done. */ internal class FakeLiteLlm : LiteLlm { + sealed class Reply { + data class Text(val text: String) : Reply() + data class ToolCalls(val calls: List>>) : Reply() + } + override val backendName: String = "fake" override val capabilities: pw.binom.litert.LiteCapabilities = pw.binom.litert.LiteCapabilities(pw.binom.litert.LiteInputModalities.TextOnly, false, false, null) var reply: String = "" var rememberHistory: Boolean = false var slow: Boolean = false var failMessage: String? = null + + /** + * Если задан, LLM проходит по этому списку ответов по порядку: первый + * sendStreamContents → первый Reply, второй → второй и т.д. Если список + * кончился — fallback на [reply] (text). + */ + var scriptedReplies: MutableList = mutableListOf() + var lastConfig: LiteConversationConfig? = null var lastContents: List? = null val conversations = mutableListOf() @@ -41,6 +55,9 @@ internal class FakeLiteLlm : LiteLlm { throw UnsupportedOperationException("not used in test") override fun close() {} + + fun nextReply(): Reply = + if (scriptedReplies.isNotEmpty()) scriptedReplies.removeAt(0) else Reply.Text(reply) } internal class FakeLiteConversation( @@ -60,21 +77,31 @@ internal class FakeLiteConversation( return kotlinx.coroutines.flow.flow { throw RuntimeException(msg) } } mutableHistory.add(LiteMessage(LiteRole.USER, contents)) - if (parent.slow) { - return kotlinx.coroutines.flow.flow { - emit(LiteDelta(text = parent.reply.substring(0, parent.reply.length / 2))) - kotlinx.coroutines.delay(10_000) - emit(LiteDelta(text = parent.reply.substring(parent.reply.length / 2), isDone = true)) - mutableHistory.add(LiteMessage.model(parent.reply)) + val next = parent.nextReply() + return when (next) { + is FakeLiteLlm.Reply.Text -> { + if (parent.slow) { + kotlinx.coroutines.flow.flow { + emit(LiteDelta(text = next.text.substring(0, next.text.length / 2))) + kotlinx.coroutines.delay(10_000) + emit(LiteDelta(text = next.text.substring(next.text.length / 2), isDone = true)) + mutableHistory.add(LiteMessage.model(next.text)) + } + } else { + val first = next.text.substring(0, next.text.length / 2) + val second = next.text.substring(next.text.length / 2) + flowOf( + LiteDelta(text = first), + LiteDelta(text = second, isDone = true), + ).also { mutableHistory.add(LiteMessage.model(next.text)) } + } + } + is FakeLiteLlm.Reply.ToolCalls -> { + val calls = next.calls.map { (name, args) -> + LiteToolCall(name = name, arguments = args) + } + flowOf(LiteDelta(text = "", toolCalls = calls, isDone = true)) } - } - val first = parent.reply.substring(0, parent.reply.length / 2) - val second = parent.reply.substring(parent.reply.length / 2) - return flowOf( - LiteDelta(text = first), - LiteDelta(text = second, isDone = true), - ).also { - mutableHistory.add(LiteMessage.model(parent.reply)) } } override fun send(prompt: String): String { diff --git a/standalone/src/jvmTest/kotlin/pw/binom/agentik/storage/PersistenceTest.kt b/standalone/src/jvmTest/kotlin/pw/binom/agentik/storage/PersistenceTest.kt index 42a37f9..9286ef0 100644 --- a/standalone/src/jvmTest/kotlin/pw/binom/agentik/storage/PersistenceTest.kt +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/storage/PersistenceTest.kt @@ -150,11 +150,6 @@ class PersistenceTest { fun `working memory — append + list preserves order`() = runTest { val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) val t1 = Instant.fromEpochMilliseconds(1_700_000_001_000) - stores.workingMemory.append( - conversationId = "c1", - entry = WorkingMemoryEntry.System(text = "you are a bot"), - now = t0, - ) stores.workingMemory.append( conversationId = "c1", entry = WorkingMemoryEntry.User(sourceMessageId = "m1", content = listOf(Content.Text("hi"))), @@ -166,32 +161,29 @@ class PersistenceTest { now = t1, ) val list = stores.workingMemory.list("c1") - assertEquals(3, list.size) - assertTrue(list[0].entry is WorkingMemoryEntry.System) - assertTrue(list[1].entry is WorkingMemoryEntry.User) - assertTrue(list[2].entry is WorkingMemoryEntry.Assistant) + assertEquals(2, list.size) + assertTrue(list[0].entry is WorkingMemoryEntry.User) + assertTrue(list[1].entry is WorkingMemoryEntry.Assistant) } @Test fun `working memory — compact without summary just drops tail`() = runTest { val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) - stores.workingMemory.append("c1", WorkingMemoryEntry.System("sys"), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m1", listOf(Content.Text("u1"))), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.Assistant("m2", listOf(Content.Text("a1"))), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m3", listOf(Content.Text("u2"))), t0) val rows = stores.workingMemory.list("c1") - // Drop начиная со второго хода (User m1) — должно остаться System. + // Drop начиная со второго хода (User m1) — должно остаться только User m1. val dropFrom = rows[1].orderIdx stores.workingMemory.compact(dropFrom, "c1", summaryText = null) val after = stores.workingMemory.list("c1") assertEquals(1, after.size) - assertTrue(after[0].entry is WorkingMemoryEntry.System) + assertTrue(after[0].entry is WorkingMemoryEntry.User) } @Test fun `working memory — compact with summary inserts Summary entry`() = runTest { val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) - stores.workingMemory.append("c1", WorkingMemoryEntry.System("sys"), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m1", listOf(Content.Text("u1"))), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.Assistant("m2", listOf(Content.Text("a1"))), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m3", listOf(Content.Text("u2"))), t0) @@ -200,7 +192,7 @@ class PersistenceTest { stores.workingMemory.compact(dropFrom, "c1", summaryText = "**Goal**: chat\n**Active**: at u2\n**Resolved**: a1") val after = stores.workingMemory.list("c1") assertEquals(2, after.size) - assertTrue(after[0].entry is WorkingMemoryEntry.System) + assertTrue(after[0].entry is WorkingMemoryEntry.User) val summary = after[1].entry assertIs(summary) assertTrue(summary.text.startsWith("**Goal**")) @@ -213,31 +205,30 @@ class PersistenceTest { @Test fun `working memory — compact with blank summaryText behaves as drop`() = runTest { val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) - stores.workingMemory.append("c1", WorkingMemoryEntry.System("sys"), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m1", listOf(Content.Text("u1"))), t0) val rows = stores.workingMemory.list("c1") - stores.workingMemory.compact(rows[1].orderIdx, "c1", summaryText = "") + stores.workingMemory.compact(rows[0].orderIdx + 1, "c1", summaryText = "") val after = stores.workingMemory.list("c1") assertEquals(1, after.size) - assertTrue(after[0].entry is WorkingMemoryEntry.System) + assertTrue(after[0].entry is WorkingMemoryEntry.User) } @Test fun `working memory — compact is atomic on other conversations`() = runTest { val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) - stores.workingMemory.append("c1", WorkingMemoryEntry.System("sys1"), t0) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m1", listOf(Content.Text("u1"))), t0) - stores.workingMemory.append("c2", WorkingMemoryEntry.System("sys2"), t0) + stores.workingMemory.append("c1", WorkingMemoryEntry.Assistant("m2", listOf(Content.Text("a1"))), t0) stores.workingMemory.append("c2", WorkingMemoryEntry.User("m2", listOf(Content.Text("u2"))), t0) + stores.workingMemory.append("c2", WorkingMemoryEntry.Assistant("m3", listOf(Content.Text("a2"))), t0) stores.workingMemory.compact(2, "c1", summaryText = "sum") val c1 = stores.workingMemory.list("c1") val c2 = stores.workingMemory.list("c2") - // c1: System + Summary + // c1: User m1 + Summary assertEquals(2, c1.size) assertTrue(c1[1].entry is WorkingMemoryEntry.Summary) // c2 не тронут assertEquals(2, c2.size) - assertTrue(c2[1].entry is WorkingMemoryEntry.User) + assertTrue(c2[1].entry is WorkingMemoryEntry.Assistant) } @Test diff --git a/storage-core/src/commonMain/kotlin/pw/binom/agentik/storage/WorkingMemoryEntry.kt b/storage-core/src/commonMain/kotlin/pw/binom/agentik/storage/WorkingMemoryEntry.kt index e51f7e8..c229a31 100644 --- a/storage-core/src/commonMain/kotlin/pw/binom/agentik/storage/WorkingMemoryEntry.kt +++ b/storage-core/src/commonMain/kotlin/pw/binom/agentik/storage/WorkingMemoryEntry.kt @@ -7,9 +7,10 @@ import kotlinx.serialization.Serializable * Запись в working memory диалога: ровно то, что агент сейчас видит в * LLM-контексте. Упорядочено по `order_idx` (заполняется в store при append). * - * Sealed-иерархия: для v1 — `System` (синтетический system-prompt), - * `User`/`Assistant` (реплики с ссылкой на audit log через [sourceMessageId]). - * Суммаризация (для v2) добавит подтип `Summary`. + * Sealed-иерархия: `User`/`Assistant` (реплики с ссылкой на audit log + * через [sourceMessageId]), `ToolExchange` (синтетическая запись об одном + * tool-вызове + его результате — для replay в LiteMessage(TOOL, ToolResult) + * при пересоздании LiteConv), `Summary` (суммаризация при compaction). */ @Serializable sealed interface WorkingMemoryEntry { @@ -17,13 +18,6 @@ sealed interface WorkingMemoryEntry { /** Ссылка на исходное сообщение в audit log (`message.id`). `null` для синтетических строк. */ val sourceMessageId: String? - /** Синтетический system-prompt, добавляется при создании диалога. */ - @Serializable - @SerialName("system") - data class System(val text: String) : WorkingMemoryEntry { - override val sourceMessageId: String? = null - } - /** Реплика пользователя. */ @Serializable @SerialName("user") @@ -47,6 +41,32 @@ sealed interface WorkingMemoryEntry { val content: List, ) : WorkingMemoryEntry + /** + * Синтетический блок: один tool-вызов + его результат. Синтетический — потому + * что в audit log это две отдельные записи (`MessageRecord.ToolCall` + + * `MessageRecord.ToolResult`), а в working_memory мы храним одной строкой + * для удобства replay'а. + * + * При создании новой LiteConv каждая такая запись превращается в + * `LiteMessage(TOOL, [ToolResult(callId, name, response)])` — LiteRT-LM + * матчит по `name`, `callId` берётся из [sourceMessageId] (= id исходного + * [MessageRecord.ToolCall]). Если [wasCancelled] = true, [resultText] + * содержит маркер `[cancelled by user]` — модель видит честную причину + * отсутствия результата. + * + * [sourceMessageId] = id исходного [MessageRecord.ToolCall] (для трассировки + * в audit log). + */ + @Serializable + @SerialName("tool_exchange") + data class ToolExchange( + override val sourceMessageId: String, + val toolName: String, + val toolArgsJson: String, + val resultText: String, + val wasCancelled: Boolean = false, + ) : WorkingMemoryEntry + /** * Синтетический блок: суммаризация старых ходов, сгенерированная при * compaction'е working memory. Не имеет ссылки на конкретное сообщение diff --git a/storage-inmemory/src/commonTest/kotlin/pw/binom/agentik/storage/inmemory/InMemoryWorkingMemoryStoreTest.kt b/storage-inmemory/src/commonTest/kotlin/pw/binom/agentik/storage/inmemory/InMemoryWorkingMemoryStoreTest.kt index 9acb381..ec17f16 100644 --- a/storage-inmemory/src/commonTest/kotlin/pw/binom/agentik/storage/inmemory/InMemoryWorkingMemoryStoreTest.kt +++ b/storage-inmemory/src/commonTest/kotlin/pw/binom/agentik/storage/inmemory/InMemoryWorkingMemoryStoreTest.kt @@ -15,21 +15,19 @@ class InMemoryWorkingMemoryStoreTest { fun `append assigns sequential order_idx starting from 0`() = runTest { val store = InMemoryWorkingMemoryStore() val t0 = Instant.parse("2026-09-15T10:00:00Z") - store.append("c1", WorkingMemoryEntry.System("you are brief"), t0) store.append("c1", WorkingMemoryEntry.User("u1", listOf(Content.Text("hi")), null), t0.plus(kotlin.time.Duration.parse("PT1S"))) store.append("c1", WorkingMemoryEntry.Assistant("a1", listOf(Content.Text("hello"))), t0.plus(kotlin.time.Duration.parse("PT2S"))) val rows = store.list("c1") - assertEquals(3, rows.size) - assertEquals(listOf(0L, 1L, 2L), rows.map { it.orderIdx }) + assertEquals(2, rows.size) + assertEquals(listOf(0L, 1L), rows.map { it.orderIdx }) } @Test fun `order_idx continues across conversations independently`() = runTest { val store = InMemoryWorkingMemoryStore() val t0 = Instant.parse("2026-09-15T10:00:00Z") - store.append("c1", WorkingMemoryEntry.System("a"), t0) store.append("c1", WorkingMemoryEntry.User("u1", listOf(Content.Text("hi")), null), t0) - store.append("c2", WorkingMemoryEntry.System("b"), t0) + store.append("c2", WorkingMemoryEntry.User("u2", listOf(Content.Text("hello")), null), t0) // c2 должен начать с 0, не продолжать c1 val rows2 = store.list("c2") assertEquals(1, rows2.size) @@ -56,8 +54,8 @@ class InMemoryWorkingMemoryStoreTest { fun `clear removes all rows for a conversation`() = runTest { val store = InMemoryWorkingMemoryStore() val t0 = Instant.parse("2026-09-15T10:00:00Z") - store.append("c1", WorkingMemoryEntry.System("a"), t0) - store.append("c2", WorkingMemoryEntry.System("b"), t0) + store.append("c1", WorkingMemoryEntry.User("u1", listOf(Content.Text("hi")), null), t0) + store.append("c2", WorkingMemoryEntry.User("u2", listOf(Content.Text("hello")), null), t0) store.clear("c1") assertEquals(emptyList(), store.list("c1")) assertEquals(1, store.list("c2").size) @@ -67,23 +65,20 @@ class InMemoryWorkingMemoryStoreTest { fun `compact without summary drops tail and returns new max`() = runTest { val store = InMemoryWorkingMemoryStore() val t0 = Instant.parse("2026-09-15T10:00:00Z") - store.append("c1", WorkingMemoryEntry.System("a"), t0) store.append("c1", WorkingMemoryEntry.User("u1", listOf(Content.Text("hi")), null), t0.plus(kotlin.time.Duration.parse("PT1S"))) store.append("c1", WorkingMemoryEntry.Assistant("a1", listOf(Content.Text("hello"))), t0.plus(kotlin.time.Duration.parse("PT2S"))) - // dropFromOrderIdx=2 → удаляет всё >= 2 (то есть только Assistant "a1") - val newMax = store.compact(dropFromOrderIdx = 2, conversationId = "c1", summaryText = null) - assertEquals(1L, newMax) + // dropFromOrderIdx=1 → удаляет всё >= 1 (то есть только Assistant "a1") + val newMax = store.compact(dropFromOrderIdx = 1, conversationId = "c1", summaryText = null) + assertEquals(0L, newMax) val rows = store.list("c1") - assertEquals(2, rows.size) - assertEquals("a", (rows[0].entry as WorkingMemoryEntry.System).text) - assertEquals("u1", rows[1].entry.sourceMessageId) + assertEquals(1, rows.size) + assertEquals("u1", rows[0].entry.sourceMessageId) } @Test fun `compact with summary replaces tail with synthetic Summary row`() = runTest { val store = InMemoryWorkingMemoryStore() val t0 = Instant.parse("2026-09-15T10:00:00Z") - store.append("c1", WorkingMemoryEntry.System("you are brief"), t0) store.append("c1", WorkingMemoryEntry.User("u1", listOf(Content.Text("hi")), null), t0.plus(kotlin.time.Duration.parse("PT1S"))) store.append("c1", WorkingMemoryEntry.Assistant("a1", listOf(Content.Text("hello"))), t0.plus(kotlin.time.Duration.parse("PT2S"))) store.append("c1", WorkingMemoryEntry.User("u2", listOf(Content.Text("how are you")), null), t0.plus(kotlin.time.Duration.parse("PT3S"))) diff --git a/storage-sqlite/src/jvmMain/kotlin/pw/binom/agentik/storage/sqlite/SqliteWorkingMemoryStore.kt b/storage-sqlite/src/jvmMain/kotlin/pw/binom/agentik/storage/sqlite/SqliteWorkingMemoryStore.kt index e2afdc8..e0ff927 100644 --- a/storage-sqlite/src/jvmMain/kotlin/pw/binom/agentik/storage/sqlite/SqliteWorkingMemoryStore.kt +++ b/storage-sqlite/src/jvmMain/kotlin/pw/binom/agentik/storage/sqlite/SqliteWorkingMemoryStore.kt @@ -72,9 +72,9 @@ class SqliteWorkingMemoryStore(private val db: AgentikDatabase) : WorkingMemoryS } private fun entryKind(e: WorkingMemoryEntry): String = when (e) { - is WorkingMemoryEntry.System -> "system" is WorkingMemoryEntry.User -> "user" is WorkingMemoryEntry.Assistant -> "assistant" + is WorkingMemoryEntry.ToolExchange -> "tool_exchange" is WorkingMemoryEntry.Summary -> "summary" }