fix: NFC-normalize project_key; retry rmtree on transient errors; skip materialize w/ custom transport

- project_key_for_directory() and _project_key_from_dir() now route through
  _canonicalize_path() (realpath + NFC) so the SDK's project_key matches the
  CLI's on filesystems that decompose Unicode (macOS HFS+). Without this,
  store.load() silently misses on paths like 'café' stored as 'cafe\u0301'.

- Add _rmtree_with_retry() in session_resume.py: retries shutil.rmtree on
  EBUSY/EPERM/EACCES/ENOTEMPTY/EMFILE/ENFILE up to 4 times with 100ms backoff,
  then falls back to ignore_errors=True. Replaces both rmtree call sites
  (failure-path cleanup and the cleanup() callback). On Windows, AV/indexer
  briefly holding .credentials.json no longer leaks the access token in temp.

- Gate materialize_resume_session() on no custom transport in both
  ClaudeSDKClient.connect() and InternalClient.process_query(). A
  pre-constructed transport never sees the materialized options, so loading
  the store and writing .credentials.json to a temp dir was wasted work that
  left the access token on disk for the session lifetime.

Tests: +5 (NFC normalization, rmtree retry on cleanup() and failure-path,
custom-transport skip on connect() and query()).

Author: qing-ant

src/claude_agent_sdk/_internal/client.py

