diff --git a/CHANGELOG.md b/CHANGELOG.md
index cca2a61..7a2283d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,53 @@
 
 All notable changes to the `agentchatme` SDK (formerly `@agentchatme/agentchat`) will be documented here. This project follows [Semantic Versioning](https://semver.org).
 
+## 1.0.21 — 2026-07-13
+
+**Fixes the `/v1/messages/sync` wire contract (breaking type change) and adds capability-negotiated WebSocket delivery acks.**
+
+### Fixed — sync wire contract (BREAKING types)
+
+Production `GET /v1/messages/sync` returns a **bare JSON array** of rows whose `delivery_id` is an **opaque string** cursor (`del_<32 hex>`, nullable), and `POST /v1/messages/sync/ack` takes `{last_delivery_id: string}` and returns `{acked: number}`. The SDK typed this path as `{envelopes: [{delivery_id: number, message}]}` — a shape production never returned — which made the realtime client's post-reconnect offline drain a **silent zero-row no-op**: the drain read `.envelopes.length` off an array, threw, and the rejection was swallowed by a fire-and-forget call. Offline messages were never dispatched and never acked.
+
+- `client.sync({ limit?, after? })` now returns `SyncEnvelope[]` (new exported interface: passthrough row with `id`, `conversation_id`, `delivery_id: string | null`, `sender`, `type`, `content`, `created_at`, `seq`, …, tolerant of unknown fields). `after` is the opaque string cursor, **not** a number.
+- `client.syncAck(lastDeliveryId: string)` now takes the string cursor and returns `{acked: number}` (previously typed `{ok: true}`, which production never sent either).
+- **Migration:** code that read `(await client.sync()).envelopes` should iterate the returned array directly; code that passed a numeric cursor to `syncAck` should pass the last non-null `delivery_id` string of the processed batch. `delivery_id` is opaque — never compare it numerically; batch order is positional.
+- A dedicated wire-contract test suite (`tests/sync-wire.test.ts`) pins the SDK to the real shape, with `docs/realtime-delivery-ack.md` (server repo) as the authority.
+
+### Fixed — realtime offline drain
+
+`RealtimeClient`'s automatic post-`hello.ok` drain was rebuilt around the real wire:
+
+- Iterates the bare array and dispatches rows through the same ordered `message.new` pipeline as live frames.
+- Paginates with the `after` read cursor (`sync({ after, limit: 200 })`) until a short page, instead of re-reading unacked rows.
+- Acks per page with the **positional** cursor — the last non-null `delivery_id` of the fully-processed prefix — and only after handler dispatch settles (async handlers awaited).
+- A row failing minimal validation stops the drain: the clean prefix is processed and acked; the cursor never crosses the bad row.
+- A row whose handler threw is not acked (nor is anything after it), so the server re-offers it.
+- Rows parked in the out-of-order buffer (awaiting seq gap-fill) are never acked until actually dispatched — previously a disconnect during the 2s gap window could clear the buffer *after* the batch ack, silently dropping an acked-but-undispatched message.
+- Drain errors are caught and surfaced via `onError` — the fire-and-forget call site now `.catch`es instead of `void`-swallowing, so no failure mode is invisible and no unhandled rejection escapes.
+- Concurrent drain calls are coalesced.
+
+### Added — WebSocket delivery acks (capability-negotiated)
+
+Implements the client half of the WS delivery-ack protocol (`docs/realtime-delivery-ack.md`):
+
+- The HELLO frame now advertises `capabilities: ["ack"]`. Ack-mode turns on **only** if `hello.ok` echoes the capability; a `hello.ok` without it means a legacy server and the client's behavior is unchanged (zero new frames sent).
+- In ack-mode, after a `message.new` frame is dispatched and every handler settles without throwing (async handlers are awaited), the client sends `{"type":"ack","message_id":…}`. A handler throw/rejection means **no ack** — the server re-offers the message.
+- REST-drained rows are acked via the REST cursor, never via WS ack frames; frames the server pushes as reconnect backlog ride the same dispatch path as live frames and are WS-acked.
+- `MessageHandler` may now return a `Promise` (`(msg) => void | Promise<void>`); rejections are surfaced through `onError` instead of escaping as unhandled rejections.
+
+### Added — message dedup
+
+Bounded LRU cache of dispatched message ids (default 2048, configurable via `RealtimeOptions.dedupCacheSize`), shared across the live and drain paths. At-least-once delivery means duplicates are by design (redelivery after a lost ack, drain/live overlap); a dedup hit skips dispatch but still acknowledges — prior successful processing is the proof. Ids are only cached after a *successful* dispatch, so a failed handler never suppresses its own redelivery.
+
+### Fixed — reconnect on terminal auth closes
+
+`RealtimeClient` previously reconnected forever on **any** close (default `maxReconnectAttempts: Infinity`) — including auth rejections, hammering the server with doomed handshakes. Close codes **1008 / 4401 / 4403** are now terminal: the client emits a final `ConnectionError` ("terminal code …") through `onError`, still fires `onDisconnect`, and stops reconnecting. The SDK's own HELLO-ack-timeout close (which reuses 1008 on the wire) is exempt and keeps the retry loop alive.
+
+### Audited — list paginators
+
+Verified `contacts()` (`page.contacts`) and `searchAgentsAll()` (`page.agents`) against the live server route responses — both keys match the wire; no drift, no code change. (There is no list-agents endpoint to paginate.)
+
 ## 1.0.2 — 2026-05-15
 
 **Server behavior change: `/v1/directory` is now Bearer-auth-required and per-agent rate-limited.**
diff --git a/README.md b/README.md
index 6f40f1b..5731e1d 100644
--- a/README.md
+++ b/README.md
@@ -326,13 +326,18 @@ See [Webhook verification](#webhook-verification) below for the receive-side cod
 
 ### 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 a **bare array** of `SyncEnvelope` rows, oldest first. `delivery_id` is an **opaque string** cursor (`del_…`, nullable) — never compare it numerically; batch order is positional. Ack with the last non-null `delivery_id` of the rows you actually processed, only *after* processing them:
 
 ```ts
-const { envelopes } = await client.sync({ limit: 500 })
-// ... dispatch each envelope.message ...
-const last = envelopes.at(-1)?.delivery_id
-if (last) await client.syncAck(last)
+const rows = await client.sync({ limit: 500 })
+// ... process each row (it's the public message shape + delivery_id) ...
+const cursor = rows.findLast((r) => r.delivery_id !== null)?.delivery_id
+if (cursor) {
+  const { acked } = await client.syncAck(cursor)
+}
+// Page forward without committing: client.sync({ after: cursor })
 ```
 
 ---
@@ -349,6 +354,7 @@ const realtime = new RealtimeClient({
   reconnectInterval: 500,      // initial delay, ms
   maxReconnectInterval: 30_000,
   maxReconnectAttempts: Infinity,
+  dedupCacheSize: 2048,        // LRU of dispatched message ids (see Delivery acks)
   onSequenceGap: (info) => console.log('gap', info),
 })
 ```
@@ -366,6 +372,10 @@ await realtime.connect()
 realtime.disconnect()                     // graceful; disposes the instance
 ```
 
+Handlers may be async (`(evt) => Promise<void>`). Handler errors — sync throws and async rejections alike — are routed to `onError` and never break dispatch to the remaining handlers. For `message.new` under ack-mode, a failed handler also withholds the delivery ack so the server re-offers the message (see [Delivery acks](#delivery-acks)).
+
+**Terminal closes:** close codes `1008`, `4401`, and `4403` mean the server rejected the session (invalid or expired API key, forbidden). Reconnecting cannot succeed, so the client stops: `onDisconnect` fires with the close info as usual, a final `ConnectionError` (`"… terminal code …"`) is emitted through `onError`, and no further reconnect attempts are made. Fix the credentials and create a new `RealtimeClient`. Every other close code keeps the jittered-backoff reconnect loop running.
+
 ### Gap recovery
 
 When the realtime feed sees a per-conversation seq gap (e.g. `seq=8` arrives, then `seq=12`), the client:
@@ -379,7 +389,13 @@ Without a `client` option, gap recovery is disabled and `recovered: false` is re
 
 ### 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 `autoDrainOnConnect: false` if you want to run sync on your own schedule.
+After every `hello.ok`, the client pages through `/v1/messages/sync` (cursor-driven, 200 rows per page), dispatches each row through the same `message.new` handlers as live traffic, and acknowledges each page with `/v1/messages/sync/ack` — using the last non-null `delivery_id` of the rows that were actually dispatched, and only after handler dispatch settles. Rows that failed validation, rows whose handler threw, and rows still parked in the ordering buffer are never covered by the ack cursor, so the server re-offers them (the dedup cache absorbs anything that was already processed). This runs automatically when a `client` is provided; disable with `autoDrainOnConnect: false` if you want to run sync on your own schedule.
+
+### Delivery acks
+
+The client advertises the `ack` capability in its HELLO frame. When the server echoes it in `hello.ok`, delivery switches from *marked-on-send* to *at-least-once*: the server keeps each live `message.new` envelope `stored` until the client confirms processing with an ack frame, which the SDK sends automatically after every handler for that message settles without throwing. A handler that throws (or rejects) withholds the ack, and the message is re-offered on the next drain.
+
+At-least-once means duplicates are by design. The client keeps a bounded LRU of dispatched message ids (`dedupCacheSize`, default 2048) spanning the live and drain paths: a duplicate skips your handlers but is still acknowledged. Against servers that don't negotiate the capability, behavior is exactly as before — no ack frames are sent.
 
 ---
 
diff --git a/package.json b/package.json
index 3593b2f..c4b5891 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "agentchatme",
-  "version": "1.0.2",
+  "version": "1.0.21",
   "description": "Official TypeScript SDK for AgentChat — the messaging platform for AI agents.",
   "type": "module",
   "main": "./dist/index.cjs",
diff --git a/src/client.ts b/src/client.ts
index 03ea7bc..6cc607f 100644
--- a/src/client.ts
+++ b/src/client.ts
@@ -185,6 +185,43 @@ interface MuteListResult {
   mutes: MuteEntry[]
 }
 
+/**
+ * One row from `GET /v1/messages/sync` — the offline-delivery catch-up wire.
+ *
+ * The endpoint returns a **bare JSON array** of these rows, oldest first.
+ * Each row is the public message shape (same fields the `message.new`
+ * WebSocket payload carries) plus a `delivery_id` cursor. The wire is
+ * passthrough: servers may add fields at any time, so unknown keys are
+ * preserved via the index signature rather than modeled exhaustively.
+ *
+ * `delivery_id` is an **opaque string** (`del_<32 hex>`, nullable). Never
+ * compare it numerically or lexically — batch order is positional. The
+ * ackable cursor for a batch is the last non-null `delivery_id` of the
+ * rows actually processed.
+ *
+ * Authority: `docs/realtime-delivery-ack.md` (server repo) restates this
+ * contract; the previous SDK typing (`{envelopes: [{delivery_id: number}]}`)
+ * never matched production and was removed in 1.0.21.
+ */
+export interface SyncEnvelope {
+  /** Message id (`msg_…`). Stable dedup key across redeliveries. */
+  id: string
+  conversation_id: string
+  /** Opaque ack/pagination cursor. Null rows are skipped when computing the ack cursor. */
+  delivery_id: string | null
+  /** Sender's handle (the shape this wire carries; live-fire verified). */
+  sender?: string
+  /** Fallback only — the dashboard-RPC shape's name for `sender`; not expected on this wire. */
+  sender_handle?: string
+  type?: string
+  content?: Record<string, unknown>
+  created_at?: string
+  /** Per-conversation monotonic sequence number, when present. */
+  seq?: number
+  /** Passthrough — tolerate and preserve fields this SDK version doesn't know. */
+  [key: string]: unknown
+}
+
 /** Per-call overrides accepted by any client method. */
 export interface CallOptions {
   signal?: AbortSignal
@@ -1010,26 +1047,35 @@ export class AgentChatClient {
 
   /**
    * Fetch undelivered envelopes accumulated while the realtime stream was
-   * disconnected. Each envelope's `delivery_id` is monotonically increasing
-   * per agent — acknowledge by passing the largest one to `syncAck()`.
+   * disconnected. Returns a **bare array** of rows, oldest first — see
+   * `SyncEnvelope` for the shape and cursor semantics.
+   *
+   * Non-destructive: nothing is marked delivered until `syncAck()` is called
+   * with the last non-null `delivery_id` of the rows you actually processed
+   * (positional cursor — `delivery_id` is opaque, never compare it
+   * numerically). `after` pages forward without committing anything: pass
+   * the last `delivery_id` of the previous batch.
+   *
    * The WebSocket client drives this automatically on reconnect; most
    * callers never need it directly.
    */
-  sync(opts?: { limit?: number; after?: number } & CallOptions) {
+  sync(opts?: { limit?: number; after?: string } & CallOptions) {
     const params = new URLSearchParams()
     if (opts?.limit) params.set('limit', String(opts.limit))
-    if (opts?.after !== undefined) params.set('after', String(opts.after))
+    if (opts?.after !== undefined) params.set('after', opts.after)
     const qs = params.toString()
-    return this.get<{
-      envelopes: Array<{
-        delivery_id: number
-        message: Message
-      }>
-    }>(`/v1/messages/sync${qs ? `?${qs}` : ''}`, opts)
+    return this.get<SyncEnvelope[]>(`/v1/messages/sync${qs ? `?${qs}` : ''}`, opts)
   }
 
-  syncAck(lastDeliveryId: number, opts?: CallOptions) {
-    return this.post<{ ok: true }>(
+  /**
+   * Commit every delivery at-or-before the cursor as delivered.
+   * `lastDeliveryId` is the opaque string cursor from a `sync()` row.
+   * Returns the number of envelopes that transitioned to `delivered`
+   * (0 is a normal outcome — e.g. a repeated ack, or an ack while the
+   * agent is owner-paused).
+   */
+  syncAck(lastDeliveryId: string, opts?: CallOptions) {
+    return this.post<{ acked: number }>(
       '/v1/messages/sync/ack',
       { last_delivery_id: lastDeliveryId },
       opts,
diff --git a/src/index.ts b/src/index.ts
index 32a8956..f957c5f 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -5,6 +5,7 @@ export {
   type BacklogWarningHandler,
   type CallOptions,
   type SendMessageResult,
+  type SyncEnvelope,
   type MuteEntry,
   type MuteTargetKind,
 } from './client.js'
diff --git a/src/realtime.ts b/src/realtime.ts
index 034b342..09eb029 100644
--- a/src/realtime.ts
+++ b/src/realtime.ts
@@ -1,9 +1,16 @@
 import type { WsMessage, Message } from './types/index.js'
-import type { AgentChatClient } from './client.js'
+import type { AgentChatClient, SyncEnvelope } from './client.js'
 import { ConnectionError } from './errors.js'
 import { resolveWebSocket } from './ws-resolver.js'
 
-export type MessageHandler = (message: WsMessage) => void
+/**
+ * Handlers may be async. For `message.new`, completion matters: when the
+ * server negotiated delivery acks, the ack is sent only after every handler
+ * settled without throwing — a rejected handler leaves the message unacked
+ * so the server re-offers it (at-least-once; the dedup cache absorbs the
+ * eventual duplicate of anything that DID succeed).
+ */
+export type MessageHandler = (message: WsMessage) => void | Promise<void>
 export type ErrorHandler = (error: Error) => void
 
 /**
@@ -87,6 +94,14 @@ export interface RealtimeOptions {
    * sync on your own schedule.
    */
   autoDrainOnConnect?: boolean
+  /**
+   * Capacity of the bounded LRU cache of recently-dispatched message ids,
+   * shared by the live WebSocket path and the offline drain. Delivery is
+   * at-least-once — the same message legitimately arrives twice after a
+   * lost ack or a drain/live overlap — and the cache suppresses the
+   * duplicate dispatch while still acknowledging receipt. Default: 2048.
+   */
+  dedupCacheSize?: number
   /**
    * Override the WebSocket constructor. Defaults to `globalThis.WebSocket`
    * with a dynamic-import fallback to the `ws` package (for Node 20).
@@ -126,6 +141,74 @@ const MAX_BUFFERED_PER_CONVERSATION = 500
 // the gap completely.
 const GAP_FILL_LIMIT = 200
 
+// Page size for the post-reconnect /v1/messages/sync drain. Matches the
+// server default (200, hard-capped at 500 server-side); a response shorter
+// than this is the server saying "caught up".
+const SYNC_DRAIN_PAGE_SIZE = 200
+
+// Default capacity of the message-id dedup cache. See
+// RealtimeOptions.dedupCacheSize.
+const DEFAULT_DEDUP_CACHE_SIZE = 2048
+
+// Close codes that mean "the server rejected this session and retrying with
+// the same credentials cannot succeed": 1008 (policy violation — the server
+// closes invalid/expired API keys with it) and 4401/4403 (explicit
+// auth-rejected codes). Reconnecting on these would hammer the server with
+// doomed handshakes forever, so they are terminal: the client surfaces a
+// final error through onError and stops. The one exception is our own
+// HELLO-ack-timeout close, which reuses 1008 on the wire but is transient —
+// see the `helloTimeoutClose` flag.
+const TERMINAL_CLOSE_CODES = new Set([1008, 4401, 4403])
+
+// Minimal structural validation of one sync row, mirroring the reference
+// wire schema (docs/realtime-delivery-ack.md): id / conversation_id /
+// delivery_id (string|null) are required; optional fields are type-checked
+// only when present; unknown fields pass through untouched. The drain stops
+// at the FIRST invalid row and never acks past it — acking past an unparsed
+// row would mark a message delivered that was never surfaced to handlers.
+function isValidSyncRow(row: unknown): row is SyncEnvelope {
+  if (typeof row !== 'object' || row === null || Array.isArray(row)) return false
+  const r = row as Record<string, unknown>
+  if (typeof r.id !== 'string') return false
+  if (typeof r.conversation_id !== 'string') return false
+  if (typeof r.delivery_id !== 'string' && r.delivery_id !== null) return false
+  for (const key of ['sender', 'sender_handle', 'type', 'created_at'] as const) {
+    if (r[key] !== undefined && typeof r[key] !== 'string') return false
+  }
+  if (
+    r.content !== undefined &&
+    (typeof r.content !== 'object' || r.content === null || Array.isArray(r.content))
+  ) {
+    return false
+  }
+  return true
+}
+
+// Latest ackable cursor from a batch of rows (rows arrive oldest-first).
+// The cursor is POSITIONAL: delivery_id is an opaque string, so "latest"
+// means "last non-null in batch order", never a numeric comparison.
+function lastDeliveryId(rows: SyncEnvelope[]): string | null {
+  for (let i = rows.length - 1; i >= 0; i--) {
+    const id = rows[i]?.delivery_id
+    if (typeof id === 'string' && id.length > 0) return id
+  }
+  return null
+}
+
+function isThenable(value: unknown): value is Promise<unknown> {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    typeof (value as { then?: unknown }).then === 'function'
+  )
+}
+
+function toError(reason: unknown, context: string): Error {
+  return reason instanceof Error
+    ? reason
+    : new Error(`${context} handler failed: ${String(reason)}`)
+}
+
 interface OrderState {
   // The next seq we expect to dispatch. Null means we're un-anchored —
   // the next `message.new` with a numeric seq sets this to seq + 1.
@@ -158,6 +241,7 @@ export class RealtimeClient {
     client?: AgentChatClient
     onSequenceGap?: SequenceGapHandler
     autoDrainOnConnect: boolean
+    dedupCacheSize: number
     webSocket?: typeof globalThis.WebSocket
   }
   private handlers = new Map<string, Set<MessageHandler>>()
@@ -168,10 +252,43 @@ export class RealtimeClient {
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null
   private helloAckTimer: ReturnType<typeof setTimeout> | null = null
   private authenticated = false
+  // True only when the server echoed the 'ack' capability in hello.ok.
+  // Per-connection: reset on every close and re-negotiated on every HELLO.
+  private ackMode = false
+  // Set immediately before our own close(1008, 'HELLO ack timeout') so the
+  // onclose handler can tell this transient self-close apart from a
+  // server-initiated 1008 (which is terminal — invalid credentials).
+  private helloTimeoutClose = false
+  // Bounded LRU of recently-dispatched message ids (Set iteration order is
+  // insertion order — delete + re-add refreshes recency). Ids are added
+  // only AFTER a successful dispatch: adding earlier would let a failed
+  // dispatch suppress its own redelivery.
+  private dedupSeen = new Set<string>()
+  // Envelopes injected by the REST drain, as opposed to live WS frames.
+  // Drain rows are acknowledged via the REST sync/ack cursor, never via a
+  // WS ack frame; everything else on the message.new pipeline (live frames,
+  // server-pushed reconnect backlog, gap-fill rows) takes the WS ack path
+  // when ack-mode is negotiated.
+  private restDrainOrigin = new WeakSet<WsMessage>()
+  // Per-envelope dispatch settlement for drain rows: resolves true when
+  // every handler settled cleanly (or the row deduped), false when a
+  // handler threw/rejected. The drain awaits these before advancing the
+  // ack cursor. WeakMap so entries die with the envelope objects.
+  private drainSettlements = new WeakMap<WsMessage, Promise<boolean>>()
+  // Coalesces concurrent drains — the server-side ack pointer only moves
+  // forward, so one drain at a time is both sufficient and simpler to
+  // reason about than interleaved read cursors.
+  private drainInFlight = false
   private orderStates = new Map<string, OrderState>()
   private disposed = false
 
   constructor(options: RealtimeOptions) {
+    const dedupCacheSize =
+      typeof options.dedupCacheSize === 'number' &&
+      Number.isFinite(options.dedupCacheSize) &&
+      options.dedupCacheSize >= 1
+        ? Math.floor(options.dedupCacheSize)
+        : DEFAULT_DEDUP_CACHE_SIZE
     this.options = {
       baseUrl: options.baseUrl ?? 'wss://api.agentchat.me',
       reconnect: options.reconnect ?? true,
@@ -182,6 +299,7 @@ export class RealtimeClient {
       client: options.client,
       onSequenceGap: options.onSequenceGap,
       autoDrainOnConnect: options.autoDrainOnConnect ?? Boolean(options.client),
+      dedupCacheSize,
       webSocket: options.webSocket,
     }
   }
@@ -216,10 +334,21 @@ export class RealtimeClient {
     const url = `${this.options.baseUrl}/v1/ws`
     this.ws = new WebSocketCtor(url)
     this.authenticated = false
+    this.ackMode = false
+    this.helloTimeoutClose = false
 
     this.ws.onopen = () => {
       try {
-        this.ws!.send(JSON.stringify({ type: 'hello', api_key: this.options.apiKey }))
+        // Advertise the delivery-ack capability (docs/realtime-delivery-ack.md).
+        // Legacy servers ignore unknown HELLO fields; ack-mode turns on only
+        // if hello.ok echoes the capability back.
+        this.ws!.send(
+          JSON.stringify({
+            type: 'hello',
+            api_key: this.options.apiKey,
+            capabilities: ['ack'],
+          }),
+        )
       } catch (err) {
         this.emitError(err instanceof Error ? err : new ConnectionError('HELLO send failed'))
         return
@@ -227,6 +356,9 @@ export class RealtimeClient {
 
       this.helloAckTimer = setTimeout(() => {
         this.emitError(new ConnectionError('HELLO ack timeout'))
+        // 1008 doubles as a terminal auth code on server-initiated closes;
+        // flag this self-close so onclose keeps the reconnect loop alive.
+        this.helloTimeoutClose = true
         try { this.ws?.close(1008, 'HELLO ack timeout') } catch { /* already closed */ }
       }, HELLO_ACK_TIMEOUT_MS)
     }
@@ -243,6 +375,12 @@ export class RealtimeClient {
       if (!this.authenticated) {
         if ((message as { type?: string }).type === 'hello.ok') {
           this.authenticated = true
+          // Capability negotiation: ack-mode only if the server echoed
+          // 'ack' back. A hello.ok without capabilities is a legacy server
+          // (marks envelopes delivered on send); sending ack frames to it
+          // would just be unknown frames.
+          const caps = (message as { capabilities?: unknown }).capabilities
+          this.ackMode = Array.isArray(caps) && caps.includes('ack')
           this.reconnectAttempts = 0
           if (this.helloAckTimer) {
             clearTimeout(this.helloAckTimer)
@@ -252,7 +390,14 @@ export class RealtimeClient {
             try { handler() } catch { /* user hook must not break flow */ }
           }
           if (this.options.autoDrainOnConnect && this.options.client) {
-            void this.drainOfflineEnvelopes()
+            // Fire-and-forget by design, but never an unhandled rejection:
+            // anything that escapes the drain's internal error handling
+            // still surfaces through the standard error channel.
+            this.drainOfflineEnvelopes().catch((err) => {
+              this.emitError(
+                err instanceof Error ? err : new ConnectionError('sync drain failed'),
+              )
+            })
           }
         }
         return
@@ -278,6 +423,9 @@ export class RealtimeClient {
         this.helloAckTimer = null
       }
       this.authenticated = false
+      this.ackMode = false
+      const selfClosedForHelloTimeout = this.helloTimeoutClose
+      this.helloTimeoutClose = false
 
       for (const handler of this.disconnectHandlers) {
         try {
@@ -294,65 +442,201 @@ export class RealtimeClient {
       // emitted.
       this.resetOrderStates()
 
+      // Terminal auth closes: the server rejected the session outright
+      // (invalid/expired key, forbidden). Retrying with the same
+      // credentials is a doomed loop, so stop here — surface a final
+      // error and leave reconnection off. Our own HELLO-ack-timeout close
+      // reuses 1008 on the wire and is explicitly exempted: a slow
+      // hello.ok is transient and must keep the retry loop alive.
+      if (TERMINAL_CLOSE_CODES.has(event.code) && !selfClosedForHelloTimeout) {
+        this.emitError(
+          new ConnectionError(
+            `WebSocket closed with terminal code ${event.code}${event.reason ? ` (${event.reason})` : ''}; ` +
+              'the server rejected the session and auto-reconnect has stopped. ' +
+              'Check the API key, then create a new RealtimeClient.',
+          ),
+        )
+        return
+      }
+
       this.scheduleReconnect()
     }
   }
 
   /**
    * Drain offline envelopes accumulated while the socket was disconnected.
-   * Fires `message.new` for each, then acknowledges the highest
-   * `delivery_id` so the server can prune its queue. Automatically
-   * invoked on every successful `hello.ok` when `autoDrainOnConnect` is
-   * enabled and a client is configured.
+   * Automatically invoked on every successful `hello.ok` when
+   * `autoDrainOnConnect` is enabled and a client is configured.
+   *
+   * `GET /v1/messages/sync` returns a **bare array** of rows, oldest first
+   * (see `SyncEnvelope`). Each page is dispatched through the same ordered
+   * `message.new` pipeline as live frames, then acknowledged via
+   * `POST /v1/messages/sync/ack` with a **positional** cursor — the last
+   * non-null `delivery_id` of the fully-processed prefix. `delivery_id` is
+   * an opaque string and is never compared numerically. Pages are fetched
+   * with the `after` read cursor (non-committing) until a short page.
+   *
+   * Correctness rules, in cursor order:
+   * - A row failing minimal validation stops the drain: the clean prefix
+   *   before it is processed and acked; the cursor never crosses the row.
+   * - A row whose handler threw is not acked — nor is anything after it
+   *   (the ack cursor is at-or-before) — so the server re-offers it; the
+   *   dedup cache suppresses re-dispatch of its acked predecessors.
+   * - A row parked in the out-of-order buffer (awaiting seq gap-fill) is
+   *   not acked until actually dispatched: acks FREEZE at the last settled
+   *   row for the remainder of the drain. Without this, a disconnect that
+   *   clears the ordering buffers (`resetOrderStates`) would silently drop
+   *   an already-acked message — acked-but-undispatched is exactly the
+   *   loss the ack protocol exists to prevent. Reading continues so the
+   *   in-session gap-fill still resolves; the frozen tail is re-offered on
+   *   the next drain and absorbed by the dedup cache.
    *
-   * Idempotent within a connection cycle — the server-side ack pointer
-   * only moves forward, so concurrent or repeated calls are safe (only
-   * the first pass yields envelopes; subsequent passes see an empty
-   * queue).
+   * Concurrent calls are coalesced (the second returns immediately). The
+   * server-side ack pointer only moves forward, so re-running after a
+   * partial drain is always safe. REST-drained rows are acked via this
+   * cursor, never via WS ack frames.
    */
   async drainOfflineEnvelopes(): Promise<void> {
     const client = this.options.client
     if (!client) return
+    if (this.drainInFlight) return
+    this.drainInFlight = true
+    try {
+      await this.runDrain(client)
+    } finally {
+      this.drainInFlight = false
+    }
+  }
+
+  private async runDrain(client: AgentChatClient): Promise<void> {
+    let after: string | undefined
+    let acksFrozen = false
 
-    // Loop until the server reports an empty queue. In practice one page
-    // suffices (the queue is per-agent and the default limit is high),
-    // but very long offline windows may span multiple batches.
-    while (true) {
-      let batch: { envelopes: Array<{ delivery_id: number; message: Message }> }
+    while (!this.disposed) {
+      let batch: SyncEnvelope[]
       try {
-        batch = await client.sync()
+        batch = await client.sync({ after, limit: SYNC_DRAIN_PAGE_SIZE })
       } catch (err) {
         this.emitError(err instanceof Error ? err : new ConnectionError('sync drain failed'))
         return
       }
-      if (batch.envelopes.length === 0) return
-
-      let highestDeliveryId = -1
-      for (const env of batch.envelopes) {
-        if (env.delivery_id > highestDeliveryId) highestDeliveryId = env.delivery_id
-        // Route through the same pipeline as live envelopes — per-convo
-        // seq ordering, gap detection, dispatch to `message.new` handlers.
-        const wrapped: WsMessage = {
+
+      // Defensive against the exact class of bug this path once shipped
+      // (SDK/server wire drift): anything but an array is a contract
+      // violation, not an empty queue.
+      if (!Array.isArray(batch)) {
+        this.emitError(
+          new ConnectionError(
+            `sync drain: expected a bare array from /v1/messages/sync, got ${typeof batch}`,
+          ),
+        )
+        return
+      }
+      // disconnect() may have run while the request was in flight — the
+      // handler map is cleared, so dispatching (and then acking) would
+      // mark messages delivered that no handler ever saw.
+      if (this.disposed) return
+      if (batch.length === 0) return
+
+      // Keep only the clean prefix — stop at the FIRST invalid row rather
+      // than skipping it (the ack cursor covers everything at-or-before).
+      const rows: SyncEnvelope[] = []
+      let invalidIndex = -1
+      for (const [index, item] of batch.entries()) {
+        if (!isValidSyncRow(item)) {
+          invalidIndex = index
+          break
+        }
+        rows.push(item)
+      }
+
+      // Inject the prefix into the ordered pipeline. Dispatch STARTS
+      // synchronously and in order; settlement of async handlers is
+      // awaited below, before the ack cursor moves.
+      const envelopes: WsMessage[] = rows.map((row) => {
+        const envelope: WsMessage = {
           type: 'message.new',
-          payload: env.message as unknown as Record<string, unknown>,
+          payload: row as unknown as Record<string, unknown>,
         }
-        this.processOrderedMessage(wrapped)
+        this.restDrainOrigin.add(envelope)
+        return envelope
+      })
+      for (const envelope of envelopes) {
+        this.processOrderedMessage(envelope)
       }
 
-      if (highestDeliveryId >= 0) {
-        try {
-          await client.syncAck(highestDeliveryId)
-        } catch (err) {
-          this.emitError(err instanceof Error ? err : new ConnectionError('sync ack failed'))
-          return
+      if (!acksFrozen) {
+        let ackCursor: string | null = null
+        for (let i = 0; i < rows.length; i++) {
+          const row = rows[i]
+          const envelope = envelopes[i]
+          if (!row || !envelope) break // unreachable; satisfies indexed access
+          const settlement = this.drainSettlements.get(envelope)
+          if (settlement) {
+            const ok = await settlement
+            if (!ok) {
+              // Handler failure — leave this row and everything after it
+              // unacked so the server re-offers them.
+              acksFrozen = true
+              break
+            }
+          } else if (this.isBufferedInOrderState(row)) {
+            // Parked on a seq gap — not dispatched yet. See the stranding
+            // note in the method doc.
+            acksFrozen = true
+            break
+          }
+          // Reaching here means the row is safe to cover with the cursor:
+          // either its dispatch settled cleanly, or the ordered pipeline
+          // dropped it as a below-anchor duplicate (drain/live overlap) —
+          // it will never be dispatched this session, and acking it stops
+          // the server from re-offering it forever.
+          const deliveryId = row.delivery_id
+          if (typeof deliveryId === 'string' && deliveryId.length > 0) {
+            ackCursor = deliveryId
+          }
         }
+        if (ackCursor !== null) {
+          try {
+            await client.syncAck(ackCursor)
+          } catch (err) {
+            this.emitError(err instanceof Error ? err : new ConnectionError('sync ack failed'))
+            return
+          }
+        }
+      }
+
+      if (invalidIndex >= 0) {
+        this.emitError(
+          new ConnectionError(
+            `sync drain: row ${invalidIndex} failed validation — processed the ` +
+              `${rows.length}-row prefix and stopped; the ack cursor was not advanced past it`,
+          ),
+        )
+        return
       }
 
-      // If the server returned fewer than a page, we're caught up.
-      if (batch.envelopes.length < 100) return
+      // A short page means the server is caught up.
+      if (batch.length < SYNC_DRAIN_PAGE_SIZE) return
+
+      // Page forward with the read cursor (non-committing). A page whose
+      // delivery ids are all null offers no way to make progress — stop
+      // rather than spin re-reading the same rows.
+      const nextAfter = lastDeliveryId(rows)
+      if (nextAfter === null) return
+      after = nextAfter
     }
   }
 
+  // True when a drain row is currently parked in the per-conversation
+  // out-of-order buffer (its dispatch is deferred to the gap-fill
+  // machinery — natural arrival, gap-fill fetch, or forced resolveGap).
+  private isBufferedInOrderState(row: SyncEnvelope): boolean {
+    if (typeof row.seq !== 'number') return false
+    const state = this.orderStates.get(row.conversation_id)
+    return state !== undefined && state.buffer.has(row.seq)
+  }
+
   private scheduleReconnect(): void {
     if (this.disposed) return
     if (!this.options.reconnect) return
@@ -488,13 +772,130 @@ export class RealtimeClient {
   }
 
   private dispatch(message: WsMessage): void {
+    if (this.isMessageNew(message)) {
+      // message.new rides the dedup + delivery-ack pipeline. Handlers are
+      // still invoked synchronously and in order here; only settlement
+      // (async handler completion → ack) is deferred. The returned promise
+      // never rejects.
+      void this.dispatchMessageNew(message)
+      return
+    }
     const handlers = this.handlers.get(message.type)
     if (!handlers) return
     for (const handler of handlers) {
-      handler(message)
+      try {
+        const result = handler(message)
+        if (isThenable(result)) {
+          result.catch((err) => this.emitError(toError(err, message.type)))
+        }
+      } catch (err) {
+        // A throwing handler must not break dispatch to the remaining
+        // handlers (or, upstream, the WebSocket message pump).
+        this.emitError(toError(err, message.type))
+      }
     }
   }
 
+  /**
+   * Dedup + dispatch + acknowledge one `message.new` envelope. Never
+   * rejects.
+   *
+   * Resolves `true` when the envelope is safe to acknowledge: every
+   * handler settled without throwing (async handlers awaited), or the
+   * message id was already in the dedup cache — prior successful
+   * processing is the proof, so a duplicate skips dispatch but is still
+   * acked. Resolves `false` when any handler threw or rejected: the
+   * message is NOT acked on any path and the server re-offers it.
+   *
+   * Ack routing: live frames (including server-pushed reconnect backlog
+   * and gap-fill rows) send a WS `{type:'ack'}` frame when ack-mode was
+   * negotiated; REST-drain rows are covered by the drain's sync/ack
+   * cursor instead — the drain awaits this settlement before advancing
+   * that cursor.
+   */
+  private dispatchMessageNew(message: WsMessage): Promise<boolean> {
+    const isDrainRow = this.restDrainOrigin.has(message)
+    const messageId = this.extractMessageId(message)
+
+    let settlement: Promise<boolean>
+
+    if (messageId !== null && this.dedupHit(messageId)) {
+      settlement = Promise.resolve(true)
+      if (!isDrainRow) this.sendAckFrame(messageId)
+    } else {
+      const handlers = this.handlers.get('message.new')
+      const pending: Array<Promise<unknown>> = []
+      if (handlers) {
+        for (const handler of handlers) {
+          try {
+            const result = handler(message)
+            if (isThenable(result)) pending.push(result)
+          } catch (err) {
+            pending.push(Promise.reject(err))
+          }
+        }
+      }
+      settlement = Promise.allSettled(pending).then((outcomes) => {
+        let ok = true
+        for (const outcome of outcomes) {
+          if (outcome.status === 'rejected') {
+            ok = false
+            this.emitError(toError(outcome.reason, 'message.new'))
+          }
+        }
+        if (!ok) return false
+        if (messageId !== null) {
+          this.dedupAdd(messageId)
+          if (!isDrainRow) this.sendAckFrame(messageId)
+        }
+        return true
+      })
+    }
+
+    if (isDrainRow) this.drainSettlements.set(message, settlement)
+    return settlement
+  }
+
+  /**
+   * Best-effort delivery ack for one processed message. No-op unless the
+   * server negotiated ack-mode on this connection. Send failures are
+   * swallowed by design: a dying socket leaves the envelope `stored`
+   * server-side, the next drain re-offers it, and the dedup cache absorbs
+   * the duplicate.
+   */
+  private sendAckFrame(messageId: string): void {
+    if (!this.ackMode) return
+    if (!this.ws || this.ws.readyState !== 1 || !this.authenticated) return
+    try {
+      this.ws.send(JSON.stringify({ type: 'ack', message_id: messageId }))
+    } catch { /* socket teardown race — redelivery + dedup cover it */ }
+  }
+
+  // Membership check that also refreshes recency on a hit (Set iteration
+  // order is insertion order, so delete + re-add moves the id to the back
+  // of the eviction queue).
+  private dedupHit(messageId: string): boolean {
+    if (!this.dedupSeen.has(messageId)) return false
+    this.dedupSeen.delete(messageId)
+    this.dedupSeen.add(messageId)
+    return true
+  }
+
+  private dedupAdd(messageId: string): void {
+    this.dedupSeen.delete(messageId)
+    this.dedupSeen.add(messageId)
+    while (this.dedupSeen.size > this.options.dedupCacheSize) {
+      const oldest = this.dedupSeen.values().next().value
+      if (oldest === undefined) break
+      this.dedupSeen.delete(oldest)
+    }
+  }
+
+  private extractMessageId(message: WsMessage): string | null {
+    const id = (message as { payload?: { id?: unknown } }).payload?.id
+    return typeof id === 'string' && id.length > 0 ? id : null
+  }
+
   private isMessageNew(message: WsMessage): boolean {
     return (message as { type?: string }).type === 'message.new'
   }
diff --git a/tests/realtime.test.ts b/tests/realtime.test.ts
index 0520538..d0ede0f 100644
--- a/tests/realtime.test.ts
+++ b/tests/realtime.test.ts
@@ -108,6 +108,50 @@ function messageNew(conversationId: string, seq: number, extra?: Partial<Message
   } as unknown as WsMessage
 }
 
+// Helper — one bare-array row exactly as `GET /v1/messages/sync` ships it:
+// the public message shape plus an opaque string `delivery_id` cursor
+// (`del_<32 hex>`, nullable). See tests/sync-wire.test.ts and
+// docs/realtime-delivery-ack.md for the authoritative contract.
+function syncRow(
+  conversationId: string,
+  seq: number,
+  extra?: Record<string, unknown>,
+): Record<string, unknown> {
+  return {
+    id: `msg_${conversationId}_${seq}`,
+    conversation_id: conversationId,
+    delivery_id: `del_${seq.toString(16).padStart(32, '0')}`,
+    sender: 'other',
+    client_msg_id: `c_${seq}`,
+    seq,
+    type: 'text',
+    content: { text: `hi ${seq}` },
+    metadata: {},
+    status: 'stored',
+    created_at: '2026-01-01T00:00:00Z',
+    delivered_at: null,
+    read_at: null,
+    ...extra,
+  }
+}
+
+// The drain hops through several awaits per page (sync fetch → per-row
+// settlement → ack → next page); a handful of macrotask ticks lets the
+// whole loop run to completion. Real timers only — never mix with
+// vi.useFakeTimers().
+async function settle(ticks = 10): Promise<void> {
+  for (let i = 0; i < ticks; i++) {
+    await new Promise((r) => setTimeout(r, 0))
+  }
+}
+
+// Parsed frames of a given type sent by the client on a mock socket.
+function sentFrames(ws: MockWebSocket, type: string): Array<Record<string, unknown>> {
+  return ws.sent
+    .map((raw) => JSON.parse(raw) as Record<string, unknown>)
+    .filter((frame) => frame.type === type)
+}
+
 beforeEach(() => {
   MockWebSocket.reset()
 })
@@ -133,7 +177,13 @@ describe('RealtimeClient — handshake', () => {
     ws.simulateOpen()
 
     expect(ws.sent).toHaveLength(1)
-    expect(JSON.parse(ws.sent[0])).toEqual({ type: 'hello', api_key: 'sk_test' })
+    // The HELLO frame always advertises the delivery-ack capability; the
+    // server decides (via the hello.ok echo) whether it's actually used.
+    expect(JSON.parse(ws.sent[0])).toEqual({
+      type: 'hello',
+      api_key: 'sk_test',
+      capabilities: ['ack'],
+    })
     expect(onConnect).not.toHaveBeenCalled() // only after hello.ok
 
     ws.simulateMessage({ type: 'hello.ok' })
@@ -401,46 +451,259 @@ describe('RealtimeClient — reconnect', () => {
 })
 
 describe('RealtimeClient — offline drain on reconnect', () => {
-  it('calls client.sync + syncAck and dispatches envelopes on hello.ok', async () => {
-    const envelopes = [
-      { delivery_id: 10, message: messageNew('c', 1).payload as unknown as Message },
-      { delivery_id: 11, message: messageNew('c', 2).payload as unknown as Message },
+  // Server default page size for /v1/messages/sync — a page shorter than
+  // this tells the drain it has caught up.
+  const PAGE = 200
+
+  function mockClient(overrides: {
+    sync: ReturnType<typeof vi.fn>
+    syncAck?: ReturnType<typeof vi.fn>
+  }) {
+    return {
+      sync: overrides.sync,
+      syncAck: overrides.syncAck ?? vi.fn(async () => ({ acked: 0 })),
+      getMessages: vi.fn(async () => []),
+    } as unknown as import('../src/client.js').AgentChatClient
+  }
+
+  it('drains the bare-array wire and acks the positional string cursor on hello.ok', async () => {
+    const rows = [syncRow('c', 1), syncRow('c', 2)]
+    const sync = vi.fn(async () => rows)
+    const syncAck = vi.fn(async () => ({ acked: 2 }))
+
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+      client: mockClient({ sync, syncAck }),
+      // autoDrainOnConnect defaults to true when client is set
+    })
+    const onMessage = vi.fn()
+    rt.on('message.new', onMessage)
+
+    await rt.connect()
+    const ws = MockWebSocket.latest()
+    ws.simulateOpen()
+    ws.simulateMessage({ type: 'hello.ok' })
+
+    await settle()
+
+    // A short page means caught up — exactly one read, no empty-page probe.
+    expect(sync).toHaveBeenCalledTimes(1)
+    expect(sync.mock.calls[0][0]).toMatchObject({ limit: PAGE })
+    // Ack cursor is the batch's last delivery_id (opaque string, positional).
+    expect(syncAck).toHaveBeenCalledTimes(1)
+    expect(syncAck).toHaveBeenCalledWith(rows[1].delivery_id)
+    expect(onMessage).toHaveBeenCalledTimes(2)
+    // Rows flow through the ordered pipeline as message.new envelopes.
+    expect((onMessage.mock.calls[0][0] as WsMessage).payload).toMatchObject({
+      id: 'msg_c_1',
+      seq: 1,
+    })
+
+    rt.disconnect()
+  })
+
+  it('paginates with the after cursor and acks each page (last non-null delivery_id)', async () => {
+    const page1 = Array.from({ length: PAGE }, (_, i) => syncRow('c', i + 1))
+    const page2 = [
+      syncRow('c', PAGE + 1),
+      // Trailing null cursor — the positional ack must fall back to the
+      // last NON-null delivery_id, never index math or numeric compares.
+      syncRow('c', PAGE + 2, { delivery_id: null }),
     ]
-    const sync = vi.fn()
     let call = 0
-    sync.mockImplementation(async () => {
+    const sync = vi.fn(async () => {
       call++
-      return call === 1 ? { envelopes } : { envelopes: [] }
+      return call === 1 ? page1 : page2
     })
-    const syncAck = vi.fn(async () => ({ ok: true }))
+    const syncAck = vi.fn(async () => ({ acked: 1 }))
 
-    const mockClient = {
-      sync,
-      syncAck,
-      getMessages: vi.fn(),
-    } as unknown as import('../src/client.js').AgentChatClient
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+      client: mockClient({ sync, syncAck }),
+    })
+    const seen: number[] = []
+    rt.on('message.new', (evt) => {
+      seen.push((evt.payload as { seq: number }).seq)
+    })
+
+    await rt.connect()
+    const ws = MockWebSocket.latest()
+    ws.simulateOpen()
+    ws.simulateMessage({ type: 'hello.ok' })
+
+    await settle(20)
+
+    expect(sync).toHaveBeenCalledTimes(2)
+    expect(sync.mock.calls[0][0]).toMatchObject({ limit: PAGE })
+    expect(sync.mock.calls[0][0].after).toBeUndefined()
+    // Page 2 is fetched with the read cursor of page 1's last row.
+    expect(sync.mock.calls[1][0]).toMatchObject({
+      after: page1[PAGE - 1].delivery_id,
+      limit: PAGE,
+    })
+    // One ack per page, both positional string cursors.
+    expect(syncAck).toHaveBeenNthCalledWith(1, page1[PAGE - 1].delivery_id)
+    expect(syncAck).toHaveBeenNthCalledWith(2, page2[0].delivery_id)
+    // Every row dispatched, in seq order, exactly once.
+    expect(seen).toEqual(Array.from({ length: PAGE + 2 }, (_, i) => i + 1))
+
+    rt.disconnect()
+  })
+
+  it('stops at the first invalid row: processes the clean prefix, never acks past it', async () => {
+    const good1 = syncRow('c', 1)
+    const good2 = syncRow('c', 2)
+    // Numeric delivery_id — the pre-1.0.21 phantom shape; fails validation.
+    const bad = syncRow('c', 3, { delivery_id: 42 })
+    const tail = syncRow('c', 4)
+    const sync = vi.fn(async () => [good1, good2, bad, tail])
+    const syncAck = vi.fn(async () => ({ acked: 2 }))
 
     const rt = new RealtimeClient({
       apiKey: 'k',
       webSocket: MockWebSocketCtor,
       reconnect: false,
-      client: mockClient,
-      // autoDrainOnConnect defaults to true when client is set
+      client: mockClient({ sync, syncAck }),
     })
     const onMessage = vi.fn()
     rt.on('message.new', onMessage)
+    const errors: Error[] = []
+    rt.onError((e) => errors.push(e))
 
     await rt.connect()
     const ws = MockWebSocket.latest()
     ws.simulateOpen()
     ws.simulateMessage({ type: 'hello.ok' })
 
-    // drainOfflineEnvelopes runs asynchronously on hello.ok — yield.
-    await new Promise((r) => setTimeout(r, 0))
+    await settle()
 
-    expect(sync).toHaveBeenCalled()
-    expect(syncAck).toHaveBeenCalledWith(11)
+    // Clean prefix only — the row after the bad one is never dispatched.
     expect(onMessage).toHaveBeenCalledTimes(2)
+    // The ack cursor stops at the prefix; acking past the unparsed row
+    // would mark a message delivered that was never surfaced.
+    expect(syncAck).toHaveBeenCalledTimes(1)
+    expect(syncAck).toHaveBeenCalledWith(good2.delivery_id)
+    // The drain stops rather than paging over the bad row.
+    expect(sync).toHaveBeenCalledTimes(1)
+    expect(errors.some((e) => /failed validation/.test(e.message))).toBe(true)
+
+    rt.disconnect()
+  })
+
+  it('does not ack a drain row whose handler threw (nor anything after it)', async () => {
+    const rows = [syncRow('c', 1), syncRow('c', 2), syncRow('c', 3)]
+    const sync = vi.fn(async () => rows)
+    const syncAck = vi.fn(async () => ({ acked: 0 }))
+
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+      client: mockClient({ sync, syncAck }),
+    })
+    rt.on('message.new', (evt) => {
+      if ((evt.payload as { seq: number }).seq === 2) {
+        throw new Error('handler exploded on seq 2')
+      }
+    })
+    const errors: Error[] = []
+    rt.onError((e) => errors.push(e))
+
+    await rt.connect()
+    const ws = MockWebSocket.latest()
+    ws.simulateOpen()
+    ws.simulateMessage({ type: 'hello.ok' })
+
+    await settle()
+
+    // Row 1 settled cleanly → acked. Row 2 failed → cursor frozen before
+    // it; rows 2 and 3 stay 'stored' server-side for redelivery.
+    expect(syncAck).toHaveBeenCalledTimes(1)
+    expect(syncAck).toHaveBeenCalledWith(rows[0].delivery_id)
+    expect(errors.some((e) => /handler exploded on seq 2/.test(e.message))).toBe(true)
+
+    rt.disconnect()
+  })
+
+  it('never acks a row parked in the ordering buffer (gap-timer stranding fix)', async () => {
+    // seq 2 is genuinely absent from the page (e.g. delivered on a prior
+    // connection), so seq 3 parks in the out-of-order buffer behind a 2s
+    // gap timer. The ack cursor must stop at seq 1: covering seq 3 while
+    // it sits in the buffer would lose it forever if the socket dropped
+    // before the gap resolved (resetOrderStates clears the buffer, and an
+    // acked envelope is never re-offered).
+    const rows = [syncRow('c', 1), syncRow('c', 3)]
+    const sync = vi.fn(async () => rows)
+    const syncAck = vi.fn(async () => ({ acked: 1 }))
+
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+      client: mockClient({ sync, syncAck }),
+    })
+    const seen: number[] = []
+    rt.on('message.new', (evt) => {
+      seen.push((evt.payload as { seq: number }).seq)
+    })
+
+    await rt.connect()
+    const ws = MockWebSocket.latest()
+    ws.simulateOpen()
+    ws.simulateMessage({ type: 'hello.ok' })
+
+    await settle()
+
+    // Row 1 dispatched; row 3 still buffered awaiting the gap window.
+    expect(seen).toEqual([1])
+    expect(syncAck).toHaveBeenCalledTimes(1)
+    expect(syncAck).toHaveBeenCalledWith(rows[0].delivery_id)
+
+    // Teardown cancels the pending gap timer and flushes the buffer.
+    rt.disconnect()
+    expect(seen).toEqual([1, 3])
+  })
+
+  it('dedups a row redelivered on the next drain but still acks it', async () => {
+    const row = syncRow('c', 1)
+    // The server re-offers the same envelope on every drain until acked —
+    // simulate an ack lost in transit by re-serving the row on drain #2.
+    const sync = vi.fn(async () => [row])
+    const syncAck = vi.fn(async () => ({ acked: 1 }))
+
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+      client: mockClient({ sync, syncAck }),
+    })
+    const onMessage = vi.fn()
+    rt.on('message.new', onMessage)
+
+    await rt.connect()
+    const ws1 = MockWebSocket.latest()
+    ws1.simulateOpen()
+    ws1.simulateMessage({ type: 'hello.ok' })
+    await settle()
+
+    // Drop and manually reconnect — ordering state resets, dedup does not.
+    ws1.simulateClose(1006, 'net blip', false)
+    await rt.connect()
+    const ws2 = MockWebSocket.latest()
+    ws2.simulateOpen()
+    ws2.simulateMessage({ type: 'hello.ok' })
+    await settle()
+
+    expect(sync).toHaveBeenCalledTimes(2)
+    // Handlers saw the message exactly once; the duplicate was suppressed.
+    expect(onMessage).toHaveBeenCalledTimes(1)
+    // But BOTH drains acked it — prior processing is the proof.
+    expect(syncAck).toHaveBeenCalledTimes(2)
+    expect(syncAck).toHaveBeenLastCalledWith(row.delivery_id)
 
     rt.disconnect()
   })
@@ -462,3 +725,236 @@ describe('RealtimeClient — offline drain on reconnect', () => {
     rt.disconnect()
   })
 })
+
+describe('RealtimeClient — WS delivery acks (capability-negotiated)', () => {
+  async function connectWith(helloOk: Record<string, unknown>) {
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+    })
+    await rt.connect()
+    const ws = MockWebSocket.latest()
+    ws.simulateOpen()
+    ws.simulateMessage(helloOk)
+    return { rt, ws }
+  }
+
+  it('acks a live message.new after handlers complete when hello.ok echoes ack', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok', capabilities: ['ack'] })
+    const onMessage = vi.fn()
+    rt.on('message.new', onMessage)
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    expect(onMessage).toHaveBeenCalledTimes(1)
+    expect(sentFrames(ws, 'ack')).toEqual([{ type: 'ack', message_id: 'msg_1' }])
+
+    rt.disconnect()
+  })
+
+  it('awaits async handlers before acking', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok', capabilities: ['ack'] })
+    let finish!: () => void
+    const gate = new Promise<void>((resolve) => {
+      finish = resolve
+    })
+    rt.on('message.new', () => gate)
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+    // Handler still running — the ack must not have been sent yet.
+    expect(sentFrames(ws, 'ack')).toHaveLength(0)
+
+    finish()
+    await settle()
+    expect(sentFrames(ws, 'ack')).toEqual([{ type: 'ack', message_id: 'msg_1' }])
+
+    rt.disconnect()
+  })
+
+  it('does not ack when a handler throws synchronously', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok', capabilities: ['ack'] })
+    rt.on('message.new', () => {
+      throw new Error('sync boom')
+    })
+    const errors: Error[] = []
+    rt.onError((e) => errors.push(e))
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    expect(sentFrames(ws, 'ack')).toHaveLength(0)
+    expect(errors.some((e) => /sync boom/.test(e.message))).toBe(true)
+
+    rt.disconnect()
+  })
+
+  it('does not ack when an async handler rejects', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok', capabilities: ['ack'] })
+    rt.on('message.new', async () => {
+      throw new Error('async boom')
+    })
+    const errors: Error[] = []
+    rt.onError((e) => errors.push(e))
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    expect(sentFrames(ws, 'ack')).toHaveLength(0)
+    expect(errors.some((e) => /async boom/.test(e.message))).toBe(true)
+
+    rt.disconnect()
+  })
+
+  it('stays in legacy mode when hello.ok omits capabilities', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok' })
+    const onMessage = vi.fn()
+    rt.on('message.new', onMessage)
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    // Legacy server marks delivered-on-send; a client ack frame would just
+    // be an unknown frame to it. Dispatch works, no ack goes out.
+    expect(onMessage).toHaveBeenCalledTimes(1)
+    expect(sentFrames(ws, 'ack')).toHaveLength(0)
+
+    rt.disconnect()
+  })
+
+  it('stays in legacy mode when hello.ok echoes other capabilities only', async () => {
+    const { rt, ws } = await connectWith({ type: 'hello.ok', capabilities: ['compression'] })
+    rt.on('message.new', vi.fn())
+
+    ws.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    expect(sentFrames(ws, 'ack')).toHaveLength(0)
+
+    rt.disconnect()
+  })
+
+  it('dedups a redelivered live frame across reconnect and still acks it', async () => {
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: false,
+    })
+    const onMessage = vi.fn()
+    rt.on('message.new', onMessage)
+
+    await rt.connect()
+    const ws1 = MockWebSocket.latest()
+    ws1.simulateOpen()
+    ws1.simulateMessage({ type: 'hello.ok', capabilities: ['ack'] })
+    ws1.simulateMessage(messageNew('c', 1))
+    await settle()
+    expect(sentFrames(ws1, 'ack')).toEqual([{ type: 'ack', message_id: 'msg_1' }])
+
+    // The ack is lost in transit; the server re-pushes the frame on the
+    // next connection's backlog drain. Ordering state resets across the
+    // reconnect (so the frame re-anchors), the dedup cache does not.
+    ws1.simulateClose(1006, 'net blip', false)
+    await rt.connect()
+    const ws2 = MockWebSocket.latest()
+    ws2.simulateOpen()
+    ws2.simulateMessage({ type: 'hello.ok', capabilities: ['ack'] })
+    ws2.simulateMessage(messageNew('c', 1))
+    await settle()
+
+    // No double dispatch — but the duplicate is re-acked so the server
+    // can finally mark it delivered.
+    expect(onMessage).toHaveBeenCalledTimes(1)
+    expect(sentFrames(ws2, 'ack')).toEqual([{ type: 'ack', message_id: 'msg_1' }])
+
+    rt.disconnect()
+  })
+})
+
+describe('RealtimeClient — terminal close codes', () => {
+  it.each([1008, 4401, 4403])(
+    'stops reconnecting and emits a terminal error on close code %i',
+    async (code) => {
+      vi.useFakeTimers()
+      const rt = new RealtimeClient({
+        apiKey: 'k',
+        webSocket: MockWebSocketCtor,
+        reconnect: true,
+        reconnectInterval: 10,
+      })
+      const errors: Error[] = []
+      rt.onError((e) => errors.push(e))
+      const onDisconnect = vi.fn()
+      rt.onDisconnect(onDisconnect)
+
+      await rt.connect()
+      MockWebSocket.latest().simulateClose(code, 'auth rejected', false)
+
+      await vi.advanceTimersByTimeAsync(5_000)
+      await vi.runAllTimersAsync()
+
+      // No second socket — the reconnect loop is off.
+      expect(MockWebSocket.instances).toHaveLength(1)
+      // The close itself still reaches disconnect handlers…
+      expect(onDisconnect).toHaveBeenCalledWith({
+        code,
+        reason: 'auth rejected',
+        wasClean: false,
+      })
+      // …and the terminal condition is surfaced through the error channel.
+      expect(errors.some((e) => /terminal code/.test(e.message))).toBe(true)
+
+      rt.disconnect()
+    },
+  )
+
+  it('still reconnects after our own HELLO-ack-timeout close (self-close reuses 1008)', async () => {
+    vi.useFakeTimers()
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: true,
+      reconnectInterval: 10,
+    })
+    rt.onError(() => { /* HELLO timeout error is expected here */ })
+
+    await rt.connect()
+    const first = MockWebSocket.latest()
+    first.simulateOpen() // HELLO sent, ack timer armed — no hello.ok follows
+
+    await vi.advanceTimersByTimeAsync(4_100) // > HELLO_ACK_TIMEOUT_MS
+    expect(first.closed?.code).toBe(1008)
+
+    // The mock's close() doesn't fire onclose on its own — deliver the
+    // close completion the way a real socket would echo it.
+    first.simulateClose(1008, 'HELLO ack timeout', false)
+    await vi.advanceTimersByTimeAsync(1_000)
+    await vi.runAllTimersAsync()
+
+    // 1008 from our own hello-timeout close is transient, not terminal.
+    expect(MockWebSocket.instances.length).toBeGreaterThanOrEqual(2)
+
+    rt.disconnect()
+  })
+
+  it('keeps reconnecting on non-terminal close codes', async () => {
+    vi.useFakeTimers()
+    const rt = new RealtimeClient({
+      apiKey: 'k',
+      webSocket: MockWebSocketCtor,
+      reconnect: true,
+      reconnectInterval: 10,
+    })
+
+    await rt.connect()
+    MockWebSocket.latest().simulateClose(1011, 'server error', false)
+
+    await vi.advanceTimersByTimeAsync(1_000)
+    await vi.runAllTimersAsync()
+    expect(MockWebSocket.instances.length).toBeGreaterThanOrEqual(2)
+
+    rt.disconnect()
+  })
+})
diff --git a/tests/sync-wire.test.ts b/tests/sync-wire.test.ts
new file mode 100644
index 0000000..19f2a52
--- /dev/null
+++ b/tests/sync-wire.test.ts
@@ -0,0 +1,174 @@
+import { describe, it, expect, vi } from 'vitest'
+import { AgentChatClient } from '../src/client.js'
+import type { SyncEnvelope } from '../src/client.js'
+
+// ─── /v1/messages/sync wire contract ────────────────────────────────────────
+//
+// AUTHORITY: docs/realtime-delivery-ack.md (server repo), section "Wire
+// contract for /v1/messages/sync (restated, authoritative)":
+//
+//   - `GET /v1/messages/sync?after=<delivery_id>&limit=<n>` → **bare JSON
+//     array**, oldest first, keyset-paginated on `(created_at, id)`.
+//     Non-destructive.
+//   - `POST /v1/messages/sync/ack` `{"last_delivery_id":"del_<32hex>"}` →
+//     `{"acked": <int>}`. Marks all `stored` envelopes at-or-before the
+//     cursor.
+//   - `delivery_id` is an **opaque string**. Clients MUST NOT compare it
+//     numerically; batch order is positional.
+//
+// SDK v1.0.2 typed this endpoint as `{envelopes: [{delivery_id: number,
+// message}]}` — a shape production never returned, which made the realtime
+// offline drain a silent zero-row no-op. These tests pin the SDK to the
+// real wire so that regression cannot come back.
+
+function scriptedFetch(
+  responses: Array<
+    Response | ((input: RequestInfo | URL, init?: RequestInit) => Response | Promise<Response>)
+  >,
+): typeof fetch {
+  let i = 0
+  return vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+    const entry = responses[i++]
+    if (!entry) throw new Error(`scriptedFetch: unexpected call #${i}`)
+    return typeof entry === 'function' ? await entry(input, init) : entry
+  }) as unknown as typeof fetch
+}
+
+const json = (status: number, body: unknown, headers: Record<string, string> = {}) =>
+  new Response(JSON.stringify(body), {
+    status,
+    headers: { 'Content-Type': 'application/json', ...headers },
+  })
+
+// Fixture captured from the production wire shape (field set per the
+// reference client schema: id, conversation_id, delivery_id string|null,
+// sender — handle string, with sender_handle only as a legacy fallback —
+// type, content, created_at, plus passthrough fields like seq / status /
+// metadata that the SDK must tolerate and preserve, not strip).
+const WIRE_FIXTURE = [
+  {
+    id: 'msg_9f2c1b7a5e304d18',
+    conversation_id: 'conv_51c9aa02f7d84b33',
+    delivery_id: 'del_0a1b2c3d4e5f60718293a4b5c6d7e8f9',
+    sender: 'aleph-null',
+    client_msg_id: '3f6f0a52-6f3d-4bfb-9a91-1f0f0f8f2a11',
+    seq: 41,
+    type: 'text',
+    content: { text: 'offline while you were away' },
+    metadata: {},
+    status: 'stored',
+    created_at: '2026-07-12T18:04:11.512Z',
+    delivered_at: null,
+    read_at: null,
+  },
+  {
+    id: 'msg_c4d0e6f2a8b1479c',
+    conversation_id: 'conv_51c9aa02f7d84b33',
+    delivery_id: 'del_ffeeddccbbaa99887766554433221100',
+    sender: 'tessera-rho',
+    client_msg_id: '7f1d9f04-2c5e-49ab-8d3a-52f7f9b0c644',
+    seq: 42,
+    type: 'structured',
+    content: { data: { kind: 'ping' } },
+    metadata: { trace_id: 'trc_123' },
+    status: 'stored',
+    created_at: '2026-07-12T18:05:02.007Z',
+    delivered_at: null,
+    read_at: null,
+    // Forward-compat: servers add fields without notice — passthrough.
+    priority: 'normal',
+  },
+  {
+    id: 'msg_5b6a79c8d0e1f234',
+    conversation_id: 'conv_e00f11223344a9b8',
+    // Nullable on the wire — cursor computation must skip null rows.
+    delivery_id: null,
+    sender: 'chatfather',
+    seq: 7,
+    type: 'system',
+    content: { text: 'group settings updated' },
+    created_at: '2026-07-12T18:06:40.901Z',
+  },
+]
+
+describe('client.sync() — bare-array wire', () => {
+  it('returns the bare array of rows exactly as the server sent them', async () => {
+    let calledUrl = ''
+    const fetch = scriptedFetch([
+      (input) => {
+        calledUrl = typeof input === 'string' ? input : input.toString()
+        return json(200, WIRE_FIXTURE)
+      },
+    ])
+    const client = new AgentChatClient({ apiKey: 'k', baseUrl: 'https://api.test', fetch })
+    const rows = await client.sync()
+
+    expect(calledUrl).toBe('https://api.test/v1/messages/sync')
+    // Bare array — NOT an {envelopes} wrapper.
+    expect(Array.isArray(rows)).toBe(true)
+    expect(rows).toHaveLength(3)
+
+    const first: SyncEnvelope = rows[0]
+    expect(first.id).toBe('msg_9f2c1b7a5e304d18')
+    expect(first.conversation_id).toBe('conv_51c9aa02f7d84b33')
+    // Opaque STRING cursor, del_<32 hex> in production.
+    expect(first.delivery_id).toBe('del_0a1b2c3d4e5f60718293a4b5c6d7e8f9')
+    expect(first.delivery_id).toMatch(/^del_[0-9a-f]{32}$/)
+    expect(first.sender).toBe('aleph-null')
+    expect(first.seq).toBe(41)
+
+    // Passthrough: unknown fields survive untouched.
+    expect(rows[1].priority).toBe('normal')
+    // Nullable cursor rows come through as-is.
+    expect(rows[2].delivery_id).toBeNull()
+  })
+
+  it('passes after + limit as query params (after is the opaque string cursor)', async () => {
+    let calledUrl = ''
+    const fetch = scriptedFetch([
+      (input) => {
+        calledUrl = typeof input === 'string' ? input : input.toString()
+        return json(200, [])
+      },
+    ])
+    const client = new AgentChatClient({ apiKey: 'k', baseUrl: 'https://api.test', fetch })
+    await client.sync({ after: 'del_0a1b2c3d4e5f60718293a4b5c6d7e8f9', limit: 500 })
+
+    const url = new URL(calledUrl)
+    expect(url.pathname).toBe('/v1/messages/sync')
+    expect(url.searchParams.get('after')).toBe('del_0a1b2c3d4e5f60718293a4b5c6d7e8f9')
+    expect(url.searchParams.get('limit')).toBe('500')
+  })
+})
+
+describe('client.syncAck() — {last_delivery_id} → {acked}', () => {
+  it('POSTs the string cursor and returns the acked count', async () => {
+    let capturedMethod = ''
+    let capturedUrl = ''
+    let capturedBody: Record<string, unknown> = {}
+    const fetch = scriptedFetch([
+      (input, init) => {
+        capturedMethod = init?.method ?? ''
+        capturedUrl = typeof input === 'string' ? input : input.toString()
+        capturedBody = JSON.parse(init!.body as string)
+        return json(200, { acked: 17 })
+      },
+    ])
+    const client = new AgentChatClient({ apiKey: 'k', baseUrl: 'https://api.test', fetch })
+    const res = await client.syncAck('del_ffeeddccbbaa99887766554433221100')
+
+    expect(capturedMethod).toBe('POST')
+    expect(capturedUrl).toBe('https://api.test/v1/messages/sync/ack')
+    expect(capturedBody).toEqual({
+      last_delivery_id: 'del_ffeeddccbbaa99887766554433221100',
+    })
+    expect(res.acked).toBe(17)
+  })
+
+  it('surfaces acked: 0 (repeated ack / owner-paused agent) without error', async () => {
+    const fetch = scriptedFetch([json(200, { acked: 0 })])
+    const client = new AgentChatClient({ apiKey: 'k', baseUrl: 'https://api.test', fetch })
+    const res = await client.syncAck('del_0a1b2c3d4e5f60718293a4b5c6d7e8f9')
+    expect(res.acked).toBe(0)
+  })
+})