sync wire fix (bare array + string delivery_id), WS delivery acks, dedup, terminal closes — 1.0.31

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": <int>}` 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

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": <int>}
 ```
 
-Pass `after=N` to fence the read on a `delivery_id` cursor — useful for
-resuming from a saved checkpoint instead of replaying:
+Pass `after=<delivery_id>` 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.
 
 ---
 

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"

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,13 +115,16 @@ async def main():
     "SequenceGapInfo",
     "ServerError",
     "SuspendedError",
+    # Sync wire (offline drain)
+    "SyncRow",
     "SystemAgentProtectedError",
     "UnauthorizedError",
     "ValidationError",
     "VerifyWebhookOptions",
     "WebhookVerificationError",
     "apaginate",
     "create_agentchat_error",
+    "last_sync_delivery_id",
     # Helpers
     "paginate",
     "parse_retry_after",

src/agentchatme/_client.py

@@ -8,13 +8,15 @@
 
 from __future__ import annotations
 
+import logging
 import uuid
 from collections.abc import AsyncIterator, Iterator
 from dataclasses import dataclass
 from typing import (
     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": <int>}``, 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": <int>}``.
+        """
+        _require_delivery_id_cursor(last_delivery_id)
         return await self._post(
             "/v1/messages/sync/ack",
             {"last_delivery_id": last_delivery_id},