proto/server/client: Agent.info, Working-маркер хода, /record, таймауты запросов
- proto: Agent.info (AgentInfo: name/description/usefulness) — человекочитаемое
имя отдельно от opaque id; сериализация и тесты
- server: GET {path} отдаёт Agent.info; GET /conversations/{id}/record —
ConversationRecord без handle'а (для клиентского кэша после AgentEvent.Created)
- outbox: Event.Working — первый event хода, эмитится из Conversation.send()
до LLM-цикла и turnLock, чтобы UI показал спиннер сразу
- standalone: AGENTIK_NAME/AGENTIK_DESCRIPTION/AGENTIK_USEFULNESS → AgentInfo
- client: AgentClient.create eagerly фетчит info (GET {baseUrl})
- client: HttpConversationStore.get читает /record (ConversationRecord,
а не ConversationSnapshot — рассинхрон типов)
- client: noReadTimeout() на POST /conversations/{id}/messages — сервер отвечает
по завершении всего хода агента (реально 0.5–144 с), дефолтные 15 с рвали
живую реплику на клиенте
- journal-api: ConversationRecord @Serializable
- ksqlite 0.1.3 → 0.1.4
- .gitignore: runtime-данные standalone-агента и hs_err-дампы
This commit is contained in:
@@ -16,6 +16,7 @@ import pw.binom.agentik.journal.ConversationStore
|
||||
import pw.binom.agentik.journal.JournalStore
|
||||
import pw.binom.agentik.outbox.OutboxStore
|
||||
import pw.binom.agentik.proto.Agent
|
||||
import pw.binom.agentik.proto.AgentInfo
|
||||
import pw.binom.agentik.proto.Conversation
|
||||
import kotlin.time.Instant
|
||||
|
||||
@@ -27,9 +28,16 @@ import kotlin.time.Instant
|
||||
* **Storage handles** ([journal], [outbox], [conversationStore]) — read-only
|
||||
* views на серверные хранилища. Запись — только через команды
|
||||
* [createConversation] / [deleteConversation] / [renameConversation].
|
||||
*
|
||||
* Конструируется через suspend [Companion.create], который **eagerly**
|
||||
* фетчит [info] (`GET {baseUrl}`) и сохраняет снимок в поле. Это убирает
|
||||
* необходимость в `lazy { runBlocking { ... } }` на горячем пути —
|
||||
* `runBlocking` живёт один раз в [Companion.create], оттуда же [AgentikAgent]
|
||||
* его и вызывает (там он приемлем: одноразовая инициализация агента).
|
||||
*/
|
||||
internal class AgentClient(
|
||||
internal class AgentClient private constructor(
|
||||
override val id: String,
|
||||
override val info: AgentInfo,
|
||||
private val baseUrl: String,
|
||||
private val httpClient: HttpClient,
|
||||
) : Agent {
|
||||
@@ -74,4 +82,32 @@ internal class AgentClient(
|
||||
override fun close() {
|
||||
httpClient.close()
|
||||
}
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* Создаёт [AgentClient] и eagerly загружает [Agent.info]
|
||||
* (`GET {baseUrl}` на серверном фасаде). Любой сбой сети на этом
|
||||
* этапе пробрасывается как исключение — агент без `info` бесполезен
|
||||
* (UI/A2A сразу упрутся в `agent.info`).
|
||||
*
|
||||
* Single-shot инициализация, `runBlocking` тут допустим (см. KDoc
|
||||
* класса). Хосты, которым нужен полностью неблокирующий старт,
|
||||
* могут обернуть вызов в свой `CoroutineScope`.
|
||||
*/
|
||||
suspend fun create(
|
||||
id: String,
|
||||
baseUrl: String,
|
||||
httpClient: HttpClient,
|
||||
): AgentClient {
|
||||
val agentUrl = baseUrl.trimEnd('/')
|
||||
val info: AgentInfo = httpClient.get(agentUrl).body()
|
||||
return AgentClient(
|
||||
id = id,
|
||||
info = info,
|
||||
baseUrl = agentUrl,
|
||||
httpClient = httpClient,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import pw.binom.agentik.journal.ConversationRecord
|
||||
@@ -95,7 +96,7 @@ fun AgentikAgent(
|
||||
token: String? = null,
|
||||
): Agent {
|
||||
val httpClient = agentikHttpClient(engineFactory = engineFactory, token = token)
|
||||
val client = AgentClient(id = id, baseUrl = baseUrl, httpClient = httpClient)
|
||||
val client = runBlocking { AgentClient.create(id = id, baseUrl = baseUrl, httpClient = httpClient) }
|
||||
return wrapWithLocalConversationCache(client, scopeClient = client)
|
||||
}
|
||||
|
||||
|
||||
@@ -59,6 +59,9 @@ internal class ConversationClient(
|
||||
httpClient.post("$convUrl/messages") {
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(SendPayload(content, context))
|
||||
// Сервер отвечает только по завершении хода агента (LLM + тулы),
|
||||
// а это минуты, а не 15 секунд дефолтного request-timeout.
|
||||
noReadTimeout()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -33,7 +33,10 @@ internal class HttpConversationStore(
|
||||
private val agentUrl: String = baseUrl.trimEnd('/')
|
||||
|
||||
override suspend fun get(id: String): ConversationRecord? {
|
||||
val response = httpClient.get("$agentUrl/conversations/$id")
|
||||
// `/record` (а НЕ `/conversations/{id}`): последний отдаёт
|
||||
// ConversationSnapshot для `AgentClient.getConversation`, у которого
|
||||
// другой shape (handle + isImageSupported, без createdAt/updatedAt).
|
||||
val response = httpClient.get("$agentUrl/conversations/$id/record")
|
||||
if (response.status == HttpStatusCode.NotFound) return null
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"conversationStore.get($id): server returned ${response.status}"
|
||||
|
||||
@@ -53,7 +53,7 @@ internal class HttpEventStore(
|
||||
append("$agentUrl/outbox/events")
|
||||
if (after != null) append("?after=$after")
|
||||
}
|
||||
httpClient.prepareGet(url) { noSseReadTimeout() }
|
||||
httpClient.prepareGet(url) { noReadTimeout() }
|
||||
.execute { response ->
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"events: server returned ${response.status}"
|
||||
@@ -74,7 +74,7 @@ internal class HttpEventStore(
|
||||
append("$agentUrl/events")
|
||||
if (after != null) append("?after=$after")
|
||||
}
|
||||
httpClient.prepareGet(url) { noSseReadTimeout() }
|
||||
httpClient.prepareGet(url) { noReadTimeout() }
|
||||
.execute { response ->
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"agentEvents: server returned ${response.status}"
|
||||
@@ -104,7 +104,7 @@ internal class HttpEventStore(
|
||||
append("$agentUrl/conversations/$conversationId/events")
|
||||
if (after != null) append("?after=$after")
|
||||
}
|
||||
httpClient.prepareGet(url) { noSseReadTimeout() }
|
||||
httpClient.prepareGet(url) { noReadTimeout() }
|
||||
.execute { response ->
|
||||
check(response.status == HttpStatusCode.OK) {
|
||||
"conversationEvents: server returned ${response.status}"
|
||||
|
||||
@@ -8,18 +8,23 @@ import io.ktor.client.request.HttpRequestBuilder
|
||||
* Отключает request/connect/socket-таймауты для конкретного запроса через
|
||||
* [HttpTimeoutCapability] со всеми таймаутами = [HttpTimeoutConfig.INFINITE_TIMEOUT_MS].
|
||||
*
|
||||
* Зачем: наш SSE-ридер ([readSse]) читает `bodyAsChannel()` руками и не
|
||||
* использует плагин `SSE`, поэтому движок не считает запрос SSE-шным
|
||||
* (`HttpRequestBuilder.supportsRequestTimeout` проверяет
|
||||
* `body is SSEClientContent`, а у нас тело — обычный GET без тела).
|
||||
* Без capability встроенный `HttpTimeoutPlugin.requestTimeoutMillis` (по умолчанию
|
||||
* **15000 мс**) молча убивает долгий idle-стрим через 15 секунд.
|
||||
* Зачем: два вида запросов живут дольше дефолтных 15 секунд:
|
||||
* - **SSE-чтение** ([readSse]) — читает `bodyAsChannel()` руками и не
|
||||
* использует плагин `SSE`, поэтому движок не считает запрос SSE-шным
|
||||
* ([HttpRequestBuilder.supportsRequestTimeout] проверяет
|
||||
* `body is SSEClientContent`, а у нас тело — обычный GET без тела).
|
||||
* - **`POST /conversations/{id}/messages`** — сервер отвечает не сразу, а
|
||||
* только когда ход агента полностью завершён (LLM + тулы). Реальный ход
|
||||
* легко длится минуты, и дефолтный request-timeout убивал бы его на
|
||||
* 15-й секунде, обрывая ещё живой ход на сервере.
|
||||
* Без capability встроенный `HttpTimeoutPlugin.requestTimeoutMillis`
|
||||
* (по умолчанию **15000 мс**) молча убивает такой запрос.
|
||||
*
|
||||
* Конфиг создаётся заново на каждый вызов — плагин `HttpTimeout` при
|
||||
* установленном capability мутирует его поля через `?:`, так что шаренный
|
||||
* инстанс мог бы утечь между запросами.
|
||||
*/
|
||||
internal fun HttpRequestBuilder.noSseReadTimeout() {
|
||||
internal fun HttpRequestBuilder.noReadTimeout() {
|
||||
setCapability(
|
||||
HttpTimeoutCapability,
|
||||
HttpTimeoutConfig(
|
||||
|
||||
Reference in New Issue
Block a user