fix(sessions): surface subagent import I/O failures

src/claude_agent_sdk/_internal/session_import.py

@@ -17,6 +17,7 @@
 
 from ..types import SessionKey, SessionStore, SessionStoreEntry
 from .sessions import (
+    _agent_metadata_sidecar_path,
     _read_agent_metadata_sidecar,
     _resolve_session_file_path,
     _validate_uuid,
@@ -65,6 +66,10 @@ async def import_session_to_store(
     Raises:
         ValueError: If ``session_id`` is not a valid UUID.
         FileNotFoundError: If the session JSONL cannot be found on disk.
+        OSError: If a transcript, sidecar, or subagent directory exists but
+            cannot be read. Import is not transactional: entries already
+            appended remain in the store, and can be safely deduplicated by
+            ``entry["uuid"]`` when the import is retried.
     """
     if not _validate_uuid(session_id):
         raise ValueError(f"Invalid session_id: {session_id}")
@@ -110,7 +115,12 @@ async def import_session_to_store(
         # recreate it and resumed subagents keep their agentType/worktreePath.
         # A missing, corrupt, or non-object sidecar is treated as absent (the
         # transcript is still imported); other read errors propagate.
-        meta = _read_agent_metadata_sidecar(file_path)
+        sidecar_path = _agent_metadata_sidecar_path(file_path)
+        meta = (
+            None
+            if sidecar_path.is_symlink()
+            else _read_agent_metadata_sidecar(file_path)
+        )
         if meta is not None:
             # Synthetic discriminator last so a stray "type" key in the
             # CLI-owned sidecar can never shadow it.
@@ -148,14 +158,20 @@ async def _append_jsonl_file_in_batches(
 def _collect_jsonl_files(base_dir: Path) -> Iterator[Path]:
     """Recursively yield all ``*.jsonl`` file paths under ``base_dir``.
 
-    Yields nothing if ``base_dir`` does not exist. Sorted per directory so
-    import order is deterministic across platforms.
+    Yields nothing if ``base_dir`` does not exist. Symlinks below ``base_dir``
+    are skipped so they cannot re-enter the tree or import transcripts from
+    outside it; ``base_dir`` itself may still be a symlink. Other ``OSError``s
+    (for example, permission denied) propagate so callers cannot mistake a
+    partial import for a complete one. Sorted per directory so import order is
+    deterministic across platforms.
     """
     try:
         dirents = sorted(base_dir.iterdir(), key=lambda p: p.name)
-    except OSError:
+    except FileNotFoundError:
         return
     for entry in dirents:
+        if entry.is_symlink():
+            continue
         if entry.is_dir():
             yield from _collect_jsonl_files(entry)
         elif entry.is_file() and entry.name.endswith(".jsonl"):

tests/test_session_import.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import errno
 import json
+from collections.abc import Iterator
 from pathlib import Path
 from unittest.mock import AsyncMock
 
@@ -51,6 +53,16 @@ def _write_jsonl(path: Path, entries: list[SessionStoreEntry]) -> None:
     path.write_text("\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8")
 
 
+def _symlink_or_skip(
+    link: Path, target: Path, *, target_is_directory: bool = True
+) -> None:
+    """Create a symlink, or skip where the platform disallows it."""
+    try:
+        link.symlink_to(target, target_is_directory=target_is_directory)
+    except (OSError, NotImplementedError) as exc:
+        pytest.skip(f"symlinks are unavailable: {exc}")
+
+
 # ---------------------------------------------------------------------------
 # Main transcript import
 # ---------------------------------------------------------------------------
@@ -229,6 +241,35 @@ async def test_meta_json_type_key_cannot_shadow_agent_metadata_marker(
             "toolUseId": "toolu_1",
         }
 
+    @pytest.mark.anyio
+    async def test_symlinked_meta_json_sidecar_is_not_followed(
+        self, claude_dir: Path, cwd: Path, project_key: str, tmp_path: Path
+    ) -> None:
+        """A derived sidecar link must not import metadata from outside the tree."""
+        _write_jsonl(claude_dir / f"{SESSION_ID}.jsonl", [_entry(0)])
+        sub_dir = claude_dir / SESSION_ID / "subagents"
+        _write_jsonl(sub_dir / "agent-abc.jsonl", [_entry(10)])
+        external_sidecar = tmp_path / "agent-abc.meta.json"
+        external_sidecar.write_text(
+            json.dumps({"agentType": "external", "worktreePath": "/outside"}),
+            encoding="utf-8",
+        )
+        _symlink_or_skip(
+            sub_dir / "agent-abc.meta.json",
+            external_sidecar,
+            target_is_directory=False,
+        )
+
+        store = InMemorySessionStore()
+        await import_session_to_store(SESSION_ID, store, directory=str(cwd))
+
+        sub_key: SessionKey = {
+            "project_key": project_key,
+            "session_id": SESSION_ID,
+            "subpath": "subagents/agent-abc",
+        }
+        assert store.get_entries(sub_key) == [_entry(10)]
+
     @pytest.mark.anyio
     @pytest.mark.parametrize("sidecar", ["not json {", "[1, 2]", "42"])
     async def test_unusable_meta_json_sidecar_is_treated_as_absent(
@@ -286,6 +327,113 @@ async def test_no_subagents_dir_is_noop(
         key: SessionKey = {"project_key": project_key, "session_id": SESSION_ID}
         assert store.get_entries(key) == [_entry(0)]
 
+    @pytest.mark.anyio
+    @pytest.mark.parametrize("nested", [False, True])
+    async def test_unreadable_subagents_dir_raises(
+        self,
+        claude_dir: Path,
+        cwd: Path,
+        project_key: str,
+        monkeypatch: pytest.MonkeyPatch,
+        nested: bool,
+    ) -> None:
+        """A failed traversal at any depth must not look like a complete import."""
+        _write_jsonl(claude_dir / f"{SESSION_ID}.jsonl", [_entry(0)])
+        subagents_dir = claude_dir / SESSION_ID / "subagents"
+        _write_jsonl(subagents_dir / "agent-abc.jsonl", [_entry(10)])
+        blocked_dir = subagents_dir / "workflows" if nested else subagents_dir
+        if nested:
+            _write_jsonl(blocked_dir / "run-1" / "agent-def.jsonl", [_entry(20)])
+
+        original_iterdir = Path.iterdir
+
+        def fail_for_subagents(path: Path) -> Iterator[Path]:
+            if path == blocked_dir:
+                raise PermissionError(errno.EACCES, "Permission denied", str(path))
+            return original_iterdir(path)
+
+        monkeypatch.setattr(Path, "iterdir", fail_for_subagents)
+
+        store = InMemorySessionStore()
+        with pytest.raises(PermissionError, match="Permission denied"):
+            await import_session_to_store(SESSION_ID, store, directory=str(cwd))
+
+        main_key: SessionKey = {
+            "project_key": project_key,
+            "session_id": SESSION_ID,
+        }
+        assert store.get_entries(main_key) == [_entry(0)]
+        assert await store.list_subkeys(main_key) == (
+            ["subagents/agent-abc"] if nested else []
+        )
+
+    @pytest.mark.anyio
+    @pytest.mark.parametrize("shape", ["cycle", "sibling", "external", "file"])
+    async def test_symlinks_below_subagents_are_not_followed(
+        self,
+        claude_dir: Path,
+        cwd: Path,
+        project_key: str,
+        tmp_path: Path,
+        shape: str,
+    ) -> None:
+        """Links must not re-enter, duplicate, or escape the transcript tree."""
+        _write_jsonl(claude_dir / f"{SESSION_ID}.jsonl", [_entry(0)])
+        subagents_dir = claude_dir / SESSION_ID / "subagents"
+        real_dir = subagents_dir / "real"
+        _write_jsonl(real_dir / "agent-abc.jsonl", [_entry(10)])
+
+        if shape == "cycle":
+            _symlink_or_skip(subagents_dir / "loop", subagents_dir)
+        elif shape == "sibling":
+            _symlink_or_skip(subagents_dir / "alias", real_dir)
+        elif shape == "external":
+            outside = tmp_path / "outside"
+            _write_jsonl(outside / "agent-foreign.jsonl", [_entry(20)])
+            _symlink_or_skip(subagents_dir / "external", outside)
+        else:
+            foreign = tmp_path / "agent-foreign.jsonl"
+            _write_jsonl(foreign, [_entry(20)])
+            _symlink_or_skip(
+                real_dir / "agent-foreign.jsonl",
+                foreign,
+                target_is_directory=False,
+            )
+
+        store = InMemorySessionStore()
+        await import_session_to_store(SESSION_ID, store, directory=str(cwd))
+
+        main_key: SessionKey = {
+            "project_key": project_key,
+            "session_id": SESSION_ID,
+        }
+        assert await store.list_subkeys(main_key) == ["subagents/real/agent-abc"]
+
+    @pytest.mark.anyio
+    async def test_symlinked_subagents_root_is_followed(
+        self,
+        claude_dir: Path,
+        cwd: Path,
+        project_key: str,
+        tmp_path: Path,
+    ) -> None:
+        """A relocated subagents root remains a supported traversal root."""
+        _write_jsonl(claude_dir / f"{SESSION_ID}.jsonl", [_entry(0)])
+        relocated = tmp_path / "relocated-subagents"
+        _write_jsonl(relocated / "agent-abc.jsonl", [_entry(10)])
+        session_dir = claude_dir / SESSION_ID
+        session_dir.mkdir(parents=True)
+        _symlink_or_skip(session_dir / "subagents", relocated)
+
+        store = InMemorySessionStore()
+        await import_session_to_store(SESSION_ID, store, directory=str(cwd))
+
+        main_key: SessionKey = {
+            "project_key": project_key,
+            "session_id": SESSION_ID,
+        }
+        assert await store.list_subkeys(main_key) == ["subagents/agent-abc"]
+
 
 # ---------------------------------------------------------------------------
 # Validation / errors