Resolve archived sessions via source files, not filename stems

claude_code_log/cache.py

@@ -2759,6 +2759,104 @@ def get_stale_sessions(
 
         return stale_sessions
 
+    def get_existing_session_ids(self, filename_session_ids: set[str]) -> set[str]:
+        """Resolve cached sessions through their actual, project-local source files.
+
+        Callers derive the "valid" session IDs from JSONL filename stems, but a
+        stem is only a reliable identifier for an uncached file. A single JSONL
+        can contain several session IDs, and none of them has to equal the
+        filename stem (for example after a file is renamed). Relying on stems
+        alone wrongly reports such sessions as archived even though their source
+        file still exists on disk.
+
+        This augments the filename-derived set with every cached session whose
+        recorded *trunk* source file still exists, scoped to this project. Only
+        non-sidechain rows count: a session's ``agent-*`` sub-transcripts live
+        in their own cached files but carry the trunk session ID, so a surviving
+        agent file must not keep a session "current" once its trunk JSONL is
+        gone (trunk discovery could not load it). It is read-only and
+        deliberately does not reconstruct transcripts.
+        """
+        existing = set(filename_session_ids)
+        if self._project_id is None:
+            return existing
+        with self._get_connection() as conn:
+            rows = conn.execute(
+                """SELECT DISTINCT m.session_id, f.file_path
+                   FROM messages m JOIN cached_files f ON f.id = m.file_id
+                   WHERE m.project_id = ? AND f.project_id = m.project_id
+                     AND m.session_id IS NOT NULL
+                     AND m._is_sidechain = 0""",
+                (self._project_id,),
+            ).fetchall()
+        path_exists: dict[str, bool] = {}
+        for row in rows:
+            sid, source = row["session_id"], row["file_path"]
+            if sid in existing:
+                continue
+            if source not in path_exists:
+                path_exists[source] = Path(source).is_file()
+            if path_exists[source]:
+                existing.add(sid)
+        return existing
+
+    def get_session_source_paths(
+        self, session_id: str, trunk_only: bool = True
+    ) -> List[str]:
+        """Existing on-disk source paths for ``session_id`` in this project.
+
+        Resolves a session's real source file(s) from the cache instead of
+        assuming the ``<session-id>.jsonl`` naming, so archive/delete act on the
+        right file for a renamed or multi-session JSONL. With ``trunk_only``
+        (default) only non-sidechain rows count, so ``agent-*`` sub-transcripts
+        are never returned — a session's agent files must not be deleted when it
+        is archived. Paths whose file no longer exists are omitted; order
+        follows the cached file id.
+        """
+        if self._project_id is None:
+            return []
+        sidechain_clause = "\n                     AND m._is_sidechain = 0" if trunk_only else ""
+        with self._get_connection() as conn:
+            rows = conn.execute(
+                """SELECT DISTINCT f.id AS fid, f.file_path AS file_path
+                   FROM messages m JOIN cached_files f ON f.id = m.file_id
+                   WHERE m.project_id = ? AND f.project_id = m.project_id
+                     AND m.session_id = ?"""
+                + sidechain_clause
+                + "\n                   ORDER BY f.id",
+                (self._project_id, session_id),
+            ).fetchall()
+        return [row["file_path"] for row in rows if Path(row["file_path"]).is_file()]
+
+    def get_sessions_sharing_file(
+        self, file_path: str, exclude_session_id: str
+    ) -> set[str]:
+        """Other non-hidden session IDs whose trunk rows live in ``file_path``.
+
+        A single JSONL can hold more than one session (a continuation written
+        into the previous session's file). Archiving or deleting that file
+        therefore affects those sessions too, so callers warn before removing a
+        shared source. Excludes ``exclude_session_id`` and hidden sessions,
+        counts trunk (non-sidechain) rows only, and is scoped to this project.
+        """
+        if self._project_id is None:
+            return set()
+        with self._get_connection() as conn:
+            rows = conn.execute(
+                """SELECT DISTINCT m.session_id AS sid
+                   FROM messages m
+                   JOIN cached_files f ON f.id = m.file_id
+                   JOIN sessions s ON s.session_id = m.session_id
+                                  AND s.project_id = m.project_id
+                   WHERE m.project_id = ? AND f.file_path = ?
+                     AND m._is_sidechain = 0
+                     AND m.session_id IS NOT NULL
+                     AND m.session_id != ?
+                     AND s.hidden = 0""",
+                (self._project_id, file_path, exclude_session_id),
+            ).fetchall()
+        return {row["sid"] for row in rows}
+
     def get_archived_session_count(self, valid_session_ids: set[str]) -> int:
         """Count sessions in cache whose JSONL files have been deleted.
 
@@ -2773,6 +2871,8 @@ def get_archived_session_count(self, valid_session_ids: set[str]) -> int:
         if self._project_id is None:
             return 0
 
+        valid_session_ids = self.get_existing_session_ids(valid_session_ids)
+
         with self._get_connection() as conn:
             cached_rows = conn.execute(
                 "SELECT session_id FROM sessions WHERE project_id = ? AND hidden = 0",
@@ -2799,6 +2899,8 @@ def get_archived_sessions(
 
         archived_sessions: Dict[str, SessionCacheData] = {}
 
+        valid_session_ids = self.get_existing_session_ids(valid_session_ids)
+
         with self._get_connection() as conn:
             session_rows = conn.execute(
                 "SELECT * FROM sessions WHERE project_id = ? AND hidden = 0",

claude_code_log/converter.py

@@ -6332,8 +6332,12 @@ def _plan_project(
     # Fast staleness check (mtime comparison only). Exclude agent
     # files - they are loaded via session references, not directly.
     jsonl_files = trunk_jsonl_files(project_dir)
-    # Valid session IDs are from existing JSONL files (file stem = session ID)
+    # Valid session IDs start from existing JSONL filename stems, then are
+    # resolved through the cache's recorded source files so renamed and
+    # multi-session JSONLs are not mistaken for deleted (archived) sources.
     valid_session_ids = {f.stem for f in jsonl_files}
+    if cache_manager:
+        valid_session_ids = cache_manager.get_existing_session_ids(valid_session_ids)
     modified_files = (
         cache_manager.get_modified_files(jsonl_files) if cache_manager else []
     )

claude_code_log/tui.py

@@ -857,6 +857,11 @@ class ArchiveConfirmScreen(ModalScreen[bool]):
         margin-bottom: 1;
     }
 
+    #archive-shared {
+        color: $warning;
+        margin-bottom: 1;
+    }
+
     #archive-buttons {
         text-align: center;
         height: auto;
@@ -870,11 +875,21 @@ class ArchiveConfirmScreen(ModalScreen[bool]):
         Binding("escape", "cancel", "Cancel", show=False),
     ]
 
-    def __init__(self, session_id: str) -> None:
+    def __init__(
+        self, session_id: str, shared_sessions: Optional[List[str]] = None
+    ) -> None:
+        """Confirm archiving ``session_id``.
+
+        ``shared_sessions`` lists other current session IDs that share this
+        session's source JSONL; when non-empty the modal warns that archiving
+        removes the file for them too.
+        """
         super().__init__()
         self.session_id = session_id
+        self.shared_sessions = shared_sessions or []
 
     def compose(self) -> ComposeResult:
+        """Lay out the archive-confirmation modal."""
         with Container(id="archive-container"):
             yield Static("Archive Session", id="archive-title")
             yield Static(
@@ -886,6 +901,13 @@ def compose(self) -> ComposeResult:
                 "The session will be archived and can be restored from cache.",
                 id="archive-info",
             )
+            if self.shared_sessions:
+                shared = ", ".join(s[:8] for s in self.shared_sessions)
+                yield Static(
+                    f"⚠ Its JSONL also holds {len(self.shared_sessions)} other "
+                    f"session(s): {shared}.\nArchiving it archives them too.",
+                    id="archive-shared",
+                )
             yield Static("\\[Enter/y] Yes  \\[Esc/n] No", id="archive-buttons")
 
     def action_confirm(self) -> None:
@@ -927,6 +949,11 @@ class DeleteConfirmScreen(ModalScreen[Optional[str]]):
         margin-bottom: 1;
     }
 
+    #delete-shared {
+        color: $error;
+        margin-bottom: 1;
+    }
+
     #delete-buttons {
         text-align: center;
         height: auto;
@@ -942,12 +969,25 @@ class DeleteConfirmScreen(ModalScreen[Optional[str]]):
         Binding("escape", "cancel", "Cancel", show=False),
     ]
 
-    def __init__(self, session_id: str, is_archived: bool = False) -> None:
+    def __init__(
+        self,
+        session_id: str,
+        is_archived: bool = False,
+        shared_sessions: Optional[List[str]] = None,
+    ) -> None:
+        """Confirm deleting ``session_id``.
+
+        ``shared_sessions`` lists other current session IDs that share this
+        session's source JSONL; when non-empty the modal warns that the "Both"
+        option deletes the file for them too.
+        """
         super().__init__()
         self.session_id = session_id
         self.is_archived = is_archived
+        self.shared_sessions = shared_sessions or []
 
     def compose(self) -> ComposeResult:
+        """Lay out the delete-confirmation modal."""
         with Container(id="delete-container"):
             yield Static("Delete Session", id="delete-title")
             yield Static(
@@ -971,6 +1011,13 @@ def compose(self) -> ComposeResult:
                     "• Both: Delete JSONL file AND cache (permanent!)",
                     id="delete-warning",
                 )
+                if self.shared_sessions:
+                    shared = ", ".join(s[:8] for s in self.shared_sessions)
+                    yield Static(
+                        f"⚠ \"Both\" also deletes the JSONL for "
+                        f"{len(self.shared_sessions)} other session(s): {shared}.",
+                        id="delete-shared",
+                    )
                 yield Static(
                     "\\[c] Cache only   \\[b] Both (permanent)   \\[Esc/n] Cancel",
                     id="delete-buttons",
@@ -1394,23 +1441,28 @@ def load_sessions(self) -> None:
                 self.log.debug(f"Cache building failed during startup: {e}")
                 return
 
-        # Only compute archived sessions if there are JSONL files to compare against
-        # (in test environments, there may be cached sessions but no JSONL files)
-        if valid_session_ids:
-            # Load archived sessions (cached but JSONL deleted)
-            self.archived_sessions = self.cache_manager.get_archived_sessions(
-                valid_session_ids
-            )
+        # Resolve filename-stem IDs through the cache's recorded source files so
+        # renamed and multi-session JSONLs count as current, not archived. This
+        # feeds both the archived lookup and the current-session filter below.
+        valid_session_ids = self.cache_manager.get_existing_session_ids(
+            valid_session_ids
+        )
 
-            # Filter current sessions to only those with existing JSONL files
-            self.sessions = {
-                sid: data
-                for sid, data in self.sessions.items()
-                if sid in valid_session_ids
-            }
-        else:
-            # No JSONL files to compare - treat all sessions as current
-            self.archived_sessions = {}
+        # Classify cached sessions against the resolved valid set. This also
+        # runs when the set is empty (e.g. the project's last source JSONL was
+        # just archived): every non-hidden cached session is then archived and
+        # offered for Restore, instead of lingering as a "current" session whose
+        # file is gone.
+        self.archived_sessions = self.cache_manager.get_archived_sessions(
+            valid_session_ids
+        )
+
+        # Keep as current only the sessions whose source JSONL still exists.
+        self.sessions = {
+            sid: data
+            for sid, data in self.sessions.items()
+            if sid in valid_session_ids
+        }
 
         # Only update UI if we're in app context
         try:
@@ -2032,22 +2084,55 @@ def action_archive_session(self) -> None:
             )
             return
 
-        # Push archive confirmation screen
+        # Push archive confirmation screen, warning if the source JSONL is
+        # shared with other current sessions.
         self.push_screen(
-            ArchiveConfirmScreen(session_id=self.selected_session_id),
+            ArchiveConfirmScreen(
+                session_id=self.selected_session_id,
+                shared_sessions=self._shared_source_sessions(
+                    self.selected_session_id
+                ),
+            ),
             callback=self._on_archive_confirm,
         )
 
+    def _resolve_source_paths(self, session_id: str) -> List[Path]:
+        """Existing trunk JSONL path(s) backing ``session_id``.
+
+        Resolves the real source files from the cache so archive/delete act on
+        a renamed or multi-session JSONL correctly, falling back to the
+        canonical ``<session-id>.jsonl`` only when the cache knows no source.
+        """
+        paths = [Path(p) for p in self.cache_manager.get_session_source_paths(session_id)]
+        if not paths:
+            legacy = self.project_path / f"{session_id}.jsonl"
+            if legacy.is_file():
+                paths = [legacy]
+        return paths
+
+    def _shared_source_sessions(self, session_id: str) -> List[str]:
+        """Other current session IDs that share ``session_id``'s source file(s)."""
+        shared: set[str] = set()
+        for path in self._resolve_source_paths(session_id):
+            shared |= self.cache_manager.get_sessions_sharing_file(
+                str(path), session_id
+            )
+        return sorted(shared & set(self.sessions))
+
     def _on_archive_confirm(self, confirmed: Optional[bool]) -> None:
         """Handle archive confirmation result."""
         if not confirmed or not self.selected_session_id:
             return
 
         try:
-            # Delete the JSONL file
-            jsonl_path = self.project_path / f"{self.selected_session_id}.jsonl"
-            if jsonl_path.exists():
-                jsonl_path.unlink()
+            # Delete the real source JSONL file(s), resolved from the cache so a
+            # renamed or multi-session file is handled, not assumed by name.
+            removed = False
+            for jsonl_path in self._resolve_source_paths(self.selected_session_id):
+                if jsonl_path.exists():
+                    jsonl_path.unlink()
+                    removed = True
+            if removed:
                 self.notify(
                     f"Session {self.selected_session_id[:8]} archived",
                     severity="information",
@@ -2076,11 +2161,18 @@ def action_delete_session(self) -> None:
         # Determine if this is an archived session (no JSONL to fall back on)
         is_archived_session = self.selected_session_id in self.archived_sessions
 
-        # Push confirmation screen
+        # Push confirmation screen, warning if "Both" would remove a JSONL that
+        # is shared with other current sessions.
+        shared = (
+            []
+            if is_archived_session
+            else self._shared_source_sessions(self.selected_session_id)
+        )
         self.push_screen(
             DeleteConfirmScreen(
                 session_id=self.selected_session_id,
                 is_archived=is_archived_session,
+                shared_sessions=shared,
             ),
             callback=self._on_delete_confirm,
         )
@@ -2093,11 +2185,18 @@ def _on_delete_confirm(self, delete_option: Optional[str]) -> None:
         try:
             deleted_what: list[str] = []
 
-            # Delete JSONL file if requested
+            # Delete JSONL file if requested, resolving the real source file(s)
+            # from the cache so a renamed or multi-session file is removed, not
+            # a path merely assumed from the session id.
             if delete_option == "both":
-                jsonl_path = self.project_path / f"{self.selected_session_id}.jsonl"
-                if jsonl_path.exists():
-                    jsonl_path.unlink()
+                removed_jsonl = False
+                for jsonl_path in self._resolve_source_paths(
+                    self.selected_session_id
+                ):
+                    if jsonl_path.exists():
+                        jsonl_path.unlink()
+                        removed_jsonl = True
+                if removed_jsonl:
                     deleted_what.append("JSONL file")
 
             # Delete from cache