Fix: session title/tag silently lost after growth past the lite-read dead zone (#1191)

src/claude_agent_sdk/_internal/sessions.py

@@ -353,6 +353,15 @@ def __init__(self, mtime: int, size: int, head: str, tail: str) -> None:
 def _read_session_lite(file_path: Path) -> _LiteSessionFile | None:
     """Opens a session file, stats it, and reads head + tail.
 
+    ``rename_session()`` / ``tag_session()`` append a standalone JSONL record
+    (``{"type":"custom-title",...}`` / ``{"type":"tag",...}``) to the file.
+    Once the transcript keeps growing past both the head and tail windows,
+    that record lands in the byte range between them and becomes invisible
+    to a plain head/tail read — the title or tag is silently lost even
+    though it is still on disk (see #1191). When that dead zone exists and
+    neither window shows a title or tag record, fall back to a full read so
+    the record stays visible for the life of the session.
+
     Returns None on any error or if file is empty.
     """
     try:
@@ -375,6 +384,14 @@ def _read_session_lite(file_path: Path) -> _LiteSessionFile | None:
                 tail_bytes = f.read(LITE_READ_BUF_SIZE)
                 tail = tail_bytes.decode("utf-8", errors="replace")
 
+            if tail_offset > LITE_READ_BUF_SIZE:
+                title_seen = '"customTitle"' in head or '"customTitle"' in tail
+                tag_seen = '{"type":"tag"' in head or '{"type":"tag"' in tail
+                if not (title_seen and tag_seen):
+                    f.seek(0)
+                    full = f.read().decode("utf-8", errors="replace")
+                    head = tail = full
+
             return _LiteSessionFile(mtime=mtime, size=size, head=head, tail=tail)
     except OSError:
         return None
@@ -1452,18 +1469,26 @@ def _type_first(e: Any) -> Any:
 def _jsonl_to_lite(jsonl: str, mtime: int) -> _LiteSessionFile:
     """Build the head/tail/size lite shape from an in-memory JSONL string.
 
-    Matches ``_read_session_lite``'s byte semantics so the store path exposes
-    the same slice to ``_parse_session_info_from_lite`` as the disk path
-    would for the same transcript.
+    Matches ``_read_session_lite``'s byte semantics, including its dead-zone
+    fallback (see #1191): a title/tag record between the head and tail
+    windows is invisible to a plain slice, so when neither window shows one
+    this falls back to the full in-memory string, which costs nothing extra
+    since ``jsonl`` is already fully materialized.
     """
     buf = jsonl.encode("utf-8")
     size = len(buf)
     head = buf[:LITE_READ_BUF_SIZE].decode("utf-8", errors="replace")
+    tail_offset = max(0, size - LITE_READ_BUF_SIZE)
     tail = (
-        buf[max(0, size - LITE_READ_BUF_SIZE) :].decode("utf-8", errors="replace")
+        buf[tail_offset:].decode("utf-8", errors="replace")
         if size > LITE_READ_BUF_SIZE
         else head
     )
+    if tail_offset > LITE_READ_BUF_SIZE:
+        title_seen = '"customTitle"' in head or '"customTitle"' in tail
+        tag_seen = '{"type":"tag"' in head or '{"type":"tag"' in tail
+        if not (title_seen and tag_seen):
+            head = tail = buf.decode("utf-8", errors="replace")
     return _LiteSessionFile(mtime=mtime, size=size, head=head, tail=tail)
 
 

tests/test_sessions.py

@@ -19,6 +19,7 @@
     list_subagents,
 )
 from claude_agent_sdk._internal.sessions import (
+    LITE_READ_BUF_SIZE,
     _build_conversation_chain,
     _extract_first_prompt_from_head,
     _extract_json_string_field,
@@ -1286,6 +1287,73 @@ def test_tag_none_when_only_tool_use_tag(
         assert len(sessions) == 1
         assert sessions[0].tag is None  # NOT "prod"
 
+    def test_title_and_tag_survive_growth_past_dead_zone(
+        self, claude_config_dir: Path, tmp_path: Path
+    ):
+        """customTitle/tag records between the head and tail windows must
+        stay visible once the transcript keeps growing (regression for #1191).
+
+        ``rename_session()``/``tag_session()`` append a standalone record.
+        A plain head+tail read only sees the first and last
+        ``LITE_READ_BUF_SIZE`` bytes, so a record written early and then
+        pushed out of both windows by further conversation used to vanish —
+        the title silently reverted to the first prompt and the tag to
+        ``None``, even though both records were still on disk.
+        """
+        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"
+
+        filler = json.dumps(
+            {"type": "user", "message": {"content": "x" * 500}}, **_COMPACT
+        )
+
+        def pad(nbytes: int) -> list[str]:
+            n = nbytes // (len(filler) + 1) + 1
+            return [filler] * n
+
+        lines = [
+            json.dumps(
+                {"type": "user", "message": {"content": "investigate the bug"}},
+                **_COMPACT,
+            ),
+        ]
+        # Push the file well past the head window before the title/tag land,
+        # then keep growing well past the tail window too, so the records
+        # sit strictly between both — the dead zone.
+        lines += pad(LITE_READ_BUF_SIZE)
+        lines.append(
+            json.dumps(
+                {
+                    "type": "custom-title",
+                    "customTitle": "Release checklist",
+                    "sessionId": sid,
+                },
+                **_COMPACT,
+            )
+        )
+        lines.append(
+            json.dumps({"type": "tag", "tag": "release", "sessionId": sid}, **_COMPACT)
+        )
+        lines += pad(LITE_READ_BUF_SIZE)
+
+        file_path.write_text("\n".join(lines) + "\n")
+        assert file_path.stat().st_size > 2 * LITE_READ_BUF_SIZE
+
+        sessions = list_sessions(directory=project_path, include_worktrees=False)
+        assert len(sessions) == 1
+        assert sessions[0].custom_title == "Release checklist"
+        assert sessions[0].tag == "release"
+
+        info = get_session_info(sid, directory=project_path)
+        assert info is not None
+        assert info.custom_title == "Release checklist"
+        assert info.tag == "release"
+
     def test_parse_session_info_from_lite_helper(self, tmp_path: Path):
         """Direct test of the refactored _parse_session_info_from_lite helper."""
         sid = str(uuid.uuid4())