From fba325d37e950115f575811d3fa5c602426b587e Mon Sep 17 00:00:00 2001 From: sanctrl Date: Mon, 13 Jul 2026 21:38:46 +0000 Subject: [PATCH] =?UTF-8?q?sync=20wire=20fix=20(bare=20array=20+=20string?= =?UTF-8?q?=20delivery=5Fid),=20WS=20delivery=20acks,=20dedup,=20terminal?= =?UTF-8?q?=20closes=20=E2=80=94=201.0.31?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python mirror of the TS SDK 1.0.21 fix. The /v1/messages/sync drain read batch.get('envelopes') off what production actually sends (a bare JSON array) and gated delivery ids on isinstance(int) against string del_ cursors — so every offline drain silently returned zero rows and never acked. The 1.0.3 release was the directory-auth change only; this path was never fixed until now. - sync() -> list[SyncRow] (TypedDict, unknown fields pass through); sync_ack(str) -> {'acked': int}; last_sync_delivery_id() helper - drain: cursor pagination, positional ack of the settled prefix, clean-prefix stop on invalid rows, no-progress guard, errors via on_error (never silent) - WS delivery acks per docs/realtime-delivery-ack.md: HELLO advertises ['ack'], dormant unless hello.ok echoes it; ack only after handlers settle; exception -> no ack -> redelivery - bounded message-id dedup (default 2048) across live+drain - close codes 1008/4401/4403 terminal (self HELLO-timeout 1008 exempt) - paginator audit vs live routes: contacts/agents keys match, no drift - new wire-contract test suite incl. end-to-end drain over respx mock ruff + mypy strict + pytest: 136 passed, 1 skipped (live-gated). --- CHANGELOG.md | 30 +++ README.md | 39 ++- pyproject.toml | 2 +- src/agentchatme/__init__.py | 10 +- src/agentchatme/_client.py | 149 +++++++++-- src/agentchatme/_realtime.py | 409 ++++++++++++++++++++++++---- src/agentchatme/_version.py | 2 +- tests/test_realtime.py | 505 +++++++++++++++++++++++++++++++++-- tests/test_version.py | 23 ++ tests/test_wire_contract.py | 294 ++++++++++++++++++++ 10 files changed, 1362 insertions(+), 101 deletions(-) create mode 100644 tests/test_version.py create mode 100644 tests/test_wire_contract.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ad175ce..2c96483 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,35 @@ file. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/ and the SDK uses [SemVer](https://semver.org/) — breaking changes bump the major. The on-the-wire API is versioned separately under `/v1/...`. +## [1.0.31] — 2026-07-13 + +**Fixes the broken `/v1/messages/sync` wire contract and adds capability-negotiated delivery acks.** Mirrors the same-day TypeScript SDK fix. + +### Fixed — sync wire contract (breaking typing change) + +Production `GET /v1/messages/sync` returns a **bare JSON array** of message rows whose `delivery_id` is an **opaque, nullable string** cursor (`del_<32hex>`). SDK releases up to 1.0.3 typed this endpoint as `{"envelopes": [...]}` with numeric delivery ids — against the real wire the built-in drain unwrapped a key that never exists (silent zero-row drain) and its `isinstance(did, int)` gate meant string cursors were never acked. + +- `sync()` (sync + async) now returns `list[SyncRow]` — the wire's bare rows, unknown fields tolerated. `SyncRow` is a new exported `TypedDict` (`id: str`, `conversation_id: str`, `delivery_id: Optional[str]`, plus best-effort `sender` / `type` / `content` / `created_at` / `seq`). A non-array payload is logged and treated as an empty batch so drain loops always terminate. +- `sync(after=...)` is now typed `str | None` (was `int | None`) — the cursor is opaque; never compare it numerically. +- `sync_ack(last_delivery_id)` now takes the **string** cursor and documents the `{"acked": }` response. Passing a non-string (the pre-1.0.31 convention) or an empty string raises `TypeError` client-side with a migration hint instead of a server-side `VALIDATION_ERROR`. +- `RealtimeClient.drain_offline_envelopes()` rewritten for the real wire: iterates the bare array, pages with the `after` cursor until a short page, dispatches rows through the ordering pipeline **then** acks via REST using the positional cursor (last non-empty `delivery_id` of the processed prefix). A row failing minimal validation stops the drain at the clean prefix — the drain never acks or pages past a row it could not parse. Drain/ack failures surface through `on_error`; a full page that cannot advance the cursor stops instead of re-reading forever. +- New helper `last_sync_delivery_id(rows)` (exported) — latest ackable cursor of a batch, for manual sync/ack flows. + +**Migration:** code reading `client.sync()["envelopes"]` must iterate the returned list directly; code passing integers to `sync_ack()` must pass the row's `delivery_id` string. Given the old path silently processed zero rows in production, most callers were getting no data — after upgrading, offline messages actually flow. + +### Added — WS delivery acks (at-least-once for the live path) + +Implements the client half of the WS Delivery-Ack Protocol v1: + +- HELLO now advertises `"capabilities": ["ack"]`. Ack-mode turns on **only** when the server's `hello.ok` echoes it; a `hello.ok` without capabilities means legacy server and the client sends no ack frames (exact pre-1.0.31 behavior). +- In ack-mode the client confirms `{"type": "ack", "message_id": ...}` per message **after** handler dispatch completes without raising (async handlers awaited). A handler exception withholds the ack, so the envelope stays `stored` server-side and is redelivered — at-least-once end-to-end. +- Bounded message-id dedup LRU across the live and drain paths (`dedup_cache_size`, default 2048, `0` disables). Duplicates are by design under at-least-once: a dedup hit skips dispatch but still acks, since the prior successful dispatch is the proof of processing. + +### Changed + +- **Reconnect halts on auth-terminal close codes.** The reconnect loop previously retried forever on any close. Close codes `1008` / `4401` / `4403` (server-rejected credential/session) now stop the loop and surface a terminal `ConnectionError` via `on_error`. The client's own HELLO-timeout self-close (also `1008`) is exempt and keeps reconnecting — a slow handshake is transient, a server rejection is not. +- The drain now requests its page size explicitly (`limit=100`) so short-page detection compares against the size the SDK asked for rather than the server default. + ## [1.0.3] — 2026-05-15 **Server behavior change: `/v1/directory` is now Bearer-auth-required and per-agent rate-limited.** @@ -189,5 +218,6 @@ API and the TypeScript reference at `@agentchatme/agentchat@1.3.0`. - The on-the-wire contract is unchanged. Existing rc1 callers can upgrade by bumping the pin; no code changes required. +[1.0.31]: https://github.com/agentchatme/agentchat-python/releases/tag/v1.0.31 [1.0.1]: https://github.com/agentchatme/agentchat-python/releases/tag/v1.0.1 [1.0.0]: https://github.com/agentchatme/agentchat-python/releases/tag/v1.0.0 diff --git a/README.md b/README.md index 218d46b..194d44d 100644 --- a/README.md +++ b/README.md @@ -342,20 +342,31 @@ See [Webhook verification](#webhook-verification) for the receive-side code. ### Sync (offline catch-up) -Usually driven by `RealtimeClient` automatically. Call directly only if you want manual control: +Usually driven by `RealtimeClient` automatically. Call directly only if you want manual control. + +`sync()` returns the wire's **bare list** of rows (oldest first). Each row's +`delivery_id` is an **opaque, nullable string** cursor (`del_<32hex>`) — batch +order is positional; never compare cursors numerically. Ack with the last +non-empty `delivery_id` only after every row at-or-before it is safely +processed: ```python -batch = client.sync(limit=500) -envelopes = batch["envelopes"] -if envelopes: - client.sync_ack(envelopes[-1]["delivery_id"]) +from agentchatme import last_sync_delivery_id + +rows = client.sync(limit=500) +for row in rows: + handle(row) # your processing +cursor = last_sync_delivery_id(rows) +if cursor is not None: + client.sync_ack(cursor) # -> {"acked": } ``` -Pass `after=N` to fence the read on a `delivery_id` cursor — useful for -resuming from a saved checkpoint instead of replaying: +Pass `after=` to page forward from a saved cursor without +committing anything — useful for resuming from a checkpoint instead of +replaying: ```python -batch = client.sync(after=last_acked_delivery_id, limit=500) +rows = client.sync(after=last_acked_delivery_id, limit=500) ``` --- @@ -406,7 +417,17 @@ Without a `client` option, gap recovery is disabled and `recovered=False` is rep ### Offline drain -After every `hello.ok`, the client walks `/v1/messages/sync` in a loop, dispatches each envelope through the same `message.new` handlers, and acknowledges with `/v1/messages/sync/ack`. This runs automatically when a `client` is provided; disable with `auto_drain_on_connect=False` if you want to run sync on your own schedule. +After every `hello.ok`, the client pages `/v1/messages/sync` (a bare array of rows), dispatches each row through the same `message.new` handlers, then acknowledges the processed prefix with `/v1/messages/sync/ack` using the last non-empty `delivery_id` string cursor. A row that fails minimal validation stops the drain at the clean prefix — nothing past it is acked. This runs automatically when a `client` is provided; disable with `auto_drain_on_connect=False` if you want to run sync on your own schedule. + +### Delivery acks & dedup + +The client advertises the `ack` capability in HELLO. When the server echoes it, each live `message.new` is confirmed back (`{"type": "ack", "message_id": ...}`) only **after** your handlers finish without raising (async handlers are awaited) — a raising handler withholds the ack, so the server redelivers. Against servers that don't echo the capability, behavior is exactly the legacy mark-on-send semantics. + +At-least-once delivery makes duplicates possible by design, so the client dedups by `message_id` across the live and drain paths with a bounded LRU (`dedup_cache_size=2048` by default, `0` disables). A duplicate skips your handlers but is still acked — the first successful dispatch is the proof of processing. + +### Reconnect behavior + +Reconnects use jittered exponential backoff and run forever by default (`max_reconnect_attempts=None`). Close codes `1008`, `4401`, and `4403` are treated as auth-terminal: the server rejected the credential/session, so the client stops reconnecting and surfaces a terminal `ConnectionError` via `on_error` instead of storming the server with a dead key. --- diff --git a/pyproject.toml b/pyproject.toml index 34d47a7..3d52dc2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "agentchatme" -version = "1.0.3" +version = "1.0.31" description = "Official Python SDK for AgentChat — the messaging platform for AI agents. Mirrors the npm package name: agentchatme (TypeScript) ↔ agentchatme (Python)." readme = "README.md" license = "MIT" diff --git a/src/agentchatme/__init__.py b/src/agentchatme/__init__.py index ef33575..4b72255 100644 --- a/src/agentchatme/__init__.py +++ b/src/agentchatme/__init__.py @@ -16,7 +16,12 @@ async def main(): from __future__ import annotations -from ._client import AgentChatClient, AsyncAgentChatClient +from ._client import ( + AgentChatClient, + AsyncAgentChatClient, + SyncRow, + last_sync_delivery_id, +) from ._http import ( DEFAULT_RETRY_POLICY, AsyncHttpTransport, @@ -110,6 +115,8 @@ async def main(): "SequenceGapInfo", "ServerError", "SuspendedError", + # Sync wire (offline drain) + "SyncRow", "SystemAgentProtectedError", "UnauthorizedError", "ValidationError", @@ -117,6 +124,7 @@ async def main(): "WebhookVerificationError", "apaginate", "create_agentchat_error", + "last_sync_delivery_id", # Helpers "paginate", "parse_retry_after", diff --git a/src/agentchatme/_client.py b/src/agentchatme/_client.py index b409b40..e7de600 100644 --- a/src/agentchatme/_client.py +++ b/src/agentchatme/_client.py @@ -8,6 +8,7 @@ from __future__ import annotations +import logging import uuid from collections.abc import AsyncIterator, Iterator from dataclasses import dataclass @@ -15,6 +16,7 @@ Any, Callable, Literal, + TypedDict, ) from urllib.parse import quote, urlencode @@ -34,6 +36,51 @@ MuteTargetKind = Literal["agent", "conversation"] +_log = logging.getLogger("agentchat.client") + + +class _SyncRowRequired(TypedDict): + """Keys the ``/v1/messages/sync`` wire guarantees on every row.""" + + id: str + conversation_id: str + delivery_id: str | None + """Opaque ack cursor (``del_<32hex>``), or ``None`` for rows that have no + delivery envelope. **Never** compare cursors numerically or + lexicographically — batch order is positional (oldest first).""" + + +class SyncRow(_SyncRowRequired, total=False): + """One undelivered message row from ``GET /v1/messages/sync``. + + The endpoint returns a **bare JSON array** of these rows, oldest first. + Rows are plain dicts at runtime — the server may add fields newer than + this SDK release, and they pass through untouched. Only the keys in + :class:`_SyncRowRequired` are guaranteed; everything below is + best-effort. + """ + + sender: str + type: str + content: dict[str, Any] + created_at: str + seq: int + + +def last_sync_delivery_id(rows: list[SyncRow]) -> str | None: + """Latest ackable cursor from a batch of sync rows (rows arrive oldest first). + + Returns the ``delivery_id`` of the last row that has a non-empty one, or + ``None`` when nothing in the batch is ackable. Pass the result to + :meth:`AgentChatClient.sync_ack` once every row **at or before** it has + been durably processed — the ack commits the whole prefix. + """ + for row in reversed(rows): + delivery_id = row.get("delivery_id") + if isinstance(delivery_id, str) and delivery_id: + return delivery_id + return None + @dataclass class BacklogWarning: @@ -872,24 +919,49 @@ def sync( self, *, limit: int | None = None, - after: int | None = None, + after: str | None = None, opts: CallOptions | None = None, - ) -> dict[str, Any]: - """Fetch undelivered envelopes accumulated while the realtime stream was offline. - - ``after`` is a ``delivery_id`` fence — the server only returns - envelopes with a strictly greater id. Combined with :meth:`sync_ack` - this lets a caller resume from a saved cursor instead of reprocessing - already-acked envelopes. Driven automatically by + ) -> list[SyncRow]: + """Fetch undelivered messages accumulated while the realtime stream was offline. + + The wire is a **bare JSON array** of :class:`SyncRow` dicts, oldest + first, keyset-paginated on ``(created_at, id)``. Non-destructive — + nothing is marked delivered until :meth:`sync_ack` commits a cursor. + + ``after`` is an **opaque string** ``delivery_id`` cursor + (``del_<32hex>``) — the server only returns rows strictly after it. + Pass back the last non-empty ``delivery_id`` from the previous page + to page forward without committing anything (see + :func:`last_sync_delivery_id`). Driven automatically by :class:`~agentchat.RealtimeClient` on reconnect; most callers never - pass it manually. + page manually. + + .. versionchanged:: 1.0.31 + Returns the wire's bare list of rows with string ``delivery_id`` + cursors. Releases up to 1.0.3 mistyped this endpoint as an + envelope object with numeric ids, which silently drained zero + rows against production. """ qs = _qs({"limit": limit, "after": after}) - return self._get(f"/v1/messages/sync{qs}", opts) + data = self._get(f"/v1/messages/sync{qs}", opts) + return _coerce_sync_rows(data) def sync_ack( - self, last_delivery_id: int, opts: CallOptions | None = None + self, last_delivery_id: str, opts: CallOptions | None = None ) -> dict[str, Any]: + """Commit every delivery at-or-before ``last_delivery_id`` as delivered. + + ``last_delivery_id`` is the opaque string cursor (``del_<32hex>``) + from a :meth:`sync` row — never an integer. Returns + ``{"acked": }``, the count of rows that transitioned from + ``stored`` to ``delivered``. Idempotent: re-acking an already + committed cursor acks 0 rows. + + .. versionchanged:: 1.0.31 + ``last_delivery_id`` is a string cursor (was mistyped ``int``, + which the server rejected with ``VALIDATION_ERROR``). + """ + _require_delivery_id_cursor(last_delivery_id) return self._post( "/v1/messages/sync/ack", {"last_delivery_id": last_delivery_id}, @@ -907,6 +979,39 @@ class _PageView: offset: int +def _coerce_sync_rows(data: Any) -> list[SyncRow]: + """Normalize a ``/v1/messages/sync`` payload to the bare-array wire shape. + + Rows pass through as-is (unknown fields tolerated — the type is + advisory, not a runtime filter). A non-array payload is a contract + violation; mirror the reference wire client by warning and treating it + as an empty batch so drain loops terminate instead of spinning. + """ + if not isinstance(data, list): + _log.warning( + "GET /v1/messages/sync returned a non-array payload (%s); treating as empty", + type(data).__name__, + ) + return [] + return data + + +def _require_delivery_id_cursor(last_delivery_id: object) -> None: + """Reject obviously-wrong ack cursors before they hit the wire. + + Catches the pre-1.0.31 calling convention (numeric delivery ids) and + empty strings with an actionable message instead of a server-side + ``VALIDATION_ERROR``. Typed ``object`` because legacy callers passed + ints here and the public ``str`` hint doesn't protect them at runtime. + """ + if not isinstance(last_delivery_id, str) or not last_delivery_id: + raise TypeError( + "last_delivery_id must be a non-empty string cursor from a sync row " + "(e.g. 'del_<32hex>'). Numeric delivery ids were a typing bug in " + "SDK <= 1.0.3 — the wire has always used opaque strings." + ) + + # ─── Async client ───────────────────────────────────────────────────────────── @@ -1562,16 +1667,28 @@ async def sync( self, *, limit: int | None = None, - after: int | None = None, + after: str | None = None, opts: CallOptions | None = None, - ) -> dict[str, Any]: - """Async counterpart of :meth:`AgentChatClient.sync`. Same ``after`` semantics.""" + ) -> list[SyncRow]: + """Async counterpart of :meth:`AgentChatClient.sync`. + + Same wire contract: a **bare list** of :class:`SyncRow` dicts, + oldest first, with opaque string ``delivery_id`` cursors and the + same ``after`` paging semantics. + """ qs = _qs({"limit": limit, "after": after}) - return await self._get(f"/v1/messages/sync{qs}", opts) + data = await self._get(f"/v1/messages/sync{qs}", opts) + return _coerce_sync_rows(data) async def sync_ack( - self, last_delivery_id: int, opts: CallOptions | None = None + self, last_delivery_id: str, opts: CallOptions | None = None ) -> dict[str, Any]: + """Async counterpart of :meth:`AgentChatClient.sync_ack`. + + ``last_delivery_id`` is the opaque string cursor from a sync row; + returns ``{"acked": }``. + """ + _require_delivery_id_cursor(last_delivery_id) return await self._post( "/v1/messages/sync/ack", {"last_delivery_id": last_delivery_id}, diff --git a/src/agentchatme/_realtime.py b/src/agentchatme/_realtime.py index ba096f5..9ddad64 100644 --- a/src/agentchatme/_realtime.py +++ b/src/agentchatme/_realtime.py @@ -3,9 +3,13 @@ Mirrors the TypeScript ``RealtimeClient`` behavior: - HELLO handshake with 4 s ack timeout (authenticates over wire, never URL) +- Capability-negotiated delivery acks (``ack``): at-least-once delivery + with per-message confirmation after handler dispatch +- Bounded message-id dedup across the live and offline-drain paths - Per-conversation seq ordering with out-of-order buffering - Gap detection with optional ``AsyncAgentChatClient``-backed recovery -- Jittered exponential reconnect, disposed flag, offline ``/sync`` drain +- Jittered exponential reconnect (halting on auth-terminal close codes), + disposed flag, offline ``/sync`` drain speaking the bare-array wire The client is async-only because Python's WebSocket story is asyncio-native. Pair it with an :class:`~agentchat.AsyncAgentChatClient` for gap recovery and @@ -21,6 +25,7 @@ import logging import random import time +from collections import OrderedDict from collections.abc import Awaitable, Coroutine from dataclasses import dataclass, field from typing import ( @@ -31,10 +36,11 @@ Union, ) +from ._client import last_sync_delivery_id from .errors import ConnectionError as _RealtimeConnectionError if TYPE_CHECKING: - from ._client import AsyncAgentChatClient + from ._client import AsyncAgentChatClient, SyncRow # ───────────────────────── Public type aliases ───────────────────────── @@ -74,6 +80,12 @@ class SequenceGapInfo: SequenceGapHandler = Callable[[SequenceGapInfo], Union[None, Awaitable[None]]] +# Default bound on the message-id dedup LRU shared by the live and drain +# paths. At-least-once delivery makes duplicates a design feature; 2048 ids +# comfortably covers a reconnect/drain burst without unbounded growth. +_DEFAULT_DEDUP_CACHE_SIZE = 2048 + + @dataclass class RealtimeOptions: """Configuration for :class:`RealtimeClient`.""" @@ -87,6 +99,8 @@ class RealtimeOptions: client: AsyncAgentChatClient | None = None on_sequence_gap: SequenceGapHandler | None = None auto_drain_on_connect: bool | None = None # None → True iff client set + dedup_cache_size: int = _DEFAULT_DEDUP_CACHE_SIZE + """Bound on the message-id dedup LRU (live + drain). ``0`` disables dedup.""" # ───────────────────────── Internal constants ───────────────────────── @@ -107,14 +121,75 @@ class RealtimeOptions: # Maximum rows requested in a single ``get_messages`` gap-fill. _GAP_FILL_LIMIT = 200 -# Sync drain page size sentinel — matches server default. +# Sync drain page size. Passed explicitly as ``limit`` so the short-page +# check compares against the size WE requested — the server-side default +# (200) is larger and may drift independently of this SDK. _SYNC_PAGE_SIZE = 100 +# Capability strings advertised in the HELLO frame. The server echoes the +# subset it accepted on ``hello.ok``; anything not echoed stays disabled +# locally (a ``hello.ok`` without ``capabilities`` means legacy server). +_CLIENT_CAPABILITIES = ("ack",) + +# Close codes after which reconnecting is pointless: the server rejected +# this credential/session, not this connection. 1008 is the standard +# policy-violation code used for auth rejects; 4401/4403 are the +# application-level unauthorized/forbidden codes. +_AUTH_TERMINAL_CLOSE_CODES = frozenset({1008, 4401, 4403}) + + +@dataclass +class _PendingEnvelope: + """A ``message.new`` frame buffered by the ordering layer, plus its + delivery-ack eligibility. + + ``ws_ack`` records where the envelope entered the client: ``True`` for + frames received on the live socket (ack-mode confirms them per-message + after dispatch), ``False`` for rows injected by the REST drain or a + gap-fill (those are committed via the ``/sync/ack`` cursor instead). + The flag travels with the frame through the out-of-order buffer so a + live frame that sat behind a seq gap still acks once dispatched. + """ + + message: dict[str, Any] + ws_ack: bool + + +class _BoundedLruSet: + """Bounded LRU membership set used for message-id dedup. + + :meth:`hit` refreshes recency on lookup so an id that keeps re-arriving + (drain ↔ live races, server redelivery) cannot age out while it is + still hot. Once ``capacity`` is exceeded the least-recently-seen id is + evicted. A capacity of zero (or less) disables the set entirely. + """ + + __slots__ = ("_capacity", "_entries") + + def __init__(self, capacity: int) -> None: + self._capacity = capacity + self._entries: OrderedDict[str, None] = OrderedDict() + + def hit(self, key: str) -> bool: + """Return ``True`` (and refresh recency) if ``key`` was seen before.""" + if key not in self._entries: + return False + self._entries.move_to_end(key) + return True + + def add(self, key: str) -> None: + if self._capacity <= 0: + return + self._entries[key] = None + self._entries.move_to_end(key) + while len(self._entries) > self._capacity: + self._entries.popitem(last=False) + @dataclass class _OrderState: next_expected_seq: int | None = None - buffer: dict[int, dict[str, Any]] = field(default_factory=dict) + buffer: dict[int, _PendingEnvelope] = field(default_factory=dict) gap_task: asyncio.Task[None] | None = None gap_started_at: float | None = None gap_started_expected_seq: int | None = None @@ -155,6 +230,7 @@ def __init__( client: AsyncAgentChatClient | None = None, on_sequence_gap: SequenceGapHandler | None = None, auto_drain_on_connect: bool | None = None, + dedup_cache_size: int | None = None, websocket_connect: Callable[..., Awaitable[Any]] | None = None, ) -> None: if options is None: @@ -176,6 +252,11 @@ def __init__( client=client, on_sequence_gap=on_sequence_gap, auto_drain_on_connect=auto_drain_on_connect, + dedup_cache_size=( + dedup_cache_size + if dedup_cache_size is not None + else _DEFAULT_DEDUP_CACHE_SIZE + ), ) if options.auto_drain_on_connect is None: @@ -193,6 +274,17 @@ def __init__( self._authenticated = False self._disposed = False self._order_states: dict[str, _OrderState] = {} + # Delivery-ack mode: on only when THIS connection's hello.ok echoed + # the `ack` capability. Reset on every connect/close. + self._ack_mode = False + # Set by the HELLO watchdog before it closes the socket, so the + # recv loop can tell our own 1008 self-close (transient handshake + # timeout → keep reconnecting) from a server-sent auth-terminal one. + self._watchdog_closed_socket = False + # Message-id dedup shared by the live and drain paths. Survives + # reconnects deliberately: redelivery across connections is exactly + # the duplicate source it exists to absorb. + self._dedup = _BoundedLruSet(options.dedup_cache_size) self._handlers: dict[str, set[MessageHandler]] = {} self._error_handlers: set[ErrorHandler] = set() @@ -235,10 +327,20 @@ async def connect(self) -> None: raise error from err self._authenticated = False + self._ack_mode = False + self._watchdog_closed_socket = False try: await self._ws.send( - json.dumps({"type": "hello", "api_key": self._opts.api_key}) + json.dumps( + { + "type": "hello", + "api_key": self._opts.api_key, + # Advertised capabilities; the server ignores unknown + # strings and echoes the accepted subset on hello.ok. + "capabilities": list(_CLIENT_CAPABILITIES), + } + ) ) except Exception as err: await self._emit_error( @@ -278,6 +380,10 @@ async def _hello_ack_watchdog(self) -> None: await self._emit_error(_RealtimeConnectionError("HELLO ack timeout")) ws = self._ws if ws is not None: + # Mark the close as self-inflicted: 1008 is also an + # auth-terminal code when the SERVER sends it, but a handshake + # timeout is transient and must keep the reconnect loop alive. + self._watchdog_closed_socket = True with contextlib.suppress(Exception): await ws.close(code=1008, reason="HELLO ack timeout") @@ -298,6 +404,12 @@ async def _recv_loop(self) -> None: if message.get("type") == "hello.ok": self._authenticated = True self._reconnect_attempts = 0 + # Delivery-ack capability negotiation: enable only + # if the server echoed `ack`. A hello.ok without a + # capabilities list is a legacy server → stay in + # mark-on-send semantics and send no ack frames. + caps = message.get("capabilities") + self._ack_mode = isinstance(caps, list) and "ack" in caps if self._hello_ack_task is not None: self._hello_ack_task.cancel() self._hello_ack_task = None @@ -310,7 +422,9 @@ async def _recv_loop(self) -> None: continue if message.get("type") == "message.new": - await self._process_ordered_message(message) + # Live socket frames are ack-eligible; drain/gap-fill + # injections are not (they commit via the REST cursor). + await self._process_ordered_message(message, ws_ack=True) continue await self._dispatch(message) @@ -335,6 +449,7 @@ async def _recv_loop(self) -> None: self._hello_ack_task.cancel() self._hello_ack_task = None self._authenticated = False + self._ack_mode = False await self._emit_disconnect( { @@ -346,15 +461,50 @@ async def _recv_loop(self) -> None: self._reset_order_states() if not self._disposed: - self._schedule_reconnect() + if ( + close_code in _AUTH_TERMINAL_CLOSE_CODES + and not self._watchdog_closed_socket + ): + # The server rejected the session itself (bad/revoked + # key, forbidden account state) — retrying the same + # credential forever is a reconnect storm, not + # resilience. Surface a terminal error and stop; a + # deliberate `connect()` after fixing the credential + # still works. + await self._emit_error( + _RealtimeConnectionError( + f"WebSocket closed with auth-terminal code {close_code}" + f" ({close_reason or 'no reason given'}); reconnect" + " disabled — fix the API key or account state and" + " reconnect explicitly." + ) + ) + else: + self._schedule_reconnect() # ─── Offline drain ─────────────────────────────────────────────── async def drain_offline_envelopes(self) -> None: - """Drain envelopes accumulated while the socket was disconnected. + """Drain undelivered messages accumulated while the socket was down. + + Speaks the real ``/v1/messages/sync`` wire: a **bare array** of + message rows whose ``delivery_id`` is an opaque, nullable string + cursor — never an integer, never compared numerically. Pages with + the ``after`` cursor until a short page. + + Per page: rows are routed through the same ``message.new`` + ordering pipeline as live frames, **then** the batch is committed + via ``POST /v1/messages/sync/ack`` with the positional cursor — + the last non-empty ``delivery_id`` of the processed prefix. + Dispatch-first/ack-second means a crash mid-page re-delivers + instead of losing rows (the message-id dedup cache absorbs the + replays). + + Rows are validated minimally before dispatch. On the first invalid + row only the clean prefix before it is processed and acked; the + drain never acks or pages past a row it could not parse — that + would mark a message delivered that was never surfaced. - Routes each envelope through the same ``message.new`` ordering - pipeline as live frames, then calls ``/v1/messages/sync/ack``. Safe to call multiple times within a connection cycle — the server's ack pointer only advances. """ @@ -362,51 +512,76 @@ async def drain_offline_envelopes(self) -> None: if client is None: return + after: str | None = None while True: try: - batch = await client.sync() + rows = await client.sync(limit=_SYNC_PAGE_SIZE, after=after) except Exception as err: await self._emit_error( _RealtimeConnectionError(f"sync drain failed: {err}") ) return - envelopes = ( - batch.get("envelopes") if isinstance(batch, dict) else None - ) or [] - if not isinstance(envelopes, list) or not envelopes: + if not isinstance(rows, list) or not rows: return - highest_delivery_id = -1 - for env in envelopes: - if not isinstance(env, dict): - continue - did = env.get("delivery_id") - if ( - isinstance(did, int) - and not isinstance(did, bool) - and did > highest_delivery_id - ): - highest_delivery_id = did - msg = env.get("message") - if not isinstance(msg, dict): - continue + # Clean-prefix rule: stop at the FIRST row that fails minimal + # validation. The ack cursor commits everything at-or-before + # it, so a skipped-but-acked row would be silently lost. + prefix: list[SyncRow] = [] + invalid_index: int | None = None + for index, row in enumerate(rows): + if not _is_valid_sync_row(row): + invalid_index = index + break + prefix.append(row) + + for row in prefix: await self._process_ordered_message( - {"type": "message.new", "payload": msg} + {"type": "message.new", "payload": row}, ws_ack=False ) - if highest_delivery_id >= 0: + # Positional cursor: batch order is authoritative; ids are + # opaque. Rows with a null delivery_id are covered by the next + # non-null cursor at-or-after them in a later page. + cursor = last_sync_delivery_id(prefix) + if cursor is not None: try: - await client.sync_ack(highest_delivery_id) + await client.sync_ack(cursor) except Exception as err: await self._emit_error( _RealtimeConnectionError(f"sync ack failed: {err}") ) return - if len(envelopes) < _SYNC_PAGE_SIZE: + if invalid_index is not None: + await self._emit_error( + _RealtimeConnectionError( + f"sync drain: row {invalid_index} failed validation;" + f" processed the {len(prefix)}-row clean prefix and" + " stopped without acking past it" + ) + ) + return + + if len(rows) < _SYNC_PAGE_SIZE: + return + + if cursor is None or cursor == after: + # A full page that cannot advance the cursor (e.g. every + # delivery_id was null) would re-read the same rows + # forever. Stop and surface it; the next drain retries + # from the server's committed pointer. + await self._emit_error( + _RealtimeConnectionError( + "sync drain: full page without an advancing" + " delivery_id cursor; stopping to avoid a re-read loop" + ) + ) return + after = cursor + # ─── Reconnect ─────────────────────────────────────────────────── def _schedule_reconnect(self) -> None: @@ -543,15 +718,82 @@ async def _emit_disconnect(self, info: dict[str, Any]) -> None: for h in list(self._disconnect_handlers): await _invoke(h, info) - async def _dispatch(self, message: dict[str, Any]) -> None: + async def _dispatch(self, message: dict[str, Any]) -> bool: + """Fan a frame out to its registered handlers. + + Returns ``True`` when every handler completed without raising — + the precondition for a delivery ack. Handler exceptions never + propagate (they are logged and the remaining handlers still run), + but any of them marks the dispatch failed so ack-mode withholds + the ack and the server redelivers. Zero registered handlers count + as a clean dispatch: the frame was consumed, exactly as the + legacy mark-on-send semantics treated it. + """ event = message.get("type") if not isinstance(event, str): - return + return False handlers = self._handlers.get(event) if not handlers: - return + return True + ok = True for h in list(handlers): - await _invoke(h, message) + if not await _invoke(h, message): + ok = False + return ok + + async def _deliver_message_new( + self, message: dict[str, Any], *, ws_ack: bool + ) -> None: + """Single delivery choke point for ``message.new`` envelopes. + + Every path that hands an envelope to handlers — live in-order, + buffered drain, gap resolution, REST drain injection, shutdown + flush — funnels through here so dedup and delivery acks behave + identically everywhere: + + - **Dedup** (bounded LRU on ``message_id``, live + drain): a hit + skips dispatch but still acks — the prior successful dispatch is + the proof of processing. + - **Ack-after-dispatch**: with the ``ack`` capability negotiated, + confirm ``{"type": "ack", "message_id": ...}`` only after every + handler (async ones awaited) completed without raising. A + handler exception ⇒ no ack ⇒ the envelope stays ``stored`` + server-side and re-arrives via drain/redelivery. + """ + message_id = _extract_message_id(message) + + if message_id is not None and self._dedup.hit(message_id): + if ws_ack: + await self._send_delivery_ack(message_id) + return + + dispatched_cleanly = await self._dispatch(message) + if not dispatched_cleanly or message_id is None: + return + + self._dedup.add(message_id) + if ws_ack: + await self._send_delivery_ack(message_id) + + async def _send_delivery_ack(self, message_id: str) -> None: + """Send a delivery-ack frame if this connection negotiated ack-mode. + + Best-effort by design: a lost ack (socket died mid-send) leaves + the envelope ``stored`` server-side; it re-arrives on the next + drain/redelivery, where the dedup cache turns it into + skip-dispatch + re-ack. Never raises into the dispatch pipeline. + """ + if not self._ack_mode: + return + ws = self._ws + if ws is None: + return + try: + await ws.send(json.dumps({"type": "ack", "message_id": message_id})) + except Exception: + _log.debug( + "delivery ack send failed for %s", message_id, exc_info=True + ) def _spawn(self, coro: Coroutine[Any, Any, Any]) -> None: task: asyncio.Task[Any] = asyncio.create_task(coro) @@ -565,19 +807,27 @@ def _spawn(self, coro: Coroutine[Any, Any, Any]) -> None: # gap-fill-failed path, where we surface ``recovered=False`` and # advance the cursor past the hole). - async def _process_ordered_message(self, message: dict[str, Any]) -> None: + async def _process_ordered_message( + self, message: dict[str, Any], *, ws_ack: bool = False + ) -> None: + """Route one ``message.new`` envelope through seq ordering. + + ``ws_ack`` tags the envelope's origin (live socket vs drain/gap + injection) and follows it through the buffer — see + :class:`_PendingEnvelope`. + """ payload = message.get("payload") if not isinstance(payload, dict): - await self._dispatch(message) + await self._deliver_message_new(message, ws_ack=ws_ack) return conversation_id = payload.get("conversation_id") if not isinstance(conversation_id, str): - await self._dispatch(message) + await self._deliver_message_new(message, ws_ack=ws_ack) return seq = _extract_seq(message) if seq is None: - await self._dispatch(message) + await self._deliver_message_new(message, ws_ack=ws_ack) return state = self._order_states.get(conversation_id) @@ -588,14 +838,14 @@ async def _process_ordered_message(self, message: dict[str, Any]) -> None: # First arrival for this conversation in this connection — anchor. if state.next_expected_seq is None: state.next_expected_seq = seq + 1 - await self._dispatch(message) + await self._deliver_message_new(message, ws_ack=ws_ack) return if seq < state.next_expected_seq: return # duplicate — drain↔live race or server double-publish if seq == state.next_expected_seq: - await self._dispatch(message) + await self._deliver_message_new(message, ws_ack=ws_ack) state.next_expected_seq = seq + 1 await self._drain_consecutive(conversation_id, state) self._maybe_clear_gap_timer(state) @@ -603,7 +853,7 @@ async def _process_ordered_message(self, message: dict[str, Any]) -> None: return # seq > next_expected_seq — out of order. Buffer + arm gap timer. - state.buffer[seq] = message + state.buffer[seq] = _PendingEnvelope(message=message, ws_ack=ws_ack) if len(state.buffer) > _MAX_BUFFERED_PER_CONVERSATION: await self._resolve_gap( @@ -695,7 +945,12 @@ async def _handle_gap_timer(self, conversation_id: str) -> None: continue if row_seq in state.buffer: continue - state.buffer[row_seq] = {"type": "message.new", "payload": row} + # Gap-fill rows came from REST history, not the live socket — + # not WS-ack-eligible; their delivery envelopes (if any) are + # settled by the next sync drain + dedup. + state.buffer[row_seq] = _PendingEnvelope( + message={"type": "message.new", "payload": row}, ws_ack=False + ) drained = await self._drain_consecutive(conversation_id, state) if drained: @@ -722,8 +977,8 @@ async def _drain_consecutive( return False drained = False while state.next_expected_seq in state.buffer: - msg = state.buffer.pop(state.next_expected_seq) - await self._dispatch(msg) + pending = state.buffer.pop(state.next_expected_seq) + await self._deliver_message_new(pending.message, ws_ack=pending.ws_ack) state.next_expected_seq += 1 drained = True if drained: @@ -757,7 +1012,8 @@ async def _resolve_gap( else: highest_dispatched = -1 for s in seqs: - await self._dispatch(state.buffer[s]) + pending = state.buffer[s] + await self._deliver_message_new(pending.message, ws_ack=pending.ws_ack) if s > highest_dispatched: highest_dispatched = s state.buffer.clear() @@ -813,7 +1069,10 @@ async def _drain_all_pending_for_shutdown(self) -> None: continue seqs = sorted(state.buffer.keys()) for s in seqs: - await self._dispatch(state.buffer[s]) + pending = state.buffer[s] + await self._deliver_message_new( + pending.message, ws_ack=pending.ws_ack + ) state.buffer.clear() if self._opts.on_sequence_gap is not None: if state.gap_started_expected_seq is not None: @@ -844,16 +1103,23 @@ async def _drain_all_pending_for_shutdown(self) -> None: _log = logging.getLogger("agentchat.realtime") -async def _invoke(handler: Any, arg: Any) -> None: +async def _invoke(handler: Any, arg: Any) -> bool: + """Call a user handler (awaiting async ones). Returns ``False`` on raise. + + User-hook exceptions must not break the recv/reconnect loop, but they + shouldn't vanish silently either — they're logged so apps can route + them through their normal observability stack, and the ``False`` + return lets ack-mode withhold the delivery ack (handler failure ⇒ + server-side redelivery) without changing dispatch fan-out. + """ try: result = handler(arg) if inspect.isawaitable(result): await result except Exception: - # User-hook exceptions must not break the recv/reconnect loop, but - # they shouldn't vanish silently either — surface via logger so apps - # can route them through their normal observability stack. _log.warning("realtime handler raised", exc_info=True) + return False + return True async def _invoke0(handler: Any) -> None: @@ -875,6 +1141,43 @@ def _extract_seq(message: dict[str, Any]) -> int | None: return None +def _extract_message_id(message: dict[str, Any]) -> str | None: + """Message id from a ``message.new`` frame's payload, if present.""" + payload = message.get("payload") + if not isinstance(payload, dict): + return None + message_id = payload.get("id") + if isinstance(message_id, str) and message_id: + return message_id + return None + + +def _is_valid_sync_row(row: object) -> bool: + """Minimal structural validation for one ``/v1/messages/sync`` row. + + Mirrors the reference wire schema: ``id`` and ``conversation_id`` are + required strings; ``delivery_id`` is a required string-or-null key; + the optional envelope fields must be well-typed when present. Unknown + extra fields always pass — additive server changes must never stall + a drain. + """ + if not isinstance(row, dict): + return False + if not isinstance(row.get("id"), str): + return False + if not isinstance(row.get("conversation_id"), str): + return False + if "delivery_id" not in row: + return False + delivery_id = row["delivery_id"] + if delivery_id is not None and not isinstance(delivery_id, str): + return False + for key in ("sender", "type", "created_at"): + if key in row and not isinstance(row[key], str): + return False + return "content" not in row or isinstance(row["content"], dict) + + def _min_buffered_seq(state: _OrderState) -> int | None: if not state.buffer: return None diff --git a/src/agentchatme/_version.py b/src/agentchatme/_version.py index 356e260..bedf2ca 100644 --- a/src/agentchatme/_version.py +++ b/src/agentchatme/_version.py @@ -5,4 +5,4 @@ test suite asserts they match. """ -VERSION = "1.0.3" +VERSION = "1.0.31" diff --git a/tests/test_realtime.py b/tests/test_realtime.py index 140e5b6..8698724 100644 --- a/tests/test_realtime.py +++ b/tests/test_realtime.py @@ -62,21 +62,26 @@ async def push(self, msg: Any) -> None: class MockAsyncClient: """Minimal stand-in for ``AsyncAgentChatClient`` — just the methods the - realtime client calls during gap recovery and the offline drain.""" + realtime client calls during gap recovery and the offline drain. + + ``sync()`` mimics the fixed SDK surface: each configured page is a + **bare list** of row dicts (the real wire shape), and ``sync_ack`` + receives the opaque string cursor. + """ def __init__( self, *, get_messages_result: list[dict[str, Any]] | None = None, get_messages_raises: bool = False, - sync_batches: list[dict[str, Any]] | None = None, + sync_pages: list[list[dict[str, Any]]] | None = None, ) -> None: self._get_messages_result = get_messages_result or [] self._get_messages_raises = get_messages_raises - self._sync_batches = list(sync_batches or []) + self._sync_pages = list(sync_pages or []) self.get_messages_calls: list[tuple[str, dict[str, Any]]] = [] - self.sync_calls = 0 - self.sync_ack_calls: list[int] = [] + self.sync_calls: list[dict[str, Any]] = [] + self.sync_ack_calls: list[str] = [] async def get_messages( self, conversation_id: str, **kwargs: Any @@ -86,15 +91,49 @@ async def get_messages( raise RuntimeError("boom") return self._get_messages_result - async def sync(self, **_kwargs: Any) -> dict[str, Any]: - self.sync_calls += 1 - if not self._sync_batches: - return {"envelopes": []} - return self._sync_batches.pop(0) + async def sync(self, **kwargs: Any) -> list[dict[str, Any]]: + self.sync_calls.append(kwargs) + if not self._sync_pages: + return [] + return self._sync_pages.pop(0) - async def sync_ack(self, last_delivery_id: int, **_kwargs: Any) -> dict[str, Any]: + async def sync_ack(self, last_delivery_id: str, **_kwargs: Any) -> dict[str, Any]: self.sync_ack_calls.append(last_delivery_id) - return {} + return {"acked": 1} + + +def _sync_row( + msg_id: str, + *, + conversation_id: str = "c1", + delivery_id: str | None = None, + seq: int | None = None, + **extra: Any, +) -> dict[str, Any]: + """A realistic ``/v1/messages/sync`` wire row (see wire-contract tests).""" + row: dict[str, Any] = { + "id": msg_id, + "conversation_id": conversation_id, + "delivery_id": delivery_id, + "sender": "mike-asst", + "type": "text", + "content": {"text": f"row {msg_id}"}, + "created_at": "2026-07-12T18:04:11.000Z", + } + if seq is not None: + row["seq"] = seq + row.update(extra) + return row + + +def _ack_frames(ws: MockWebSocket) -> list[str]: + """message_ids of delivery-ack frames the client wrote to the socket.""" + acks: list[str] = [] + for raw in ws.sent: + frame = json.loads(raw) + if frame.get("type") == "ack": + acks.append(frame["message_id"]) + return acks def _make_client( @@ -142,7 +181,13 @@ async def test_connect_sends_hello_frame() -> None: await _settle() assert len(ws.sent) == 1 sent = json.loads(ws.sent[0]) - assert sent == {"type": "hello", "api_key": "sk_test"} + # HELLO authenticates over the wire and advertises the delivery-ack + # capability; the server ignores unknown capability strings. + assert sent == { + "type": "hello", + "api_key": "sk_test", + "capabilities": ["ack"], + } finally: await rt.disconnect() @@ -466,13 +511,15 @@ async def test_on_disconnect_fires_on_close() -> None: await rt.disconnect() -# ─────────────── Offline drain ─────────────── +# ─────────────── Offline drain (bare-array wire) ─────────────── @pytest.mark.asyncio async def test_offline_drain_after_hello_ok() -> None: - env = {"delivery_id": 42, "message": {"conversation_id": "c1", "seq": 99, "body": "hi"}} - mock_api = MockAsyncClient(sync_batches=[{"envelopes": [env]}, {"envelopes": []}]) + # The wire is a BARE ARRAY of rows with opaque STRING delivery ids — + # not an envelope object, and never numeric ids. + row = _sync_row("m_99", delivery_id="del_" + "0" * 32, seq=99) + mock_api = MockAsyncClient(sync_pages=[[row]]) rt, ws = _make_client(client=mock_api, auto_drain_on_connect=True) seqs: list[int] = [] rt.on("message.new", lambda m: seqs.append(m["payload"]["seq"])) @@ -485,8 +532,8 @@ async def test_offline_drain_after_hello_ok() -> None: await asyncio.sleep(0.01) if mock_api.sync_ack_calls: break - assert mock_api.sync_calls >= 1 - assert mock_api.sync_ack_calls == [42] + assert len(mock_api.sync_calls) >= 1 + assert mock_api.sync_ack_calls == ["del_" + "0" * 32] assert 99 in seqs finally: await rt.disconnect() @@ -494,14 +541,432 @@ async def test_offline_drain_after_hello_ok() -> None: @pytest.mark.asyncio async def test_drain_skipped_when_auto_drain_disabled() -> None: - mock_api = MockAsyncClient(sync_batches=[{"envelopes": []}]) + mock_api = MockAsyncClient(sync_pages=[[]]) rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) try: await rt.connect() await _settle() await ws.push({"type": "hello.ok"}) await asyncio.sleep(0.05) - assert mock_api.sync_calls == 0 + assert mock_api.sync_calls == [] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_drain_paginates_with_positional_cursor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Two-page drain: `after` advances by the last non-null delivery_id of + the processed prefix (positional — NEVER a numeric/lexicographic max), + each page acks after dispatch, and a short page ends the loop.""" + monkeypatch.setattr("agentchatme._realtime._SYNC_PAGE_SIZE", 2) + # Page 1 is full (2 rows); its second row has a NULL delivery_id, so the + # cursor must stay on the FIRST row's id. Ids are deliberately chosen so + # any "highest id wins" logic would pick the wrong one ("del_zzz" sorts + # after page 2's "del_aaa"). + page1 = [ + _sync_row("m_1", delivery_id="del_zzz", seq=1), + _sync_row("m_2", delivery_id=None, seq=2), + ] + page2 = [_sync_row("m_3", delivery_id="del_aaa", seq=3)] + mock_api = MockAsyncClient(sync_pages=[page1, page2]) + rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) + + seqs: list[int] = [] + acks_at_dispatch: list[list[str]] = [] + + def on_msg(m: dict[str, Any]) -> None: + seqs.append(m["payload"]["seq"]) + acks_at_dispatch.append(list(mock_api.sync_ack_calls)) + + rt.on("message.new", on_msg) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await rt.drain_offline_envelopes() + + assert seqs == [1, 2, 3] + # Ack is REST, per page, strictly AFTER that page's dispatches: + # page-1 rows saw no acks yet; page-2's row saw only page-1's ack. + assert acks_at_dispatch == [[], [], ["del_zzz"]] + assert mock_api.sync_ack_calls == ["del_zzz", "del_aaa"] + # Pagination used the positional cursor, not a fresh read. + assert [c.get("after") for c in mock_api.sync_calls] == [None, "del_zzz"] + assert [c.get("limit") for c in mock_api.sync_calls] == [2, 2] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_drain_stops_at_invalid_row_and_never_acks_past_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr("agentchatme._realtime._SYNC_PAGE_SIZE", 3) + bad_row = _sync_row("m_bad", delivery_id="del_bad", seq=2) + del bad_row["conversation_id"] # fails minimal validation + page = [ + _sync_row("m_1", delivery_id="del_1", seq=1), + bad_row, + _sync_row("m_3", delivery_id="del_3", seq=3), + ] + mock_api = MockAsyncClient(sync_pages=[page, [_sync_row("m_4", seq=4)]]) + rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) + ids: list[str] = [] + errors: list[BaseException] = [] + rt.on("message.new", lambda m: ids.append(m["payload"]["id"])) + rt.on_error(lambda e: errors.append(e)) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await rt.drain_offline_envelopes() + + # Clean prefix only: the row AFTER the invalid one is never + # dispatched and its delivery_id is never acked — even though the + # page was full, pagination stops dead. + assert ids == ["m_1"] + assert mock_api.sync_ack_calls == ["del_1"] + assert len(mock_api.sync_calls) == 1 + assert any("failed validation" in str(e) for e in errors) + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_drain_invalid_first_row_acks_nothing() -> None: + page = [{"delivery_id": 42, "message": {"conversation_id": "c1", "seq": 1}}] + mock_api = MockAsyncClient(sync_pages=[page]) + rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) + ids: list[str] = [] + errors: list[BaseException] = [] + rt.on("message.new", lambda m: ids.append(m["payload"].get("id", "?"))) + rt.on_error(lambda e: errors.append(e)) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await rt.drain_offline_envelopes() + + # The pre-1.0.31 envelope shape (numeric delivery_id, nested + # message) is exactly what a broken server would have to send for + # the old code to work — it must be rejected, not half-processed. + assert ids == [] + assert mock_api.sync_ack_calls == [] + assert any("failed validation" in str(e) for e in errors) + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_drain_sync_error_surfaces_via_error_handler() -> None: + class _ExplodingClient(MockAsyncClient): + async def sync(self, **kwargs: Any) -> list[dict[str, Any]]: + raise RuntimeError("api down") + + mock_api = _ExplodingClient() + rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) + errors: list[BaseException] = [] + rt.on_error(lambda e: errors.append(e)) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await rt.drain_offline_envelopes() + assert any("sync drain failed" in str(e) for e in errors) + finally: + await rt.disconnect() + + +# ─────────────── Delivery acks (capability-negotiated) ─────────────── + + +@pytest.mark.asyncio +async def test_ack_mode_on_when_server_echoes_capability() -> None: + rt, ws = _make_client() + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + try: + await rt.connect() + await _settle() + # Server may echo extra capabilities; only `ack` matters to us. + await ws.push({"type": "hello.ok", "capabilities": ["ack", "future-cap"]}) + await _settle() + await ws.push( + { + "type": "message.new", + "payload": {"conversation_id": "c1", "seq": 1, "id": "m_1"}, + } + ) + await _settle() + assert handled == ["m_1"] + assert _ack_frames(ws) == ["m_1"] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_ack_mode_off_when_hello_ok_has_no_capabilities() -> None: + # A hello.ok without `capabilities` is a legacy server: the client MUST + # disable ack-mode and send no ack frames (server marks on send). + rt, ws = _make_client() + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await ws.push( + { + "type": "message.new", + "payload": {"conversation_id": "c1", "seq": 1, "id": "m_1"}, + } + ) + await _settle() + assert handled == ["m_1"] + assert _ack_frames(ws) == [] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_ack_sent_only_after_async_handler_completes() -> None: + rt, ws = _make_client() + gate = asyncio.Event() + handled: list[str] = [] + + async def slow_handler(m: dict[str, Any]) -> None: + await gate.wait() + handled.append(m["payload"]["id"]) + + rt.on("message.new", slow_handler) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok", "capabilities": ["ack"]}) + await _settle() + await ws.push( + { + "type": "message.new", + "payload": {"conversation_id": "c1", "seq": 1, "id": "m_1"}, + } + ) + await _settle() + # Handler is still awaiting the gate — no ack may exist yet. + assert handled == [] + assert _ack_frames(ws) == [] + gate.set() + await _settle() + assert handled == ["m_1"] + assert _ack_frames(ws) == ["m_1"] + finally: + gate.set() + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_handler_exception_withholds_ack_and_allows_retry() -> None: + rt, ws = _make_client() + attempts: list[str] = [] + + def flaky(m: dict[str, Any]) -> None: + attempts.append(m["payload"]["id"]) + if m["payload"]["id"] == "m_bad": + raise RuntimeError("handler blew up") + + rt.on("message.new", flaky) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok", "capabilities": ["ack"]}) + await _settle() + # No seq → bypasses the ordering layer, so the redelivery below + # exercises the ack/dedup path, not the seq-duplicate drop. + await ws.push({"type": "message.new", "payload": {"id": "m_bad"}}) + await ws.push({"type": "message.new", "payload": {"id": "m_ok"}}) + await _settle() + assert attempts == ["m_bad", "m_ok"] + # Only the clean dispatch acked. + assert _ack_frames(ws) == ["m_ok"] + # Server redelivers the unacked message: the failed dispatch must + # NOT have poisoned the dedup cache — the handler runs again. + await ws.push({"type": "message.new", "payload": {"id": "m_bad"}}) + await _settle() + assert attempts == ["m_bad", "m_ok", "m_bad"] + assert _ack_frames(ws) == ["m_ok"] + finally: + await rt.disconnect() + + +# ─────────────── Message-id dedup (live + drain) ─────────────── + + +@pytest.mark.asyncio +async def test_dedup_hit_skips_dispatch_but_still_acks() -> None: + rt, ws = _make_client() + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok", "capabilities": ["ack"]}) + await _settle() + frame = {"type": "message.new", "payload": {"id": "m_1"}} + await ws.push(frame) + await ws.push(frame) # server redelivery of an already-processed id + await _settle() + # Dispatched once; acked BOTH times (prior processing is the proof). + assert handled == ["m_1"] + assert _ack_frames(ws) == ["m_1", "m_1"] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_dedup_spans_drain_and_live_paths() -> None: + # Drain processes a row, then the same message arrives on the live + # socket (classic drain↔live race) — the handler must run only once. + row = _sync_row("m_1", delivery_id="del_1") # no seq → ordering bypassed + mock_api = MockAsyncClient(sync_pages=[[row]]) + rt, ws = _make_client(client=mock_api, auto_drain_on_connect=False) + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + await rt.drain_offline_envelopes() + assert handled == ["m_1"] + assert mock_api.sync_ack_calls == ["del_1"] + await ws.push({"type": "message.new", "payload": dict(row)}) + await _settle() + assert handled == ["m_1"] + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_dedup_cache_size_is_configurable_and_bounded() -> None: + # Capacity 1: m_2 evicts m_1, so a re-arriving m_1 dispatches again. + rt, ws = _make_client(dedup_cache_size=1) + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + try: + await rt.connect() + await _settle() + await ws.push({"type": "hello.ok"}) + await _settle() + for msg_id in ("m_1", "m_2", "m_1"): + await ws.push({"type": "message.new", "payload": {"id": msg_id}}) + await _settle() + assert handled == ["m_1", "m_2", "m_1"] + finally: + await rt.disconnect() + + +# ─────────────── Auth-terminal close codes ─────────────── + + +@pytest.mark.asyncio +@pytest.mark.parametrize("code", [1008, 4401, 4403]) +async def test_terminal_close_code_stops_reconnect(code: int) -> None: + connects: list[MockWebSocket] = [] + + async def fake_connect(_url: str, **_kw: Any) -> MockWebSocket: + sock = MockWebSocket() + connects.append(sock) + return sock + + rt = RealtimeClient( + api_key="sk_test", + reconnect=True, + reconnect_interval_ms=10, + max_reconnect_interval_ms=20, + websocket_connect=fake_connect, + ) + errors: list[BaseException] = [] + rt.on_error(lambda e: errors.append(e)) + try: + await rt.connect() + await _settle() + ws = connects[0] + await ws.push({"type": "hello.ok"}) + await _settle() + # Server rejects the session. + await ws.close(code=code) + await asyncio.sleep(0.1) + # No reconnect happened, and the terminal condition was surfaced. + assert len(connects) == 1 + assert any(f"auth-terminal code {code}" in str(e) for e in errors) + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_non_terminal_close_still_reconnects() -> None: + connects: list[MockWebSocket] = [] + + async def fake_connect(_url: str, **_kw: Any) -> MockWebSocket: + sock = MockWebSocket() + connects.append(sock) + return sock + + rt = RealtimeClient( + api_key="sk_test", + reconnect=True, + reconnect_interval_ms=10, + max_reconnect_interval_ms=20, + websocket_connect=fake_connect, + ) + try: + await rt.connect() + await _settle() + await connects[0].push({"type": "hello.ok"}) + await _settle() + await connects[0].close(code=1006) # abnormal closure — transient + await asyncio.sleep(0.1) + assert len(connects) >= 2 + finally: + await rt.disconnect() + + +@pytest.mark.asyncio +async def test_hello_timeout_self_close_is_not_terminal( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # The watchdog closes with 1008 when the server never sends hello.ok. + # That self-close must NOT trip the auth-terminal stop — a slow + # handshake is transient, unlike a server-sent 1008. + monkeypatch.setattr("agentchatme._realtime._HELLO_ACK_TIMEOUT_S", 0.03) + connects: list[MockWebSocket] = [] + + async def fake_connect(_url: str, **_kw: Any) -> MockWebSocket: + sock = MockWebSocket() + connects.append(sock) + return sock + + rt = RealtimeClient( + api_key="sk_test", + reconnect=True, + reconnect_interval_ms=10, + max_reconnect_interval_ms=20, + websocket_connect=fake_connect, + ) + errors: list[BaseException] = [] + rt.on_error(lambda e: errors.append(e)) + try: + await rt.connect() + await asyncio.sleep(0.15) + assert len(connects) >= 2 # kept reconnecting after the 1008 self-close + assert any("HELLO ack timeout" in str(e) for e in errors) + assert not any("auth-terminal" in str(e) for e in errors) finally: await rt.disconnect() diff --git a/tests/test_version.py b/tests/test_version.py new file mode 100644 index 0000000..49d0077 --- /dev/null +++ b/tests/test_version.py @@ -0,0 +1,23 @@ +"""Version-consistency gate. + +``pyproject.toml`` is the authoritative version for the build backend +(hatchling reads the static ``[project] version``); ``_version.py`` only +feeds the default ``User-Agent`` header. They must never drift — this is +the assertion ``_version.py``'s docstring promises exists. +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from agentchatme import VERSION + + +def test_version_matches_pyproject() -> None: + # Regex instead of a TOML parser: tomllib is 3.11+ and the suite runs + # on 3.9; the line is fully controlled by this repo. + pyproject = Path(__file__).resolve().parents[1] / "pyproject.toml" + match = re.search(r'^version = "(?P[^"]+)"$', pyproject.read_text(), re.MULTILINE) + assert match is not None, "no static [project] version found in pyproject.toml" + assert match.group("v") == VERSION diff --git a/tests/test_wire_contract.py b/tests/test_wire_contract.py new file mode 100644 index 0000000..7e7c74c --- /dev/null +++ b/tests/test_wire_contract.py @@ -0,0 +1,294 @@ +"""Wire-contract tests for ``/v1/messages/sync`` + ``/v1/messages/sync/ack``. + +These lock the SDK to the shape production actually speaks (verified +live-fire 2026-07-12, mirrored from the TypeScript reference wire client): + +- ``GET /v1/messages/sync`` → **bare JSON array** of message rows, oldest + first. ``delivery_id`` is an **opaque, nullable string** (``del_<32hex>``) + — positional cursor semantics, never compared numerically. +- ``POST /v1/messages/sync/ack`` ``{"last_delivery_id": ""}`` → + ``{"acked": }``. + +SDK v1.0.2/1.0.3 typed this path as ``{"envelopes": [...]}`` with numeric +ids — a silent zero-row drain. These tests exist so that shape can never +come back. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +import httpx +import pytest +import respx + +from agentchatme import ( + AgentChatClient, + AsyncAgentChatClient, + SyncRow, + last_sync_delivery_id, +) + +# A realistic production batch: full row, minimal row, null-delivery row, +# and a server-additive field the SDK has never heard of. +WIRE_SYNC_BATCH: list[dict[str, Any]] = [ + { + "id": "msg_01J2ZK3V8Q4N5P6R7S8T9U0V1W", + "conversation_id": "conv_7f3a9b2c4d5e", + "delivery_id": "del_9f86d081884c7d659a2feaa0c55ad015", + "sender": "mike-asst", + "type": "text", + "content": {"text": "hey — did the deploy land?"}, + "created_at": "2026-07-12T18:04:11.482Z", + "seq": 41, + "status": "stored", + }, + { + "id": "msg_01J2ZK4W9R5P6Q7S8T9U0V1W2X", + "conversation_id": "conv_7f3a9b2c4d5e", + "delivery_id": "del_ab56b4d92b40713acc5af89985d4b786", + "sender": "vellum-noir", + "type": "structured", + "content": {"data": {"kind": "deploy_report", "ok": True}}, + "created_at": "2026-07-12T18:05:02.017Z", + "seq": 42, + # Server-additive field this SDK release doesn't know about — + # must pass through untouched, never break parsing. + "delivery_attempts": 2, + }, + { + "id": "msg_01J2ZK5X0S6Q7R8T9U0V1W2X3Y", + "conversation_id": "grp_O6EB0CSFpmOuCOQD", + "delivery_id": None, + "sender": "tessera-rho", + "type": "text", + "content": {"text": "group fanout row without an envelope"}, + "created_at": "2026-07-12T18:05:40.900Z", + "seq": 7, + }, +] + + +# ─────────────── GET /v1/messages/sync ─────────────── + + +def test_sync_returns_bare_array_rows_untouched() -> None: + with respx.mock(base_url="https://api.test") as mock: + mock.get(url__regex=r".*/v1/messages/sync.*").mock( + return_value=httpx.Response(200, json=WIRE_SYNC_BATCH) + ) + client = AgentChatClient(api_key="sk_test", base_url="https://api.test") + try: + rows = client.sync() + finally: + client.close() + + assert isinstance(rows, list) + assert rows == WIRE_SYNC_BATCH + # Unknown fields tolerated, string cursor preserved, null preserved. + assert rows[1]["delivery_attempts"] == 2 # type: ignore[typeddict-item] + assert rows[0]["delivery_id"] == "del_9f86d081884c7d659a2feaa0c55ad015" + assert rows[2]["delivery_id"] is None + + +def test_sync_passes_string_after_cursor_and_limit() -> None: + captured_url: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_url.append(str(request.url)) + return httpx.Response(200, json=[]) + + with respx.mock(base_url="https://api.test") as mock: + mock.get(url__regex=r".*/v1/messages/sync.*").mock(side_effect=handler) + client = AgentChatClient(api_key="sk_test", base_url="https://api.test") + try: + rows = client.sync(limit=100, after="del_9f86d081884c7d659a2feaa0c55ad015") + finally: + client.close() + + assert rows == [] + url = captured_url[0] + assert "limit=100" in url + assert "after=del_9f86d081884c7d659a2feaa0c55ad015" in url + + +def test_sync_non_array_payload_is_empty_batch_with_warning( + caplog: pytest.LogCaptureFixture, +) -> None: + # The pre-1.0.31 envelope-object shape (or any other non-array body) is + # a contract violation: warn + empty so drain loops terminate instead + # of spinning or crashing. + with respx.mock(base_url="https://api.test") as mock: + mock.get(url__regex=r".*/v1/messages/sync.*").mock( + return_value=httpx.Response(200, json={"envelopes": []}) + ) + client = AgentChatClient(api_key="sk_test", base_url="https://api.test") + try: + with caplog.at_level(logging.WARNING, logger="agentchat.client"): + rows = client.sync() + finally: + client.close() + + assert rows == [] + assert any("non-array" in rec.message for rec in caplog.records) + + +@pytest.mark.asyncio +async def test_async_sync_returns_bare_array() -> None: + with respx.mock(base_url="https://api.test") as mock: + mock.get(url__regex=r".*/v1/messages/sync.*").mock( + return_value=httpx.Response(200, json=WIRE_SYNC_BATCH) + ) + async with AsyncAgentChatClient( + api_key="sk_test", base_url="https://api.test" + ) as client: + rows = await client.sync(limit=100) + + assert rows == WIRE_SYNC_BATCH + + +# ─────────────── POST /v1/messages/sync/ack ─────────────── + + +def test_sync_ack_posts_string_cursor_and_parses_acked_count() -> None: + captured_body: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.append(json.loads(request.content.decode())) + return httpx.Response(200, json={"acked": 2}) + + with respx.mock(base_url="https://api.test") as mock: + mock.post("/v1/messages/sync/ack").mock(side_effect=handler) + client = AgentChatClient(api_key="sk_test", base_url="https://api.test") + try: + result = client.sync_ack("del_ab56b4d92b40713acc5af89985d4b786") + finally: + client.close() + + assert captured_body == [ + {"last_delivery_id": "del_ab56b4d92b40713acc5af89985d4b786"} + ] + assert result == {"acked": 2} + + +@pytest.mark.asyncio +async def test_async_sync_ack_posts_string_cursor() -> None: + captured_body: list[dict[str, Any]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.append(json.loads(request.content.decode())) + return httpx.Response(200, json={"acked": 1}) + + with respx.mock(base_url="https://api.test") as mock: + mock.post("/v1/messages/sync/ack").mock(side_effect=handler) + async with AsyncAgentChatClient( + api_key="sk_test", base_url="https://api.test" + ) as client: + result = await client.sync_ack("del_9f86d081884c7d659a2feaa0c55ad015") + + assert captured_body[0]["last_delivery_id"] == ( + "del_9f86d081884c7d659a2feaa0c55ad015" + ) + assert result["acked"] == 1 + + +@pytest.mark.parametrize("bad_cursor", [42, "", None, 41.0]) +def test_sync_ack_rejects_non_string_cursors(bad_cursor: Any) -> None: + # Legacy (<=1.0.3) callers passed numeric ids — fail fast client-side + # with a migration hint rather than a server-side VALIDATION_ERROR. + client = AgentChatClient(api_key="sk_test", base_url="https://api.test") + try: + with pytest.raises(TypeError, match="non-empty string cursor"): + client.sync_ack(bad_cursor) + finally: + client.close() + + +@pytest.mark.asyncio +async def test_async_sync_ack_rejects_non_string_cursors() -> None: + async with AsyncAgentChatClient( + api_key="sk_test", base_url="https://api.test" + ) as client: + with pytest.raises(TypeError, match="non-empty string cursor"): + await client.sync_ack(42) # type: ignore[arg-type] + + +# ─────────────── Realtime drain ↔ HTTP client integration ─────────────── + + +@pytest.mark.asyncio +async def test_realtime_drain_speaks_real_wire_end_to_end() -> None: + """Drain through a REAL ``AsyncAgentChatClient`` against the mocked + production wire — catches signature/shape drift between the realtime + drain and the HTTP client, the exact bug class that shipped in 1.0.2 + (drain unwrapped ``batch["envelopes"]`` that the wire never sends). + """ + from agentchatme import RealtimeClient + + sync_urls: list[str] = [] + ack_bodies: list[dict[str, Any]] = [] + + def sync_handler(request: httpx.Request) -> httpx.Response: + sync_urls.append(str(request.url)) + return httpx.Response(200, json=WIRE_SYNC_BATCH) + + def ack_handler(request: httpx.Request) -> httpx.Response: + ack_bodies.append(json.loads(request.content.decode())) + return httpx.Response(200, json={"acked": 3}) + + with respx.mock(base_url="https://api.test") as mock: + mock.get(url__regex=r".*/v1/messages/sync(\?.*)?$").mock( + side_effect=sync_handler + ) + mock.post("/v1/messages/sync/ack").mock(side_effect=ack_handler) + + async with AsyncAgentChatClient( + api_key="sk_test", base_url="https://api.test" + ) as client: + rt = RealtimeClient( + api_key="sk_test", + client=client, + auto_drain_on_connect=False, + reconnect=False, + ) + handled: list[str] = [] + rt.on("message.new", lambda m: handled.append(m["payload"]["id"])) + await rt.drain_offline_envelopes() + await rt.disconnect() + + # Every wire row reached the handlers, through the real HTTP client. + assert handled == [row["id"] for row in WIRE_SYNC_BATCH] + # One page (3 rows < the drain's requested limit of 100 → short page). + assert len(sync_urls) == 1 + assert "limit=100" in sync_urls[0] + # Acked once, with the positional cursor: the LAST NON-NULL delivery_id + # (row 3's null is skipped), as a string. + assert ack_bodies == [ + {"last_delivery_id": "del_ab56b4d92b40713acc5af89985d4b786"} + ] + + +# ─────────────── Cursor helper ─────────────── + + +def test_last_sync_delivery_id_is_positional() -> None: + rows: list[SyncRow] = [ + {"id": "m1", "conversation_id": "c1", "delivery_id": "del_zzz"}, + {"id": "m2", "conversation_id": "c1", "delivery_id": "del_aaa"}, + {"id": "m3", "conversation_id": "c1", "delivery_id": None}, + ] + # Last non-null wins by POSITION — "del_aaa" despite sorting before + # "del_zzz"; the trailing null is skipped, not treated as a reset. + assert last_sync_delivery_id(rows) == "del_aaa" + + +def test_last_sync_delivery_id_handles_unackable_batches() -> None: + assert last_sync_delivery_id([]) is None + all_null: list[SyncRow] = [ + {"id": "m1", "conversation_id": "c1", "delivery_id": None}, + {"id": "m2", "conversation_id": "c1", "delivery_id": ""}, + ] + # Nulls and empty strings are both unackable. + assert last_sync_delivery_id(all_null) is None