sdk(session_store): bounded retry on mirror append + uuid idempotency docs (#857)

## Summary
Adds bounded retry (3 attempts, (0.2s, 0.8s) backoff) to
`TranscriptMirrorBatcher._do_flush` so transient `SessionStore.append()`
failures don't immediately drop the batch. Only after retries are
exhausted is the batch dropped and `MirrorErrorMessage` emitted. Also
documents that adapters should treat each entry's `uuid` as an
idempotency key.

## Changes
- `_internal/transcript_mirror_batcher.py`: retry loop, constants
`MIRROR_APPEND_MAX_ATTEMPTS=3` / `MIRROR_APPEND_BACKOFF_S=(0.2, 0.8)`
- `types.py`: `SessionStore.append` docstring
- `tests/test_transcript_mirror.py`: 2 new retry tests; patched
`asyncio.sleep` in 3 existing failure tests

Paired TS SDK change:
anthropics/claude-cli-internal#30980

<!-- CHANGELOG:START -->
- `SessionStore.append()` failures are now retried up to 3 times with
short backoff before the batch is dropped and `MirrorErrorMessage` is
emitted
<!-- CHANGELOG:END -->

Author: qing-ant

src/claude_agent_sdk/_internal/transcript_mirror_batcher.py

@@ -27,6 +27,11 @@
 MAX_PENDING_BYTES = 1 << 20  # 1 MiB
 SEND_TIMEOUT_SECONDS = 60.0
 
+# Bounded retry for transient adapter failures. Backoff list length must be
+# MAX_ATTEMPTS - 1 (one delay between each pair of attempts).
+MIRROR_APPEND_MAX_ATTEMPTS = 3
+MIRROR_APPEND_BACKOFF_S = (0.2, 0.8)
+
 
 @dataclass
 class _MirrorEntry:
@@ -44,9 +49,14 @@ class TranscriptMirrorBatcher:
     an eager flush fires in the background so memory stays flat during long
     turns where no ``result`` (and thus no explicit ``flush()``) arrives.
 
-    Adapter failures are reported via ``on_error`` and the failed batch is
-    dropped (at-most-once delivery). Failures never raise — the local-disk
-    transcript is already durable so the session must continue unaffected.
+    Adapter failures are retried (``MIRROR_APPEND_MAX_ATTEMPTS`` attempts
+    total) with short backoff; timeouts are not retried since the in-flight
+    call may still land. Only after the final attempt fails is the batch
+    dropped and reported via ``on_error``. Failures never raise — the
+    local-disk transcript is already durable so the session must continue
+    unaffected. Adapters should dedupe by ``entry["uuid"]`` when present
+    (some entry types lack a uuid) since a retried batch may partially
+    overlap a prior partial write.
     """
 
     store: SessionStore
@@ -163,12 +173,46 @@ async def _do_flush(
                     self.projects_dir,
                 )
                 continue
-            try:
-                await asyncio.wait_for(
-                    self.store.append(key, entries), timeout=self.send_timeout
-                )
-            except Exception as e:
+            last_err: Exception | None = None
+            succeeded = False
+            for attempt in range(MIRROR_APPEND_MAX_ATTEMPTS):
+                if attempt > 0:
+                    await asyncio.sleep(MIRROR_APPEND_BACKOFF_S[attempt - 1])
+                try:
+                    await asyncio.wait_for(
+                        self.store.append(key, entries), timeout=self.send_timeout
+                    )
+                    succeeded = True
+                    break
+                except asyncio.TimeoutError as e:
+                    # Don't retry on timeout: wait_for cancels the task but
+                    # cancellation is best-effort for adapters wrapping
+                    # non-cancellable I/O, so the in-flight call may still
+                    # land — a retry would launch a concurrent duplicate.
+                    # Also keeps worst-case lock hold at ~send_timeout rather
+                    # than ~3×send_timeout + backoff.
+                    last_err = e
+                    logger.debug(
+                        "[TranscriptMirrorBatcher] append timed out after "
+                        "%.1fs for %s — not retrying",
+                        self.send_timeout,
+                        file_path,
+                    )
+                    break
+                except Exception as e:  # noqa: BLE001 - adapter is user code
+                    last_err = e
+                    logger.debug(
+                        "[TranscriptMirrorBatcher] append attempt %d/%d failed "
+                        "for %s: %s",
+                        attempt + 1,
+                        MIRROR_APPEND_MAX_ATTEMPTS,
+                        file_path,
+                        e,
+                    )
+            if not succeeded:
                 logger.error(
-                    "[TranscriptMirrorBatcher] flush failed for %s: %s", file_path, e
+                    "[TranscriptMirrorBatcher] flush failed for %s: %s",
+                    file_path,
+                    last_err,
                 )
-                errors.append((key, str(e)))
+                errors.append((key, str(last_err)))

src/claude_agent_sdk/types.py

@@ -1274,8 +1274,13 @@ async def append(self, key: SessionKey, entries: list[SessionStoreEntry]) -> Non
         Within a single process, persist entries in append-call order; across
         concurrent processes, order is by storage commit time, not call time.
 
-        Exceptions are logged; the subprocess continues unaffected.
-        At-most-once delivery — failed batches are not retried.
+        Most entries carry a stable ``uuid`` that adapters should treat as an
+        idempotency key (upsert / ignore-duplicate). Entries without a
+        ``uuid`` (e.g. titles, tags, mode markers) should be appended without
+        dedup. Exceptions are logged and the subprocess continues unaffected
+        — failed batches are retried (3 attempts total) with short backoff
+        before being dropped and surfaced as a ``MirrorErrorMessage``;
+        timeouts are not retried since the in-flight call may still land.
         """
         ...
 

tests/test_transcript_mirror.py

@@ -151,6 +151,12 @@ async def _noop_error(_key: SessionKey | None, _err: str) -> None:
     pass
 
 
+# Patch target for the retry backoff — the batcher does ``import asyncio`` so
+# patching this attribute swaps the global ``asyncio.sleep`` for the duration
+# of the ``with`` block.
+_BATCHER_SLEEP = "claude_agent_sdk._internal.transcript_mirror_batcher.asyncio.sleep"
+
+
 class _RecordingStore(InMemorySessionStore):
     """InMemorySessionStore that records each append call separately."""
 
@@ -260,17 +266,22 @@ async def on_error(key: SessionKey | None, err: str) -> None:
             store=FailingStore(), projects_dir=PROJECTS_DIR, on_error=on_error
         )
         batcher.enqueue(_main_path(), [{"type": "x"}])