@@ -60,7 +60,12 @@ async def process_query(
 
         # resume/continue + session_store: load the session from the store
         # into a temp CLAUDE_CONFIG_DIR for the subprocess to resume from.
-        materialized = await materialize_resume_session(options)
+        # Skipped when a custom transport was supplied — the materialized
+        # options never reach a pre-constructed transport, so loading the
+        # store and writing .credentials.json to a temp dir would be wasted.
+        materialized = (
+            await materialize_resume_session(options) if transport is None else None
+        )
         try:
             async for msg in self._process_query_inner(
                 prompt, options, transport, materialized

src/claude_agent_sdk/_internal/session_resume.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import asyncio
+import errno
 import getpass
 import json
 import logging
@@ -119,11 +120,11 @@ async def materialize_resume_session(
         # so asyncio.CancelledError (BaseException since 3.8) also triggers
         # cleanup — callers can't compensate because the assignment raises
         # before completing.
-        shutil.rmtree(tmp_base, ignore_errors=True)
+        await _rmtree_with_retry(tmp_base)
         raise
 
     async def cleanup() -> None:
-        shutil.rmtree(tmp_base, ignore_errors=True)
+        await _rmtree_with_retry(tmp_base)
 
     return MaterializedResume(
         config_dir=tmp_base,
@@ -136,6 +137,52 @@ async def cleanup() -> None:
 # Helpers
 # ---------------------------------------------------------------------------
 
+# OSError errnos that indicate a transiently-held handle (Windows AV/indexer
+# scanning a freshly-written file) rather than a permanent failure.
+_RETRYABLE_RMTREE_ERRNOS = frozenset(
+    {
+        errno.EBUSY,
+        errno.EMFILE,
+        errno.ENFILE,
+        errno.ENOTEMPTY,
+        errno.EPERM,
+        errno.EACCES,
+    }
+)
+
+
+async def _rmtree_with_retry(
+    path: Path, *, retries: int = 4, delay: float = 0.1
+) -> None:
+    """Best-effort ``shutil.rmtree`` with retries on transient lock errors.
+
+    On Windows, AV/indexer can briefly hold a handle on freshly-written
+    files (notably ``.credentials.json``), causing rmtree to fail with
+    EBUSY/EPERM. Retry a few times with a short backoff; after exhausting
+    retries, fall back to ``ignore_errors=True`` (matches the previous
+    behavior, but gives the handle a chance to release first so the access
+    token doesn't leak in temp). Never raises.
+    """
+    if not path.exists():
+        return
+    for _ in range(retries):
+        try:
+            shutil.rmtree(path)
+            return
+        except OSError as e:
+            if e.errno not in _RETRYABLE_RMTREE_ERRNOS and not isinstance(
+                e, PermissionError
+            ):
+                break
+        try:
+            await asyncio.sleep(delay)
+        except asyncio.CancelledError:
+            # Best-effort final sweep before propagating cancellation so a
+            # cancelled connect() doesn't leak the temp dir.
+            shutil.rmtree(path, ignore_errors=True)
+            raise
+    shutil.rmtree(path, ignore_errors=True)
+
 
 async def _load_candidate(
     store: SessionStore, project_key: str, session_id: str, timeout_s: float

src/claude_agent_sdk/_internal/session_store.py

@@ -13,7 +13,7 @@
     SessionStoreEntry,
     SessionStoreListEntry,
 )
-from .sessions import _sanitize_path
+from .sessions import _canonicalize_path, _sanitize_path
 
 
 def _key_to_string(key: SessionKey) -> str:
@@ -150,9 +150,11 @@ def file_path_to_session_key(file_path: str, projects_dir: str) -> SessionKey |
 def project_key_for_directory(directory: str | Path | None = None) -> str:
     """Derive the :class:`SessionStore` ``project_key`` for a directory.
 
-    Defaults to the current working directory. Uses the same djb2-hashed
-    sanitization the CLI uses for project directory names, so keys match
-    between local-disk transcripts and store-mirrored transcripts.
+    Defaults to the current working directory. Uses the same realpath + NFC
+    normalization + djb2-hashed sanitization the CLI uses for project
+    directory names, so keys match between local-disk transcripts and
+    store-mirrored transcripts even on filesystems that decompose Unicode
+    (macOS HFS+).
     """
-    abs_path = Path(directory if directory is not None else ".").resolve()
-    return _sanitize_path(str(abs_path))
+    abs_path = _canonicalize_path(str(directory) if directory is not None else ".")
+    return _sanitize_path(abs_path)

src/claude_agent_sdk/_internal/sessions.py

@@ -1410,8 +1410,8 @@ def _project_key_from_dir(directory: str | None) -> str:
     :func:`project_key_for_directory` but is defined locally to avoid a
     circular import with ``session_store.py``.
     """
-    abs_path = Path(directory if directory is not None else ".").resolve()
-    return _sanitize_path(str(abs_path))
+    abs_path = _canonicalize_path(directory if directory is not None else ".")
+    return _sanitize_path(abs_path)
 
 
 def _entries_to_jsonl(entries: list[Any]) -> str:

src/claude_agent_sdk/client.py

@@ -121,8 +121,15 @@ async def _empty_stream() -> AsyncIterator[dict[str, Any]]:
         # into a temp CLAUDE_CONFIG_DIR for the subprocess to resume from.
         # When materialized, override resume/continue/env on a copy of options
         # so the subprocess points at the temp dir; when None, fall through
-        # to normal handling (fresh session or local-disk resume).
-        self._materialized = await materialize_resume_session(self.options)
+        # to normal handling (fresh session or local-disk resume). Skipped
+        # when a custom transport was supplied — the materialized options
+        # never reach a pre-constructed transport, so loading the store and
+        # writing .credentials.json to a temp dir would be wasted work.
+        self._materialized = (
+            await materialize_resume_session(self.options)
+            if self._custom_transport is None
+            else None
+        )
         try:
             await self._connect_inner(prompt, actual_prompt)
         except BaseException:

tests/test_session_resume.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+import errno
 import json
+import shutil
 import tempfile
 import uuid
 from pathlib import Path
@@ -755,6 +757,79 @@ def capture_transport(*, prompt, options):
         # Cleanup removed the temp dir.
         assert not Path(config_dir).exists()
 
+    @pytest.mark.asyncio
+    async def test_custom_transport_skips_materialization(
+        self,
+        cwd: Path,
+        project_key: str,
+        isolated_home: Path,
+        track_resume_dirs: list[Path],
+    ) -> None:
+        """A pre-constructed custom transport never sees the materialized
+        options, so loading the store and writing .credentials.json to a
+        temp dir would be wasted (and leave the access token on disk for
+        the session lifetime). connect() must skip materialization."""
+
+        class SpyStore(InMemorySessionStore):
+            load_calls = 0
+
+            async def load(self, key):  # type: ignore[override]
+                SpyStore.load_calls += 1
+                return await super().load(key)
+
+        store = SpyStore()
+        await store.append(
+            {"project_key": project_key, "session_id": SESSION_ID},
+            [{"type": "user", "uuid": "u1"}],
+        )
+
+        opts = ClaudeAgentOptions(cwd=cwd, session_store=store, resume=SESSION_ID)
+        client = ClaudeSDKClient(options=opts, transport=_make_mock_transport())
+
+        with patch(
+            "claude_agent_sdk._internal.query.Query.initialize",
+            new_callable=AsyncMock,
+        ):
+            await client.connect()
+            assert SpyStore.load_calls == 0
+            assert not track_resume_dirs  # mkdtemp never ran
+            assert client._materialized is None
+            await client.disconnect()
+
+    @pytest.mark.asyncio
+    async def test_query_custom_transport_skips_materialization(
+        self,
+        cwd: Path,
+        project_key: str,
+        isolated_home: Path,
+        track_resume_dirs: list[Path],
+    ) -> None:
+        """Same gate for the one-shot ``query()`` path."""
+
+        class SpyStore(InMemorySessionStore):
+            load_calls = 0
+
+            async def load(self, key):  # type: ignore[override]
+                SpyStore.load_calls += 1
+                return await super().load(key)
+
+        store = SpyStore()
+        await store.append(
+            {"project_key": project_key, "session_id": SESSION_ID},
+            [{"type": "user", "uuid": "u1"}],
+        )
+
+        opts = ClaudeAgentOptions(cwd=cwd, session_store=store, resume=SESSION_ID)
+        custom = _make_mock_transport()
+        custom.connect = AsyncMock(side_effect=OSError("spawn failed"))
+        with pytest.raises(OSError, match="spawn failed"):
+            async for _ in query(prompt="hi", options=opts, transport=custom):
+                pass  # pragma: no cover
+
+        # Gate runs before transport.connect(); materialization never happened.
+        assert SpyStore.load_calls == 0
+        assert not track_resume_dirs
+
     @pytest.mark.asyncio
     async def test_connect_no_materialization_passthrough(
         self, cwd: Path, isolated_home: Path
@@ -815,6 +890,84 @@ class TestSpawnFailureCleanup:
     removed even when transport.connect() raises before any try/finally that
     normally guards cleanup."""
 
+    @pytest.mark.asyncio
+    async def test_cleanup_retries_on_transient_os_error(
+        self,
+        cwd: Path,
+        project_key: str,
+        isolated_home: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Windows AV/indexer can briefly hold ``.credentials.json`` open;
+        ``cleanup()`` must retry rmtree on EPERM/EBUSY so the access token
+        doesn't leak in temp."""
+        store = InMemorySessionStore()
+        await store.append(
+            {"project_key": project_key, "session_id": SESSION_ID},
+            [{"type": "user", "uuid": "u1"}],
+        )
+        opts = ClaudeAgentOptions(cwd=cwd, session_store=store, resume=SESSION_ID)
+        m = await materialize_resume_session(opts)
+        assert m is not None
+        config_dir = m.config_dir
+
+        calls: list[Any] = []
+        real_rmtree = shutil.rmtree
+
+        def fake_rmtree(p: Any, **kw: Any) -> None:
+            calls.append((p, kw))
+            if len(calls) <= 2 and not kw.get("ignore_errors"):
+                raise PermissionError(errno.EPERM, "held by indexer")
+            if Path(p).exists():
+                real_rmtree(p, **kw)
+
+        monkeypatch.setattr(shutil, "rmtree", fake_rmtree)
+        await m.cleanup()
+
+        assert not config_dir.exists()
+        assert len(calls) >= 3  # 2 failures + 1 success
+
+    @pytest.mark.asyncio
+    async def test_failure_path_retries_rmtree(
+        self,
+        cwd: Path,
+        project_key: str,
+        isolated_home: Path,
+        monkeypatch: pytest.MonkeyPatch,
+        track_resume_dirs: list[Path],
+    ) -> None:
+        """The except-BaseException cleanup path also retries on EPERM."""
+
+        class FailLateStore(InMemorySessionStore):
+            async def list_subkeys(self, key):  # type: ignore[override]
+                raise OSError("boom")
+
+        store = FailLateStore()
+        await store.append(
+            {"project_key": project_key, "session_id": SESSION_ID},
+            [{"type": "user", "uuid": "u1"}],
+        )
+
+        calls: list[Any] = []
+        real_rmtree = shutil.rmtree
+
+        def fake_rmtree(p: Any, **kw: Any) -> None:
+            calls.append((p, kw))
+            if len(calls) <= 2 and not kw.get("ignore_errors"):
+                raise PermissionError(errno.EPERM, "held by indexer")
+            if Path(p).exists():
+                real_rmtree(p, **kw)
+
+        monkeypatch.setattr(shutil, "rmtree", fake_rmtree)
+
+        opts = ClaudeAgentOptions(cwd=cwd, session_store=store, resume=SESSION_ID)
+        with pytest.raises(RuntimeError, match="boom"):
+            await materialize_resume_session(opts)
+
+        assert track_resume_dirs
+        assert not track_resume_dirs[0].exists()
+        assert len(calls) >= 3
+
     @pytest.mark.asyncio
     async def test_client_connect_failure_removes_temp_dir(
         self,

tests/test_session_store_conformance.py

@@ -228,6 +228,20 @@ def test_relative_dir_resolved_to_absolute_before_hashing(self) -> None:
         assert key == _sanitize_path(str(Path().resolve()))
         assert key != _sanitize_path(".")
 
+    def test_nfc_normalizes_decomposed_unicode(self, tmp_path) -> None:
+        """Parity with TS: the CLI canonicalizes via realpath + NFC. On macOS
+        HFS+ a path like ``café`` may be stored decomposed (``cafe\\u0301``);
+        without NFC normalization the SDK's project_key would mismatch the
+        CLI's and store.load() would silently miss."""
+        import unicodedata
+
+        nfc = tmp_path / unicodedata.normalize("NFC", "café")
+        nfd = tmp_path / unicodedata.normalize("NFD", "café")
+        nfc.mkdir(exist_ok=True)
+        assert project_key_for_directory(str(nfc)) == project_key_for_directory(
+            str(nfd)
+        )
+
     def test_long_path_uses_portable_djb2_suffix(self) -> None:
         """Parity with TS: paths > MAX_SANITIZED_LENGTH get a djb2 hash
         suffix (runtime-portable so parent and subprocess derive the