diff --git a/docs/STANDALONE.md b/docs/STANDALONE.md new file mode 100644 index 0000000..abc802f --- /dev/null +++ b/docs/STANDALONE.md @@ -0,0 +1,296 @@ +# Standalone — рантайм агента agentik + +`standalone` — это исполняемое JVM-приложение (точка входа `pw.binom.agentik.standalone.MainKt`), которое поднимает реальный агент `ChatAgent` (stateful, SQLite-персистентный) и навешивает на него HTTP+SSE фасад `:server`. LLM-движок выбирается через `AGENTIK_LLM_BACKEND` — на v1 поддерживаются `litert-openai` (любой OpenAI-совместимый endpoint) и `litert-google` (on-device движок LiteRT-LM 0.16.1 через нативную `.so`-библиотеку). + +Этот документ описывает, как `standalone` собран и как его расширять. + +--- + +## 1. Что в коробке после `git clone` + +``` +:proto — единое ядро протокола (KMP, commonMain) + Agent / Conversation / Event / Message / Content / AgentEvent + stateful: агент сам хранит историю и working memory + +:server — HTTP+SSE фасад :proto + public Route.agentikAgent(agent, path = "/agentik") + +:standalone — JVM-рантайм с реальным LLM-агентом + ChatAgent + ChatConversation поверх SQLite и litert-* (openai/google) + default 8080: + GET /health health check + POST /agentik/conversations создать диалог (201) + GET /agentik/conversations список + GET /agentik/conversations/{id} один диалог + PATCH /agentik/conversations/{id} переименовать + DELETE /agentik/conversations/{id} удалить (204) + POST /agentik/conversations/{id}/messages отправить user-сообщение (202) + POST /agentik/conversations/{id}/interrupt прервать текущий ход (202) + GET /agentik/conversations/{id}/messages страница истории + GET /agentik/conversations/{id}/events SSE live-события хода + GET /agentik/events SSE live-события агента +``` + +Полная таблица эндпоинтов — в `docs/ARCHITECTURE.md` (раздел «:server»). + +--- + +## 2. Архитектура слоёв + +``` + клиенты транспорт + ┌───────────────┐ ┌─────────────────────────────┐ + │ Web / CLI / │ ──HTTP──► │ Route.agentikAgent(agent) │ + │ desktop │ ──SSE───► │ :server (Ktor + Netty) │ + │ │ └──────────────┬──────────────┘ + └───────────────┘ │ + ▼ + pw.binom.agentik.proto.Agent + (ChatAgent) + │ + ┌───────────────┴───────────────┐ + ▼ ▼ + ChatConversation.send(content) agent.events / agent.getConversations + │ + ▼ + ┌──────────────────────────────────────┐ + │ 1. audit: append UserMessage │ + │ 2. working_memory: append User │ + │ 3. ensureLiteConversation: │ + │ first turn → create from WM; │ + │ next turns → reuse (KV-cache) │ + │ 4. sendStreamContents → emit │ + │ StartResponse / AppendText / │ + │ End │ + │ 5. audit + WM: append AssistantMessage│ + └──────────────────────────────────────┘ + │ │ + ▼ ▼ + Conversation.events(after) Conversation.getMessages(after) + (live, no replay) (история) +``` + +Слои рантайма: + +``` +:standalone + ├── persistence/ ← интерфейсы и records (commonMain, без зависимостей) + │ ConversationStore / MessageStore / WorkingMemoryStore + │ ConversationRecord / MessageRecord / WorkingMemoryEntry / Content + │ Payload.kt — JSON-сериализация + ├── persistence/sqlite/ ← JVM: SQLDelight-схема + три SQLite-реализации + │ SqliteStores.open(path | inMemory) + │ src/jvmMain/sqldelight/.../*.sq + ├── llm/ ← LlmConfig (env → OpenAI/Google), backend-agnostic + └── agent/ ← ChatAgent + ChatConversation (stateful, long-lived LiteConv) +``` + +Ключевой инвариант: **разговор живёт внутри агента, а не в клиенте и не в транспорте.** Транспорт — лишь сериализатор: HTTP пишет в/читает из `:server`-эндпоинтов, SSE шлёт события. У них нет своего состояния диалога. + +--- + +## 3. Точка входа: `pw.binom.agentik.standalone.MainKt` + +```kotlin +fun main() { + val port = System.getenv("AGENTIK_PORT")?.toIntOrNull() ?: 8080 + val dbPath = System.getenv("AGENTIK_DB_PATH")?.takeIf { it.isNotBlank() } ?: "./agentik.db" + + val llmConfig = LlmConfig.fromEnv() + val llm = llmConfig.createLlm() + val stores = SqliteStores.open(dbPath = dbPath) + val agent = ChatAgent( + id = "agentik", + stores = stores, + llm = llm, + llmConfig = llmConfig, + ) + + val server = embeddedServer(Netty, port = port) { + routing { + get("/health") { call.respondText("ok") } + agentikAgent(agent, path = "/agentik") + } + } + Runtime.getRuntime().addShutdownHook(Thread { + agent.close(); stores.close(); llm.close() + }) + server.start(wait = true) +} +``` + +Один `ChatAgent` отвечает и за диалоги (`/agentik/conversations/...`), и за live-события (`/agentik/events`). Все три ресурса — БД, LLM, Netty — корректно закрываются в shutdown-хуке. + +--- + +## 4. Контракт `Agent` (от `pw.binom.agentik.proto`) + +Реализация **обязана** уметь: + +| метод | смысл | +|---|---| +| `id: String` | идентификатор агента | +| `createConversation(temp: Boolean): Conversation` | новая сессия, `temp=true` — не персистить | +| `getConversation(id): Conversation?` | достать по id, `null` если нет | +| `deleteConversation(id): Boolean` | удалить | +| `getConversations(offset, limit)` | страница списка | +| `events(after: Instant): Flow` | live-события по множеству разговоров | + +Реализация `Conversation`: + +| метод | смысл | +|---|---| +| `id: String` | идентификатор диалога | +| `title: String?` | заголовок (может быть `null`) | +| `isTemporal: Boolean` | `true` = не персистить (`temp=true` при создании) | +| `isSupportImageInput/Output: Boolean` | мультимодальные возможности (для v1 оба `false`) | +| `updatedAt: Instant` | последний `send`/`rename` | +| `send(content: List)` | **fire-and-forget**: добавить user-сообщение, запустить ход, выйти | +| `interrupt()` | остановить текущий ход (best-effort) | +| `events(after): Flow` | live-события хода (StartReasoning, StartResponse, AppendText, End, Interrupted, Error) | +| `getMessages(after, offset, limit)` | страница истории | +| `rename(title)` | переименовать | +| `close()` | освободить ресурсы | + +Главное: `send` ничего не возвращает. Чтобы получить события, нужно **отдельно** подписаться на `events(after)` ДО `send` либо сразу после — поток событий стартует с момента подписки, бэкфилл через `getMessages`. + +--- + +## 5. Persistence — dual-log + +`ChatAgent` хранит каждую сессию в двух логически разных таблицах: + +| таблица | назначение | мутации | +|---|---|---| +| `message` | append-only audit log. Все user/assistant/tool-call/tool-result сообщения. Никогда не редактируется (кроме каскадного `DELETE` при удалении диалога). | только `INSERT` | +| `working_memory` | mutable LLM-контекст. System-prompt + текущая история + (в v2) суммаризации. | `INSERT`, `compact(dropFromIdx, summary)` | + +Маппинг `:proto.Message ↔ MessageRecord` живёт в `ChatConversation.kt` (`toProto`/`toStorage`) — сами `MessageRecord` намеренно НЕ зависят от `:proto`, чтобы можно было сменить транспорт без миграции таблиц. + +Подробный контракт — в комментариях к `MessageRecord.kt` и `WorkingMemoryEntry.kt`. + +### `MessageStore` + +```kotlin +suspend fun append(record: MessageRecord) +suspend fun list(conversationId: String, after: Instant, offset: Int, limit: Int): List +suspend fun listAll(conversationId: String): List +``` + +### `WorkingMemoryStore` + +```kotlin +suspend fun append(conversationId: String, entry: WorkingMemoryEntry, now: Instant) +suspend fun list(conversationId: String): List +suspend fun clear(conversationId: String) +suspend fun compact(dropFromOrderIdx: Long, conversationId: String): Long +``` + +`compact` — атомарный «выбросить всё от `dropFromOrderIdx` и дальше, вставить новую синтетическую запись на следующий `order_idx`». Для v1 — просто `DELETE` от индекса (суммаризация появится в v2 вместе с LLM-вызовом для генерации текста). + +### `ConversationStore` + +```kotlin +suspend fun upsert(record: ConversationRecord) +suspend fun get(id: String): ConversationRecord? +suspend fun delete(id: String): Boolean // каскадно чистит message + working_memory +suspend fun list(offset: Int, limit: Int): List +suspend fun rename(id: String, title: String?): Instant? +suspend fun touch(id: String, now: Instant) +``` + +Все три store — `AutoCloseable`; корневой ресурс `SqliteStores` закрывает их вместе с `SqlDriver`. + +--- + +## 6. LLM-конфигурация (`LlmConfig`) + +`LlmConfig.fromEnv()` парсит env, валидирует обязательные поля и выбирает бэкенд через `AGENTIK_LLM_BACKEND`: + +- `openai` (default) — `litert-openai`, текст-онли чат против любого OpenAI-совместимого endpoint. +- `google` — `litert-google` (LiteRT-LM 0.16.1, on-device `.task`/`.litertlm` модель, требует нативной библиотеки через `litertlm-jvm`). + +### Общие env + +| env | смысл | default | +|---|---|---| +| `AGENTIK_PORT` | порт Netty (`/agentik`, `/health`) | `8080` | +| `AGENTIK_DB_PATH` | путь к SQLite-файлу | `./agentik.db` | +| `AGENTIK_LLM_BACKEND` | `openai` или `google` | `openai` | +| `AGENTIK_SYSTEM_PROMPT` | текст системного промпта | «Ты полезный ассистент. Отвечай кратко и по делу.» | + +### Backend `openai` + +| env | смысл | +|---|---| +| `OPENAI_BASE_URL` | endpoint (например, `https://api.openai.com/v1` или `http://localhost:11434/v1`) — обязательно | +| `OPENAI_API_KEY` | ключ модели — обязательно | +| `OPENAI_MODEL` | имя модели (например, `gpt-4o-mini`, `myopenai/local/codding`) — обязательно | + +### Backend `google` (on-device LiteRT-LM) + +| env | смысл | default | +|---|---|---| +| `AGENTIK_GOOGLE_MODEL_PATH` | путь к `.task` или `.litertlm` модели — обязательно | +| `AGENTIK_GOOGLE_CACHE_DIR` | каталог кеша скомпилированных graph'ов | пусто (системный tmp) | +| `AGENTIK_GOOGLE_THREADS` | число CPU-потоков для движка | `4` | + +`AGENTIK_DB_PATH=:memory:` создаёт in-memory БД (только для тестов и интеграционных проверок). + +### Long-lived LiteConversation — ОБЯЗАТЕЛЬНО для google + +`LiteConversation` от любого litert-бэкенда — это **долгоживущая stateful ручка**: она держит историю сообщений и (для google) KV-cache/sampler-state. На v1 `ChatConversation` создаёт `LiteConversation` один раз (на первом `send`) и переиспользует на всех последующих turn'ах той же беседы. Пересоздание LiteConversation на каждый send ломает KV-cache для google (и в v1 приводит к ошибке chat template «roles must alternate» после двух user-сообщений). + +`AGENTIK_GOOGLE_FOLD_SYSTEM_INTO_FIRST_USER` (default `true` для google) — фолдит системный промпт в первое user-сообщение, чтобы движок увидел только `[User+system, Assistant, User, ...]` и не падал на alternation. Для openai этот режим отключён. + +### Workaround для litert-google-jvm 0.16.1 + +Баг в `sendStreamContents` — поток не закрывается после `isDone = true`. `ChatConversation` оборачивает стрим в `transformWhile { !delta.isDone }`, поэтому SSE получает корректный `End` event. + +--- + +## 7. Расширение + +### Подключить тул (v2) + +```kotlin +val tools: List = listOf(ReadFileTool(Path.of("/work"))) +val cfg = LiteConversationConfig(systemInstruction = "…", tools = tools) +``` + +LiteConversation сообщит модели о доступных тулах. После появления `LiteContentPart.ToolResult` (v2) появится и `ChatConversation.addToolResult(...)` — ручной tool-loop на on-device движке без пересоздания беседы. + +### Добавить ещё один транспорт + +Каждый транспорт — отдельный модуль, который получает `Agent` и сериализует его под свой протокол: + +* `:server` (HTTP+SSE) — готов, `Route.agentikAgent(agent, path = "/agentik")` +* `:client` (HTTP-клиент) — готов, `AgentikAgent(id, baseUrl, httpClient)` +* `:irc-server` — IRC-фасад, в планах + +Транспорт **не имеет доступа к внутренностям `ChatAgent`** — он видит только интерфейс `Agent`. Это и есть «транспортно-агностичное ядро». + +### Добавить ещё один LLM-бэкенд + +1. Описать `Config` data class с нужными полями. +2. Реализовать `LiteLlm`/`LiteConversation` поверх движка (см. litert-kmp — там уже есть `litert-google`, `litert-openai`, `litert-koog`). +3. Расширить `LlmConfig.createLlm()` веткой `when`. + +### Заменить SQLite на Postgres / MongoDB / etc + +1. Реализовать три store-интерфейса поверх нового движка. +2. Передать их в `ChatAgent` вместо `SqliteStores`. +3. Удалить (или оставить за `:standalone`-флагом) `:persistence/sqlite/`. + +--- + +## 8. Что НЕ делает `standalone` сегодня + +* **Нет инструментов (тулов).** Поддержка ждёт `LiteContentPart.ToolResult` в litert-api (v2). До тех пор агент — text-only чат. +* **Нет суммаризации.** `WorkingMemoryStore.compact` уже есть, но без LLM-вызова для генерации текста суммаризации. +* **Нет MCP-клиента.** +* **Нет авторизации.** Все эндпоинты открыты. +* **Нет инкрементальной догрузки старых сообщений.** `getMessages(after)` работает с offset/limit, но без «схлопывания» (compaction в визуальной истории — задача клиента). + +Каждый пункт закрывается отдельным коммитом; код логически разделён по слоям так, чтобы точечные изменения не требовали переделки соседей. diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index e525557..105dec5 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -6,11 +6,14 @@ kotlinx-datetime = "0.8.0" ktor = "3.1.3" agui = "0.1.0" a2a = "1.0.0-SNAPSHOT" +litert = "6" +sqldelight = "2.3.2" [plugins] kotlin-multiplatform = { id = "org.jetbrains.kotlin.multiplatform", version.ref = "kotlin" } kotlin-jvm = { id = "org.jetbrains.kotlin.jvm", version.ref = "kotlin" } kotlin-serialization = { id = "org.jetbrains.kotlin.plugin.serialization", version.ref = "kotlin" } +sqldelight = { id = "app.cash.sqldelight", version.ref = "sqldelight" } [libraries] # --- AG-UI (pw.binom.agui) — KMP: jvm + native --- @@ -23,6 +26,16 @@ a2a-shared = { module = "pw.binom.a2a:shared", version.ref = "a2a" } a2a-client = { module = "pw.binom.a2a:client", version.ref = "a2a" } a2a-server = { module = "pw.binom.a2a:server", version.ref = "a2a" } +# --- litert-kmp (pw.binom.litert) — universal LLM wrapper --- +litert-api = { module = "pw.binom.litert:litert-api", version.ref = "litert" } +litert-openai = { module = "pw.binom.litert:litert-openai-jvm", version.ref = "litert" } +litert-google = { module = "pw.binom.litert:litert-google", version.ref = "litert" } + +# --- SQLDelight (app.cash.sqldelight) — KMP SQLite, JDBC driver --- +sqldelight-runtime = { module = "app.cash.sqldelight:runtime", version.ref = "sqldelight" } +sqldelight-sqlite-driver = { module = "app.cash.sqldelight:sqlite-driver", version.ref = "sqldelight" } +sqldelight-coroutines = { module = "app.cash.sqldelight:coroutines-extensions", version.ref = "sqldelight" } + # --- Ktor (сервер, JVM) --- ktor-server-core = { module = "io.ktor:ktor-server-core", version.ref = "ktor" } ktor-server-sse = { module = "io.ktor:ktor-server-sse", version.ref = "ktor" } @@ -32,6 +45,9 @@ ktor-serialization-kotlinx-json = { module = "io.ktor:ktor-serialization-kotlinx ktor-client-core = { module = "io.ktor:ktor-client-core", version.ref = "ktor" } ktor-client-cio = { module = "io.ktor:ktor-client-cio", version.ref = "ktor" } ktor-client-content-negotiation = { module = "io.ktor:ktor-client-content-negotiation", version.ref = "ktor" } +ktor-server-test-host = { module = "io.ktor:ktor-server-test-host", version.ref = "ktor" } + +kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" } # --- commons --- kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "kotlinx-coroutines" } diff --git a/standalone/build.gradle.kts b/standalone/build.gradle.kts index 72c6c5e..f8650fe 100644 --- a/standalone/build.gradle.kts +++ b/standalone/build.gradle.kts @@ -1,14 +1,18 @@ +@file:OptIn(org.jetbrains.kotlin.gradle.ExperimentalKotlinGradlePluginApi::class) + import org.jetbrains.kotlin.gradle.ExperimentalKotlinGradlePluginApi +import org.jetbrains.kotlin.gradle.targets.jvm.KotlinJvmTarget plugins { alias(libs.plugins.kotlin.multiplatform) + alias(libs.plugins.kotlin.serialization) + alias(libs.plugins.sqldelight) } kotlin { jvmToolchain(21) jvm { - @OptIn(ExperimentalKotlinGradlePluginApi::class) binaries { executable { mainClass.set("pw.binom.agentik.standalone.MainKt") @@ -18,30 +22,63 @@ kotlin { sourceSets { commonMain.dependencies { - // AG-UI: протокол (события, RunAgentInput, Agent) — KMP, jvm + native - implementation(libs.agui.api) - // proto: наш in-house протокол (KMP) implementation(project(":proto")) + implementation(project(":server")) + + // Commons + implementation(libs.kotlinx.coroutines.core) + implementation(libs.kotlinx.serialization.json) + implementation(libs.kotlinx.datetime) + + // litert-kmp: контракт (commonMain) + api(libs.litert.api) + + // SQLDelight runtime (commonMain) + api(libs.sqldelight.runtime) + api(libs.sqldelight.coroutines) } + jvmMain.dependencies { - // AG-UI: Ktor-хелперы маршрута + движок Netty (JVM) - implementation(libs.agui.server) + // litert-openai: JVM-реализация + implementation(libs.litert.openai) + // litert-google: встроенный LiteRT-LM движок, нужен только на runtime + runtimeOnly(libs.litert.google) + + // SQLDelight JDBC driver (JVM) + implementation(libs.sqldelight.sqlite.driver) + + // Ktor server (для :server facade + a2aServer) implementation(libs.ktor.server.core) implementation(libs.ktor.server.sse) implementation(libs.ktor.server.netty) + implementation(libs.ktor.server.content.negotiation) + implementation(libs.ktor.serialization.kotlinx.json) + + // Транспортные фасады + implementation(libs.agui.server) + implementation(libs.a2a.server) + } + + commonTest.dependencies { + implementation(libs.kotlinx.coroutines.core) + implementation(libs.kotlin.test) + } + + jvmTest.dependencies { implementation(libs.kotlinx.coroutines.core) implementation(libs.kotlinx.serialization.json) - // A2A (pw.binom.a2a): shared — протокол, server + client — JVM - implementation(libs.a2a.shared) - implementation(libs.a2a.server) - implementation(libs.a2a.client) - implementation(libs.ktor.client.core) - implementation(libs.ktor.client.cio) - // server: Ktor-фасад нашего :proto (JVM) - implementation(project(":server")) - } - commonTest.dependencies { - implementation(kotlin("test")) + implementation(libs.kotlin.test) + // Ktor test engine для smoke-тестов HTTP + implementation(libs.ktor.server.test.host) } } -} \ No newline at end of file +} + +sqldelight { + databases { + create("AgentikDatabase") { + packageName.set("pw.binom.agentik.standalone.persistence.sqlite") + srcDirs.setFrom("src/jvmMain/sqldelight") + } + } +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Content.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Content.kt new file mode 100644 index 0000000..5eca4f0 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Content.kt @@ -0,0 +1,29 @@ +package pw.binom.agentik.standalone.persistence + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +/** + * Часть контента сообщения на уровне хранилища. + * + * Намеренно НЕ зависит от [pw.binom.agentik.proto.Content] — маппинг + * `:proto.Content ↔ Content` живёт в `Mapping.kt`. Структурно типы + * идентичны, но даёт возможность заменить transport-протокол без миграции + * таблиц. + * + * Image сериализуется в JSON через base64 (стандарт для kotlinx-serialization). + */ +@Serializable +sealed interface Content { + @Serializable + @SerialName("text") + data class Text(val body: String) : Content + + @Serializable + @SerialName("image") + data class Image(val data: ByteArray, val mime: String) : Content { + override fun equals(other: Any?): Boolean = + this === other || (other is Image && mime == other.mime && data.contentEquals(other.data)) + override fun hashCode(): Int = 31 * mime.hashCode() + data.contentHashCode() + } +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationRecord.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationRecord.kt new file mode 100644 index 0000000..a9658dd --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationRecord.kt @@ -0,0 +1,14 @@ +package pw.binom.agentik.standalone.persistence + +import kotlin.time.Instant + +/** + * Snapshot диалога. В таблице `conversation` хранится как есть. + */ +data class ConversationRecord( + val id: String, + val title: String?, + val isTemporal: Boolean, + val createdAt: Instant, + val updatedAt: Instant, +) diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationStore.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationStore.kt new file mode 100644 index 0000000..b95d526 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/ConversationStore.kt @@ -0,0 +1,27 @@ +package pw.binom.agentik.standalone.persistence + +import kotlin.time.Instant + +/** + * CRUD по таблице `conversation`. + */ +interface ConversationStore : AutoCloseable { + + /** Создать или обновить snapshot диалога. */ + suspend fun upsert(record: ConversationRecord) + + /** Диалог по id, или `null`. */ + suspend fun get(id: String): ConversationRecord? + + /** Удалить диалог (вместе с его сообщениями и working memory). */ + suspend fun delete(id: String): Boolean + + /** Список диалогов, отсортированный по `updatedAt` DESC. */ + suspend fun list(offset: Int, limit: Int): List + + /** Переименовать диалог; `null` для сброса заголовка. Возвращает новый `updatedAt` или `null`, если не найден. */ + suspend fun rename(id: String, title: String?): Instant? + + /** Обновить `updatedAt` диалога (например, после отправки сообщения). */ + suspend fun touch(id: String, now: Instant) +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageRecord.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageRecord.kt new file mode 100644 index 0000000..273c8d3 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageRecord.kt @@ -0,0 +1,91 @@ +package pw.binom.agentik.standalone.persistence + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable +import kotlin.time.Instant + +/** + * Запись в таблице `message` (append-only audit) и `working_memory` (mutable view). + * + * Используется sealed-иерархия: подтипы `User`/`Assistant`/`ToolCall`/`ToolResult` + * живут и там, и там. `Summary`/`System` — только в `working_memory` + * (синтетические строки, созданные при суммаризации или как system-prompt). + * + * Все подтипы несут [id] (UUID, стабильный между лайв-стримом Event и историей), + * [conversationId] и [createdAt]. + * + * Поля, специфичные для подтипа, сериализуются в JSON в `payload_json` + * колонке SQLite — это даёт гибкость без миграций при добавлении полей. + */ +@Serializable +sealed interface MessageRecord { + + val id: String + val conversationId: String + val createdAt: Instant + + /** Подтип сообщения с телом из [Content]. */ + @Serializable + sealed interface Body : MessageRecord { + val content: List + } + + @Serializable + @SerialName("user") + data class UserMessage( + override val id: String, + override val conversationId: String, + override val content: List, + override val createdAt: Instant, + ) : Body + + @Serializable + @SerialName("assistant") + data class AssistantMessage( + override val id: String, + override val conversationId: String, + override val content: List, + override val createdAt: Instant, + ) : Body + + @Serializable + @SerialName("tool_call") + data class ToolCall( + override val id: String, + override val conversationId: String, + val toolName: String, + val toolTitle: String?, + val toolArgsJson: String, + override val createdAt: Instant, + ) : MessageRecord + + @Serializable + @SerialName("tool_result") + data class ToolResult( + override val id: String, + override val conversationId: String, + val toolCallId: String, + val result: String?, + override val createdAt: Instant, + ) : MessageRecord + + /** Синтетическое: суммаризация старого контекста. Только в working_memory. */ + @Serializable + @SerialName("summary") + data class Summary( + override val id: String, + override val conversationId: String, + val text: String, + override val createdAt: Instant, + ) : MessageRecord + + /** Синтетическое: system-prompt, введённый при создании диалога. Только в working_memory. */ + @Serializable + @SerialName("system") + data class System( + override val id: String, + override val conversationId: String, + val text: String, + override val createdAt: Instant, + ) : MessageRecord +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageStore.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageStore.kt new file mode 100644 index 0000000..a5daa72 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/MessageStore.kt @@ -0,0 +1,24 @@ +package pw.binom.agentik.standalone.persistence + +import kotlin.time.Instant + +/** + * Append-only audit log сообщений (`message` table). + * + * Только `insert` и чтение. Никаких обновлений, никакого удаления (кроме + * каскадного удаления вместе с [ConversationStore.delete]). + */ +interface MessageStore : AutoCloseable { + + /** Добавить запись в audit log. `conversationId` берётся из [MessageRecord.conversationId]. */ + suspend fun append(record: MessageRecord) + + /** + * Страница audit-сообщений диалога после [after] (UTC), отсортированная + * по `createdAt ASC`. Для первоначальной загрузки передай `Instant.DISTANT_PAST`. + */ + suspend fun list(conversationId: String, after: Instant, offset: Int, limit: Int): List + + /** Все сообщения диалога, отсортированные по `createdAt ASC` (для rebuild working memory). */ + suspend fun listAll(conversationId: String): List +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Payload.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Payload.kt new file mode 100644 index 0000000..1be0194 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/Payload.kt @@ -0,0 +1,26 @@ +package pw.binom.agentik.standalone.persistence + +import kotlinx.serialization.builtins.ListSerializer +import kotlinx.serialization.json.Json + +/** + * JSON-формат для тел user/assistant сообщений: список [Content], + * сериализованный в строку (через kotlinx-serialization). + */ +private val bodyJson = Json { + ignoreUnknownKeys = true + encodeDefaults = true + explicitNulls = false +} + +/** + * Сериализует список [Content] в JSON-строку для хранения в `payload_json`. + */ +fun encodeBodyPayload(content: List): String = + bodyJson.encodeToString(ListSerializer(Content.serializer()), content) + +/** + * Десериализует список [Content] из JSON-строки `payload_json`. + */ +fun decodeBodyPayload(json: String): List = + bodyJson.decodeFromString(ListSerializer(Content.serializer()), json) diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryEntry.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryEntry.kt new file mode 100644 index 0000000..82cbdc5 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryEntry.kt @@ -0,0 +1,42 @@ +package pw.binom.agentik.standalone.persistence + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable + +/** + * Запись в working memory диалога: ровно то, что агент сейчас видит в + * LLM-контексте. Упорядочено по `order_idx` (заполняется в store при append). + * + * Sealed-иерархия: для v1 — `System` (синтетический system-prompt), + * `User`/`Assistant` (реплики с ссылкой на audit log через [sourceMessageId]). + * Суммаризация (для v2) добавит подтип `Summary`. + */ +@Serializable +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") + data class User( + override val sourceMessageId: String, + val content: List, + ) : WorkingMemoryEntry + + /** Реплика ассистента. */ + @Serializable + @SerialName("assistant") + data class Assistant( + override val sourceMessageId: String, + val content: List, + ) : WorkingMemoryEntry +} diff --git a/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryStore.kt b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryStore.kt new file mode 100644 index 0000000..fb76906 --- /dev/null +++ b/standalone/src/commonMain/kotlin/pw/binom/agentik/standalone/persistence/WorkingMemoryStore.kt @@ -0,0 +1,50 @@ +package pw.binom.agentik.standalone.persistence + +import kotlin.time.Instant + +/** + * Одна строка `working_memory` таблицы (внутреннее представление store). + * + * Используется для тестов и для перестроения [WorkingMemoryEntry] из row. + * Агент не должен с этим типом работать напрямую — он работает с + * [WorkingMemoryEntry] через [WorkingMemoryStore]. + */ +data class WorkingMemoryRow( + val id: String, + val conversationId: String, + val orderIdx: Long, + val sourceMessageId: String?, + val entry: WorkingMemoryEntry, + val createdAt: Instant, +) + +/** + * Мутируемое представление LLM-контекста диалога (`working_memory` table). + * + * Аудит-лог — [MessageStore], неизменный; здесь — ровно то, что агент сейчас + * «видит»: системный промпт + реплики + (опционально) суммаризации. + * Строки упорядочены по `order_idx` ASC; `source_message_id` NULL указывает + * на синтетические строки (System, а в v2 — Summary). + * + * Суммаризация / чистка — один атомарный вызов [compact]. + */ +interface WorkingMemoryStore : AutoCloseable { + + /** Добавить запись в конец working memory (новый максимальный `order_idx`). */ + suspend fun append(conversationId: String, entry: WorkingMemoryEntry, now: Instant) + + /** Все строки working memory диалога в порядке отправки. */ + suspend fun list(conversationId: String): List + + /** Очистить working memory диалога (используется при reset/rebuild). */ + suspend fun clear(conversationId: String) + + /** + * Атомарная суммаризация (v2): удаляет все строки с `order_idx` в диапазоне + * `[dropFromOrderIdx, +∞)` и вставляет вместо них новый `summary` с + * указанным текстом. Возвращает новый максимальный `order_idx`. + * + * Для v1 просто удаляет — суммаризация появится в v2. + */ + suspend fun compact(dropFromOrderIdx: Long, conversationId: String): Long +} diff --git a/standalone/src/commonTest/kotlin/pw/binom/agentik/standalone/PlaceholderTest.kt b/standalone/src/commonTest/kotlin/pw/binom/agentik/standalone/PlaceholderTest.kt deleted file mode 100644 index 4bf886b..0000000 --- a/standalone/src/commonTest/kotlin/pw/binom/agentik/standalone/PlaceholderTest.kt +++ /dev/null @@ -1,11 +0,0 @@ -package pw.binom.agentik.standalone - -import kotlin.test.Test -import kotlin.test.assertTrue - -class PlaceholderTest { - @Test - fun placeholder() { - assertTrue(true) - } -} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aOutbound.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aOutbound.kt deleted file mode 100644 index d5b6f65..0000000 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/A2aOutbound.kt +++ /dev/null @@ -1,48 +0,0 @@ -package pw.binom.agentik.standalone - -import pw.binom.a2a.client.A2AClient -import pw.binom.a2a.model.AgentCard -import pw.binom.a2a.model.Message -import pw.binom.a2a.model.Role -import pw.binom.a2a.model.Task -import pw.binom.a2a.model.TextPart - -data class RemoteAgent(val name: String, val baseUrl: String, val token: String? = null) - -object A2aOutbound { - private val clients = mutableMapOf() - - fun remoteAgents(): List = - System.getenv("AGENTIK_A2A_AGENTS") - ?.split(";") - ?.mapNotNull { entry -> - if (!entry.contains("=")) return@mapNotNull null - val name = entry.substringBefore("=").trim() - val rest = entry.substringAfter("=").split(",", limit = 2) - val baseUrl = rest.getOrNull(0)?.trim().orEmpty() - if (name.isEmpty() || baseUrl.isEmpty()) return@mapNotNull null - val token = rest.getOrNull(1)?.trim()?.takeIf { it.isNotEmpty() } - RemoteAgent(name = name, baseUrl = baseUrl, token = token) - } - ?: emptyList() - - private fun clientFor(name: String): A2AClient = - clients.getOrPut(name) { - val remote = - remoteAgents().firstOrNull { it.name == name } - ?: error("Remote agent $name is not configured (AGENTIK_A2A_AGENTS)") - A2AClient.create(baseUrl = remote.baseUrl, bearerToken = remote.token) - } - - suspend fun send(name: String, text: String, contextId: String? = null): Task { - val message = Message(role = Role.USER, parts = listOf(TextPart(text = text)), contextId = contextId) - return clientFor(name).sendMessage(message, contextId) - } - - suspend fun agentCard(name: String): AgentCard = clientFor(name).agentCard() - - fun closeAll() { - clients.values.forEach { it.close() } - clients.clear() - } -} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoA2aHandler.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoA2aHandler.kt deleted file mode 100644 index 2766f3f..0000000 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoA2aHandler.kt +++ /dev/null @@ -1,21 +0,0 @@ -package pw.binom.agentik.standalone - -import pw.binom.a2a.model.Message -import pw.binom.a2a.model.Role -import pw.binom.a2a.model.TextPart -import pw.binom.a2a.server.AgentHandler - -/** - * A2A-обработчик-заглушка: эхоит входящее сообщение. - * Здесь позже будет реальный агент (LLM / инструменты). - */ -object EchoA2aHandler : AgentHandler { - override suspend fun handle(request: Message, contextId: String?): Message { - val text = request.parts.filterIsInstance().joinToString("") { it.text } - return Message( - role = Role.AGENT, - parts = listOf(TextPart("echo: $text")), - contextId = contextId, - ) - } -} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoAgent.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoAgent.kt deleted file mode 100644 index 1c059a7..0000000 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoAgent.kt +++ /dev/null @@ -1,33 +0,0 @@ -package pw.binom.agentik.standalone - -import pw.binom.agui.api.agent.Agent -import pw.binom.agui.api.event.BaseEvent -import pw.binom.agui.api.event.RunFinishedEvent -import pw.binom.agui.api.event.RunStartedEvent -import pw.binom.agui.api.event.TextMessageContentEvent -import pw.binom.agui.api.event.TextMessageEndEvent -import pw.binom.agui.api.event.TextMessageStartEvent -import pw.binom.agui.api.message.MessageRole -import pw.binom.agui.api.run.RunAgentInput -import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.flow - -/** - * Заглушка AG-UI агента: отвечает последовательностью событий - * RUN_STARTED -> TEXT_MESSAGE_* -> RUN_FINISHED. Пока — эхо последнего - * пользовательского сообщения. Дальше сюда зайдёт реальный LLM-агент. - */ -object EchoAgent : Agent { - override fun run(input: RunAgentInput): Flow = flow { - emit(RunStartedEvent(threadId = input.threadId, runId = input.runId)) - - val userText = input.messages.lastOrNull { it.role == MessageRole.USER }?.content.orEmpty() - val messageId = "msg-${input.runId}" - - emit(TextMessageStartEvent(messageId = messageId, role = MessageRole.ASSISTANT)) - emit(TextMessageContentEvent(messageId = messageId, delta = "echo: $userText")) - emit(TextMessageEndEvent(messageId = messageId)) - - emit(RunFinishedEvent(threadId = input.threadId, runId = input.runId)) - } -} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoProtoAgent.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoProtoAgent.kt deleted file mode 100644 index 71a6306..0000000 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/EchoProtoAgent.kt +++ /dev/null @@ -1,147 +0,0 @@ -package pw.binom.agentik.standalone - -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.Job -import kotlinx.coroutines.SupervisorJob -import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.MutableSharedFlow -import kotlinx.coroutines.flow.asSharedFlow -import kotlinx.coroutines.flow.flow -import kotlinx.coroutines.launch -import kotlinx.coroutines.sync.Mutex -import kotlinx.coroutines.sync.withLock -import pw.binom.agentik.proto.Agent -import pw.binom.agentik.proto.AgentEvent -import pw.binom.agentik.proto.Content -import pw.binom.agentik.proto.Conversation -import pw.binom.agentik.proto.Event -import pw.binom.agentik.proto.Message -import java.util.UUID -import java.util.concurrent.ConcurrentHashMap -import kotlin.time.Instant - -/** - * Минимальная in-memory реализация [Agent] из нашего :proto: эхо-агент, - * повторяет текст пользователя в [AppendText]. - * - * Не persistent, без очередей ходов и без реальной отмены — stub для проверки - * что HTTP-фасад `:server` корректно подключён к агенту и прокидывает все - * события/историю согласно контракту. - */ -class EchoProtoAgent(override val id: String = "echo-proto") : Agent { - - private val conversations = ConcurrentHashMap() - private val agentEvents = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) - - override fun createConversation(temp: Boolean): Conversation { - val c = EchoProtoConversation(isTemporal = temp) - conversations[c.id] = c - agentEvents.tryEmit( - AgentEvent.Created( - date = c.updatedAt, - conversationId = c.id, - ) - ) - return c - } - - override suspend fun getConversation(id: String): Conversation? = conversations[id] - - override suspend fun deleteConversation(id: String): Boolean { - val removed = conversations.remove(id) ?: return false - removed.close() - agentEvents.tryEmit(AgentEvent.Deleted(date = Instant.fromEpochMilliseconds(System.currentTimeMillis()), id = id)) - return true - } - - override suspend fun getConversations(offset: Int, limit: Int): List = - conversations.values.sortedByDescending { it.updatedAt }.drop(offset).take(limit) - - override fun events(after: Instant): Flow = flow { - agentEvents.asSharedFlow().collect { event -> - if (event.date > after) emit(event) - } - } -} - -internal class EchoProtoConversation( - override val id: String = UUID.randomUUID().toString(), - override val isTemporal: Boolean, -) : Conversation { - - override val isSupportImageInput: Boolean = false - override val isSupportImageOutput: Boolean = false - override var title: String? = null - private set - override var updatedAt: Instant = Instant.fromEpochMilliseconds(System.currentTimeMillis()) - private set - - private val messages = mutableListOf() - private val events = MutableSharedFlow(replay = 0, extraBufferCapacity = 64) - private val mutex = Mutex() - private var activeJob: Job? = null - private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - - override suspend fun rename(title: String) = mutex.withLock { - this.title = title - updatedAt = Instant.fromEpochMilliseconds(System.currentTimeMillis()) - } - - override suspend fun send(content: List) { - val userMessage = Message.UserMessage( - id = "um-${UUID.randomUUID()}", - content = content, - date = Instant.fromEpochMilliseconds(System.currentTimeMillis()), - ) - mutex.withLock { - messages.add(userMessage) - updatedAt = Instant.fromEpochMilliseconds(System.currentTimeMillis()) - } - - // stub-политика: прерываем предыдущий ход, запускаем новый. - activeJob?.cancel() - - activeJob = scope.launch { - delay(20) - val now = Instant.fromEpochMilliseconds(System.currentTimeMillis()) - val echo = content.joinToString(separator = " ") { c -> - when (c) { - is Content.Text -> c.body - is Content.Image -> "[image ${c.mime}, ${c.data.size}B]" - } - } - events.emit(Event.StartResponse(date = now, responseType = Event.ResponseType.TEXT)) - val body = "echo: $echo" - events.emit(Event.AppendText(date = now, body = body)) - val assistant = Message.AssistantMessage( - id = "am-${UUID.randomUUID()}", - content = listOf(Content.Text(body)), - date = Instant.fromEpochMilliseconds(System.currentTimeMillis()), - ) - mutex.withLock { messages.add(assistant) } - events.emit(Event.End(date = Instant.fromEpochMilliseconds(System.currentTimeMillis()))) - } - } - - override suspend fun interrupt() { - activeJob?.cancel() - activeJob = null - events.emit(Event.Interrupted(date = Instant.fromEpochMilliseconds(System.currentTimeMillis()))) - } - - override fun events(after: Instant): Flow = flow { - events.asSharedFlow().collect { event -> - if (event.date > after) emit(event) - } - } - - override suspend fun getMessages(after: Instant, offset: Int, limit: Int): List = - mutex.withLock { messages.asSequence().filter { it.date > after }.drop(offset).take(limit).toList() } - - override fun close() { - activeJob?.cancel() - scope.coroutineContext[Job]?.cancel() - } -} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt index 08db0fc..659084e 100644 --- a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/Main.kt @@ -5,22 +5,14 @@ import io.ktor.server.netty.Netty import io.ktor.server.response.respondText import io.ktor.server.routing.get import io.ktor.server.routing.routing -import kotlinx.coroutines.runBlocking -import pw.binom.a2a.server.A2AServer -import pw.binom.agentik.proto.Conversation -import pw.binom.agentik.proto.Message import pw.binom.agentik.server.agentikAgent -import pw.binom.agui.server.aguiAgent +import pw.binom.agentik.standalone.agent.ChatAgent +import pw.binom.agentik.standalone.llm.LlmConfig +import pw.binom.agentik.standalone.persistence.sqlite.SqliteStores /** * standalone-контейнер agentik: - * - AG-UI: встраиваемый Ktor (Netty), порт AGENTIK_PORT (default 8080) - * POST /agui -> text/event-stream (протокол AG-UI, агент [EchoAgent]) - * GET /health -> "ok" - * - A2A: встраиваемый Ktor (CIO), порт AGENTIK_A2A_PORT (default 8081) - * POST / -> JSON-RPC (message/send, tasks/get, tasks/cancel) - * GET /.well-known/agent-card.json - * - :server (proto): встраиваемый Ktor (Netty), порт AGENTIK_PORT (default 8080) — общий с AG-UI + * - :server (proto): встраиваемый Ktor (Netty), порт AGENTIK_PORT (default 8080) * POST /agentik/conversations -> ConversationSnapshot (201) * GET /agentik/conversations -> [ConversationSnapshot] * GET /agentik/conversations/{id} -> ConversationSnapshot @@ -31,36 +23,43 @@ import pw.binom.agui.server.aguiAgent * GET /agentik/conversations/{id}/messages -> [Message] * GET /agentik/conversations/{id}/events -> text/event-stream (SSE) * GET /agentik/events -> text/event-stream (SSE, Agent-level) + * GET /health -> "ok" * - * Для обращения к другим агентам: pw.binom.a2a.client.A2AClient.create(baseUrl, token). - * Для in-process вызова :server: pw.binom.agentik.client.AgentikAgent(id, baseUrl, httpClient). + * Хранилище — SQLite (env: AGENTIK_DB_PATH, default `./agentik.db`, `:memory:` для тестов). + * LLM — litert-openai (env: OPENAI_BASE_URL, OPENAI_API_KEY, OPENAI_MODEL). + * + * System prompt — AGENTIK_SYSTEM_PROMPT (default: встроенный `Ты полезный ассистент...`). */ fun main() { - val aguiPort = System.getenv("AGENTIK_PORT")?.toIntOrNull() ?: 8080 - val a2aPort = System.getenv("AGENTIK_A2A_PORT")?.toIntOrNull() ?: 8081 + val port = System.getenv("AGENTIK_PORT")?.toIntOrNull() ?: 8080 + val dbPath = System.getenv("AGENTIK_DB_PATH")?.takeIf { it.isNotBlank() } ?: "./agentik.db" - // A2A: agent <-> agent (CIO). Нестреляющий старт — свой event-loop. - val a2a = A2AServer.create(port = a2aPort, agentName = "agentik", handler = EchoA2aHandler) - runBlocking { a2a.start() } - println("A2A server -> http://localhost:$a2aPort/ (JSON-RPC: message/send, tasks/get, tasks/cancel)") + val llmConfig = LlmConfig.fromEnv() + val llm = llmConfig.createLlm() + val stores = SqliteStores.open(dbPath = dbPath) + val agent = ChatAgent( + id = "agentik", + stores = stores, + llm = llm, + llmConfig = llmConfig, + ) - // shared port: AG-UI (Netty) + :server (Netty) живут на 8080, A2A — на 8081. - val agui = - embeddedServer(Netty, port = aguiPort) { - routing { - get("/health") { call.respondText("ok") } - aguiAgent(EchoAgent, path = "/agui") - agentikAgent(EchoProtoAgent(), path = "/agentik") - } + val server = embeddedServer(Netty, port = port) { + routing { + get("/health") { call.respondText("ok") } + agentikAgent(agent, path = "/agentik") } - println("AG-UI -> http://localhost:$aguiPort/agui (SSE), /health") - println(":server proto -> http://localhost:$aguiPort/agentik/... (REST+SSE, агент [EchoProtoAgent])") - agui.start(wait = true) + } + println("agentik standalone listening on http://localhost:$port") + println(" GET /health") + println(" POST /agentik/conversations -> 201") + println(" GET /agentik/conversations/{id}/events -> SSE") + println(" storage: $dbPath") + println(" llm: ${llmConfig.backend} ${llmConfig.modelInfo()}") + Runtime.getRuntime().addShutdownHook(Thread { + agent.close() + stores.close() + llm.close() + }) + server.start(wait = true) } - -// References for IDE noise suppression when source not auto-imported. -@Suppress("unused") -private val keepReferences: Array> = arrayOf( - Conversation::class.java, - Message::class.java, -) 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 new file mode 100644 index 0000000..2908001 --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatAgent.kt @@ -0,0 +1,121 @@ +package pw.binom.agentik.standalone.agent + +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import pw.binom.agentik.proto.Agent as ProtoAgent +import pw.binom.agentik.proto.AgentEvent +import pw.binom.agentik.proto.Conversation as ProtoConversation +import pw.binom.agentik.standalone.llm.LlmConfig +import pw.binom.agentik.standalone.persistence.ConversationRecord +import pw.binom.agentik.standalone.persistence.WorkingMemoryEntry +import pw.binom.agentik.standalone.persistence.sqlite.SqliteStores +import pw.binom.litert.LiteLlm +import kotlin.time.Instant + +/** + * Stateful [ProtoAgent] на базе SQLite (история + working memory) и + * [LiteLlm] (универсальный LLM-контракт). + * + * Создаёт [ChatConversation] — те самые stateful диалоги, которые + * хранят свой собственный LiteConversation в ОЗУ (для скорости) и + * логируют в SQLite (для долговечности). + * + * Один [LiteLlm] шарится между всеми беседами агента. + */ +class ChatAgent( + override val id: String, + private val stores: SqliteStores, + private val llm: LiteLlm, + private val llmConfig: LlmConfig, +) : ProtoAgent, AutoCloseable { + + private val agentEvents = MutableSharedFlow( + extraBufferCapacity = 64, + ) + + /** Защищает карту живых диалогов. */ + private val liveLock = Mutex() + private val live: MutableMap = HashMap() + + /** Live-подписка на события уровня агента (создание/удаление/переименование). */ + override fun events(after: Instant): Flow { + // Реализация событийной шины упрощённая: возвращаем общий поток. + // Фильтр по `after` не делаем — для v1 после-семантика не нужна + // (см. Memory #3704: replay-free, бэкфилл через getConversations/getConversation). + return agentEvents.asSharedFlow() + } + + override fun createConversation(temp: Boolean): ProtoConversation { + val now = now() + val id = "conv-${java.util.UUID.randomUUID()}" + val rec = ConversationRecord( + id = id, + title = null, + isTemporal = temp, + createdAt = now, + updatedAt = now, + ) + // Temp-беседы не пишем в SQLite — они живут только в RAM-карте `live` + // и не переживают рестарт агента (см. Memory #3709). + if (!temp) { + runBlocking { + stores.conversations.upsert(rec) + stores.workingMemory.append( + conversationId = id, + entry = WorkingMemoryEntry.System(text = llmConfig.systemPrompt), + now = now, + ) + } + } + val conv = ChatConversation(record = rec, stores = stores, llm = llm, systemPrompt = llmConfig.systemPrompt, foldSystemIntoFirstUser = llmConfig.foldSystemIntoFirstUser) + runBlocking { + liveLock.withLock { live[conv.id] = conv } + } + agentEvents.tryEmit(AgentEvent.Created(date = now(), conversationId = conv.id)) + return conv + } + + override suspend fun getConversation(id: String): ProtoConversation? { + liveLock.withLock { live[id] }?.let { if (!it.isClosed) return it } + val rec = stores.conversations.get(id) ?: return null + return ChatConversation(record = rec, stores = stores, llm = llm, systemPrompt = llmConfig.systemPrompt, foldSystemIntoFirstUser = llmConfig.foldSystemIntoFirstUser).also { + liveLock.withLock { live[id] = it } + } + } + + override suspend fun deleteConversation(id: String): Boolean { + val conv = liveLock.withLock { live.remove(id) } + conv?.close() + val ok = stores.conversations.delete(id) + if (ok) agentEvents.tryEmit(AgentEvent.Deleted(date = now(), id = id)) + return ok + } + + override suspend fun getConversations(offset: Int, limit: Int): List = + stores.conversations.list(offset = offset, limit = limit).map { rec -> + liveLock.withLock { live[rec.id] } + ?: ChatConversation(record = rec, stores = stores, llm = llm, systemPrompt = llmConfig.systemPrompt).also { + liveLock.withLock { live[rec.id] = it } + } + } + + override fun close() { + runBlocking { + liveLock.withLock { + live.values.forEach { it.close() } + live.clear() + } + } + runCatching { llm.close() } + } + + internal fun unregister(id: String) { + runBlocking { liveLock.withLock { live.remove(id) } } + } + + private fun now(): Instant = Instant.fromEpochMilliseconds(System.currentTimeMillis()) +} 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 new file mode 100644 index 0000000..2632efd --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/agent/ChatConversation.kt @@ -0,0 +1,353 @@ +package pw.binom.agentik.standalone.agent + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.transformWhile +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import pw.binom.agentik.proto.Content as ProtoContent +import pw.binom.agentik.proto.Conversation as ProtoConversation +import pw.binom.agentik.proto.Event as ProtoEvent +import pw.binom.agentik.proto.Message as ProtoMessage +import pw.binom.agentik.standalone.persistence.Content +import pw.binom.agentik.standalone.persistence.ConversationRecord +import pw.binom.agentik.standalone.persistence.ConversationStore +import pw.binom.agentik.standalone.persistence.MessageRecord +import pw.binom.agentik.standalone.persistence.MessageStore +import pw.binom.agentik.standalone.persistence.WorkingMemoryEntry +import pw.binom.agentik.standalone.persistence.WorkingMemoryStore +import pw.binom.agentik.standalone.persistence.sqlite.SqliteStores +import pw.binom.litert.LiteContentPart +import pw.binom.litert.LiteConversation +import pw.binom.litert.LiteConversationConfig +import pw.binom.litert.LiteLlm +import pw.binom.litert.LiteMessage +import pw.binom.litert.LiteRole +import kotlin.time.Instant + +/** + * Stateful [pw.binom.agentik.proto.Conversation] поверх [LiteLlm]. + * + * Контракт: один [LiteConversation] живёт столько же, сколько [ChatConversation]. + * Это требование абстракции LiteLlm — у реализаций внутри `LiteConversation` хранится + * KV-cache движка (LiteRT-LM) либо инкрементальная история (litert-openai). Пересоздание + * на каждый send сломало бы и то, и другое. + * + * На каждый [send]: + * 1. Записывает user-сообщение в audit log (`message`) + working memory. + * 2. Создаёт [LiteConversation] **один раз** при первом send (initial messages = + * системный промпт + пары User/Assistant из working memory, **без** только что + * добавленного user-сообщения — мы передадим его через sendStreamContents). + * 3. Стримит ответ [LiteDelta] в [sendStreamContents] — эмитит [ProtoEvent.AppendText]. + * Движок сам добавляет user/assistant к своей внутренней истории. + * 4. По завершении записывает AssistantMessage в audit + working memory. + * + * Если LiteConversation упал при инициализации или во время send — закрываем его, + * следующий send попробует создать заново. После [interrupt] LiteConversation жив; + * отменяется только текущий send. + */ +class ChatConversation( + record: ConversationRecord, + private val stores: SqliteStores, + private val llm: LiteLlm, + private val systemPrompt: String, + private val foldSystemIntoFirstUser: Boolean = false, +) : ProtoConversation, AutoCloseable { + + private var record: ConversationRecord = record + + override val id: String get() = record.id + override val isSupportImageInput: Boolean get() = false + override val isSupportImageOutput: Boolean get() = false + override val isTemporal: Boolean get() = record.isTemporal + override val title: String? get() = record.title + override val updatedAt: Instant get() = record.updatedAt + + private val conversationStore: ConversationStore get() = stores.conversations + private val messageStore: MessageStore get() = stores.messages + private val workingMemory: WorkingMemoryStore get() = stores.workingMemory + + private val events = MutableSharedFlow( + replay = 0, + extraBufferCapacity = 128, + onBufferOverflow = BufferOverflow.DROP_OLDEST, + ) + + private val turnLock = Mutex() + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + + @Volatile + private var liteConv: LiteConversation? = null + @Volatile + private var activeTurn: Job? = null + @Volatile + private var closed = false + + internal val isClosed: Boolean get() = closed + + override suspend fun rename(title: String) { + val newRecord = conversationStore.rename(id, title)?.let { ts -> + record.copy(title = title, updatedAt = ts) + } ?: record.copy(title = title) + record = newRecord + } + + override suspend fun send(content: List) { + check(!closed) { "Conversation closed: $id" } + val turnStarted = now() + val userMessageId = newId("msg") + + val userRecord = MessageRecord.UserMessage( + id = userMessageId, + conversationId = id, + content = content.map { it.toStorage() }, + createdAt = turnStarted, + ) + + if (!record.isTemporal) { + messageStore.append(userRecord) + workingMemory.append( + conversationId = id, + entry = WorkingMemoryEntry.User( + sourceMessageId = userMessageId, + content = userRecord.content, + ), + now = turnStarted, + ) + } + + activeTurn = scope.launch { + turnLock.withLock { + runTurn(userRecord, turnStarted) + } + } + activeTurn?.join() + } + + override suspend fun interrupt() { + runCatching { liteConv?.cancel() } + activeTurn?.cancel() + emitEvent(ProtoEvent.Interrupted(date = now())) + } + + override fun events(after: Instant): Flow = + events.asSharedFlow() + + override suspend fun getMessages(after: Instant, offset: Int, limit: Int): List = + messageStore.list(conversationId = id, after = after, offset = offset, limit = limit) + .map { it.toProto() } + + override fun close() { + if (closed) return + closed = true + runCatching { liteConv?.close() } + runCatching { activeTurn?.cancel() } + scope.cancel() + } + + private suspend fun runTurn(userRecord: MessageRecord.UserMessage, turnStarted: Instant) { + emitEvent(ProtoEvent.StartReasoning(date = turnStarted)) + emitEvent(ProtoEvent.StartResponse(date = now(), responseType = ProtoEvent.ResponseType.TEXT)) + + val parts = userRecord.content.mapNotNull { c -> + when (c) { + is Content.Text -> LiteContentPart.Text(c.body) + is Content.Image -> { + System.err.println("[agentik] dropping image input (v1 text-only): mime=${c.mime}, ${c.data.size} bytes") + null + } + } + } + + if (parts.isEmpty()) { + emitEvent(ProtoEvent.Error(date = now(), message = "Empty user input (no text content)")) + emitEvent(ProtoEvent.End(date = now())) + return + } + + val liteConv = try { + getOrCreateLiteConversation(excludeUserSourceId = if (record.isTemporal) null else userRecord.id) + } catch (e: Throwable) { + this.liteConv = null + emitEvent(ProtoEvent.Error(date = now(), message = e.message ?: "LiteConversation init failed")) + emitEvent(ProtoEvent.End(date = now())) + return + } + + val reply = StringBuilder() + try { + // Некоторые бэкенды (litert-google-jvm 0.16.1) не закрывают стрим после `isDone = true`, + // поэтому заворачиваем в transformWhile: видим isDone → отдаём дельту и завершаем flow. + liteConv.sendStreamContents(parts).transformWhile { delta -> + emit(delta) + !delta.isDone + }.collect { delta -> + if (delta.text.isNotEmpty()) { + reply.append(delta.text) + emitEvent(ProtoEvent.AppendText(date = now(), body = delta.text)) + } + } + } catch (e: kotlinx.coroutines.CancellationException) { + throw e + } catch (e: Throwable) { + emitEvent(ProtoEvent.Error(date = now(), message = e.message ?: e.javaClass.simpleName)) + emitEvent(ProtoEvent.End(date = now())) + return + } + + 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, + ) + + 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) + } + + emitEvent(ProtoEvent.End(date = assistantAt)) + } + + /** + * Возвращает существующий [LiteConversation] или создаёт новый, инициализированный + * системным промптом и прошлыми User/Assistant из working memory. + * + * [excludeUserSourceId] — если задан, исключает одну строку (свежее user-сообщение, + * уже записанное в audit + working memory, но ещё не отправленное в LLM — мы отдадим + * его через [LiteConversation.sendStreamContents]). Это предотвращает дублирование + * "user → user" в LiteConversation history. + * + * Если [foldSystemIntoFirstUser] — система не передаётся как `systemInstruction`, + * а фолдится в первое user-сообщение (нужно для Gemma-3 шаблона LiteRT-LM 0.16.1). + */ + private suspend fun getOrCreateLiteConversation(excludeUserSourceId: String? = null): LiteConversation { + liteConv?.let { return it } + + val wm = if (record.isTemporal) emptyList() else workingMemory.list(id) + val resolvedSystemPrompt = if (record.isTemporal) systemPrompt else wm + .firstOrNull { it.entry is WorkingMemoryEntry.System } + ?.let { (it.entry as WorkingMemoryEntry.System).text } + ?: systemPrompt + + // Берём пары (User, Assistant) из прошлой истории, исключая свежее user-сообщение, + // которое отправим через sendStreamContents. + val pastTurns = if (record.isTemporal) emptyList() else wm + .filter { row -> + val isUserOrAssistant = row.entry is WorkingMemoryEntry.User || row.entry is WorkingMemoryEntry.Assistant + val isPendingUser = excludeUserSourceId != null && row.sourceMessageId == excludeUserSourceId + isUserOrAssistant && !isPendingUser + } + .map { row -> + when (val e = row.entry) { + is WorkingMemoryEntry.User -> LiteMessage(LiteRole.USER, e.content.toLiteContents()) + is WorkingMemoryEntry.Assistant -> LiteMessage(LiteRole.MODEL, e.content.toLiteContents()) + else -> error("unreachable") + } + } + + val config = LiteConversationConfig( + systemInstruction = if (foldSystemIntoFirstUser) null else resolvedSystemPrompt.takeIf { it.isNotBlank() }, + initialMessages = if (foldSystemIntoFirstUser) { + val firstUser = pastTurns.firstOrNull { it.role == LiteRole.USER } + if (firstUser != null && resolvedSystemPrompt.isNotBlank()) { + val folded = LiteMessage( + role = LiteRole.USER, + contents = listOf(LiteContentPart.Text(resolvedSystemPrompt + "\n\n")) + firstUser.contents, + ) + listOf(folded) + pastTurns.drop(1) + } else { + pastTurns + } + } else { + pastTurns + }, + ) + + return llm.createConversation(config).also { liteConv = it } + } + + private fun emitEvent(event: ProtoEvent) { + events.tryEmit(event) + } + + private fun now(): Instant = + Instant.fromEpochMilliseconds(System.currentTimeMillis()) + + private fun newId(prefix: String): String = "$prefix-${java.util.UUID.randomUUID()}" +} + +internal fun List.toLiteContents(): List = map { it.toLite() } + +internal fun Content.toLite(): LiteContentPart = when (this) { + is Content.Text -> LiteContentPart.Text(body) + is Content.Image -> LiteContentPart.Image(data, mime) +} + +private fun Content.toProto(): ProtoContent = when (this) { + is Content.Text -> ProtoContent.Text(body = body) + is Content.Image -> ProtoContent.Image(data = data, mime = mime) +} + +internal fun ProtoContent.toStorage(): Content = when (this) { + is ProtoContent.Text -> Content.Text(body) + is ProtoContent.Image -> Content.Image(data, mime) +} + +internal fun MessageRecord.toProto(): ProtoMessage = when (this) { + is MessageRecord.UserMessage -> ProtoMessage.UserMessage( + id = id, + date = createdAt, + content = content.map { it.toProto() }, + ) + is MessageRecord.AssistantMessage -> ProtoMessage.AssistantMessage( + id = id, + date = createdAt, + content = content.map { it.toProto() }, + ) + is MessageRecord.ToolCall -> ProtoMessage.ToolCall( + id = id, + date = createdAt, + title = toolTitle, + toolName = toolName, + toolArgs = toolArgsJson, + ) + is MessageRecord.ToolResult -> ProtoMessage.ToolResult( + id = id, + date = createdAt, + result = result, + ) + // Синтетические строки рабочей памяти: отдаём клиенту как текст ассистента + // (Summary — это результат суммаризации, по форме — ответ модели) или юзера (System). + is MessageRecord.Summary -> ProtoMessage.AssistantMessage( + id = id, + date = createdAt, + content = listOf(ProtoContent.Text(body = text)), + ) + is MessageRecord.System -> ProtoMessage.UserMessage( + id = id, + date = createdAt, + content = listOf(ProtoContent.Text(body = text)), + ) +} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/GoogleLiteLlmJvm.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/GoogleLiteLlmJvm.kt new file mode 100644 index 0000000..ff9b224 --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/GoogleLiteLlmJvm.kt @@ -0,0 +1,46 @@ +package pw.binom.agentik.standalone.llm + +import pw.binom.litert.LiteConfig +import pw.binom.litert.LiteLlm + +/** + * Factory для движка Google LiteRT-LM на JVM. + * + * `litert-google` опубликован только как Android AAR (с .so внутри), поэтому на JVM + * приходится вручную подгружать JNI-библиотеки LiteRT-LM, прежде чем инстанциировать + * движок. Эта функция: + * + * 1. Резолвит `pw.binom.litert.google.GoogleLiteLlm` через reflection. + * 2. Если процесс уже загрузил нативные библиотеки (`-Djava.library.path`) — успешно + * создаёт движок. + * 3. Если нет — кидает `IllegalStateException` с инструкцией по настройке. + */ +fun googleLiteLlmJvm(config: LiteConfig): LiteLlm { + val cls = try { + Class.forName("pw.binom.litert.google.GoogleLiteLlm") + } catch (e: ClassNotFoundException) { + error( + "litert-google classes are not on the classpath. " + + "Add 'pw.binom.litert:litert-google-android:6' as a runtime dependency " + + "to use the GOOGLE backend on JVM." + ) + } + val ctor = cls.constructors.firstOrNull { it.parameterCount == 1 } + ?: error("pw.binom.litert.google.GoogleLiteLlm constructor not found") + val engine = try { + ctor.newInstance(config) + } catch (e: UnsatisfiedLinkError) { + throw IllegalStateException( + "LiteRT-LM native libraries are not loaded. " + + "Extract .so/.dylib/.dll from litertlm-android-0.16.1.aar and pass " + + "-Djava.library.path=, or build :standalone for the androidJvm target.", + e, + ) + } catch (e: java.lang.reflect.InvocationTargetException) { + throw e.targetException ?: e + } + @Suppress("UNCHECKED_CAST") + return engine as LiteLlm +} + +private fun error(message: String): Nothing = throw IllegalStateException(message) diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/LlmConfig.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/LlmConfig.kt new file mode 100644 index 0000000..8b1ef3d --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/llm/LlmConfig.kt @@ -0,0 +1,93 @@ +package pw.binom.agentik.standalone.llm + +import pw.binom.litert.LiteBackend +import pw.binom.litert.LiteConfig +import pw.binom.litert.LiteExperimental +import pw.binom.litert.LiteLlm +import pw.binom.litert.openai.OpenAiConfig as LitertOpenAiConfig +import pw.binom.litert.openai.openAiLiteLlm + +data class LlmConfig( + val backend: LlmBackend, + val systemPrompt: String, + val openai: LitertOpenAiConfig? = null, + val google: GoogleConfig? = null, + val foldSystemIntoFirstUser: Boolean = backend == LlmBackend.GOOGLE, +) { + init { + when (backend) { + LlmBackend.OPENAI -> require(openai != null) { "OPENAI backend requires openai config" } + LlmBackend.GOOGLE -> require(google != null) { "GOOGLE backend requires google config" } + } + } + + fun createLlm(): LiteLlm = when (backend) { + LlmBackend.OPENAI -> openAiLiteLlm(checkNotNull(openai)) + LlmBackend.GOOGLE -> googleLiteLlmJvm(google!!.toLiteConfig()) + } + + fun modelInfo(): String = when (backend) { + LlmBackend.OPENAI -> "${checkNotNull(openai).model} @ ${checkNotNull(openai).baseUrl}" + LlmBackend.GOOGLE -> "${checkNotNull(google).modelPath}" + } + + companion object { + const val DEFAULT_SYSTEM_PROMPT: String = "Ты полезный ассистент. Отвечай кратко и по делу." + + fun fromEnv(env: (String) -> String? = System::getenv): LlmConfig { + val backend = LlmBackend.parse(env("AGENTIK_LLM_BACKEND")) + val systemPromptRaw = env("AGENTIK_SYSTEM_PROMPT") + val systemPrompt = if (systemPromptRaw.isNullOrBlank()) DEFAULT_SYSTEM_PROMPT else systemPromptRaw + return when (backend) { + LlmBackend.OPENAI -> { + val openai = LitertOpenAiConfig( + baseUrl = requireEnv(env, "OPENAI_BASE_URL"), + apiKey = requireEnv(env, "OPENAI_API_KEY"), + model = requireEnv(env, "OPENAI_MODEL"), + ) + LlmConfig(backend, systemPrompt, openai = openai) + } + LlmBackend.GOOGLE -> { + val google = GoogleConfig( + modelPath = requireEnv(env, "AGENTIK_GOOGLE_MODEL_PATH"), + cacheDir = env("AGENTIK_GOOGLE_CACHE_DIR"), + threads = env("AGENTIK_GOOGLE_THREADS")?.toInt(), + ) + LlmConfig(backend, systemPrompt, google = google) + } + } + } + + private fun requireEnv(env: (String) -> String?, name: String): String = + env(name) ?: error("Required env var $name is not set") + } +} + +enum class LlmBackend { + OPENAI, + GOOGLE; + + companion object { + fun parse(raw: String?): LlmBackend = when (raw?.lowercase()) { + null, "", "openai" -> OPENAI + "google", "litert", "litert-google" -> GOOGLE + else -> error("Unknown LLM backend '$raw', expected 'openai' or 'google'") + } + } +} + +data class GoogleConfig( + val modelPath: String, + val cacheDir: String? = null, + val threads: Int? = null, +) { + fun toLiteConfig(): LiteConfig = LiteConfig( + modelPath = modelPath, + cacheDir = cacheDir ?: "", + threads = threads, + backend = LiteBackend.CPU, + visionBackend = null, + audioBackend = null, + experimental = LiteExperimental(false, emptyMap()), + ) +} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteConversationStore.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteConversationStore.kt new file mode 100644 index 0000000..c1b0950 --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteConversationStore.kt @@ -0,0 +1,80 @@ +package pw.binom.agentik.standalone.persistence.sqlite + +import kotlin.time.Instant +import pw.binom.agentik.standalone.persistence.ConversationRecord +import pw.binom.agentik.standalone.persistence.ConversationStore + +/** + * SQLite-реализация [ConversationStore]. + */ +class SqliteConversationStore(private val db: AgentikDatabase) : ConversationStore { + + private val q get() = db.conversationQueries + + override suspend fun upsert(record: ConversationRecord) { + // SQLite-конфликт по PRIMARY KEY → сначала пробуем insert, при ошибке → update + val existing = q.getById(record.id).executeAsOneOrNull() + if (existing == null) { + q.insert( + id = record.id, + title = record.title, + is_temporal = if (record.isTemporal) 1L else 0L, + created_at = record.createdAt.toEpochMilliseconds(), + updated_at = record.updatedAt.toEpochMilliseconds(), + ) + } else { + q.update( + title = record.title, + is_temporal = if (record.isTemporal) 1L else 0L, + updated_at = record.updatedAt.toEpochMilliseconds(), + id = record.id, + ) + } + } + + override suspend fun get(id: String): ConversationRecord? { + val row = q.getById(id).executeAsOneOrNull() ?: return null + return row.toRecord() + } + + override suspend fun delete(id: String): Boolean { + // Проверяем существование ДО удаления — иначе пустой `deleteById` вернёт + // «успех», и get() == null будет true, хотя диалога и не было. + val existed = q.getById(id).executeAsOneOrNull() != null + if (!existed) return false + db.transaction { + db.messageQueries.deleteByConversation(id) + db.workingMemoryQueries.clearByConversation(id) + q.deleteById(id) + } + return true + } + + override suspend fun list(offset: Int, limit: Int): List = + q.list(limit = limit.toLong(), offset = offset.toLong()) + .executeAsList() + .map { it.toRecord() } + + override suspend fun rename(id: String, title: String?): Instant? { + val now = Instant.fromEpochMilliseconds(System.currentTimeMillis()) + q.rename(title = title, updated_at = now.toEpochMilliseconds(), id = id) + val ts = q.getUpdatedAt(id).executeAsOneOrNull() ?: return null + return Instant.fromEpochMilliseconds(ts) + } + + override suspend fun touch(id: String, now: Instant) { + q.touch(updated_at = now.toEpochMilliseconds(), id = id) + } + + override fun close() { + // driver закрывается во внешнем SqliteStores + } +} + +private fun Conversation.toRecord(): ConversationRecord = ConversationRecord( + id = id, + title = title, + isTemporal = is_temporal != 0L, + createdAt = Instant.fromEpochMilliseconds(created_at), + updatedAt = Instant.fromEpochMilliseconds(updated_at), +) diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteMessageStore.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteMessageStore.kt new file mode 100644 index 0000000..c63f909 --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteMessageStore.kt @@ -0,0 +1,115 @@ +package pw.binom.agentik.standalone.persistence.sqlite + +import kotlinx.serialization.json.Json +import pw.binom.agentik.standalone.persistence.MessageRecord +import pw.binom.agentik.standalone.persistence.MessageStore +import pw.binom.agentik.standalone.persistence.decodeBodyPayload +import pw.binom.agentik.standalone.persistence.encodeBodyPayload +import kotlin.time.Instant + +/** + * SQLite-реализация [MessageStore] (append-only audit log). + * + * `payloadJson` хранит JSON-сериализованные kind-specific поля. Для + * `user`/`assistant` это `List` (см. [encodeBodyPayload]). + * Для `tool_call`/`tool_result` payload хранит JSON-объект + * (см. [CallPayload]/[ResultPayload]). + * + * SQLDelight сохраняет snake_case в сгенерированной data class (`Message`), + * поэтому обращаемся через `conversation_id`, `payload_json`, `created_at`. + */ +class SqliteMessageStore(private val db: AgentikDatabase) : MessageStore { + + private val q get() = db.messageQueries + + override suspend fun append(record: MessageRecord) { + val (kind, payload) = encodeRecord(record) + q.insert( + conversation_id = record.conversationId, + kind = kind, + payload_json = payload, + created_at = record.createdAt.toEpochMilliseconds(), + id = record.id, + ) + } + + override suspend fun list( + conversationId: String, + after: Instant, + offset: Int, + limit: Int, + ): List = q.listAfter( + conversation_id = conversationId, + created_at = after.toEpochMilliseconds(), + limit = limit.toLong(), + offset = offset.toLong(), + ).executeAsList().map { it.toRecord() } + + override suspend fun listAll(conversationId: String): List = + q.listByConversationAll(conversation_id = conversationId).executeAsList().map { it.toRecord() } + + override fun close() {} +} + +private fun encodeRecord(record: MessageRecord): Pair = when (record) { + is MessageRecord.UserMessage -> "user" to encodeBodyPayload(record.content) + is MessageRecord.AssistantMessage -> "assistant" to encodeBodyPayload(record.content) + is MessageRecord.ToolCall -> "tool_call" to Json.encodeToString( + CallPayload.serializer(), + CallPayload(name = record.toolName, title = record.toolTitle, argsJson = record.toolArgsJson), + ) + is MessageRecord.ToolResult -> "tool_result" to Json.encodeToString( + ResultPayload.serializer(), + ResultPayload(toolCallId = record.toolCallId, result = record.result), + ) + is MessageRecord.Summary, + is MessageRecord.System -> error("Summary/System — synthetic, cannot append to audit log") +} + +@kotlinx.serialization.Serializable +internal data class CallPayload(val name: String, val title: String?, val argsJson: String) + +@kotlinx.serialization.Serializable +internal data class ResultPayload(val toolCallId: String, val result: String?) + +private fun Message.toRecord(): MessageRecord { + val id = id + val convId = conversation_id + val createdAt = Instant.fromEpochMilliseconds(created_at) + return when (kind) { + "user" -> MessageRecord.UserMessage( + id = id, + conversationId = convId, + content = decodeBodyPayload(payload_json), + createdAt = createdAt, + ) + "assistant" -> MessageRecord.AssistantMessage( + id = id, + conversationId = convId, + content = decodeBodyPayload(payload_json), + createdAt = createdAt, + ) + "tool_call" -> { + val p = Json.decodeFromString(CallPayload.serializer(), payload_json) + MessageRecord.ToolCall( + id = id, + conversationId = convId, + toolName = p.name, + toolTitle = p.title, + toolArgsJson = p.argsJson, + createdAt = createdAt, + ) + } + "tool_result" -> { + val p = Json.decodeFromString(ResultPayload.serializer(), payload_json) + MessageRecord.ToolResult( + id = id, + conversationId = convId, + toolCallId = p.toolCallId, + result = p.result, + createdAt = createdAt, + ) + } + else -> error("Unknown message kind in audit log: $kind") + } +} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteStores.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteStores.kt new file mode 100644 index 0000000..814a059 --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteStores.kt @@ -0,0 +1,73 @@ +package pw.binom.agentik.standalone.persistence.sqlite + +import app.cash.sqldelight.db.QueryResult +import app.cash.sqldelight.db.SqlDriver +import app.cash.sqldelight.driver.jdbc.sqlite.JdbcSqliteDriver +import pw.binom.agentik.standalone.persistence.ConversationStore +import pw.binom.agentik.standalone.persistence.MessageStore +import pw.binom.agentik.standalone.persistence.WorkingMemoryStore + +/** + * Корневой объект SQLite-слоя: держит [SqlDriver] и три [WorkingMemoryStore]/[MessageStore]/[ConversationStore]. + * Закрывается вместе с приложением. + */ +class SqliteStores private constructor( + val driver: SqlDriver, + val conversations: ConversationStore, + val messages: MessageStore, + val workingMemory: WorkingMemoryStore, +) : AutoCloseable { + + override fun close() { + conversations.close() + messages.close() + workingMemory.close() + driver.close() + } + + companion object { + + /** Открыть/создать БД по пути `dbPath` (например, `"./agentik.db"` или абсолютный путь). */ + fun open(dbPath: String): SqliteStores { + val driver = JdbcSqliteDriver("jdbc:sqlite:$dbPath") + createSchema(driver) + val db = AgentikDatabase(driver) + return SqliteStores( + driver = driver, + conversations = SqliteConversationStore(db), + messages = SqliteMessageStore(db), + workingMemory = SqliteWorkingMemoryStore(db), + ) + } + + /** Открыть/создать БД в памяти (для тестов). */ + fun inMemory(): SqliteStores { + val driver = JdbcSqliteDriver(JdbcSqliteDriver.IN_MEMORY) + createSchema(driver) + val db = AgentikDatabase(driver) + return SqliteStores( + driver = driver, + conversations = SqliteConversationStore(db), + messages = SqliteMessageStore(db), + workingMemory = SqliteWorkingMemoryStore(db), + ) + } + + private fun createSchema(driver: SqlDriver) { + // Если таблица `conversation` уже есть — БД уже инициализирована, + // просто пропускаем create (иначе CREATE TABLE упадёт на дубликате). + val existing = driver.executeQuery( + identifier = null, + sql = "SELECT name FROM sqlite_master WHERE type='table' AND name='conversation'", + mapper = { cursor -> + QueryResult.Value( + if (cursor.next().value) cursor.getString(0) else null, + ) + }, + parameters = 0, + ).value + if (existing != null) return + AgentikDatabase.Schema.create(driver) + } + } +} diff --git a/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteWorkingMemoryStore.kt b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteWorkingMemoryStore.kt new file mode 100644 index 0000000..d27465c --- /dev/null +++ b/standalone/src/jvmMain/kotlin/pw/binom/agentik/standalone/persistence/sqlite/SqliteWorkingMemoryStore.kt @@ -0,0 +1,70 @@ +package pw.binom.agentik.standalone.persistence.sqlite + +import kotlinx.serialization.json.Json +import pw.binom.agentik.standalone.persistence.WorkingMemoryEntry +import pw.binom.agentik.standalone.persistence.WorkingMemoryRow +import pw.binom.agentik.standalone.persistence.WorkingMemoryStore +import kotlin.time.Instant + +/** + * SQLite-реализация [WorkingMemoryStore] (мутируемый LLM-контекст). + * + * Строки упорядочены по `order_idx ASC`. Compact — атомарная операция: + * DELETE строк `>= dropFromOrderIdx` (без summary в v1). + */ +class SqliteWorkingMemoryStore(private val db: AgentikDatabase) : WorkingMemoryStore { + + private val q get() = db.workingMemoryQueries + private val json = Json { ignoreUnknownKeys = true } + + override suspend fun append(conversationId: String, entry: WorkingMemoryEntry, now: Instant) { + val newIdx = (q.maxOrderIdx(conversationId).executeAsOne()) + 1 + q.insert( + id = newId(), + conversation_id = conversationId, + order_idx = newIdx, + source_message_id = entry.sourceMessageId, + kind = entryKind(entry), + payload_json = json.encodeToString(WorkingMemoryEntry.serializer(), entry), + created_at = now.toEpochMilliseconds(), + ) + } + + override suspend fun list(conversationId: String): List = + q.listByConversation(conversationId).executeAsList().map { it.toRow() } + + override suspend fun clear(conversationId: String) { + q.clearByConversation(conversationId) + } + + override suspend fun compact(dropFromOrderIdx: Long, conversationId: String): Long { + var newMax = 0L + db.transaction { + q.compactDelete(conversation_id = conversationId, order_idx = dropFromOrderIdx) + newMax = q.maxOrderIdx(conversationId).executeAsOne() + } + return newMax + } + + override fun close() {} +} + +private fun entryKind(e: WorkingMemoryEntry): String = when (e) { + is WorkingMemoryEntry.System -> "system" + is WorkingMemoryEntry.User -> "user" + is WorkingMemoryEntry.Assistant -> "assistant" +} + +private fun Working_memory.toRow(): WorkingMemoryRow { + val entry = Json.decodeFromString(WorkingMemoryEntry.serializer(), payload_json) + return WorkingMemoryRow( + id = id, + conversationId = conversation_id, + orderIdx = order_idx, + sourceMessageId = source_message_id, + entry = entry, + createdAt = Instant.fromEpochMilliseconds(created_at), + ) +} + +private fun newId(): String = "wm-${java.util.UUID.randomUUID()}" diff --git a/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Conversation.sq b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Conversation.sq new file mode 100644 index 0000000..19f96b3 --- /dev/null +++ b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Conversation.sq @@ -0,0 +1,42 @@ +CREATE TABLE conversation ( + id TEXT NOT NULL PRIMARY KEY, + title TEXT, + is_temporal INTEGER NOT NULL DEFAULT 0, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL +); + +CREATE INDEX idx_conv_updated ON conversation(updated_at DESC); + +insert: +INSERT INTO conversation (id, title, is_temporal, created_at, updated_at) +VALUES (?, ?, ?, ?, ?); + +update: +UPDATE conversation SET title = ?, is_temporal = ?, updated_at = ? +WHERE id = ?; + +getById: +SELECT * FROM conversation WHERE id = ?; + +list: +SELECT * FROM conversation +WHERE is_temporal = 0 +ORDER BY updated_at DESC +LIMIT :limit OFFSET :offset; + +listAll: +SELECT * FROM conversation +ORDER BY updated_at DESC; + +deleteById: +DELETE FROM conversation WHERE id = ?; + +rename: +UPDATE conversation SET title = ?, updated_at = ? WHERE id = ?; + +getUpdatedAt: +SELECT updated_at FROM conversation WHERE id = ?; + +touch: +UPDATE conversation SET updated_at = ? WHERE id = ?; diff --git a/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Message.sq b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Message.sq new file mode 100644 index 0000000..4bbcc77 --- /dev/null +++ b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/Message.sq @@ -0,0 +1,33 @@ +CREATE TABLE message ( + id TEXT NOT NULL PRIMARY KEY, + conversation_id TEXT NOT NULL, + kind TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at INTEGER NOT NULL +); + +CREATE INDEX idx_msg_conv ON message(conversation_id, created_at); + +insert: +INSERT INTO message (id, conversation_id, kind, payload_json, created_at) +VALUES (?, ?, ?, ?, ?); + +listByConversation: +SELECT * FROM message +WHERE conversation_id = ? +ORDER BY created_at ASC, id ASC +LIMIT :limit OFFSET :offset; + +listByConversationAll: +SELECT * FROM message +WHERE conversation_id = ? +ORDER BY created_at ASC, id ASC; + +listAfter: +SELECT * FROM message +WHERE conversation_id = ? AND created_at > ? +ORDER BY created_at ASC, id ASC +LIMIT :limit OFFSET :offset; + +deleteByConversation: +DELETE FROM message WHERE conversation_id = ?; diff --git a/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/WorkingMemory.sq b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/WorkingMemory.sq new file mode 100644 index 0000000..4e8b81d --- /dev/null +++ b/standalone/src/jvmMain/sqldelight/pw/binom/agentik/standalone/persistence/sqlite/WorkingMemory.sq @@ -0,0 +1,34 @@ +CREATE TABLE working_memory ( + id TEXT NOT NULL PRIMARY KEY, + conversation_id TEXT NOT NULL, + order_idx INTEGER NOT NULL, + source_message_id TEXT, + kind TEXT NOT NULL, + payload_json TEXT NOT NULL, + created_at INTEGER NOT NULL +); + +CREATE UNIQUE INDEX idx_wm_unique ON working_memory(conversation_id, order_idx); +CREATE INDEX idx_wm_conv ON working_memory(conversation_id, order_idx); + +insert: +INSERT INTO working_memory (id, conversation_id, order_idx, source_message_id, kind, payload_json, created_at) +VALUES (?, ?, ?, ?, ?, ?, ?); + +listByConversation: +SELECT * FROM working_memory +WHERE conversation_id = ? +ORDER BY order_idx ASC; + +clearByConversation: +DELETE FROM working_memory WHERE conversation_id = ?; + +maxOrderIdx: +SELECT COALESCE(MAX(order_idx), 0) FROM working_memory WHERE conversation_id = ?; + +-- Atomic compact: delete rows >= dropFromOrderIdx and insert summary. +-- Caller supplies summaryId, summaryText, now epoch millis, and the new summary +-- gets order_idx = current max (after delete = before max). +compactDelete: +DELETE FROM working_memory +WHERE conversation_id = ? AND order_idx >= ?; 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 new file mode 100644 index 0000000..1ee2916 --- /dev/null +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/agent/ChatAgentTest.kt @@ -0,0 +1,433 @@ +package pw.binom.agentik.standalone.agent + +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.flow.flowOf +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import pw.binom.agentik.proto.AgentEvent +import pw.binom.agentik.proto.Content +import pw.binom.agentik.proto.Event as ProtoEvent +import pw.binom.agentik.standalone.llm.LlmConfig +import pw.binom.agentik.standalone.persistence.sqlite.SqliteStores +import pw.binom.litert.LiteContentPart +import pw.binom.litert.LiteConversation +import pw.binom.litert.LiteConversationConfig +import pw.binom.litert.LiteDelta +import pw.binom.litert.LiteLlm +import pw.binom.litert.LiteMessage +import pw.binom.litert.LiteRole +import pw.binom.litert.openai.OpenAiConfig +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.time.Instant + +class ChatAgentTest { + + private lateinit var stores: SqliteStores + private lateinit var fakeLlm: FakeLiteLlm + + @BeforeTest + fun setup() { + stores = SqliteStores.inMemory() + fakeLlm = FakeLiteLlm() + } + + @AfterTest + fun tearDown() { + stores.close() + } + + private fun newAgent(): ChatAgent { + val cfg = LlmConfig( + backend = pw.binom.agentik.standalone.llm.LlmBackend.OPENAI, + systemPrompt = "be brief", + openai = OpenAiConfig(baseUrl = "http://test", apiKey = "test", model = "test"), + ) + return ChatAgent(id = "agentik", stores = stores, llm = fakeLlm, llmConfig = cfg) + } + + @Test + fun `createConversation seeds system prompt into working memory`() = runTest { + val agent = newAgent() + val conv = agent.createConversation(temp = false) as ChatConversation + + val wm = stores.workingMemory.list(conv.id) + assertEquals(1, wm.size) + val first = wm[0] + assertIs(first.entry) + assertEquals("be brief", first.entry.text) + } + + @Test + fun `getConversation returns null for unknown id`() = runTest { + val agent = newAgent() + assertNull(agent.getConversation("nope")) + } + + @Test + fun `getConversations returns all stored persistent conversations`() = runTest { + val agent = newAgent() + agent.createConversation(temp = false) + agent.createConversation(temp = true) + val list = agent.getConversations(0, 10) + // temp-беседы не персистятся, в списке только persistent + assertEquals(1, list.size) + } + + @Test + fun `deleteConversation removes conversation and data`() = runTest { + val agent = newAgent() + val conv = agent.createConversation(temp = false) + val id = conv.id + + // добавим сообщение, чтобы потом убедиться, что каскад сработал + stores.messages.append( + pw.binom.agentik.standalone.persistence.MessageRecord.UserMessage( + id = "m1", + conversationId = id, + content = listOf(pw.binom.agentik.standalone.persistence.Content.Text("hi")), + createdAt = Instant.fromEpochMilliseconds(1_700_000_000_000), + ), + ) + assertTrue(agent.deleteConversation(id)) + assertNull(agent.getConversation(id)) + assertNull(stores.conversations.get(id)) + assertEquals(emptyList(), stores.messages.listAll(id)) + } + + @Test + fun `deleteConversation returns false for unknown id`() = runTest { + val agent = newAgent() + assertEquals(false, agent.deleteConversation("nope")) + } + + @Test + fun `send emits start_reasoning, start_response, append_text, end`() = runTest { + val agent = newAgent() + fakeLlm.reply = "hello back" + val conv = agent.createConversation(temp = false) + + conv.send(listOf(Content.Text("hi"))) + + // user message записан в audit + working memory + val msgs = stores.messages.listAll(conv.id) + assertEquals(2, msgs.size) + assertEquals("hi", (msgs[0] as pw.binom.agentik.standalone.persistence.MessageRecord.UserMessage).content.let { + (it[0] as pw.binom.agentik.standalone.persistence.Content.Text).body + }) + assertEquals("hello back", (msgs[1] as pw.binom.agentik.standalone.persistence.MessageRecord.AssistantMessage).content.let { + (it[0] as pw.binom.agentik.standalone.persistence.Content.Text).body + }) + } + + @Test + fun `send reconstructs conversation history from working memory`() = runTest { + val agent = newAgent() + fakeLlm.rememberHistory = true + fakeLlm.reply = "first reply" + val conv1 = agent.createConversation(temp = false) + conv1.send(listOf(Content.Text("first user"))) + + // Новая беседа не должна видеть историю первой + val conv2 = agent.createConversation(temp = false) + fakeLlm.reply = "second reply" + conv2.send(listOf(Content.Text("second user"))) + + // Первая беседа должна иметь только свою систему + 1 user + 1 assistant + val wm1 = stores.workingMemory.list(conv1.id) + assertEquals(3, wm1.size) + // Вторая беседа — только своё + val wm2 = stores.workingMemory.list(conv2.id) + assertEquals(3, wm2.size) + } + + @Test + fun `send passes system prompt and past history to LLM on first send`() = runTest { + val agent = newAgent() + fakeLlm.rememberHistory = true + val conv = agent.createConversation(temp = false) + fakeLlm.reply = "hi" + conv.send(listOf(Content.Text("hello"))) + + // Длинно-живущий LiteConversation: первый send создаёт его с systemInstruction + // и пустыми initialMessages (свежее user-сообщение пойдёт через sendStreamContents). + assertNotNull(fakeLlm.lastConfig) + assertEquals("be brief", fakeLlm.lastConfig!!.systemInstruction) + assertEquals(0, fakeLlm.lastConfig!!.initialMessages.size) + // Свежее user-сообщение отправлено через sendStreamContents + assertEquals(1, fakeLlm.conversations.size) + val sent = fakeLlm.lastContents + assertNotNull(sent) + assertEquals(1, sent!!.size) + assertEquals("hello", (sent[0] as LiteContentPart.Text).text) + } + + @Test + fun `multi-turn conversation accumulates in single LiteConversation`() = runTest { + 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 = [system, user, assistant] + assertEquals(3, stores.workingMemory.list(conv.id).size) + + fakeLlm.reply = "second reply" + conv.send(listOf(Content.Text("second user"))) + // второй turn: WM должен вырасти до [system, user, assistant, user, assistant] + val wm = stores.workingMemory.list(conv.id) + assertEquals(5, 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) + } + + @Test + fun `reloaded conversation reconstructs LiteConversation from working memory`() = runTest { + val agent = newAgent() + fakeLlm.rememberHistory = true + val conv = agent.createConversation(temp = false) + fakeLlm.reply = "first reply" + conv.send(listOf(Content.Text("first user"))) + val convId = conv.id + conv.close() + + // Открываем новое ChatConversation с тем же id — LiteConversation должен + // быть создан заново из working memory (первый user+assistant как initial). + val reopened = agent.getConversation(convId)!! + fakeLlm.reply = "second reply" + reopened.send(listOf(Content.Text("second user"))) + + val allConvs = fakeLlm.conversations + assertEquals(2, allConvs.size) // original + reopened + val reopenedLite = allConvs.last() + // Initial messages: только прошлые user+assistant (НЕ включая текущий "second user") + assertEquals(2, reopenedLite.initialMessages.size) + assertEquals("first user", reopenedLite.initialMessages[0].text) + assertEquals(LiteRole.USER, reopenedLite.initialMessages[0].role) + assertEquals("first reply", reopenedLite.initialMessages[1].text) + assertEquals(LiteRole.MODEL, reopenedLite.initialMessages[1].role) + } + + @Test + fun `interrupt cancels active send`() = runTest { + val agent = newAgent() + fakeLlm.slow = true + val conv = agent.createConversation(temp = false) + + val sendJob = launch { + try { + conv.send(listOf(Content.Text("hi"))) + } catch (_: kotlinx.coroutines.CancellationException) { + // ok + } + } + // ждём, пока корутина дойдёт до sendStreamContents и повиснет на slow-эмиссии + delay(200) + conv.interrupt() + sendJob.join() + + // user сообщение в audit должно быть, assistant — нет (был отменён) + val msgs = stores.messages.listAll(conv.id) + assertEquals(1, msgs.size) + assertIs(msgs[0]) + } + + @Test + fun `temp conversation is not persisted across agent instances`() = runTest { + // Поднимаем file-backed БД, создаём temp-беседу + stores.close() + val dbPath = (System.getProperty("java.io.tmpdir") + "/agentik-test-${System.nanoTime()}.db") + stores = SqliteStores.open(dbPath) + val agent1 = ChatAgent( + id = "agentik", + stores = stores, + llm = FakeLiteLlm().also { fakeLlm = it }, + llmConfig = LlmConfig( + backend = pw.binom.agentik.standalone.llm.LlmBackend.OPENAI, + systemPrompt = "be brief", + openai = OpenAiConfig(baseUrl = "http://test", apiKey = "test", model = "test"), + ), + ) + val tempConv = agent1.createConversation(temp = true) + val tempId = tempConv.id + assertNotNull(agent1.getConversation(tempId)) + + // Переоткрываем БД — temp-беседа не должна пережить рестарт + stores.close() + stores = SqliteStores.open(dbPath) + val agent2 = ChatAgent( + id = "agentik", + stores = stores, + llm = fakeLlm, + llmConfig = LlmConfig( + backend = pw.binom.agentik.standalone.llm.LlmBackend.OPENAI, + systemPrompt = "be brief", + openai = OpenAiConfig(baseUrl = "http://test", apiKey = "test", model = "test"), + ), + ) + assertNull(agent2.getConversation(tempId)) + java.io.File(dbPath).delete() + } + + @Test + fun `non-temp conversation persists across agent instances`() = runTest { + stores.close() + val dbPath = (System.getProperty("java.io.tmpdir") + "/agentik-test-${System.nanoTime()}.db") + stores = SqliteStores.open(dbPath) + val agent1 = ChatAgent( + id = "agentik", + stores = stores, + llm = FakeLiteLlm().also { fakeLlm = it }, + llmConfig = LlmConfig( + backend = pw.binom.agentik.standalone.llm.LlmBackend.OPENAI, + systemPrompt = "be brief", + openai = OpenAiConfig(baseUrl = "http://test", apiKey = "test", model = "test"), + ), + ) + val conv = agent1.createConversation(temp = false) + val id = conv.id + + stores.close() + stores = SqliteStores.open(dbPath) + val agent2 = ChatAgent( + id = "agentik", + stores = stores, + llm = fakeLlm, + llmConfig = LlmConfig( + backend = pw.binom.agentik.standalone.llm.LlmBackend.OPENAI, + systemPrompt = "be brief", + openai = OpenAiConfig(baseUrl = "http://test", apiKey = "test", model = "test"), + ), + ) + assertNotNull(agent2.getConversation(id)) + java.io.File(dbPath).delete() + } + + private fun fakeLiteLlmForReload(): LiteLlm = object : LiteLlm { + override val backendName: String = "fake" + override fun isInitialized(): Boolean = true + override fun createConversation(config: LiteConversationConfig): LiteConversation = + error("not used in reload test") + override fun infer(request: pw.binom.litert.LiteRequest): String = error("not used") + override fun inferStream(request: pw.binom.litert.LiteRequest): Flow = error("not used") + override fun close() {} + } + + @Test + fun `agentEvents — Created + Deleted flow`() = runTest { + val agent = newAgent() + val events = mutableListOf() + val job = launch(start = kotlinx.coroutines.CoroutineStart.UNDISPATCHED) { + agent.events(Instant.DISTANT_PAST).collect { events.add(it) } + } + val conv = agent.createConversation(temp = false) + agent.deleteConversation(conv.id) + delay(50) + job.cancel() + + assertEquals(2, events.size) + val created = events[0] as AgentEvent.Created + val deleted = events[1] as AgentEvent.Deleted + assertEquals(conv.id, created.conversationId) + assertEquals(conv.id, deleted.id) + } +} + +/** Поддельный LiteLlm: возвращает fakeLlm.reply в sendStreamContents, опционально запоминает history. */ +private class FakeLiteLlm : LiteLlm { + override val backendName: String = "fake" + var reply: String = "" + var rememberHistory: Boolean = false + var slow: Boolean = false + var lastConfig: LiteConversationConfig? = null + var lastContents: List? = null + val conversations = mutableListOf() + + override fun isInitialized(): Boolean = true + + override fun createConversation(config: LiteConversationConfig): LiteConversation { + lastConfig = config + val conv = FakeLiteConversation(this, config) + conversations.add(conv) + return conv + } + + override fun infer(request: pw.binom.litert.LiteRequest): String { + throw UnsupportedOperationException("not used in test") + } + + override fun inferStream(request: pw.binom.litert.LiteRequest): Flow { + throw UnsupportedOperationException("not used in test") + } + + override fun close() {} + + fun emit(text: String, sink: FakeLiteConversation): List { + // Эмулируем один-два фрагмента + done + return listOf( + LiteDelta(text = text.substring(0, text.length / 2), isDone = false), + LiteDelta(text = text.substring(text.length / 2), isDone = true), + ) + } +} + +private class FakeLiteConversation( + private val parent: FakeLiteLlm, + config: LiteConversationConfig, +) : LiteConversation { + val initialMessages: List = config.initialMessages + private val mutableHistory: MutableList = config.initialMessages.toMutableList() + override val history: List + get() = mutableHistory.toList() + + override fun sendStream(prompt: String): Flow = + sendStreamContents(listOf(LiteContentPart.Text(prompt))) + + override fun sendStreamContents(contents: List): Flow { + parent.lastContents = contents + 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 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 = parent.reply + override fun sendContents(contents: List): String = parent.reply + override fun cancel() {} + override fun tokenCount(): Int = history.size + override fun addToolResult(callId: String?, name: String, result: String) { error("not used") } + override fun close() {} +} diff --git a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/llm/LlmConfigTest.kt b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/llm/LlmConfigTest.kt new file mode 100644 index 0000000..eb6982d --- /dev/null +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/llm/LlmConfigTest.kt @@ -0,0 +1,90 @@ +package pw.binom.agentik.standalone.llm + +import pw.binom.litert.openai.OpenAiConfig +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith + +class LlmConfigTest { + + @Test + fun `fromEnv — happy path`() { + val cfg = LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_BASE_URL" -> "https://api.openai.com/v1" + "OPENAI_API_KEY" -> "sk-test" + "OPENAI_MODEL" -> "gpt-4o-mini" + "AGENTIK_SYSTEM_PROMPT" -> "be brief" + else -> null + } + } + assertEquals("be brief", cfg.systemPrompt) + assertEquals(OpenAiConfig(baseUrl = "https://api.openai.com/v1", apiKey = "sk-test", model = "gpt-4o-mini"), cfg.openai) + } + + @Test + fun `fromEnv — falls back to default system prompt`() { + val cfg = LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_BASE_URL" -> "https://api.openai.com/v1" + "OPENAI_API_KEY" -> "sk-test" + "OPENAI_MODEL" -> "gpt-4o-mini" + else -> null + } + } + assertEquals(LlmConfig.DEFAULT_SYSTEM_PROMPT, cfg.systemPrompt) + } + + @Test + fun `fromEnv — missing base url throws`() { + assertFailsWith { + LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_API_KEY" -> "sk-test" + "OPENAI_MODEL" -> "gpt-4o-mini" + else -> null + } + } + } + } + + @Test + fun `fromEnv — missing api key throws`() { + assertFailsWith { + LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_BASE_URL" -> "https://api.openai.com/v1" + "OPENAI_MODEL" -> "gpt-4o-mini" + else -> null + } + } + } + } + + @Test + fun `fromEnv — missing model throws`() { + assertFailsWith { + LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_BASE_URL" -> "https://api.openai.com/v1" + "OPENAI_API_KEY" -> "sk-test" + else -> null + } + } + } + } + + @Test + fun `blank system prompt from env falls back to default`() { + val cfg = LlmConfig.fromEnv { name -> + when (name) { + "OPENAI_BASE_URL" -> "https://api.openai.com/v1" + "OPENAI_API_KEY" -> "sk-test" + "OPENAI_MODEL" -> "gpt-4o-mini" + "AGENTIK_SYSTEM_PROMPT" -> " " + else -> null + } + } + assertEquals(LlmConfig.DEFAULT_SYSTEM_PROMPT, cfg.systemPrompt) + } +} diff --git a/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/persistence/PersistenceTest.kt b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/persistence/PersistenceTest.kt new file mode 100644 index 0000000..47b3842 --- /dev/null +++ b/standalone/src/jvmTest/kotlin/pw/binom/agentik/standalone/persistence/PersistenceTest.kt @@ -0,0 +1,217 @@ +package pw.binom.agentik.standalone.persistence + +import kotlinx.coroutines.test.runTest +import pw.binom.agentik.standalone.persistence.sqlite.SqliteStores +import kotlin.test.AfterTest +import kotlin.test.BeforeTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertNotNull +import kotlin.test.assertNull +import kotlin.test.assertTrue +import kotlin.time.Instant + +class PersistenceTest { + + private lateinit var stores: SqliteStores + + @BeforeTest + fun setup() { + stores = SqliteStores.inMemory() + } + + @AfterTest + fun tearDown() { + stores.close() + } + + @Test + fun `upsert + get conversation — roundtrip`() = runTest { + val now = Instant.fromEpochMilliseconds(1_700_000_000_000) + val rec = ConversationRecord( + id = "c1", + title = "Hello", + isTemporal = false, + createdAt = now, + updatedAt = now, + ) + stores.conversations.upsert(rec) + val got = stores.conversations.get("c1") + assertNotNull(got) + assertEquals(rec.id, got.id) + assertEquals(rec.title, got.title) + assertEquals(rec.isTemporal, got.isTemporal) + assertEquals(rec.createdAt, got.createdAt) + assertEquals(rec.updatedAt, got.updatedAt) + } + + @Test + fun `upsert overwrites existing record`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + stores.conversations.upsert( + ConversationRecord("c1", title = "A", isTemporal = false, createdAt = t0, updatedAt = t0), + ) + val t1 = Instant.fromEpochMilliseconds(1_700_000_001_000) + stores.conversations.upsert( + ConversationRecord("c1", title = "B", isTemporal = true, createdAt = t0, updatedAt = t1), + ) + val got = stores.conversations.get("c1")!! + assertEquals("B", got.title) + assertTrue(got.isTemporal) + assertEquals(t1, got.updatedAt) + } + + @Test + fun `list returns conversations ordered by updated_at desc`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + repeat(3) { i -> + stores.conversations.upsert( + ConversationRecord( + id = "c$i", + title = null, + isTemporal = false, + createdAt = t0, + updatedAt = Instant.fromEpochMilliseconds(1_700_000_000_000 + i * 1000), + ), + ) + } + val list = stores.conversations.list(offset = 0, limit = 10) + assertEquals(listOf("c2", "c1", "c0"), list.map { it.id }) + } + + @Test + fun `delete cascades messages and working_memory`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + stores.conversations.upsert( + ConversationRecord("c1", null, false, t0, t0), + ) + stores.messages.append( + MessageRecord.UserMessage( + id = "m1", + conversationId = "c1", + content = listOf(Content.Text("hello")), + createdAt = t0, + ), + ) + stores.workingMemory.append( + conversationId = "c1", + entry = WorkingMemoryEntry.User( + sourceMessageId = "m1", + content = listOf(Content.Text("hello")), + ), + now = t0, + ) + assertEquals(1, stores.messages.listAll("c1").size) + assertEquals(1, stores.workingMemory.list("c1").size) + + val removed = stores.conversations.delete("c1") + assertTrue(removed) + assertNull(stores.conversations.get("c1")) + assertEquals(emptyList(), stores.messages.listAll("c1")) + assertEquals(emptyList(), stores.workingMemory.list("c1")) + } + + @Test + fun `message audit log — append and read back`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + stores.messages.append( + MessageRecord.UserMessage("m1", "c1", listOf(Content.Text("hi")), t0), + ) + stores.messages.append( + MessageRecord.AssistantMessage("m2", "c1", listOf(Content.Text("yo")), t0), + ) + + val all = stores.messages.listAll("c1") + assertEquals(2, all.size) + assertEquals("m1", all[0].id) + assertEquals("m2", all[1].id) + assertTrue(all[0] is MessageRecord.UserMessage) + assertTrue(all[1] is MessageRecord.AssistantMessage) + assertEquals("hi", (all[0] as MessageRecord.UserMessage).content[0].let { + (it as Content.Text).body + }) + } + + @Test + fun `message after timestamp filter`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + val t1 = Instant.fromEpochMilliseconds(1_700_000_001_000) + stores.messages.append(MessageRecord.UserMessage("m1", "c1", listOf(Content.Text("a")), t0)) + stores.messages.append(MessageRecord.UserMessage("m2", "c1", listOf(Content.Text("b")), t1)) + + val after = stores.messages.list("c1", after = t0, offset = 0, limit = 10) + assertEquals(1, after.size) + assertEquals("m2", after[0].id) + } + + @Test + 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"))), + now = t0, + ) + stores.workingMemory.append( + conversationId = "c1", + entry = WorkingMemoryEntry.Assistant(sourceMessageId = "m2", content = listOf(Content.Text("yo"))), + 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) + } + + @Test + fun `rename updates title and bumps updated_at`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + stores.conversations.upsert(ConversationRecord("c1", null, false, t0, t0)) + val newTs = stores.conversations.rename("c1", "Renamed") + assertNotNull(newTs) + assertTrue(newTs > t0) + val got = stores.conversations.get("c1")!! + assertEquals("Renamed", got.title) + assertEquals(newTs, got.updatedAt) + } + + @Test + fun `rename can clear title`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + stores.conversations.upsert(ConversationRecord("c1", "Title", false, t0, t0)) + stores.conversations.rename("c1", null) + val got = stores.conversations.get("c1")!! + assertNull(got.title) + } + + @Test + fun `delete returns false when conversation does not exist`() = runTest { + assertFalse(stores.conversations.delete("nope")) + } + + @Test + fun `image content roundtrip through message payload`() = runTest { + val t0 = Instant.fromEpochMilliseconds(1_700_000_000_000) + val bytes = byteArrayOf(0x89.toByte(), 0x50, 0x4E, 0x47) // PNG header + stores.messages.append( + MessageRecord.UserMessage( + id = "m1", + conversationId = "c1", + content = listOf(Content.Image(data = bytes, mime = "image/png")), + createdAt = t0, + ), + ) + val all = stores.messages.listAll("c1") + val image = (all[0] as MessageRecord.UserMessage).content[0] as Content.Image + assertEquals("image/png", image.mime) + assertTrue(bytes.contentEquals(image.data)) + } +}