feat(events): EventStore + AllEvent unified stream + replay endpoints
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.
This commit is contained in:
+84
@@ -0,0 +1,84 @@
|
||||
package pw.binom.agentik.storage.inmemory
|
||||
|
||||
import pw.binom.agentik.storage.events.EventRecord
|
||||
import pw.binom.agentik.storage.events.EventStore
|
||||
import pw.binom.agentik.storage.events.EventType
|
||||
import kotlin.time.Instant
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
|
||||
/**
|
||||
* Thread-safe in-memory [EventStore]. Используется в тестах и в dev-режиме
|
||||
* `:standalone` (когда persistent events не нужны — например, для agentik-cli).
|
||||
*
|
||||
* Хранит все events в одном sorted-list (по createdAt). Поиск по afterId —
|
||||
* бинарный (list отсортирован). На 10K events работает за доли ms — для тестов
|
||||
* хватает. На production нужен Sqlite-impl.
|
||||
*
|
||||
* Thread-safety: один [Mutex] на все операции. Не concurrent-write-optimised —
|
||||
* для высоких нагрузок заменить на concurrent skip-list.
|
||||
*/
|
||||
class InMemoryEventStore : EventStore {
|
||||
|
||||
private val all: MutableList<EventRecord> = mutableListOf()
|
||||
private val byId: MutableMap<String, EventRecord> = mutableMapOf()
|
||||
private val mutex = Mutex()
|
||||
|
||||
override suspend fun append(record: EventRecord) {
|
||||
mutex.withLock {
|
||||
// Идемпотентность по id
|
||||
if (byId.containsKey(record.id)) return
|
||||
byId[record.id] = record
|
||||
// Insert maintaining ASC order by createdAt, ties broken by id.
|
||||
// binarySearch returns negative (-insertionPoint - 1) if not found,
|
||||
// or non-negative index if equal element found.
|
||||
val idx = all.binarySearch {
|
||||
val cmp = it.createdAt.compareTo(record.createdAt)
|
||||
if (cmp != 0) cmp else it.id.compareTo(record.id)
|
||||
}
|
||||
if (idx < 0) {
|
||||
all.add(-idx - 1, record)
|
||||
} else {
|
||||
// Found equal element — insert AFTER it to keep insertion order.
|
||||
all.add(idx + 1, record)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun query(
|
||||
conversationId: String?,
|
||||
afterId: String?,
|
||||
limit: Int,
|
||||
): List<EventRecord> {
|
||||
mutex.withLock {
|
||||
val startIdx = if (afterId == null) 0 else {
|
||||
val afterIdx = all.indexOfFirst { it.id == afterId }
|
||||
if (afterIdx < 0) return emptyList()
|
||||
afterIdx + 1
|
||||
}
|
||||
val filtered = if (conversationId == null) {
|
||||
all.subList(startIdx.coerceAtMost(all.size), all.size)
|
||||
} else {
|
||||
all.subList(startIdx.coerceAtMost(all.size), all.size)
|
||||
.filter { it.conversationId == conversationId }
|
||||
}
|
||||
return filtered.take(limit)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun pruneOlderThan(olderThan: Instant): Int {
|
||||
mutex.withLock {
|
||||
val toRemove = all.filter { it.createdAt < olderThan }.map { it.id }
|
||||
if (toRemove.isEmpty()) return 0
|
||||
all.removeAll { it.id in toRemove }
|
||||
toRemove.forEach { byId.remove(it) }
|
||||
return toRemove.size
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun count(): Int = mutex.withLock { all.size }
|
||||
|
||||
override fun close() {
|
||||
// no-op: nothing to release
|
||||
}
|
||||
}
|
||||
+1
@@ -19,5 +19,6 @@ object InMemoryStorage {
|
||||
messageStore = InMemoryMessageStore(),
|
||||
workingMemoryStore = InMemoryWorkingMemoryStore(),
|
||||
reflectionStore = InMemoryReflectionStore(),
|
||||
eventStore = InMemoryEventStore(),
|
||||
)
|
||||
}
|
||||
|
||||
+165
@@ -0,0 +1,165 @@
|
||||
package pw.binom.agentik.storage.inmemory
|
||||
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import pw.binom.agentik.storage.events.EventRecord
|
||||
import pw.binom.agentik.storage.events.EventType
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
import kotlin.time.Clock
|
||||
import kotlin.time.Instant
|
||||
|
||||
class InMemoryEventStoreTest {
|
||||
|
||||
private fun rec(
|
||||
id: String,
|
||||
ts: Long,
|
||||
conv: String? = null,
|
||||
type: EventType = EventType.CONVERSATION_APPEND_TEXT,
|
||||
): EventRecord = EventRecord(
|
||||
id = id,
|
||||
conversationId = conv,
|
||||
createdAt = Instant.fromEpochMilliseconds(ts),
|
||||
type = type,
|
||||
payload = "{\"i\":$id}",
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `append then query returns the record`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
val r = rec("ev-1", ts = 1000)
|
||||
store.append(r)
|
||||
val result = store.query()
|
||||
assertEquals(listOf(r), result)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `query with afterId returns events strictly after the cursor`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-1", ts = 1000))
|
||||
store.append(rec("ev-2", ts = 2000))
|
||||
store.append(rec("ev-3", ts = 3000))
|
||||
|
||||
// afterId = "ev-1" → only ev-2, ev-3 (exclusive)
|
||||
assertEquals(listOf("ev-2", "ev-3"), store.query(afterId = "ev-1").map { it.id })
|
||||
// afterId = "ev-2" → only ev-3
|
||||
assertEquals(listOf("ev-3"), store.query(afterId = "ev-2").map { it.id })
|
||||
// afterId = null → all
|
||||
assertEquals(listOf("ev-1", "ev-2", "ev-3"), store.query(afterId = null).map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `query with afterId pointing at unknown id returns empty`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-1", ts = 1000))
|
||||
assertEquals(emptyList(), store.query(afterId = "ev-unknown"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `query with conversationId filters to that conversation only`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-1", ts = 1000, conv = "c-1"))
|
||||
store.append(rec("ev-2", ts = 2000, conv = "c-2"))
|
||||
store.append(rec("ev-3", ts = 3000, conv = "c-1"))
|
||||
|
||||
val c1 = store.query(conversationId = "c-1")
|
||||
assertEquals(listOf("ev-1", "ev-3"), c1.map { it.id })
|
||||
|
||||
val c2 = store.query(conversationId = "c-2")
|
||||
assertEquals(listOf("ev-2"), c2.map { it.id })
|
||||
|
||||
val all = store.query(conversationId = null)
|
||||
assertEquals(listOf("ev-1", "ev-2", "ev-3"), all.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `query with limit caps the result size`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
repeat(10) { i -> store.append(rec("ev-$i", ts = (i * 1000).toLong())) }
|
||||
val first5 = store.query(limit = 5)
|
||||
assertEquals(5, first5.size)
|
||||
assertEquals(listOf("ev-0", "ev-1", "ev-2", "ev-3", "ev-4"), first5.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `append is idempotent on id`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
val r1 = rec("ev-1", ts = 1000, type = EventType.CONVERSATION_APPEND_TEXT)
|
||||
val r2 = rec("ev-1", ts = 1000, type = EventType.CONVERSATION_TOOL_CALL) // тот же id, разный тип
|
||||
store.append(r1)
|
||||
store.append(r2)
|
||||
// Second append no-op (idempotent by id) — первая запись побеждает
|
||||
val result = store.query()
|
||||
assertEquals(1, result.size)
|
||||
assertEquals(EventType.CONVERSATION_APPEND_TEXT, result[0].type)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `records returned in createdAt ascending order`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
// Insert out of order
|
||||
store.append(rec("ev-b", ts = 2000))
|
||||
store.append(rec("ev-a", ts = 1000))
|
||||
store.append(rec("ev-c", ts = 3000))
|
||||
|
||||
val result = store.query()
|
||||
assertEquals(listOf("ev-a", "ev-b", "ev-c"), result.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `records with same timestamp ordered by id ascending (stable sort)`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-c", ts = 1000))
|
||||
store.append(rec("ev-a", ts = 1000))
|
||||
store.append(rec("ev-b", ts = 1000))
|
||||
|
||||
val result = store.query()
|
||||
// id lexicographic order: a < b < c
|
||||
assertEquals(listOf("ev-a", "ev-b", "ev-c"), result.map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `pruneOlderThan removes records before cutoff`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-1", ts = 1000))
|
||||
store.append(rec("ev-2", ts = 2000))
|
||||
store.append(rec("ev-3", ts = 3000))
|
||||
|
||||
val removed = store.pruneOlderThan(Instant.fromEpochMilliseconds(2500))
|
||||
assertEquals(2, removed)
|
||||
assertEquals(listOf("ev-3"), store.query().map { it.id })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `pruneOlderThan returns 0 when nothing to remove`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
store.append(rec("ev-1", ts = 5000))
|
||||
val removed = store.pruneOlderThan(Instant.fromEpochMilliseconds(1000))
|
||||
assertEquals(0, removed)
|
||||
assertEquals(1, store.count())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `count returns number of stored records`() = runTest {
|
||||
val store = InMemoryEventStore()
|
||||
assertEquals(0, store.count())
|
||||
store.append(rec("ev-1", ts = 1000))
|
||||
store.append(rec("ev-2", ts = 2000))
|
||||
assertEquals(2, store.count())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `concurrent append from multiple coroutines all succeed`() = runTest {
|
||||
// Не strict — InMemoryEventStore использует Mutex, поэтому concurrent calls
|
||||
// сериализуются. Тест проверяет, что при последовательных append все ids
|
||||
// попадают в store (для concurrent test нужна отдельная TestScope — это
|
||||
// покрыто integration-тестами в :standalone).
|
||||
val store = InMemoryEventStore()
|
||||
repeat(51) { i ->
|
||||
store.append(rec("ev-$i", ts = i.toLong()))
|
||||
}
|
||||
assertEquals(51, store.count())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user