-        await batcher.flush()  # must not raise
+        with patch(_BATCHER_SLEEP, new=AsyncMock()):
+            await batcher.flush()  # must not raise
 
         assert len(errors) == 1
         assert errors[0][0] == {"project_key": "proj", "session_id": "sess"}
         assert "boom" in errors[0][1]
 
     @pytest.mark.asyncio
     async def test_append_timeout_calls_on_error(self) -> None:
+        """Timeout → on_error fires once, append is NOT retried (1 attempt)."""
+        calls: list[int] = []
+
         class HangingStore(InMemorySessionStore):
             async def append(self, key, entries):
-                await asyncio.sleep(10)
+                calls.append(1)
+                await asyncio.Event().wait()  # never resolves
 
         errors: list[str] = []
 
@@ -284,8 +295,115 @@ async def on_error(_key: SessionKey | None, err: str) -> None:
             send_timeout=0.05,
         )
         batcher.enqueue(_main_path(), [{"type": "x"}])
+        sleep_mock = AsyncMock()
+        with patch(_BATCHER_SLEEP, new=sleep_mock):
+            await batcher.flush()
+        assert len(calls) == 1  # not retried on timeout
+        assert len(errors) == 1
+        sleep_mock.assert_not_awaited()  # no backoff sleep
+
+    @pytest.mark.asyncio
+    async def test_append_timeout_no_concurrent_retry(self) -> None:
+        """A slow append that outlives send_timeout is attempted exactly once;
+        no retry overlaps the still-in-flight first call."""
+        in_flight = 0
+        max_in_flight = 0
+        calls = 0
+
+        class SlowStore(InMemorySessionStore):
+            async def append(self, key, entries):
+                nonlocal in_flight, max_in_flight, calls
+                calls += 1
+                in_flight += 1
+                max_in_flight = max(max_in_flight, in_flight)
+                try:
+                    # Outlives send_timeout=0.02 and shields against the
+                    # cancellation wait_for issues — models a non-cancellable
+                    # adapter (e.g. sync I/O in a thread).
+                    await asyncio.shield(asyncio.sleep(0.1))
+                finally:
+                    in_flight -= 1
+
+        errors: list[str] = []
+
+        async def on_error(_key: SessionKey | None, err: str) -> None:
+            errors.append(err)
+
+        batcher = TranscriptMirrorBatcher(
+            store=SlowStore(),
+            projects_dir=PROJECTS_DIR,
+            on_error=on_error,
+            send_timeout=0.02,
+        )
+        batcher.enqueue(_main_path(), [{"type": "x"}])
         await batcher.flush()
