fix: load_range path builds _LiteSessionFile directly (head/tail isolated, file_size=None); ReorderingStore overrides load_range; bump conformance count to 14

Author: qing-ant

src/claude_agent_sdk/_internal/sessions.py

@@ -346,7 +346,7 @@ class _LiteSessionFile:
 
     __slots__ = ("mtime", "size", "head", "tail")
 
-    def __init__(self, mtime: int, size: int, head: str, tail: str) -> None:
+    def __init__(self, mtime: int, size: int | None, head: str, tail: str) -> None:
         self.mtime = mtime
         self.size = size
         self.head = head
@@ -1506,16 +1506,18 @@ def _filter_transcript_entries(entries: list[Any]) -> list[_TranscriptEntry]:
     return result
 
 
-async def _load_store_entries_as_jsonl(
+async def _load_store_session_lite(
     store: SessionStore, session_id: str, directory: str | None
-) -> str | None:
-    """Load entries from a SessionStore and serialize to a JSONL string.
+) -> _LiteSessionFile | None:
+    """Load a session from a SessionStore as a ``_LiteSessionFile``.
 
     Only the head/tail slice needed for lite-parse summary derivation is
     fetched when the store implements :meth:`SessionStore.load_range`;
-    otherwise falls back to a full :meth:`SessionStore.load`. Either way the
-    result is fed to ``_jsonl_to_lite`` which itself slices to a 64KB
-    head/tail, so the parse path is identical.
+    otherwise falls back to a full :meth:`SessionStore.load` and slices via
+    ``_jsonl_to_lite``. Head and tail are serialized separately on the
+    ``load_range`` path so head-only scans (e.g. ``first_prompt``) never see
+    tail entries; ``size`` is ``None`` on that path since only a slice was
+    fetched.
 
     Returns ``None`` if the session has no entries.
     """
@@ -1525,12 +1527,24 @@ async def _load_store_entries_as_jsonl(
         head, tail = await store.load_range(
             key, head=_LITE_LOAD_HEAD_ENTRIES, tail=_LITE_LOAD_TAIL_ENTRIES
         ) or ([], [])
-        entries_for_lite: list[Any] | None = list(head) + list(tail)
-    else:
-        entries_for_lite = await store.load(key)
-    if not entries_for_lite:
+        if not head and not tail:
+            return None
+        head_jsonl = _entries_to_jsonl(list(head)) if head else ""
+        # Mirror the disk path's "tail = head when small" semantics so
+        # adapters that return an empty tail for short sessions still expose
+        # the head entries to tail-scanning extractors.
+        tail_jsonl = _entries_to_jsonl(list(tail)) if tail else head_jsonl
+        return _LiteSessionFile(
+            mtime=_mtime_from_jsonl_tail(tail_jsonl or head_jsonl),
+            size=None,
+            head=head_jsonl,
+            tail=tail_jsonl,
+        )
+    entries = await store.load(key)
+    if not entries:
         return None
-    return _entries_to_jsonl(entries_for_lite)
+    jsonl = _entries_to_jsonl(entries)
+    return _jsonl_to_lite(jsonl, _mtime_from_jsonl_tail(jsonl))
 
 
 async def list_sessions_from_store(
@@ -1595,9 +1609,9 @@ async def list_sessions_from_store(
     # disk path.
     sem = asyncio.Semaphore(_STORE_LIST_LOAD_CONCURRENCY)
 
-    async def _bounded_load(sid: str) -> str | None:
+    async def _bounded_load(sid: str) -> _LiteSessionFile | None:
         async with sem:
-            return await _load_store_entries_as_jsonl(session_store, sid, directory)
+            return await _load_store_session_lite(session_store, sid, directory)
 
     settled = await asyncio.gather(
         *(_bounded_load(e["session_id"]) for e in listing),
@@ -1614,9 +1628,7 @@ async def _bounded_load(sid: str) -> str | None:
             continue
         if outcome is None:
             continue
-        parsed = _parse_session_info_from_lite(
-            sid, _jsonl_to_lite(outcome, mtime), project_path
-        )
+        parsed = _parse_session_info_from_lite(sid, outcome, project_path)
         if parsed is None:
             # Sidechain or no extractable summary — drop, matching the
             # filesystem path.
@@ -1648,10 +1660,9 @@ async def get_session_info_from_store(
     """
     if not _validate_uuid(session_id):
         return None
-    jsonl = await _load_store_entries_as_jsonl(session_store, session_id, directory)
-    if jsonl is None:
+    lite = await _load_store_session_lite(session_store, session_id, directory)
+    if lite is None:
         return None
-    lite = _jsonl_to_lite(jsonl, _mtime_from_jsonl_tail(jsonl))
     project_path = _canonicalize_path(str(directory) if directory is not None else ".")
     return _parse_session_info_from_lite(session_id, lite, project_path)
 

src/claude_agent_sdk/testing/session_store_conformance.py

@@ -1,9 +1,10 @@
 """Shared conformance test suite for :class:`SessionStore` adapters.
 
 Call :func:`run_session_store_conformance` from an async test to assert the
-13 behavioral contracts every adapter must satisfy. Tests for optional
-methods (``list_sessions``, ``delete``, ``list_subkeys``) are skipped when
-named in ``skip_optional`` or when the store does not override that method.
+14 behavioral contracts every adapter must satisfy. Tests for optional
+methods (``list_sessions``, ``delete``, ``list_subkeys``, ``load_range``) are
+skipped when named in ``skip_optional`` or when the store does not override
+that method.
 
 Example::
 
@@ -53,12 +54,12 @@ async def run_session_store_conformance(
     *,
     skip_optional: frozenset[str] = frozenset(),
 ) -> None:
-    """Assert the 13 :class:`SessionStore` behavioral contracts.
+    """Assert the 14 :class:`SessionStore` behavioral contracts.
 
     ``make_store`` is invoked once per contract to provide isolation. It may be
     sync or async. Contracts for optional methods (``list_sessions``,
-    ``delete``, ``list_subkeys``) are skipped when named in ``skip_optional``
-    or when the store does not override that method.
+    ``delete``, ``list_subkeys``, ``load_range``) are skipped when named in
+    ``skip_optional`` or when the store does not override that method.
     """
     invalid = skip_optional - _OPTIONAL_METHODS
     assert not invalid, f"unknown optional methods in skip_optional: {invalid}"

tests/test_session_helpers_store.py

@@ -553,13 +553,18 @@ async def test_tag_survives_adapter_key_reordering(self) -> None:
         object keys (Postgres JSONB does this). The tag extractor must not
         depend on ``type`` being the first key in the serialized line."""
 
+        def _reorder(entries):
+            # Sort keys alphabetically — deep-equal but different order.
+            return [dict(sorted(e.items())) for e in entries]
+
         class ReorderingStore(InMemorySessionStore):
             async def load(self, key):  # type: ignore[override]
                 entries = await super().load(key)
-                if entries is None:
-                    return None
-                # Sort keys alphabetically — deep-equal but different order.
-                return [dict(sorted(e.items())) for e in entries]
+                return None if entries is None else _reorder(entries)
+
+            async def load_range(self, key, *, head=0, tail=0):  # type: ignore[override]
+                r = await super().load_range(key, head=head, tail=tail)
+                return None if r is None else (_reorder(r[0]), _reorder(r[1]))
 
         store = ReorderingStore()
         sid = str(uuid_mod.uuid4())

tests/test_session_store_load_range.py

@@ -294,6 +294,9 @@ async def test_first_prompt_and_title_derived_from_head_tail(self) -> None:
         assert info.tag == "important"
         # created_at parsed from first entry's timestamp.
         assert info.created_at is not None
+        # Only a slice was fetched, so file_size is unknown — None, not the
+        # misleading byte-count of the ~30-entry slice.
+        assert info.file_size is None
         # Only load_range was used — full transcript never fetched.
         assert store.load_range_calls == 1
         assert store.load_calls == 0
@@ -305,5 +308,40 @@ async def test_first_prompt_and_title_derived_from_head_tail(self) -> None:
         assert len(sessions) == 1
         assert sessions[0].summary == "My Big Session"
         assert sessions[0].first_prompt == "prompt 0"
+        assert sessions[0].file_size is None
         assert store.load_range_calls == 1
         assert store.load_calls == 0
+
+    async def test_first_prompt_does_not_leak_from_tail(self) -> None:
+        """When the head slice contains only skippable noise (isMeta/system),
+        ``first_prompt`` must be ``None`` rather than a recent user prompt
+        pulled from the tail. Regression for head/tail being concatenated
+        before ``_jsonl_to_lite`` so the head scan saw tail entries."""
+        store = _SpyStoreWithRange()
+        sid = str(uuid_mod.uuid4())
+        key: SessionKey = {"project_key": PROJECT_KEY, "session_id": sid}
+
+        # head=10 will be these 10 system/meta noise entries.
+        noise: list[dict[str, Any]] = [
+            {"type": "system", "uuid": f"n{i}", "isMeta": True} for i in range(10)
+        ]
+        # Middle filler so the head/tail slices don't overlap.
+        filler: list[dict[str, Any]] = [
+            {"type": "system", "uuid": f"f{i}"} for i in range(50)
+        ]
+        # tail=20 will include this recent user prompt + a customTitle so
+        # the session has an extractable summary.
+        u = str(uuid_mod.uuid4())
+        recent = [
+            _user("recent prompt", u, None, sid, "2024-01-01T00:59:00Z"),
+            {"type": "custom-title", "customTitle": "Noise Head"},
+        ]
+        await store.append(key, noise + filler + recent)  # type: ignore[arg-type]
+
+        info = await get_session_info_from_store(store, sid, directory=DIR)
+        assert info is not None
+        assert info.summary == "Noise Head"
+        # Head contained no user prompt → first_prompt must not be sourced
+        # from the tail's "recent prompt".
+        assert info.first_prompt is None
+        assert info.file_size is None