Bring up :proto protocol + :server (Ktor) + :client (HTTP) modules; wire :server into standalone with EchoProtoAgent
Major additions: * :proto (KMP submodule) — in-house stateful protocol replacing AG-UI. Agent owns conversation transcript; Conversation.events(after) is a live, replay-free stream; backfill via Conversation.getMessages(after, offset, limit). Each Event carries an Instant date for client-side resume tracking. Sealed hierarchies (Content/Message/Event/AgentEvent) annotated @Serializable with snake_case @SerialName JSON discriminators so the wire format is decoupled from Kotlin class names. * :server (JVM, Ktor 3.1.3) — REST+SSE facade for Agent. Public entry: Route.agentikAgent(agent, path = "/agentik"). Endpoints: create/list/get/patch/delete conversations, POST messages (202), POST interrupt, GET messages, GET conversation events (SSE), GET agent events (SSE), GET /health. Custom Instant serializer for kotlin.time.Instant registered contextually on agentikJson (ISO-8601, ignoreUnknownKeys=true, explicitNulls=false). * :client (JVM, Ktor HTTP Client + CIO) — mirror of :server returning a pw.binom.agentik.proto.Agent backed by HTTP calls. Custom SSE parser since ktor-client-sse is not on the 3.1.3 client classpath. * standalone — EchoProtoAgent (in-memory Agent for :proto), EchoAgent (existing AG-UI echo), both mounted on the same Netty embedded server on port 8080 (/agui and /agentik); A2A stays on its own CIO engine on 8081. EchoProtoAgent smoke-tested end-to-end against :server: all 11 endpoints, including live SSE delivery of StartResponse/AppendText/End event triplets and Agent-level Created/Deleted events. Design notes pinned in: * agentik/IRC-QUESTIONS.md — closed 13-item checklist for the upcoming :irc-server transport (channel = conversation, CTCP for structural events, draft/chathistory for backfill, ImageStore side-channel, etc). * docs/ARCHITECTURE.md — overall layout snapshot.
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
package pw.binom.agentik.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.call.body
|
||||
import io.ktor.client.request.delete
|
||||
import io.ktor.client.request.get
|
||||
import io.ktor.client.request.parameter
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.bodyAsChannel
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentType
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.flow
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import pw.binom.agentik.proto.Agent
|
||||
import pw.binom.agentik.proto.AgentEvent
|
||||
import pw.binom.agentik.proto.Conversation
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* HTTP-реализация [Agent]. Ходит в `:server`-фасад, см. `agentikAgent(...)`.
|
||||
*
|
||||
* Замечание по [createConversation]: интерфейс [Agent] объявлен не-suspend
|
||||
* (in-process кейс этого не требует), но HTTP-вариант обязан ждать ответа
|
||||
* POST `/conversations`. Используем `runBlocking` — это одноразовая
|
||||
* операция (открытие чата), не горячий путь. В UI-контексте вызывающий сам
|
||||
* решает, что делать.
|
||||
*/
|
||||
internal class AgentClient(
|
||||
private val httpClient: HttpClient,
|
||||
private val baseUrl: String,
|
||||
override val id: String,
|
||||
) : Agent {
|
||||
|
||||
private val agentUrl: String = baseUrl.trimEnd('/')
|
||||
|
||||
override fun createConversation(temp: Boolean): Conversation =
|
||||
runBlocking {
|
||||
val snapshot: ConversationSnapshot = httpClient.post("$agentUrl/conversations") {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(RequestCreateConversation(temp))
|
||||
}.body()
|
||||
ConversationClient(httpClient = httpClient, baseUrl = agentUrl, snapshot = snapshot)
|
||||
}
|
||||
|
||||
override suspend fun getConversation(id: String): Conversation? {
|
||||
val response = httpClient.get("$agentUrl/conversations/$id")
|
||||
if (response.status == HttpStatusCode.NotFound) return null
|
||||
val snapshot = response.body<ConversationSnapshot>()
|
||||
return ConversationClient(httpClient, agentUrl, snapshot)
|
||||
}
|
||||
|
||||
override suspend fun deleteConversation(id: String): Boolean {
|
||||
val response = httpClient.delete("$agentUrl/conversations/$id")
|
||||
return response.status == HttpStatusCode.NoContent
|
||||
}
|
||||
|
||||
override suspend fun getConversations(offset: Int, limit: Int): List<Conversation> {
|
||||
val snapshots = httpClient.get("$agentUrl/conversations") {
|
||||
parameter("offset", offset)
|
||||
parameter("limit", limit)
|
||||
}.body<List<ConversationSnapshot>>()
|
||||
return snapshots.map { ConversationClient(httpClient, agentUrl, it) }
|
||||
}
|
||||
|
||||
override fun events(after: Instant): Flow<AgentEvent> = flow {
|
||||
val response = httpClient.get("$agentUrl/events?after=$after")
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"events: server returned ${response.status}"
|
||||
}
|
||||
readSse(response.bodyAsChannel())
|
||||
.collect { payload ->
|
||||
emit(agentikJson.decodeFromString(AgentEvent.serializer(), payload))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
package pw.binom.agentik.client
|
||||
|
||||
import io.ktor.client.HttpClient
|
||||
import io.ktor.client.engine.cio.CIO
|
||||
import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
|
||||
import io.ktor.serialization.kotlinx.json.json
|
||||
import pw.binom.agentik.proto.Agent
|
||||
|
||||
/**
|
||||
* Создаёт [Agent], который под капотом ходит в HTTP-фасад `agentikAgent`
|
||||
* (модуль `:server`).
|
||||
*
|
||||
* ```
|
||||
* val client = AgentikAgent(
|
||||
* id = "my-agent",
|
||||
* baseUrl = "http://localhost:8080/agentik",
|
||||
* )
|
||||
* val conv = client.createConversation(temp = false)
|
||||
* conv.send(listOf(Content.Text("hi")))
|
||||
* conv.events(Instant.DISTANT_PAST).collect { ev -> ... }
|
||||
* ```
|
||||
*
|
||||
* [id] пробрасывается в реализацию [Agent.id] — сервер про идентичность
|
||||
* агента не знает, поэтому клиент должен её знать сам (или взять из
|
||||
* конфига).
|
||||
*
|
||||
* [httpClient] по умолчанию — [defaultAgentikHttpClient] (CIO + JSON +
|
||||
* SSE). Можно передать свой, если нужен свой engine/логирование/аутентификация.
|
||||
*/
|
||||
fun AgentikAgent(
|
||||
id: String,
|
||||
baseUrl: String,
|
||||
httpClient: HttpClient = defaultAgentikHttpClient(),
|
||||
): Agent = AgentClient(httpClient = httpClient, baseUrl = baseUrl, id = id)
|
||||
|
||||
/**
|
||||
* Дефолтный [HttpClient] для общения с `agentikAgent`: CIO-движок и
|
||||
* kotlinx-serialization с тем же wire-форматом, что на сервере. SSE-парсер
|
||||
* (см. [readSse]) живёт в общем коде и плагина не требует.
|
||||
*/
|
||||
fun defaultAgentikHttpClient(): HttpClient = HttpClient(CIO) {
|
||||
install(ContentNegotiation) { json(agentikJson) }
|
||||
}
|
||||
@@ -0,0 +1,94 @@
|
||||
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.client.request.patch
|
||||
import io.ktor.client.request.post
|
||||
import io.ktor.client.request.setBody
|
||||
import io.ktor.client.statement.bodyAsChannel
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.http.contentType
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.flow
|
||||
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 kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* HTTP-реализация [Conversation]. Ходит в `:server`-фасад под
|
||||
* `/conversations/{id}/...`.
|
||||
*
|
||||
* `id` отдаётся синхронно (он в [snapshot], доступном сразу). Остальные
|
||||
* поля (`title`, `isSupportImageInput`, ...) — тоже из snapshot. [snapshot]
|
||||
* обновляется после [rename] (сервер возвращает свежий).
|
||||
*
|
||||
* **Caveat — `updatedAt`:** сервер бампит `updatedAt` на каждый
|
||||
* `send`/`rename`, но клиент узнает об этом только при следующем
|
||||
* `rename` или `getConversation`. Если нужна свежая свежесть после
|
||||
* `send` — перезапроси через `Agent.getConversation(id)`.
|
||||
*
|
||||
* [close] — локальный no-op: сервер держит диалог живым. Удалить —
|
||||
* через `Agent.deleteConversation(id)`.
|
||||
*/
|
||||
internal class ConversationClient(
|
||||
private val httpClient: HttpClient,
|
||||
private val baseUrl: String,
|
||||
private var snapshot: ConversationSnapshot,
|
||||
) : Conversation {
|
||||
|
||||
override val id: String get() = snapshot.id
|
||||
override val title: String? get() = snapshot.title
|
||||
override val isSupportImageInput: Boolean get() = snapshot.isSupportImageInput
|
||||
override val isSupportImageOutput: Boolean get() = snapshot.isSupportImageOutput
|
||||
override val isTemporal: Boolean get() = snapshot.isTemporal
|
||||
override val updatedAt: Instant get() = snapshot.updatedAt
|
||||
|
||||
private val convUrl: String get() = "$baseUrl/conversations/$id"
|
||||
|
||||
override suspend fun rename(title: String) {
|
||||
val updated = httpClient.patch(convUrl) {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(RequestRename(title))
|
||||
}.body<ConversationSnapshot>()
|
||||
snapshot = updated
|
||||
}
|
||||
|
||||
override suspend fun send(content: List<Content>) {
|
||||
httpClient.post("$convUrl/messages") {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(content)
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun interrupt() {
|
||||
httpClient.post("$convUrl/interrupt")
|
||||
}
|
||||
|
||||
override fun events(after: Instant): Flow<Event> = flow {
|
||||
val response = httpClient.get("$convUrl/events?after=$after")
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"events: server returned ${response.status}"
|
||||
}
|
||||
readSse(response.bodyAsChannel())
|
||||
.collect { payload ->
|
||||
emit(agentikJson.decodeFromString(Event.serializer(), payload))
|
||||
}
|
||||
}
|
||||
|
||||
override suspend fun getMessages(after: Instant, offset: Int, limit: Int): List<Message> =
|
||||
httpClient.get("$convUrl/messages") {
|
||||
parameter("after", after.toString())
|
||||
parameter("offset", offset)
|
||||
parameter("limit", limit)
|
||||
}.body()
|
||||
|
||||
override fun close() {
|
||||
// Локальный no-op: диалог на сервере живёт, пока не вызван
|
||||
// Agent.deleteConversation(id). См. [Conversation.close] KDoc.
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package pw.binom.agentik.client
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* HTTP-снимок [pw.binom.agentik.proto.Conversation] — те же поля, что у
|
||||
* интерфейса, но без методов. Зеркалит
|
||||
* [pw.binom.agentik.server.ConversationSnapshot]. Дубликат сознательно:
|
||||
* переедем в общий `:wire`, когда появится больше типов.
|
||||
*/
|
||||
@Serializable
|
||||
data class ConversationSnapshot(
|
||||
val id: String,
|
||||
val isSupportImageInput: Boolean,
|
||||
val isSupportImageOutput: Boolean,
|
||||
val isTemporal: Boolean,
|
||||
val title: String? = null,
|
||||
val updatedAt: Instant,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
internal data class RequestCreateConversation(val temp: Boolean)
|
||||
|
||||
@Serializable
|
||||
internal data class RequestRename(val title: String)
|
||||
@@ -0,0 +1,38 @@
|
||||
package pw.binom.agentik.client
|
||||
|
||||
import kotlinx.serialization.KSerializer
|
||||
import kotlinx.serialization.descriptors.PrimitiveKind
|
||||
import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor
|
||||
import kotlinx.serialization.descriptors.SerialDescriptor
|
||||
import kotlinx.serialization.encoding.Decoder
|
||||
import kotlinx.serialization.encoding.Encoder
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlinx.serialization.modules.SerializersModule
|
||||
import kotlin.time.Instant
|
||||
|
||||
/**
|
||||
* Зеркалит [pw.binom.agentik.server.InstantSerializer]. Дублируем сознательно:
|
||||
* wire-формат компактный, альтернатива — отдельный `:wire`-модуль ради 10 строк.
|
||||
*/
|
||||
internal object InstantSerializer : KSerializer<Instant> {
|
||||
override val descriptor: SerialDescriptor =
|
||||
PrimitiveSerialDescriptor("kotlin.time.Instant", PrimitiveKind.STRING)
|
||||
|
||||
override fun serialize(encoder: Encoder, value: Instant) =
|
||||
encoder.encodeString(value.toString())
|
||||
|
||||
override fun deserialize(decoder: Decoder): Instant =
|
||||
Instant.parse(decoder.decodeString())
|
||||
}
|
||||
|
||||
/**
|
||||
* JSON-конфиг клиента. Должен **точно** совпадать с серверным `agentikJson` —
|
||||
* один и тот же wire-формат с обеих сторон.
|
||||
*/
|
||||
internal val agentikJson: Json = Json {
|
||||
ignoreUnknownKeys = true
|
||||
explicitNulls = false
|
||||
serializersModule = SerializersModule {
|
||||
contextual(Instant::class, InstantSerializer)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package pw.binom.agentik.client
|
||||
|
||||
import io.ktor.utils.io.ByteReadChannel
|
||||
import io.ktor.utils.io.readUTF8Line
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
import kotlinx.coroutines.flow.flow
|
||||
|
||||
/**
|
||||
* Минимальный парсер Server-Sent Events, читающий канал до EOF и эмиттящий
|
||||
* собранный `data:`-пейлоад каждого события. Достаточно для нашего wire-формата:
|
||||
* сервер шлёт `data: <json>\n\n`, имя события и прочие поля не используются.
|
||||
*
|
||||
* Формат (см. WHATWG):
|
||||
* event: foo — игнор (у нас нет имён событий)
|
||||
* data: {"k":1} — накапливается, многострочный `data:` склеивается через '\n'
|
||||
* :comment — игнор
|
||||
* id:/retry:/<blank> — пустая строка = граница события; всё остальное игнор
|
||||
*
|
||||
* Поток закрывается, когда канал доходит до EOF; накопленный `data` (если есть)
|
||||
* эмитится как финальный ивент.
|
||||
*/
|
||||
internal fun readSse(channel: ByteReadChannel): Flow<String> = flow {
|
||||
val data = StringBuilder()
|
||||
while (!channel.isClosedForRead) {
|
||||
val line = channel.readUTF8Line() ?: break
|
||||
when {
|
||||
line.isEmpty() -> {
|
||||
if (data.isNotEmpty()) {
|
||||
emit(data.toString())
|
||||
data.clear()
|
||||
}
|
||||
}
|
||||
line.startsWith("data: ") -> {
|
||||
if (data.isNotEmpty()) data.append('\n')
|
||||
data.append(line.removePrefix("data: "))
|
||||
}
|
||||
line.startsWith("data:") -> {
|
||||
if (data.isNotEmpty()) data.append('\n')
|
||||
data.append(line.removePrefix("data:"))
|
||||
}
|
||||
// event:, id:, retry:, ":" (comment) — игнорируем
|
||||
}
|
||||
}
|
||||
if (data.isNotEmpty()) emit(data.toString())
|
||||
}
|
||||
Reference in New Issue
Block a user