+        # Let any (incorrectly) shielded/retried task observe overlap.
+        await asyncio.sleep(0.15)
+
+        assert calls == 1
+        assert max_in_flight == 1
+        assert len(errors) == 1
+
+    @pytest.mark.asyncio
+    async def test_append_retries_then_succeeds_no_error_reported(self) -> None:
+        """Transient outage: append raises twice then succeeds on the 3rd
+        attempt — batch is delivered, no mirror error reported."""
+        attempts: list[int] = []
+
+        class FlakyStore(InMemorySessionStore):
+            async def append(self, key, entries):
+                attempts.append(1)
+                if len(attempts) < 3:
+                    raise RuntimeError("transient")
+                await super().append(key, entries)
+
+        errors: list[tuple[SessionKey | None, str]] = []
+
+        async def on_error(key: SessionKey | None, err: str) -> None:
+            errors.append((key, err))
+
+        store = FlakyStore()
+        batcher = TranscriptMirrorBatcher(
+            store=store, projects_dir=PROJECTS_DIR, on_error=on_error
+        )
+        batcher.enqueue(_main_path(), [{"type": "x"}])
+        sleep_mock = AsyncMock()
+        with patch(_BATCHER_SLEEP, new=sleep_mock):
+            await batcher.flush()
+
+        assert len(attempts) == 3
+        assert errors == []
+        assert await store.load({"project_key": "proj", "session_id": "sess"}) == [
+            {"type": "x"}
+        ]
+        # Backoff schedule honoured between attempts.
+        assert [c.args[0] for c in sleep_mock.await_args_list] == [0.2, 0.8]
+
+    @pytest.mark.asyncio
+    async def test_append_retries_exhausted_reports_error_once(self) -> None:
+        """append raises on all 3 attempts → exactly one mirror error."""
+        attempts: list[int] = []
+
+        class AlwaysFailingStore(InMemorySessionStore):
+            async def append(self, key, entries):
+                attempts.append(1)
+                raise RuntimeError("boom")
+
+        errors: list[tuple[SessionKey | None, str]] = []
+
+        async def on_error(key: SessionKey | None, err: str) -> None:
+            errors.append((key, err))
+
+        batcher = TranscriptMirrorBatcher(
+            store=AlwaysFailingStore(), projects_dir=PROJECTS_DIR, on_error=on_error
+        )
+        batcher.enqueue(_main_path(), [{"type": "x"}])
+        with patch(_BATCHER_SLEEP, new=AsyncMock()):
+            await batcher.flush()
+
+        assert len(attempts) == 3
         assert len(errors) == 1
+        assert "boom" in errors[0][1]
 
     @pytest.mark.asyncio
     async def test_close_flushes_pending(self) -> None:
@@ -656,6 +774,7 @@ async def append(self, key, entries):
                     "claude_agent_sdk._internal.session_resume._get_projects_dir",
                     return_value=PROJECTS_DIR,
                 ),
+                patch(_BATCHER_SLEEP, new=AsyncMock()),
             ):
                 mock_cls.return_value = mock_transport
                 messages = [