fix: gap-fill list_session_summaries fast-path for sessions missing a sidecar

Author: qing-ant

src/claude_agent_sdk/_internal/sessions.py

@@ -1514,6 +1514,53 @@ async def _load_store_entries_as_jsonl(
     return _entries_to_jsonl(entries)
 
 
+async def _derive_infos_via_load(
+    session_store: SessionStore,
+    listing: list[Any],
+    directory: str | None,
+    project_path: str,
+) -> list[SDKSessionInfo]:
+    """Derive ``SDKSessionInfo`` for each ``listing`` entry via per-session
+    ``store.load()`` + lite-parse.
+
+    Loads run concurrently with a fixed bound so large listings don't exhaust
+    adapter connection pools or hit backend rate limits; adapter errors degrade
+    that row to an empty summary instead of failing the whole list. Sidechain
+    and no-summary sessions are dropped.
+    """
+    sem = asyncio.Semaphore(_STORE_LIST_LOAD_CONCURRENCY)
+
+    async def _bounded_load(sid: str) -> str | None:
+        async with sem:
+            return await _load_store_entries_as_jsonl(session_store, sid, directory)
+
+    settled = await asyncio.gather(
+        *(_bounded_load(e["session_id"]) for e in listing),
+        return_exceptions=True,
+    )
+    results: list[SDKSessionInfo] = []
+    for entry, outcome in zip(listing, settled, strict=True):
+        sid = entry["session_id"]
+        mtime = entry["mtime"]
+        if isinstance(outcome, BaseException):
+            results.append(
+                SDKSessionInfo(session_id=sid, summary="", last_modified=mtime)
+            )
+            continue
+        if outcome is None:
+            continue
+        parsed = _parse_session_info_from_lite(
+            sid, _jsonl_to_lite(outcome, mtime), project_path
+        )
+        if parsed is None:
+            # Sidechain or no extractable summary — drop, matching the
+            # filesystem path.
+            continue
+        parsed.last_modified = mtime
+        results.append(parsed)
+    return results
+
+
 async def list_sessions_from_store(
     session_store: SessionStore,
     directory: str | None = None,
@@ -1556,6 +1603,7 @@ async def list_sessions_from_store(
     """
     project_path = _canonicalize_path(str(directory) if directory is not None else ".")
     project_key = _sanitize_path(project_path)
+    has_list_sessions = _store_implements(session_store, "list_sessions")
 
     # Fast path: if the store maintains incremental summaries, fetch them in
     # one call instead of N per-session load()s.
@@ -1572,54 +1620,36 @@ async def list_sessions_from_store(
                 for s in summaries
                 if (info := summary_entry_to_sdk_info(s, project_path)) is not None
             ]
+            # Gap-fill: a store may have entries for sessions appended before
+            # it adopted list_session_summaries (no sidecar yet). Enumerate
+            # via list_sessions() and derive the missing ones via the
+            # per-session load() path so they aren't silently dropped.
+            if has_list_sessions:
+                summary_ids = {s["session_id"] for s in summaries}
+                listing = list(await session_store.list_sessions(project_key))
+                missing = [e for e in listing if e["session_id"] not in summary_ids]
+                if missing:
+                    infos.extend(
+                        await _derive_infos_via_load(
+                            session_store, missing, directory, project_path
+                        )
+                    )
             return _apply_sort_limit_offset(infos, limit, offset)
 
-    if not _store_implements(session_store, "list_sessions"):
+    if not has_list_sessions:
         raise ValueError(
             "session_store does not implement list_sessions() -- cannot list "
             "sessions. Provide a store with a list_sessions() method."
         )
-    raw = await session_store.list_sessions(project_key)
     # Copy — store.list_sessions() may return a reference to internal state.
-    listing = list(raw)
-
+    listing = list(await session_store.list_sessions(project_key))
     # Derive a real summary per session by loading its entries and reusing
-    # the filesystem path's lite-parse. Loads run concurrently with a fixed
-    # bound so large listings don't exhaust adapter connection pools or hit
-    # backend rate limits; adapter errors degrade that row instead of failing
-    # the whole list. Filtering (sidechain/empty drop) happens before
-    # pagination so ``limit``/``offset`` index the same filtered set as the
-    # disk path.
-    sem = asyncio.Semaphore(_STORE_LIST_LOAD_CONCURRENCY)
-
-    async def _bounded_load(sid: str) -> str | None:
-        async with sem:
-            return await _load_store_entries_as_jsonl(session_store, sid, directory)
-
-    settled = await asyncio.gather(
-        *(_bounded_load(e["session_id"]) for e in listing),
-        return_exceptions=True,
+    # the filesystem path's lite-parse. Filtering (sidechain/empty drop)
+    # happens before pagination so ``limit``/``offset`` index the same
+    # filtered set as the disk path.
+    results = await _derive_infos_via_load(
+        session_store, listing, directory, project_path
     )
-    results: list[SDKSessionInfo] = []
-    for entry, outcome in zip(listing, settled, strict=True):
-        sid = entry["session_id"]
-        mtime = entry["mtime"]
-        if isinstance(outcome, BaseException):
-            results.append(
-                SDKSessionInfo(session_id=sid, summary="", last_modified=mtime)
-            )
-            continue
-        if outcome is None:
-            continue
-        parsed = _parse_session_info_from_lite(
-            sid, _jsonl_to_lite(outcome, mtime), project_path
-        )
-        if parsed is None:
-            # Sidechain or no extractable summary — drop, matching the
-            # filesystem path.
-            continue
-        parsed.last_modified = mtime
-        results.append(parsed)
     return _apply_sort_limit_offset(results, limit, offset)
 
 

tests/test_session_summary.py

@@ -465,6 +465,82 @@ async def list_session_summaries(self, project_key: str):  # noqa: ANN201
         assert len(sessions) == 1
         assert sessions[0].summary == "hi"
 
+    async def test_mixed_sessions_gap_filled(self) -> None:
+        """A store with summaries for only SOME sessions (e.g. adopted the
+        method mid-stream) must have the rest gap-filled via per-session
+        load() so old sessions aren't silently dropped."""
+        sid_with = str(uuid_mod.uuid4())
+        sid_without = str(uuid_mod.uuid4())
+
+        class PartialStore(InMemorySessionStore):
+            load_calls: list[str] = []
+
+            async def list_session_summaries(self, project_key: str):  # noqa: ANN201
+                full = await super().list_session_summaries(project_key)
+                return [s for s in full if s["session_id"] == sid_with]
+
+            async def load(self, key):  # noqa: ANN001, ANN201
+                self.load_calls.append(key["session_id"])
+                return await super().load(key)
+
+        store = PartialStore()
+        await store.append(
+            {"project_key": PROJECT_KEY, "session_id": sid_with},
+            [_user("has sidecar", ts="2024-01-02T00:00:00Z")],
+        )
+        await store.append(
+            {"project_key": PROJECT_KEY, "session_id": sid_without},
+            [_user("no sidecar", ts="2024-01-01T00:00:00Z")],
+        )
+
+        sessions = await list_sessions_from_store(store, directory=DIR)
+        by_id = {s.session_id: s for s in sessions}
+        assert set(by_id) == {sid_with, sid_without}
+        assert by_id[sid_with].summary == "has sidecar"
+        assert by_id[sid_without].summary == "no sidecar"
+        # Only the missing session should have been load()ed.
+        assert store.load_calls == [sid_without]
+
+    async def test_gap_fill_bounded_concurrency(self) -> None:
+        """Gap-fill reuses the bounded per-session load helper, so
+        ``_STORE_LIST_LOAD_CONCURRENCY`` applies to the missing-session set."""
+        import asyncio
+
+        from claude_agent_sdk._internal import sessions as _sessions
+
+        in_flight = 0
+        peak = 0
+        gate = asyncio.Event()
+
+        class PartialSlowStore(InMemorySessionStore):
+            async def list_session_summaries(self, project_key: str):  # noqa: ANN201
+                return []  # everything is "missing"
+
+            async def load(self, key):  # noqa: ANN001, ANN201
+                nonlocal in_flight, peak
+                in_flight += 1
+                peak = max(peak, in_flight)
+                await gate.wait()
+                in_flight -= 1
+                return await super().load(key)
+
+        store = PartialSlowStore()
+        n = _sessions._STORE_LIST_LOAD_CONCURRENCY * 2
+        for i in range(n):
+            await InMemorySessionStore.append(
+                store,
+                {"project_key": PROJECT_KEY, "session_id": str(uuid_mod.uuid4())},
+                [_user(f"p{i}")],
+            )
+
+        task = asyncio.create_task(list_sessions_from_store(store, directory=DIR))
+        for _ in range(5):
+            await asyncio.sleep(0)
+        assert 0 < peak <= _sessions._STORE_LIST_LOAD_CONCURRENCY
+        gate.set()
+        await task
+        assert peak == _sessions._STORE_LIST_LOAD_CONCURRENCY
+
 
 # ---------------------------------------------------------------------------
 # Parity: incremental fold == batch lite-parse