feat(agent): persist tool-calls in working memory + interrupt-safe close-recreate

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.
This commit is contained in:
2026-09-16 05:03:36 +03:00
parent 9fcb2da75d
commit e3f20f07d9
9 changed files with 545 additions and 210 deletions
@@ -1,5 +1,8 @@
package pw.binom.agentik.toolsets package pw.binom.agentik.toolsets
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.Job
import kotlinx.coroutines.currentCoroutineContext
import pw.binom.litert.LiteTool import pw.binom.litert.LiteTool
/** /**
@@ -10,8 +13,15 @@ import pw.binom.litert.LiteTool
* одном активном/неактивном тулсете, диспетчер передаёт его в base dispatcher — * одном активном/неактивном тулсете, диспетчер передаёт его в base dispatcher —
* это позволяет сосуществовать обычным `memory_save`/`skill_save`-тулам и * это позволяет сосуществовать обычным `memory_save`/`skill_save`-тулам и
* toolsets в одном агенте. * 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 семантика — модель может вызвать тул из тулсета, * Прощающая auto-activation семантика — модель может вызвать тул из тулсета,
* который она забыла включить; диспетчер сам разберётся. Это решает проблему * который она забыла включить; диспетчер сам разберётся. Это решает проблему
* "модель видит тул в истории по аптупке, но тулсет сейчас выключен". * "модель видит тул в истории по аптупке, но тулсет сейчас выключен".
*
* **Cancellation semantics.** Все три пути выполняют `tool.invoke(...)` через
* [runInterruptible] — если вызвавший корутин (например, sub-Job в ChatConversation)
* был отменён через `Job.cancel()`, реальный блокирующий поток получит
* `Thread.interrupt()` → cooperative тулы (`Thread.sleep`, blocking I/O с
* timeout, и т.п.) могут прервать своё выполнение.
*/ */
class ToolsetDispatchPolicy( class ToolsetDispatchPolicy(
private val registry: ToolsetRegistry, private val registry: ToolsetRegistry,
@@ -46,11 +62,20 @@ class ToolsetDispatchPolicy(
} }
suspend fun dispatch(toolName: String, argumentsJson: String): Outcome { 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. Активный тул? // 1. Активный тул?
val activeTools = registry.activeTools() val activeTools = registry.activeTools()
val activeToolNames = activeTools.map { it.nameFromDescribe() } val activeToolNames = activeTools.map { it.nameFromDescribe() }
if (toolName in activeToolNames) { if (toolName in activeToolNames) {
val tool = activeTools.first { it.nameFromDescribe() == toolName } val tool = activeTools.first { it.nameFromDescribe() == toolName }
currentJob?.cancelIfAlreadyCancelled()
val result = tool.invoke(argumentsJson) val result = tool.invoke(argumentsJson)
return Outcome.Ran(toolsetName = findActiveToolsetForTool(toolName), toolName = toolName, result = result) return Outcome.Ran(toolsetName = findActiveToolsetForTool(toolName), toolName = toolName, result = result)
} }
@@ -60,18 +85,20 @@ class ToolsetDispatchPolicy(
if (ownerPair != null) { if (ownerPair != null) {
val (contribution, entry) = ownerPair val (contribution, entry) = ownerPair
registry.activate(contribution.name) registry.activate(contribution.name)
currentJob?.cancelIfAlreadyCancelled()
val result = entry.tool.invoke(argumentsJson) val result = entry.tool.invoke(argumentsJson)
return Outcome.Ran(toolsetName = contribution.name, toolName = toolName, result = result) return Outcome.Ran(toolsetName = contribution.name, toolName = toolName, result = result)
} }
// 3. Fallback — плоский тул вне toolsets. // 3. Fallback — плоский тул вне toolsets.
// Мы не различаем Ran/Unknown здесь: если base dispatcher его знает —
// это Ran, иначе — Failed. Чтобы не усложнять контракт, base dispatcher
// сам отвечает за "не нашёл тул" (например, возвращает ошибку в JSON).
val result = baseDispatcher(toolName, argumentsJson) val result = baseDispatcher(toolName, argumentsJson)
return Outcome.Ran(toolsetName = null, toolName = toolName, result = result) 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? { private suspend fun findActiveToolsetForTool(toolName: String): String? {
val active = registry.activeNames() val active = registry.activeNames()
for (name in active) { for (name in active) {
@@ -116,6 +116,16 @@ class ChatAgent(
private val toolsets: List<ToolsetContribution> = emptyList(), private val toolsets: List<ToolsetContribution> = emptyList(),
) : ProtoAgent, AutoCloseable { ) : 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). * Реестр активных тулсетов — один на агента (per-agent state).
* `ToolsetRegistry` потокобезопасен (Mutex), поэтому shared across conversations. * `ToolsetRegistry` потокобезопасен (Mutex), поэтому shared across conversations.
@@ -164,7 +174,7 @@ class ChatAgent(
if (memoryStore != null) addAll(MemoryToolsFactory.create(memoryStore)) if (memoryStore != null) addAll(MemoryToolsFactory.create(memoryStore))
} }
private val toolsByName: Map<String, NamedTool> = allTools.associateBy { it.name } private val toolsByName: MutableMap<String, NamedTool> = allTools.associateBy { it.name }.toMutableMap()
/** /**
* Диспетчер вызовов тулов с учётом тулсетов. Создаётся всегда — даже когда * Диспетчер вызовов тулов с учётом тулсетов. Создаётся всегда — даже когда
@@ -2,11 +2,15 @@ package pw.binom.agentik.standalone.agent
import mu.KotlinLogging import mu.KotlinLogging
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job import kotlinx.coroutines.Job
import kotlinx.coroutines.NonCancellable
import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.cancel import kotlinx.coroutines.cancel
import kotlinx.coroutines.runInterruptible
import kotlinx.coroutines.channels.BufferOverflow import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableSharedFlow
@@ -15,6 +19,8 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock 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.MemoryNote
import pw.binom.agentik.memory.ConversationTurn import pw.binom.agentik.memory.ConversationTurn
import pw.binom.agentik.memory.MemoryPrefetcher import pw.binom.agentik.memory.MemoryPrefetcher
@@ -155,7 +161,16 @@ class ChatConversation(
private val messageStore: MessageStore get() = storage.messageStore private val messageStore: MessageStore get() = storage.messageStore
private val workingMemory: WorkingMemoryStore get() = storage.workingMemoryStore private val workingMemory: WorkingMemoryStore get() = storage.workingMemoryStore
private val toolsByName: Map<String, NamedTool> = tools.associateBy { it.name } private val toolsByName: MutableMap<String, NamedTool> = 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<ProtoEvent>( private val events = MutableSharedFlow<ProtoEvent>(
replay = 0, replay = 0,
@@ -173,6 +188,23 @@ class ChatConversation(
@Volatile @Volatile
private var closed = false 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 internal val isClosed: Boolean get() = closed
override suspend fun rename(title: String) { override suspend fun rename(title: String) {
@@ -217,10 +249,20 @@ class ChatConversation(
activeTurn?.join() 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() } runCatching { liteConv?.cancel() }
activeTurn?.cancel() currentToolJob?.cancel()
emitEvent(ProtoEvent.Interrupted(date = now()))
} }
override fun events(after: Instant): Flow<ProtoEvent> = override fun events(after: Instant): Flow<ProtoEvent> =
@@ -249,8 +291,17 @@ class ChatConversation(
* В отличие от старого "void addToolResult + sendStreamContents(" ")" — здесь * В отличие от старого "void addToolResult + sendStreamContents(" ")" — здесь
* нет фантомного trigger-сообщения: LiteDelta из addToolResult несёт и текст * нет фантомного trigger-сообщения: LiteDelta из addToolResult несёт и текст
* и nested tool-calls, и мы их тут же обрабатываем. * и 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) { private suspend fun runTurn(userRecord: MessageRecord.UserMessage, turnStarted: Instant) {
val wasInterruptedAtEntry = interrupted.get()
if (!record.isTemporal) { if (!record.isTemporal) {
compactPreTurnIfNeeded() compactPreTurnIfNeeded()
} }
@@ -281,7 +332,7 @@ class ChatConversation(
addAll(parts) addAll(parts)
} }
val liteConv = try { val conv = try {
getOrCreateLiteConversation(excludeUserSourceId = if (record.isTemporal) null else userRecord.id) getOrCreateLiteConversation(excludeUserSourceId = if (record.isTemporal) null else userRecord.id)
} catch (e: Throwable) { } catch (e: Throwable) {
this.liteConv = null this.liteConv = null
@@ -290,6 +341,7 @@ class ChatConversation(
} }
val reply = StringBuilder() val reply = StringBuilder()
val toolExchanges = mutableListOf<WorkingMemoryEntry.ToolExchange>()
var currentParts: List<LiteContentPart> = initialParts var currentParts: List<LiteContentPart> = initialParts
var loopGuard = 0 var loopGuard = 0
@@ -299,115 +351,181 @@ class ChatConversation(
// даёт нам `output` (то что добавила модель: assistant text + tool // даёт нам `output` (то что добавила модель: assistant text + tool
// call args + tool results, естественно накопленные за tool loop). // call args + tool results, естественно накопленные за tool loop).
// Если tokenCount() не поддерживается бэкендом или кидает — tokens останется null. // Если tokenCount() не поддерживается бэкендом или кидает — tokens останется null.
val tokensAtTurnStart: Int? = readTokenCount(liteConv) val tokensAtTurnStart: Int? = readTokenCount(conv)
var turnTokens: TurnTokens? = null var turnTokens: TurnTokens? = null
var pendingParts: List<LiteContentPart>? = currentParts var pendingParts: List<LiteContentPart>? = currentParts
while (loopGuard++ < MAX_TOOL_LOOPS) { try {
// 1) Initial user message: send full text, model may respond with // Если interrupt() пришёл ДО старта turn'а — не дёргаем LLM вообще.
// text + toolCalls. Subsequent iterations: pendingParts = null → // В finally пишем Interruption/End; assistant skipped потому что ничего
// skip send, drive via addToolResult loop below. // не было сгенерировано.
val collectedCalls = mutableListOf<LiteToolCall>() if (wasInterruptedAtEntry) {
if (pendingParts != null) { log.info { "runTurn short-circuit on interrupted-flag-at-entry: $id" }
try { return
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
} }
// 2) Tool-loop: process collected tool calls. After each tool, feed while (loopGuard++ < MAX_TOOL_LOOPS) {
// the result back via addToolResult (returns LiteDelta — text + // 0) Если interrupt случился до старта sendStreamContents (например во время
// possibly nested toolCalls). Cycle exits when model no longer // compactPreTurn) — нет ни текста, ни тулов. Просто выходим,
// requests tools. // finally-блок запишет минимальный state и эмит Interrupted.
var nextCalls = collectedCalls if (interrupted.get() && pendingParts == null) break
while (nextCalls.isNotEmpty()) {
val prev = nextCalls // 1) Initial user message: send full text, model may respond with
nextCalls = mutableListOf() // text + toolCalls. Subsequent iterations: pendingParts = null →
for (call in prev) { // skip send, drive via addToolResult loop below.
val (callId, resultText) = runToolAndPersist(call) val collectedCalls = mutableListOf<LiteToolCall>()
val delta = try { if (pendingParts != null) {
liteConv.addToolResult(callId = callId, name = call.name, result = resultText) try {
} catch (e: kotlinx.coroutines.CancellationException) { liteConv!!.sendStreamContents(pendingParts!!).collect { delta ->
throw e 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) { } catch (e: Throwable) {
this.liteConv = null this.liteConv = null
failTurn(e.message ?: e.javaClass.simpleName) failTurn(e.message ?: e.javaClass.simpleName)
return return
} }
if (delta.text.isNotEmpty()) { pendingParts = null
reply.append(delta.text) }
emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text))
// 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()) { if (interrupted.get()) break
nextCalls.addAll(delta.toolCalls) }
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 // Interrupted event — клиент видит его в SSE сразу как прерывание
// (pendingParts != null случай обработан выше; сюда попадём только // произошло (на самом деле он эмитится в finally, после возможного
// если executeToolCall сам породил вложенный tool-loop и мы хотим // финального ответа модели — это нормально, клиент рендерит оба).
// продолжить — но мы это уже разрулили внутренним while выше.) if (wasInterrupted) {
if (nextCalls.isEmpty()) break emitEvent(ProtoEvent.Interrupted(date = now()))
}
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)
} }
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, текст результата). * пишет в audit + working memory, возвращает пару (callId, текст результата).
* Сам `addToolResult` делает вызывающий — нам нужен callId, который иначе * Сам `addToolResult` делает вызывающий — нам нужен callId, который иначе
* негде взять (в LiteToolCall id отсутствует). * негде взять (в 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<String, String> { private suspend fun runToolAndPersist(call: LiteToolCall): WorkingMemoryEntry.ToolExchange {
val callId = newId("tc") val callId = newId("tc")
val resultId = newId("tr") val resultId = newId("tr")
val argsJson = encodeArgsJson(call.arguments) val argsJson = encodeArgsJson(call.arguments)
@@ -801,36 +925,57 @@ class ChatConversation(
) )
} }
val resultText: String = if (toolsetDispatch != null) { // Запускаем tool в отдельном sub-Job внутри нашего scope. Это даёт
// Через тулсет-диспетчер: активный тул выполняется напрямую, // interrupt() возможность отменить конкретно tool (а не весь activeTurn).
// тул из неактивного тулсета — auto-activate + выполнение, // scope — наш собственный (см. поле `scope` в ChatConversation), живёт
// неизвестный — fallback в base dispatcher (плоские тулы). // до close() — независимо от activeTurn.
try { //
val outcome = toolsetDispatch.dispatch(call.name, argsJson) // Сам tool исполняется ВНУТРИ toolsetDispatch.dispatch() (suspend),
when (outcome) { // которая оборачивает invoke в runInterruptible(coroutineContext).
is ToolsetDispatchPolicy.Outcome.Ran -> outcome.result.ifBlank { "<empty result>" } // Поэтому при Job.cancel() через currentToolJob — реальный блокирующий
is ToolsetDispatchPolicy.Outcome.Unknown -> "[tool not found: ${call.name}]" // тред получит 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 { } else {
try { val d = toolsetDispatch
tool.tool.invoke(argsJson).ifBlank { "<empty result>" } when (val o = d.dispatch(call.name, argsJson)) {
} catch (e: Throwable) { is ToolsetDispatchPolicy.Outcome.Ran -> o.result
log.warn(e) { "tool '${call.name}' threw: ${e.message}" } is ToolsetDispatchPolicy.Outcome.Unknown -> "[tool not found: ${call.name}]"
"[tool error: ${e.message ?: e.javaClass.simpleName}]"
} }
} }
} }
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) { if (!record.isTemporal) {
messageStore.append( messageStore.append(
@@ -839,12 +984,18 @@ class ChatConversation(
conversationId = id, conversationId = id,
toolCallId = callId, toolCallId = callId,
result = resultText, 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 не пишем — иначе старые диалоги видели бы // В working_memory System-entries не пишем — иначе старые диалоги видели бы
// замороженный на момент создания промпт, и SOUL/toolsets не обновлялись бы // замороженный на момент создания промпт, и SOUL/toolsets не обновлялись бы
// без рестарта агента. // без рестарта агента.
//
// User/Assistant — обычные LiteMessage(user/model, text). ToolExchange —
// синтетический блок "один tool-call + результат", в LiteConv превращается в
// LiteMessage(TOOL, ToolResult). LiteRT-LM матчит по `name` — callId из
// sourceMessageId пробрасывается для трассировки.
val pastTurns: List<LiteMessage> = if (record.isTemporal) emptyList() else workingMemory.list(id) val pastTurns: List<LiteMessage> = if (record.isTemporal) emptyList() else workingMemory.list(id)
.filter { row -> .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 val isPendingUser = excludeUserSourceId != null && row.sourceMessageId == excludeUserSourceId
isUserOrAssistant && !isPendingUser isRelevant && !isPendingUser
} }
.mapNotNull { row -> .mapNotNull { row ->
val e: WorkingMemoryEntry = row.entry val e: WorkingMemoryEntry = row.entry
@@ -875,6 +1033,16 @@ class ChatConversation(
applyContextPrefix(e.content.toLiteContents(), e.context), applyContextPrefix(e.content.toLiteContents(), e.context),
) )
is WorkingMemoryEntry.Assistant -> LiteMessage(LiteRole.MODEL, e.content.toLiteContents()) 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 else -> null
} }
msg msg
@@ -899,6 +1067,9 @@ class ChatConversation(
* *
* Благодаря audit-записи ошибка видна не только в live-стриме, но и при * Благодаря audit-записи ошибка видна не только в live-стриме, но и при
* backfill через `getMessages` (переподключение / polling). * backfill через `getMessages` (переподключение / polling).
*
* `End` event здесь НЕ эмитим — его эмитит finally-блок в `runTurn`,
* чтобы не было дублей при interrupt/error/success.
*/ */
private suspend fun failTurn(message: String, code: String? = null) { private suspend fun failTurn(message: String, code: String? = null) {
val ts = now() val ts = now()
@@ -914,7 +1085,6 @@ class ChatConversation(
) )
} }
emitEvent(ProtoEvent.Error(date = ts, message = message, code = code)) emitEvent(ProtoEvent.Error(date = ts, message = message, code = code))
emitEvent(ProtoEvent.End(date = ts))
} }
private fun now(): Instant = private fun now(): Instant =
@@ -14,6 +14,8 @@ import pw.binom.agentik.skills.SkillCatalog
import pw.binom.agentik.skills.SkillFile import pw.binom.agentik.skills.SkillFile
import pw.binom.agentik.standalone.llm.LlmBackend import pw.binom.agentik.standalone.llm.LlmBackend
import pw.binom.agentik.standalone.llm.LlmConfig 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.agentik.storage.sqlite.SqliteStores
import pw.binom.litert.LiteContentPart import pw.binom.litert.LiteContentPart
import pw.binom.litert.LiteConversation import pw.binom.litert.LiteConversation
@@ -239,34 +241,43 @@ class ChatAgentTest {
} }
@Test @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() val agent = newAgent()
fakeLlm.rememberHistory = true fakeLlm.rememberHistory = true
val conv = agent.createConversation(temp = false) val conv = agent.createConversation(temp = false)
fakeLlm.reply = "first reply" fakeLlm.reply = "first reply"
conv.send(listOf(Content.Text("first user"))) 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) assertEquals(2, storage.workingMemoryStore.list(conv.id).size)
fakeLlm.reply = "second reply" fakeLlm.reply = "second reply"
conv.send(listOf(Content.Text("second user"))) conv.send(listOf(Content.Text("second user")))
// второй turn: WM должен вырасти до [user, assistant, user, assistant] // второй turn: WM должен вырасти до [user, assistant, user, assistant]
val wm = storage.workingMemoryStore.list(conv.id) 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) assertEquals(4, wm.size)
// Длинно-живущий LiteConversation: один на ChatConversation, история // Новая семантика: один LiteConv на turn → два LiteConv после двух send'ов.
// накапливается через sendStreamContents, без пересоздания. assertEquals(2, fakeLlm.conversations.size)
assertEquals(1, fakeLlm.conversations.size) // Второй LiteConv создан с initialMessages из working memory, ИСКЛЮЧАЯ pending user2
val history = fakeLlm.conversations[0].history // (он передаётся в sendStreamContents, чтобы не дублироваться).
assertEquals(4, history.size) val reopened = fakeLlm.conversations.last()
assertEquals("first user", history[0].text) assertEquals(2, reopened.initialMessages.size)
assertEquals(LiteRole.USER, history[0].role) assertEquals("first user", reopened.initialMessages[0].text)
assertEquals("first reply", history[1].text) assertEquals(LiteRole.USER, reopened.initialMessages[0].role)
assertEquals(LiteRole.MODEL, history[1].role) assertEquals("first reply", reopened.initialMessages[1].text)
assertEquals("second user", history[2].text) assertEquals(LiteRole.MODEL, reopened.initialMessages[1].role)
assertEquals(LiteRole.USER, history[2].role) // После sendStreamContents (с user2 + сгенерированный asst2) mutableHistory = 4
assertEquals("second reply", history[3].text) assertEquals(4, reopened.history.size)
assertEquals(LiteRole.MODEL, history[3].role) assertEquals("second reply", reopened.history.last().text)
assertEquals(LiteRole.MODEL, reopened.history.last().role)
} }
@Test @Test
@@ -328,11 +339,22 @@ class ChatAgentTest {
} }
@Test @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() val agent = newAgent()
fakeLlm.slow = true fakeLlm.slow = true
val conv = agent.createConversation(temp = false) val conv = agent.createConversation(temp = false)
// Подписываемся на events ДО send() — SharedFlow без replay, после
// отправки событий подписка ничего не увидит.
val events = mutableListOf<ProtoEvent>()
val eventsJob = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) {
conv.events(Instant.DISTANT_PAST).collect { events.add(it) }
}
val sendJob = launch { val sendJob = launch {
try { try {
conv.send(listOf(Content.Text("hi"))) conv.send(listOf(Content.Text("hi")))
@@ -344,11 +366,84 @@ class ChatAgentTest {
delay(200) delay(200)
conv.interrupt() conv.interrupt()
sendJob.join() sendJob.join()
eventsJob.cancel()
// user сообщение в audit должно быть, assistant — нет (был отменён) // audit: только user (assistant не успел сгенериться)
val msgs = storage.messageStore.listAll(conv.id) val msgs = storage.messageStore.listAll(conv.id)
assertEquals(1, msgs.size) assertEquals(1, msgs.size)
assertIs<pw.binom.agentik.storage.MessageRecord.UserMessage>(msgs[0]) assertIs<pw.binom.agentik.storage.MessageRecord.UserMessage>(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<ProtoEvent>()
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<pw.binom.agentik.storage.MessageRecord.ToolResult>().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 @Test
@@ -9,18 +9,32 @@ import pw.binom.litert.LiteDelta
import pw.binom.litert.LiteLlm import pw.binom.litert.LiteLlm
import pw.binom.litert.LiteMessage import pw.binom.litert.LiteMessage
import pw.binom.litert.LiteRole import pw.binom.litert.LiteRole
import pw.binom.litert.LiteToolCall
/** /**
* Тестовая [LiteLlm], запоминающая последний конфиг/контент и отвечающая * Тестовая [LiteLlm], запоминающая последний конфиг/контент и отвечающая
* заданной строкой [reply] двумя фрагментами + done. * заданной строкой [reply] двумя фрагментами + done.
*/ */
internal class FakeLiteLlm : LiteLlm { internal class FakeLiteLlm : LiteLlm {
sealed class Reply {
data class Text(val text: String) : Reply()
data class ToolCalls(val calls: List<Pair<String, Map<String, Any?>>>) : Reply()
}
override val backendName: String = "fake" override val backendName: String = "fake"
override val capabilities: pw.binom.litert.LiteCapabilities = pw.binom.litert.LiteCapabilities(pw.binom.litert.LiteInputModalities.TextOnly, false, false, null) override val capabilities: pw.binom.litert.LiteCapabilities = pw.binom.litert.LiteCapabilities(pw.binom.litert.LiteInputModalities.TextOnly, false, false, null)
var reply: String = "" var reply: String = ""
var rememberHistory: Boolean = false var rememberHistory: Boolean = false
var slow: Boolean = false var slow: Boolean = false
var failMessage: String? = null var failMessage: String? = null
/**
* Если задан, LLM проходит по этому списку ответов по порядку: первый
* sendStreamContents → первый Reply, второй → второй и т.д. Если список
* кончился — fallback на [reply] (text).
*/
var scriptedReplies: MutableList<Reply> = mutableListOf()
var lastConfig: LiteConversationConfig? = null var lastConfig: LiteConversationConfig? = null
var lastContents: List<LiteContentPart>? = null var lastContents: List<LiteContentPart>? = null
val conversations = mutableListOf<FakeLiteConversation>() val conversations = mutableListOf<FakeLiteConversation>()
@@ -41,6 +55,9 @@ internal class FakeLiteLlm : LiteLlm {
throw UnsupportedOperationException("not used in test") throw UnsupportedOperationException("not used in test")
override fun close() {} override fun close() {}
fun nextReply(): Reply =
if (scriptedReplies.isNotEmpty()) scriptedReplies.removeAt(0) else Reply.Text(reply)
} }
internal class FakeLiteConversation( internal class FakeLiteConversation(
@@ -60,21 +77,31 @@ internal class FakeLiteConversation(
return kotlinx.coroutines.flow.flow { throw RuntimeException(msg) } return kotlinx.coroutines.flow.flow { throw RuntimeException(msg) }
} }
mutableHistory.add(LiteMessage(LiteRole.USER, contents)) mutableHistory.add(LiteMessage(LiteRole.USER, contents))
if (parent.slow) { val next = parent.nextReply()
return kotlinx.coroutines.flow.flow { return when (next) {
emit(LiteDelta(text = parent.reply.substring(0, parent.reply.length / 2))) is FakeLiteLlm.Reply.Text -> {
kotlinx.coroutines.delay(10_000) if (parent.slow) {
emit(LiteDelta(text = parent.reply.substring(parent.reply.length / 2), isDone = true)) kotlinx.coroutines.flow.flow {
mutableHistory.add(LiteMessage.model(parent.reply)) 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 { override fun send(prompt: String): String {
@@ -150,11 +150,6 @@ class PersistenceTest {
fun `working memory — append + list preserves order`() = runTest { fun `working memory — append + list preserves order`() = runTest {
val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000)
val t1 = Instant.fromEpochMilliseconds(1_700_000_001_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( stores.workingMemory.append(
conversationId = "c1", conversationId = "c1",
entry = WorkingMemoryEntry.User(sourceMessageId = "m1", content = listOf(Content.Text("hi"))), entry = WorkingMemoryEntry.User(sourceMessageId = "m1", content = listOf(Content.Text("hi"))),
@@ -166,32 +161,29 @@ class PersistenceTest {
now = t1, now = t1,
) )
val list = stores.workingMemory.list("c1") val list = stores.workingMemory.list("c1")
assertEquals(3, list.size) assertEquals(2, list.size)
assertTrue(list[0].entry is WorkingMemoryEntry.System) assertTrue(list[0].entry is WorkingMemoryEntry.User)
assertTrue(list[1].entry is WorkingMemoryEntry.User) assertTrue(list[1].entry is WorkingMemoryEntry.Assistant)
assertTrue(list[2].entry is WorkingMemoryEntry.Assistant)
} }
@Test @Test
fun `working memory — compact without summary just drops tail`() = runTest { fun `working memory — compact without summary just drops tail`() = runTest {
val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) 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.User("m1", listOf(Content.Text("u1"))), t0)
stores.workingMemory.append("c1", WorkingMemoryEntry.Assistant("m2", listOf(Content.Text("a1"))), 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) stores.workingMemory.append("c1", WorkingMemoryEntry.User("m3", listOf(Content.Text("u2"))), t0)
val rows = stores.workingMemory.list("c1") val rows = stores.workingMemory.list("c1")
// Drop начиная со второго хода (User m1) — должно остаться System. // Drop начиная со второго хода (User m1) — должно остаться только User m1.
val dropFrom = rows[1].orderIdx val dropFrom = rows[1].orderIdx
stores.workingMemory.compact(dropFrom, "c1", summaryText = null) stores.workingMemory.compact(dropFrom, "c1", summaryText = null)
val after = stores.workingMemory.list("c1") val after = stores.workingMemory.list("c1")
assertEquals(1, after.size) assertEquals(1, after.size)
assertTrue(after[0].entry is WorkingMemoryEntry.System) assertTrue(after[0].entry is WorkingMemoryEntry.User)
} }
@Test @Test
fun `working memory — compact with summary inserts Summary entry`() = runTest { fun `working memory — compact with summary inserts Summary entry`() = runTest {
val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) 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.User("m1", listOf(Content.Text("u1"))), t0)
stores.workingMemory.append("c1", WorkingMemoryEntry.Assistant("m2", listOf(Content.Text("a1"))), 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) 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") stores.workingMemory.compact(dropFrom, "c1", summaryText = "**Goal**: chat\n**Active**: at u2\n**Resolved**: a1")
val after = stores.workingMemory.list("c1") val after = stores.workingMemory.list("c1")
assertEquals(2, after.size) assertEquals(2, after.size)
assertTrue(after[0].entry is WorkingMemoryEntry.System) assertTrue(after[0].entry is WorkingMemoryEntry.User)
val summary = after[1].entry val summary = after[1].entry
assertIs<WorkingMemoryEntry.Summary>(summary) assertIs<WorkingMemoryEntry.Summary>(summary)
assertTrue(summary.text.startsWith("**Goal**")) assertTrue(summary.text.startsWith("**Goal**"))
@@ -213,31 +205,30 @@ class PersistenceTest {
@Test @Test
fun `working memory — compact with blank summaryText behaves as drop`() = runTest { fun `working memory — compact with blank summaryText behaves as drop`() = runTest {
val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) 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.User("m1", listOf(Content.Text("u1"))), t0)
val rows = stores.workingMemory.list("c1") 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") val after = stores.workingMemory.list("c1")
assertEquals(1, after.size) assertEquals(1, after.size)
assertTrue(after[0].entry is WorkingMemoryEntry.System) assertTrue(after[0].entry is WorkingMemoryEntry.User)
} }
@Test @Test
fun `working memory — compact is atomic on other conversations`() = runTest { fun `working memory — compact is atomic on other conversations`() = runTest {
val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) 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("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.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") stores.workingMemory.compact(2, "c1", summaryText = "sum")
val c1 = stores.workingMemory.list("c1") val c1 = stores.workingMemory.list("c1")
val c2 = stores.workingMemory.list("c2") val c2 = stores.workingMemory.list("c2")
// c1: System + Summary // c1: User m1 + Summary
assertEquals(2, c1.size) assertEquals(2, c1.size)
assertTrue(c1[1].entry is WorkingMemoryEntry.Summary) assertTrue(c1[1].entry is WorkingMemoryEntry.Summary)
// c2 не тронут // c2 не тронут
assertEquals(2, c2.size) assertEquals(2, c2.size)
assertTrue(c2[1].entry is WorkingMemoryEntry.User) assertTrue(c2[1].entry is WorkingMemoryEntry.Assistant)
} }
@Test @Test
@@ -7,9 +7,10 @@ import kotlinx.serialization.Serializable
* Запись в working memory диалога: ровно то, что агент сейчас видит в * Запись в working memory диалога: ровно то, что агент сейчас видит в
* LLM-контексте. Упорядочено по `order_idx` (заполняется в store при append). * LLM-контексте. Упорядочено по `order_idx` (заполняется в store при append).
* *
* Sealed-иерархия: для v1 — `System` (синтетический system-prompt), * Sealed-иерархия: `User`/`Assistant` (реплики с ссылкой на audit log
* `User`/`Assistant` (реплики с ссылкой на audit log через [sourceMessageId]). * через [sourceMessageId]), `ToolExchange` (синтетическая запись об одном
* Суммаризация (для v2) добавит подтип `Summary`. * tool-вызове + его результате — для replay в LiteMessage(TOOL, ToolResult)
* при пересоздании LiteConv), `Summary` (суммаризация при compaction).
*/ */
@Serializable @Serializable
sealed interface WorkingMemoryEntry { sealed interface WorkingMemoryEntry {
@@ -17,13 +18,6 @@ sealed interface WorkingMemoryEntry {
/** Ссылка на исходное сообщение в audit log (`message.id`). `null` для синтетических строк. */ /** Ссылка на исходное сообщение в audit log (`message.id`). `null` для синтетических строк. */
val sourceMessageId: String? val sourceMessageId: String?
/** Синтетический system-prompt, добавляется при создании диалога. */
@Serializable
@SerialName("system")
data class System(val text: String) : WorkingMemoryEntry {
override val sourceMessageId: String? = null
}
/** Реплика пользователя. */ /** Реплика пользователя. */
@Serializable @Serializable
@SerialName("user") @SerialName("user")
@@ -47,6 +41,32 @@ sealed interface WorkingMemoryEntry {
val content: List<Content>, val content: List<Content>,
) : WorkingMemoryEntry ) : 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. Не имеет ссылки на конкретное сообщение * compaction'е working memory. Не имеет ссылки на конкретное сообщение
@@ -15,21 +15,19 @@ class InMemoryWorkingMemoryStoreTest {
fun `append assigns sequential order_idx starting from 0`() = runTest { fun `append assigns sequential order_idx starting from 0`() = runTest {
val store = InMemoryWorkingMemoryStore() val store = InMemoryWorkingMemoryStore()
val t0 = Instant.parse("2026-09-15T10:00:00Z") 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.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.Assistant("a1", listOf(Content.Text("hello"))), t0.plus(kotlin.time.Duration.parse("PT2S")))
val rows = store.list("c1") val rows = store.list("c1")
assertEquals(3, rows.size) assertEquals(2, rows.size)
assertEquals(listOf(0L, 1L, 2L), rows.map { it.orderIdx }) assertEquals(listOf(0L, 1L), rows.map { it.orderIdx })
} }
@Test @Test
fun `order_idx continues across conversations independently`() = runTest { fun `order_idx continues across conversations independently`() = runTest {
val store = InMemoryWorkingMemoryStore() val store = InMemoryWorkingMemoryStore()
val t0 = Instant.parse("2026-09-15T10:00:00Z") 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("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 // c2 должен начать с 0, не продолжать c1
val rows2 = store.list("c2") val rows2 = store.list("c2")
assertEquals(1, rows2.size) assertEquals(1, rows2.size)
@@ -56,8 +54,8 @@ class InMemoryWorkingMemoryStoreTest {
fun `clear removes all rows for a conversation`() = runTest { fun `clear removes all rows for a conversation`() = runTest {
val store = InMemoryWorkingMemoryStore() val store = InMemoryWorkingMemoryStore()
val t0 = Instant.parse("2026-09-15T10:00:00Z") 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)
store.clear("c1") store.clear("c1")
assertEquals(emptyList(), store.list("c1")) assertEquals(emptyList(), store.list("c1"))
assertEquals(1, store.list("c2").size) assertEquals(1, store.list("c2").size)
@@ -67,23 +65,20 @@ class InMemoryWorkingMemoryStoreTest {
fun `compact without summary drops tail and returns new max`() = runTest { fun `compact without summary drops tail and returns new max`() = runTest {
val store = InMemoryWorkingMemoryStore() val store = InMemoryWorkingMemoryStore()
val t0 = Instant.parse("2026-09-15T10:00:00Z") 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.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.Assistant("a1", listOf(Content.Text("hello"))), t0.plus(kotlin.time.Duration.parse("PT2S")))
// dropFromOrderIdx=2 → удаляет всё >= 2 (то есть только Assistant "a1") // dropFromOrderIdx=1 → удаляет всё >= 1 (то есть только Assistant "a1")
val newMax = store.compact(dropFromOrderIdx = 2, conversationId = "c1", summaryText = null) val newMax = store.compact(dropFromOrderIdx = 1, conversationId = "c1", summaryText = null)
assertEquals(1L, newMax) assertEquals(0L, newMax)
val rows = store.list("c1") val rows = store.list("c1")
assertEquals(2, rows.size) assertEquals(1, rows.size)
assertEquals("a", (rows[0].entry as WorkingMemoryEntry.System).text) assertEquals("u1", rows[0].entry.sourceMessageId)
assertEquals("u1", rows[1].entry.sourceMessageId)
} }
@Test @Test
fun `compact with summary replaces tail with synthetic Summary row`() = runTest { fun `compact with summary replaces tail with synthetic Summary row`() = runTest {
val store = InMemoryWorkingMemoryStore() val store = InMemoryWorkingMemoryStore()
val t0 = Instant.parse("2026-09-15T10:00:00Z") 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.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.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"))) store.append("c1", WorkingMemoryEntry.User("u2", listOf(Content.Text("how are you")), null), t0.plus(kotlin.time.Duration.parse("PT3S")))
@@ -72,9 +72,9 @@ class SqliteWorkingMemoryStore(private val db: AgentikDatabase) : WorkingMemoryS
} }
private fun entryKind(e: WorkingMemoryEntry): String = when (e) { private fun entryKind(e: WorkingMemoryEntry): String = when (e) {
is WorkingMemoryEntry.System -> "system"
is WorkingMemoryEntry.User -> "user" is WorkingMemoryEntry.User -> "user"
is WorkingMemoryEntry.Assistant -> "assistant" is WorkingMemoryEntry.Assistant -> "assistant"
is WorkingMemoryEntry.ToolExchange -> "tool_exchange"
is WorkingMemoryEntry.Summary -> "summary" is WorkingMemoryEntry.Summary -> "summary"
} }