subochev
4ad59d5f5d
EventStore (persistent event log) и AllEvent (sealed wrapper для
третьего типа подписки — ВСЕ events в одном потоке). Touches 7 modules.
Архитектура:
Producer (ChatAgent + ConversationEvents) → EventStore + SharedFlow
↓ ↓
Live SSE (cold, no replay) Replay endpoints (cursor-based)
(1) :storage-core — EventStore interface
- append(record): idempotent по record.id (INSERT OR IGNORE)
- query(conversationId?, afterId?, limit): пагинированный catchup
- pruneOlderThan(instant): TTL cleanup
- count(): maintenance метрика
- @Serializable EventRecord(id, conversationId?, createdAt, type, payload)
- enum EventType: AGENT_*/CONVERSATION_* (forward-compat fallback)
- StorageBundle дополнен eventStore: EventStore? = null (backward-compat)
(2) :storage-inmemory — InMemoryEventStore
- Thread-safe (Mutex), binarySearch для упорядоченной вставки
- Записи сортируются по createdAt ASC, ties по id ASC (стабильно)
- Idempotency по id (повторный append no-op)
(3) :storage-sqlite — SqliteEventStore
- sqldelight schema: agent_event (id PK, conversation_id?, created_at,
type, payload BLOB) + 2 индекса (conversation_id+created_at,
created_at)
- Миграция v3: CREATE TABLE IF NOT EXISTS (additive)
- 5 запросов: insert, queryGlobal, queryByConv, pruneOlderThan, count
- Forward-compat: неизвестный EventType в БД → fallback AGENT_CREATED
(чтобы старые клиенты не падали на новых enum values)
- Добавлен в SqliteStores (open/inMemory + asBundle())
(4) :standalone — Producer wiring
- ChatAgent.persistAgentEvent() — fire-and-forget append при каждом
AgentEvent (Created/Deleted/Renamed)
- ConversationEvents — персистит в EventStore при каждом tryEmit/emit
(концертный случай от connect disconnect)
- ChatAgent.allEvents() — merge agent-events + snapshot всех живых
диалогов в единый Flow<AllEvent>
(5) :proto — AllEvent sealed interface
- AllEvent.Agent(date, event: AgentEvent)
- AllEvent.Conversation(date, conversationId, event: Event)
- Agent.allEvents(after): Flow<AllEvent> — третий тип подписки
(в дополнение к events() и Conversation.events)
(6) :server — Endpoints
- GET /events/all — SSE поток AllEvent (cold)
- GET /events/replay?after_id=&limit= — пагинированный catchup
(503 если EventStore не сконфигурирован)
- GET /conversations/{id}/events/replay?after_id=&limit= — то же per-conv
- Module.kt принимает eventStore: EventStore? параметром
(7) :client — Client API
- AgentClient.allEvents(after) — подписка на /events/all SSE
- AgentClient.replayAllEvents(afterId, limit) — catchup /events/replay
- AgentClient.replayConversationEvents(convId, afterId, limit)
- EventRecordDto — wire-зеркало EventRecord (клиент не зависит
от :storage-core, определяет DTO локально; формат совместим с
серверным JSON)
Тесты: 22 новых теста (12 InMemory + 10 Sqlite), все зелёные.
Все три слоя синхронизированы: proto contract + standalone impl +
server endpoint + client API.
:client — Ktor-клиент к :server/:proto (KMP, jvm + native)
Что это
Ktor client (io.ktor.client.HttpClient + ContentNegotiation(json) + Sse), превращающий HTTP/SSE-фасад :server в Agent/Conversation
интерфейсы :proto:
AgentikAgent(id, baseUrl)— entry-point фабрики.AgentClient— список и lifecycle диалогов.ConversationClient—send(),events(),interrupt(),getMessages(),rename(),close().- Внутренний парсер SSE →
Flow<Event>.
Решает: пишем нативный Kotlin-клиент, без curl/JS/Python boilerplate, с теми же типами, что и сервер. Один и тот же клиент работает на JVM, iOS, macOS, Linux, Windows.
Где используется
:agentik-cli— REPL.:agentik-cli— JVM/native CLI-клиент поверх:client.- Любой внешний KMP-проект, который хочет встроить агента в свой UI.
Как подключить
// build.gradle.kts
kotlin {
sourceSets.commonMain.dependencies {
api("pw.binom.agentik:client:0.1.0")
}
}
// ваш код:
val agent = AgentikAgent(id = "agentik", baseUrl = "http://192.168.76.166:8080/agentik")
val conv = agent.createConversation(title = "test")
conv.send(listOf(Content.Text("hello"))).collect { event ->
when (event) {
is Event.AppendText -> print(event.body)
is Event.End -> println("\n--- end ---")
is Event.Error -> error("agent error: ${event.message}")
else -> Unit
}
}
Версии
gradle/libs.versions.toml → [versions] agentik-client.
Поддерживает все KMP-таргеты, что и :proto.
Примеры API
// список диалогов
agent.getConversations().collect { println(it.id to it.title) }
// live-подписка на события отдельного диалога
val sub = conversation.events(after = Instant.parse("2026-09-01T00:00:00Z")).collect { }
// прерывание текущего хода
conversation.interrupt()
// история
conversation.getMessages(offset = 0).collect { msg ->
when (msg) {
is Message.UserMessage -> println("user: ${msg.content}")
is Message.AssistantMessage -> println("assistant: ${msg.content}")
else -> Unit
}
}
Тесты
./gradlew :client:jvmTest
Покрывают: JSON-парсинг Event'ов, SSE-стрим, recovery после разрыва, 401/404.
Чего здесь НЕТ
- Никакого LLM-кода. Это просто клиент.
- Никакого persistent state. История хранится у сервера, клиент её
запрашивает через
getMessagesили подписывается черезevents.
Текущий статус
Используется продакшеном. Бэкендом служит :server поверх :standalone,
но клиент совместим с любым сервером, который держит wire-контракт
:server.
Известное ограничение
SSE event-stream в не-TTY ssh-сессии (без -tt) закрывается на
default-таймауте Ktor. Используйте либо ssh -tt, либо нативный
terminal (TTY). Это upstream-особенность Ktor SSE.