Keep session metadata when the first record exceeds the lite read window

src/claude_agent_sdk/_internal/session_mutations.py

@@ -41,6 +41,7 @@
     _find_project_dir,
     _get_projects_dir,
     _get_worktree_paths,
+    _lite_head_bytes,
     _validate_uuid,
     project_key_for_directory,
 )
@@ -311,9 +312,7 @@ def fork_session(
 
     def _derive_title() -> str | None:
         buf_len = len(content)
-        head = content[: min(buf_len, LITE_READ_BUF_SIZE)].decode(
-            "utf-8", errors="replace"
-        )
+        head = _lite_head_bytes(content).decode("utf-8", errors="replace")
         tail = content[max(0, buf_len - LITE_READ_BUF_SIZE) :].decode(
             "utf-8", errors="replace"
         )

src/claude_agent_sdk/_internal/sessions.py

@@ -31,6 +31,9 @@
 
 # Size of the head/tail buffer for lite metadata reads.
 LITE_READ_BUF_SIZE = 65536
+# Cap on growing the head to complete an oversized first record, so a
+# pathological file cannot blow up a listing read.
+LITE_HEAD_MAX_SIZE = 16 * LITE_READ_BUF_SIZE
 
 # Upper bound on concurrent ``store.load()`` calls issued by
 # ``list_sessions_from_store``. Keeps large project listings from exhausting
@@ -256,6 +259,27 @@ def _extract_last_json_string_field(text: str, key: str) -> str | None:
 # ---------------------------------------------------------------------------
 
 
+def _extract_first_top_level_timestamp(head: str) -> str | None:
+    """First top-level "timestamp" value among the head's complete records.
+
+    A raw text scan can match a nested timestamp inside another record's
+    payload and yield a wrong value, so parse whole lines instead. Lines
+    that fail to parse (including a trailing cut-off fragment) are skipped.
+    Records split on "\n" only: str.splitlines() also breaks on U+2028 and
+    friends, which JSON.stringify emits unescaped inside strings.
+    """
+    for line in head.split("\n"):
+        try:
+            entry = json.loads(line)
+        except (json.JSONDecodeError, ValueError):
+            continue
+        if isinstance(entry, dict):
+            timestamp = entry.get("timestamp")
+            if isinstance(timestamp, str):
+                return timestamp
+    return None
+
+
 def _extract_first_prompt_from_head(head: str) -> str:
     """Extracts the first meaningful user prompt from a JSONL head chunk.
 
@@ -338,6 +362,15 @@ def _extract_first_prompt_from_head(head: str) -> str:
 # ---------------------------------------------------------------------------
 
 
+def _lite_head_bytes(buf: bytes) -> bytes:
+    """First-window head of in-memory content, grown (bounded) to complete a
+    record cut by the window. Mirrors ``_read_session_lite``'s growth."""
+    if len(buf) > LITE_READ_BUF_SIZE and not buf[:LITE_READ_BUF_SIZE].endswith(b"\n"):
+        newline_at = buf.find(b"\n", LITE_READ_BUF_SIZE, LITE_HEAD_MAX_SIZE)
+        return buf[: newline_at + 1] if newline_at >= 0 else buf[:LITE_HEAD_MAX_SIZE]
+    return buf[:LITE_READ_BUF_SIZE]
+
+
 class _LiteSessionFile:
     """Result of reading a session file's head, tail, mtime and size."""
 
@@ -365,6 +398,27 @@ def _read_session_lite(file_path: Path) -> _LiteSessionFile | None:
             if not head_bytes:
                 return None
 
+            # A single record can exceed the window (a pasted log or stack
+            # trace) and get cut mid-line; the parsers then skip it as
+            # malformed. The CLI writes bookkeeping records first, so the
+            # oversized record is rarely line 1. The trigger is the window
+            # ending mid-record, not the window lacking a newline. Grow the
+            # head (bounded) one chunk at a time until the cut record closes.
+            if size > LITE_READ_BUF_SIZE and not head_bytes.endswith(b"\n"):
+                search_from = len(head_bytes)
+                while len(head_bytes) < LITE_HEAD_MAX_SIZE:
+                    chunk = f.read(
+                        min(LITE_READ_BUF_SIZE, LITE_HEAD_MAX_SIZE - len(head_bytes))
+                    )
+                    if not chunk:
+                        break
+                    head_bytes += chunk
+                    newline_at = head_bytes.find(b"\n", search_from)
+                    if newline_at >= 0:
+                        head_bytes = head_bytes[: newline_at + 1]
+                        break
+                    search_from = len(head_bytes)
+
             head = head_bytes.decode("utf-8", errors="replace")
 
             tail_offset = max(0, size - LITE_READ_BUF_SIZE)
@@ -484,7 +538,7 @@ def _parse_session_info_from_lite(
     # with no timestamp field; the first user/assistant record that follows
     # does carry one.
     created_at: int | None = None
-    first_timestamp = _extract_json_string_field(head, "timestamp")
+    first_timestamp = _extract_first_top_level_timestamp(head)
     if first_timestamp:
         try:
             # Python 3.10's fromisoformat doesn't support trailing 'Z'
@@ -1458,7 +1512,7 @@ def _jsonl_to_lite(jsonl: str, mtime: int) -> _LiteSessionFile:
     """
     buf = jsonl.encode("utf-8")
     size = len(buf)
-    head = buf[:LITE_READ_BUF_SIZE].decode("utf-8", errors="replace")
+    head = _lite_head_bytes(buf).decode("utf-8", errors="replace")
     tail = (
         buf[max(0, size - LITE_READ_BUF_SIZE) :].decode("utf-8", errors="replace")
         if size > LITE_READ_BUF_SIZE

tests/test_session_store_anyio.py

@@ -202,3 +202,19 @@ async def test_list_sessions_from_store_one_load_fails(tmp_path: Path) -> None:
     by_sid = {r.session_id: r for r in result}
     assert by_sid[_SID_A].summary == "hello"
     assert by_sid[_SID_B].summary == ""
+
+
+async def test_list_sessions_from_store_oversized_first_record(
+    tmp_path: Path,
+) -> None:
+    """The load() fallback keeps first_prompt when the first record is larger
+    than the lite read window, matching what the disk path now returns."""
+    big_paste = "review this crash log\n" + "ERROR connection reset\n" * 4000
+    store = _ListStore({_SID_A: [_user_entry(big_paste)]})
+
+    result = await list_sessions_from_store(
+        cast(SessionStore, store), directory=str(tmp_path)
+    )
+    assert len(result) == 1
+    assert result[0].first_prompt is not None
+    assert result[0].first_prompt.startswith("review this crash log")

tests/test_sessions.py

@@ -600,6 +600,94 @@ def test_git_branch_from_tail_preferred(
         assert len(sessions) == 1
         assert sessions[0].git_branch == "new-branch"
 
+    def test_oversized_first_record(self, claude_config_dir: Path, tmp_path: Path):
+        """A first record larger than the lite read window keeps its metadata.
+
+        The CLI writes `message` before the record's metadata keys, so a big
+        first message used to push the record's own cwd/gitBranch past the
+        64 KiB head window and drop first_prompt entirely.
+        """
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid = str(uuid.uuid4())
+        file_path = project_dir / f"{sid}.jsonl"
+        big_paste = "review this crash log\n" + "ERROR connection reset\n" * 4000
+        lines = [
+            json.dumps(
+                {
+                    "type": "user",
+                    "message": {
+                        "role": "user",
+                        "content": [{"type": "text", "text": big_paste}],
+                    },
+                    "timestamp": "2026-01-15T10:30:00.000Z",
+                    "cwd": project_path,
+                    "gitBranch": "main",
+                },
+                **_COMPACT,
+            ),
+            json.dumps({"type": "assistant", "message": {"content": "ok"}}, **_COMPACT),
+        ]
+        file_path.write_text("\n".join(lines) + "\n")
+
+        sessions = list_sessions(directory=project_path, include_worktrees=False)
+        assert len(sessions) == 1
+        s = sessions[0]
+        assert s.first_prompt is not None
+        assert s.first_prompt.startswith("review this crash log")
+        assert s.cwd == project_path
+        assert s.git_branch == "main"
+        assert s.created_at == 1768473000000
+
+    def test_oversized_record_after_leading_record(
+        self, claude_config_dir: Path, tmp_path: Path
+    ):
+        """Metadata survives when a small record precedes the oversized one.
+
+        The CLI usually writes a bookkeeping record (e.g. queue-operation)
+        before the first user turn, so the oversized record is rarely line 1.
+        A single leading newline in the window must not disable the head
+        growth.
+        """
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid = str(uuid.uuid4())
+        file_path = project_dir / f"{sid}.jsonl"
+        big_paste = "review this crash log\n" + "ERROR connection reset\n" * 4000
+        lines = [
+            json.dumps({"type": "queue-operation", "operation": "dequeue"}, **_COMPACT),
+            json.dumps(
+                {
+                    "type": "user",
+                    "message": {
+                        "role": "user",
+                        "content": [{"type": "text", "text": big_paste}],
+                    },
+                    "timestamp": "2026-01-15T10:30:00.000Z",
+                    "cwd": project_path,
+                    "gitBranch": "main",
+                },
+                **_COMPACT,
+            ),
+            json.dumps({"type": "assistant", "message": {"content": "ok"}}, **_COMPACT),
+        ]
+        file_path.write_text("\n".join(lines) + "\n")
+
+        sessions = list_sessions(directory=project_path, include_worktrees=False)
+        assert len(sessions) == 1
+        s = sessions[0]
+        assert s.first_prompt is not None
+        assert s.first_prompt.startswith("review this crash log")
+        assert s.cwd == project_path
+        assert s.git_branch == "main"
+        assert s.created_at == 1768473000000
+
 
 class TestSDKSessionInfoType:
     """Tests for the SDKSessionInfo dataclass."""
@@ -1413,6 +1501,82 @@ def test_created_at_when_first_line_lacks_timestamp(
         assert len(sessions) == 1
         assert sessions[0].created_at == 1768473000000
 
+    def test_created_at_ignores_nested_timestamp(
+        self, claude_config_dir: Path, tmp_path: Path
+    ):
+        """created_at comes from a top-level timestamp, not a nested one.
+
+        A raw text scan of the head can match a "timestamp" key inside
+        another record's payload (e.g. a file-history-snapshot) and produce
+        a wrong created_at rather than a missing one.
+        """
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid = str(uuid.uuid4())
+        file_path = project_dir / f"{sid}.jsonl"
+        lines = [
+            json.dumps(
+                {
+                    "type": "file-history-snapshot",
+                    "snapshot": {"timestamp": "2020-01-01T00:00:00.000Z"},
+                }
+            ),
+            json.dumps(
+                {
+                    "type": "user",
+                    "message": {"content": "hello"},
+                    "timestamp": "2026-01-15T10:30:00.000Z",
+                }
+            ),
+        ]
+        file_path.write_text("\n".join(lines) + "\n")
+
+        sessions = list_sessions(directory=project_path, include_worktrees=False)
+        assert len(sessions) == 1
+        assert sessions[0].created_at == 1768473000000
+
+    def test_created_at_survives_unicode_line_separator(
+        self, claude_config_dir: Path, tmp_path: Path
+    ):
+        """A U+2028 inside a JSON string must not split the record.
+
+        JSON.stringify writes U+2028/U+2029 (and U+0085) unescaped inside
+        strings, and str.splitlines() breaks on them, cutting the record
+        into fragments that fail to parse and losing its timestamp.
+        """
+        project_path = str(tmp_path / "proj")
+        Path(project_path).mkdir(parents=True)
+        project_dir = _make_project_dir(
+            claude_config_dir, os.path.realpath(project_path)
+        )
+        sid = str(uuid.uuid4())
+        file_path = project_dir / f"{sid}.jsonl"
+        lines = [
+            json.dumps(
+                {
+                    "type": "user",
+                    "message": {"content": "before\u2028after"},
+                    "timestamp": "2026-01-15T10:30:00.000Z",
+                },
+                ensure_ascii=False,
+            ),
+            json.dumps(
+                {
+                    "type": "assistant",
+                    "message": {"content": "ok"},
+                    "timestamp": "2026-01-15T11:00:00.000Z",
+                }
+            ),
+        ]
+        file_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+        sessions = list_sessions(directory=project_path, include_worktrees=False)
+        assert len(sessions) == 1
+        assert sessions[0].created_at == 1768473000000
+
     def test_created_at_none_when_missing(
         self, claude_config_dir: Path, tmp_path: Path
     ):