feat(client): implement HttpJournalStore and integrate journal endpoints
ci / JVM build + tests (push) Failing after 1m8s
ci / JVM build + tests (push) Failing after 1m8s
- Added `HttpJournalStore` as an HTTP-backed implementation of `JournalStore` for read-only access to the audit log.
- Integrated `GET /journal/conversations/{id}/messages` endpoint to fetch conversation transcripts with full payloads.
- Updated `AgentClient` to expose `HttpJournalStore` as the `journal` property.
- Adjusted `HttpEventStore` to align with updated endpoint structure (`/outbox/events`).
This commit is contained in:
@@ -22,6 +22,7 @@ kotlin {
|
|||||||
commonMain.dependencies {
|
commonMain.dependencies {
|
||||||
api(project(":proto"))
|
api(project(":proto"))
|
||||||
api(project(":event-store"))
|
api(project(":event-store"))
|
||||||
|
api(project(":journal-api"))
|
||||||
|
|
||||||
api(libs.ktor.client.core)
|
api(libs.ktor.client.core)
|
||||||
implementation(libs.ktor.client.content.negotiation)
|
implementation(libs.ktor.client.content.negotiation)
|
||||||
|
|||||||
@@ -46,13 +46,10 @@ internal class AgentClient(
|
|||||||
override val outbox: OutboxStore = HttpEventStore(httpClient = httpClient, baseUrl = agentUrl)
|
override val outbox: OutboxStore = HttpEventStore(httpClient = httpClient, baseUrl = agentUrl)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* HTTP-фасад для journal пока не реализован: на стороне `:server` ещё
|
* HTTP-фасад для journal: ходит в `:server`'s `GET /journal/conversations/{id}/messages`.
|
||||||
* не выставлены endpoint'ы `/journal/conversations/{id}/messages`.
|
* См. [HttpJournalStore] и [pw.binom.agentik.server.journalRoutes].
|
||||||
* Как только появятся — заменить на `HttpJournalStore(httpClient, agentUrl)`.
|
|
||||||
*/
|
*/
|
||||||
override val journal: JournalStore = error(
|
override val journal: JournalStore = HttpJournalStore(httpClient = httpClient, baseUrl = agentUrl)
|
||||||
"HttpJournalStore ещё не реализован — дождаться :server endpoint'а /journal/...",
|
|
||||||
)
|
|
||||||
|
|
||||||
override fun createConversation(temp: Boolean): Conversation =
|
override fun createConversation(temp: Boolean): Conversation =
|
||||||
runBlocking {
|
runBlocking {
|
||||||
|
|||||||
@@ -17,21 +17,18 @@ import kotlin.time.Instant
|
|||||||
* HTTP-реализация [EventStore] (= [pw.binom.agentik.outbox.OutboxStore]),
|
* HTTP-реализация [EventStore] (= [pw.binom.agentik.outbox.OutboxStore]),
|
||||||
* ходящая в `:server`-фасад.
|
* ходящая в `:server`-фасад.
|
||||||
*
|
*
|
||||||
* **Хитрый план**: вместо того, чтобы все методы шли в один общий endpoint и
|
* **Endpoint-раскладка** (новый дизайн — storage handles на [Agent]):
|
||||||
* фильтровали client-side ([EventStore.events]/[filterIsInstance]), эта
|
* - [events] → `GET {baseUrl}/outbox/events?after=` (полный поток
|
||||||
* реализация бьёт запросы по URL'ам в зависимости от того, какой класс
|
* [CommonEvent], bounded-tail + live SSE, см. [pw.binom.agentik.server.outboxRoutes])
|
||||||
* событий нужен:
|
* - [agentEvents] → `GET {baseUrl}/events?after=` (legacy proto-роут:
|
||||||
* - [events] → `GET /events/all` (полный поток CommonEvent)
|
* сервер пробрасывает [pw.binom.agentik.outbox.agentEvents] и распаковывает
|
||||||
* - [agentEvents] → `GET /events` (только lifecycle диалогов)
|
* `.event` для обратной совместимости с форматом AgentEvent)
|
||||||
* - [conversationEvents] с `conversationId != null` → `GET /conversations/{id}/events`
|
* - [conversationEvents] с `conversationId != null` → `GET /conversations/{id}/events`
|
||||||
*
|
*
|
||||||
* Так серверный фильтр (SQL `WHERE` или разные буферы) работает на своей стороне,
|
|
||||||
* а клиент получает ровно тот срез, который ему нужен, без лишнего трафика.
|
|
||||||
*
|
|
||||||
* Для [conversationEvents] с `conversationId == null` (события всех диалогов)
|
* Для [conversationEvents] с `conversationId == null` (события всех диалогов)
|
||||||
* fallback на default [EventStore.conversationEvents] — общий поток `/events/all`
|
* fallback на default [EventStore.conversationEvents] — общий поток
|
||||||
* + фильтр client-side. Это редкий кейс (admin-дашборды), и оптимизировать его
|
* `/outbox/events` + filter. Это редкий кейс (admin-дашборды), и
|
||||||
* отдельно нерационально.
|
* оптимизировать его отдельно нерационально.
|
||||||
*
|
*
|
||||||
* [earliestEventDate] не имеет своего endpoint'а; возвращает `Clock.System.now()`
|
* [earliestEventDate] не имеет своего endpoint'а; возвращает `Clock.System.now()`
|
||||||
* (см. KDoc [EventStore.earliestEventDate] — для пустого буфера это и есть
|
* (см. KDoc [EventStore.earliestEventDate] — для пустого буфера это и есть
|
||||||
@@ -53,7 +50,7 @@ internal class HttpEventStore(
|
|||||||
|
|
||||||
override fun events(after: Instant?): Flow<CommonEvent> = flow {
|
override fun events(after: Instant?): Flow<CommonEvent> = flow {
|
||||||
val url = buildString {
|
val url = buildString {
|
||||||
append("$agentUrl/events/all")
|
append("$agentUrl/outbox/events")
|
||||||
if (after != null) append("?after=$after")
|
if (after != null) append("?after=$after")
|
||||||
}
|
}
|
||||||
httpClient.prepareGet(url) { noSseReadTimeout() }
|
httpClient.prepareGet(url) { noSseReadTimeout() }
|
||||||
|
|||||||
@@ -0,0 +1,60 @@
|
|||||||
|
package pw.binom.agentik.client
|
||||||
|
|
||||||
|
import io.ktor.client.HttpClient
|
||||||
|
import io.ktor.client.call.body
|
||||||
|
import io.ktor.client.request.get
|
||||||
|
import io.ktor.client.request.parameter
|
||||||
|
import io.ktor.http.HttpStatusCode
|
||||||
|
import pw.binom.agentik.journal.JournalStore
|
||||||
|
import pw.binom.agentik.journal.MessageRecord
|
||||||
|
import kotlin.time.Instant
|
||||||
|
|
||||||
|
/**
|
||||||
|
* HTTP-реализация [JournalStore] (append-only audit log сообщений диалога),
|
||||||
|
* ходящая в `:server`-фасад.
|
||||||
|
*
|
||||||
|
* **Endpoint**: `GET {baseUrl}/journal/conversations/{id}/messages?after=&offset=&limit=`
|
||||||
|
* (см. [pw.binom.agentik.server.journalRoutes]).
|
||||||
|
*
|
||||||
|
* Возвращает raw [MessageRecord] (все типы: UserMessage / AssistantMessage /
|
||||||
|
* ToolCall / ToolResult / Error). В отличие от `GET /conversations/{id}/messages`
|
||||||
|
* в `:server`'s proto-роутах (который отдаёт project'нутые
|
||||||
|
* [pw.binom.agentik.proto.Message]), здесь клиент получает полный transcript
|
||||||
|
* с tool-call/tool-result/error payload'ами, turn-tokens и context'ом.
|
||||||
|
*
|
||||||
|
* **listFlow** — default cold-flow paging через [list] (N+1 round-trip,
|
||||||
|
* дефолтная реализация из [JournalStore]). Для remote/SQL-backed store'а
|
||||||
|
* это OK: server-side paging + client-side flow compose'ится естественно.
|
||||||
|
*
|
||||||
|
* **Read-only**: [JournalStore] не имеет `append` — запись только через
|
||||||
|
* writer-референс, который ChatAgent держит внутри (тип
|
||||||
|
* `MutableJournalStore`, не выставлен наружу через [pw.binom.agentik.proto.Agent]).
|
||||||
|
*/
|
||||||
|
internal class HttpJournalStore(
|
||||||
|
private val httpClient: HttpClient,
|
||||||
|
private val baseUrl: String,
|
||||||
|
) : JournalStore {
|
||||||
|
|
||||||
|
private val agentUrl: String = baseUrl.trimEnd('/')
|
||||||
|
|
||||||
|
override suspend fun list(
|
||||||
|
conversationId: String,
|
||||||
|
after: Instant,
|
||||||
|
offset: Int,
|
||||||
|
limit: Int,
|
||||||
|
): List<MessageRecord> {
|
||||||
|
val response = httpClient.get("$agentUrl/journal/conversations/$conversationId/messages") {
|
||||||
|
parameter("after", after.toString())
|
||||||
|
parameter("offset", offset)
|
||||||
|
parameter("limit", limit)
|
||||||
|
}
|
||||||
|
check(response.status == HttpStatusCode.OK) {
|
||||||
|
"journal.list: server returned ${response.status}"
|
||||||
|
}
|
||||||
|
return response.body<List<MessageRecord>>()
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun close() {
|
||||||
|
// HttpClient закрывает владелец (AgentClient / AgentikAgent).
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user