feat(client): add ReconnectingOutbox with parallel connectionStatus flow
ci / JVM build + tests (push) Successful in 6m3s
release / Publish KMP libraries → caffeine Nexus (release) Failing after 23s

Android-client review item 9: every client reimplements SSE reconnect
with cursor preservation, exponential backoff, and connection-status UI
signals. ReconnectingOutbox extracts that into the lib.

Design:
- wraps any OutboxStore (HttpEventStore or local InMemoryJournalStore)
- two INDEPENDENT parallel flows — never mixed:
  - events(after): Flow<CommonEvent> with auto-reconnect, cursor
    (lastSeen) preserved across retries, so client never loses events
  - connectionStatus(): Flow<ConnectionStatus> = Connecting(attempt) /
    Connected(since) / Disconnected(reason, willRetryIn) / Failed(cause)
    — for UI banner / spinner; NOT emitted into CommonEvent stream
- BackoffPolicy.Default: initial=1s, max=30s, multiplier=2.0,
  jitter=0.2 (±20% spread), maxAttempts=∞
- BackoffPolicy.Fixed(delay, attempts) for tests
- After maxAttempts exhaustion: Failed + flow closes
- recon.close() cancels background job, both flows terminate

Tests (4 cases, all green):
- first event → Connecting(1) + Connected + event delivered
- disconnect mid-stream → Disconnected → Connecting(2) → resume from
  lastSeen cursor (no duplicate)
- exhausted attempts → Failed + 0 events
- close() → background loop cancelled, no further emissions

client/README.md: new 'Auto-reconnect для живого outbox' section with
usage example (two parallel scope.launch blocks) + parameter table.

jvmTest green (95 tasks, includes 4 new ReconnectingOutboxTest cases).
This commit is contained in:
2026-09-22 02:55:19 +03:00
parent 29851c047a
commit c0a933d251
3 changed files with 522 additions and 1 deletions
+72 -1
View File
@@ -16,6 +16,11 @@
- `HttpJournalStore` — `list(convId, after, offset, limit)` → `List<MessageRecord>`
со всеми типами записей (User/Assistant/ToolCall/ToolResult/Error + tokens).
- `HttpEventStore` — `events` / `agentEvents` / `conversationEvents` (SSE).
- `ReconnectingOutbox(outbox, scope, policy)` — обёртка над `OutboxStore` с
авто-reconnect при обрыве стрима (exponential backoff). Два независимых
потока: `events()` (те же `CommonEvent`) и `connectionStatus()`
(`Connecting`/`Connected`/`Disconnected`/`Failed`) — статус НЕ мешается
с основным потоком событий. См. ниже.
`Agent` — `AutoCloseable`; `agent.close()` закрывает HttpClient. Не нужно
вручную создавать `HttpClient` и накатывать на него JSON/Bearer-плагины.
@@ -36,6 +41,30 @@ dependencies {
}
```
## Что клиент хранит локально (persistence)
Либа **не** имеет `SettingsRepository` / `Config` — это намеренно: UI-фреймворки хранят настройки по-разному (JSON-файл, Keychain, Android DataStore, NSUserDefaults, ...). Либа не навязывает формат, но клиент должен сериализовать у себя минимум:
| Поле | Что это | Где взять |
|---|---|---|
| `clientId` (параметр `id` в `AgentikAgent`) | Идентичность клиента в логах сервера (X-Client-Id header). Не user-id в агенте, не device-id — это **произвольная строка клиента**, обычно `<app-name>-<installation-uuid>`. Сервер использует для log multiplexing и не интерпретирует. | Генерируется один раз при первом запуске (`UUID.randomUUID().toString()`) и сохраняется. Никогда не меняется. |
| `baseUrl` | URL сервера (`http://host:8080/agentik`). Должен включать path-prefix фасада, не только хост. | Из настроек пользователя / дефолт |
| `token` | Bearer-токен. `null` = анонимный доступ (если сервер разрешает). | Из настроек пользователя / secure-storage |
Опционально (для UX): `engineFactory` — обычно compile-time выбор по платформе (`CIO` JVM/Native, `OkHttp` JVM, `Darwin` iOS/macOS).
Минимальный JSON для UI, который хранит в файле:
```json
{
"clientId": "my-android-app-550e8400-e29b-41d4-a716-446655440000",
"baseUrl": "https://agent.example.com/agentik",
"token": "s3cret"
}
```
⚠️ `clientId` **генерируется один раз** при установке и больше не меняется — иначе сломается log multiplexing на сервере.
## Быстрый старт: свой клиент за 5 минут
Один self-contained пример: создаём агента, открываем диалог,
@@ -322,7 +351,49 @@ UI-обновление списка — отдельная задача, реш
```
Покрывают: JSON-парсинг `Event`-ов, SSE-стрим, recovery после разрыва,
401/404.
401/404, reconnect-cycle `ReconnectingOutbox` (4 кейса: успех / обрыв +
reconnect / exhausted attempts → Failed / close → cancel).
## Auto-reconnect для живого outbox
Базовый `OutboxStore.events(after)` — cold SSE-стрим, при обрыве (мобильная
сеть, рестарт сервера) клиент сам должен реконнектиться с `after = lastEventDate`.
Это повторяется в каждом клиенте. `ReconnectingOutbox` берёт это на себя:
```kotlin
val recon = ReconnectingOutbox(
outbox = agent.outbox, // или HttpEventStore
scope = myScreenScope,
policy = BackoffPolicy.Default, // 1s → 2s → ... → 30s, ±20% jitter
)
scope.launch { recon.events(Instant.DISTANT_PAST).collect { handle(it) } }
scope.launch {
recon.connectionStatus().collect { status ->
when (status) {
is Connecting -> ui.showBanner("connecting...")
is Connected -> ui.hideBanner()
is Disconnected -> ui.showBanner("reconnecting in ${status.willRetryIn}…")
is Failed -> ui.showError(status.cause)
}
}
}
// На выходе (например, navigation back):
recon.close() // отменяет background-loop, потоки терминируются
```
Два потока **независимы** — `events()` содержит только `CommonEvent`,
`connectionStatus()` содержит только `ConnectionStatus`. Никакого
"мешающего" `Connecting`/`Disconnected` в потоке событий.
Параметры backoff (см. `BackoffPolicy`):
- `initial` / `max` — границы задержки
- `multiplier` — множитель на каждом шаге
- `jitter` — рандом-разброс (по умолчанию 20%)
- `maxAttempts` — лимит попыток; после — `Failed` + закрытие потока
Если нужен фиксированный delay для тестов — `BackoffPolicy.Fixed(10.milliseconds, attempts = 3)`.
## Известное ограничение