diff --git a/claude_code_log/cache.py b/claude_code_log/cache.py index 564b4494..6fd81ec6 100644 --- a/claude_code_log/cache.py +++ b/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 ``.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", diff --git a/claude_code_log/converter.py b/claude_code_log/converter.py index 53c5573b..f487cfc5 100644 --- a/claude_code_log/converter.py +++ b/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 [] ) diff --git a/claude_code_log/tui.py b/claude_code_log/tui.py index f812711a..cafca6a1 100644 --- a/claude_code_log/tui.py +++ b/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 ``.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 diff --git a/test/test_archive_detection.py b/test/test_archive_detection.py new file mode 100644 index 00000000..3c83501b --- /dev/null +++ b/test/test_archive_detection.py @@ -0,0 +1,241 @@ +#!/usr/bin/env python3 +"""Archive detection resolves sessions through real source files, not stems. + +A cached session is "archived" only when its source JSONL has actually been +deleted. Because a JSONL can hold several session IDs, and its filename stem +need not equal any of them (e.g. after a rename), matching session IDs against +filename stems alone wrongly flags live sessions as archived. These tests pin +`CacheManager.get_existing_session_ids` and the archive lookups that use it. +""" + +import contextlib +import sqlite3 +import tempfile +import unittest +from pathlib import Path + +from claude_code_log.cache import CacheManager + + +class ArchiveDetectionTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.db = sqlite3.connect(":memory:") + self.db.row_factory = sqlite3.Row + self.addCleanup(self.db.close) + self.db.executescript( + """ + CREATE TABLE cached_files(id INTEGER PRIMARY KEY, project_id INTEGER, file_path TEXT); + CREATE TABLE messages(project_id INTEGER, file_id INTEGER, session_id TEXT, + _is_sidechain INTEGER DEFAULT 0); + CREATE TABLE sessions(project_id INTEGER, session_id TEXT, hidden INTEGER DEFAULT 0, + summary TEXT, ai_title TEXT, first_timestamp TEXT DEFAULT '', last_timestamp TEXT DEFAULT '', + message_count INTEGER DEFAULT 0, first_user_message TEXT DEFAULT '', cwd TEXT, + total_input_tokens INTEGER DEFAULT 0, total_output_tokens INTEGER DEFAULT 0, + total_cache_creation_tokens INTEGER DEFAULT 0, total_cache_read_tokens INTEGER DEFAULT 0, + team_name TEXT); + """ + ) + # Build a CacheManager without touching its real on-disk schema/connection: + # exercise the archive logic in isolation against the stub connection. + self.manager = object.__new__(CacheManager) + self.manager._project_id = 1 + self.manager._get_connection = lambda: contextlib.nullcontext(self.db) + + def source( + self, + sid, + name="different.jsonl", + project=1, + present=True, + hidden=0, + sidechain=0, + add_session=True, + ): + """Record a cached session `sid` sourced from file `name` in `project`. + + `sidechain=1` models an ``agent-*`` sub-transcript row (the trunk + session ID recorded against an agent file), which must not keep a + session current once its trunk JSONL is gone. Pass `add_session=False` + when adding a second source file for a session that already has its row + (the real `sessions` table is unique per `(project, session_id)`). + """ + path = self.root / name + if present: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{}\n") + fid = self.db.execute( + "INSERT INTO cached_files(project_id,file_path) VALUES (?,?)", + (project, str(path)), + ).lastrowid + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id,_is_sidechain) " + "VALUES (?,?,?,?)", + (project, fid, sid, sidechain), + ) + if add_session: + self.db.execute( + "INSERT INTO sessions(project_id,session_id,hidden) VALUES (?,?,?)", + (project, sid, hidden), + ) + return path + + def test_renamed_source_is_not_archived(self): + # Session id differs from the filename stem; the file still exists. + p = self.source("embedded-id") + self.assertEqual(self.manager.get_archived_session_count({p.stem}), 0) + self.assertEqual(self.manager.get_archived_sessions({p.stem}), {}) + self.assertIn("embedded-id", self.manager.get_existing_session_ids({p.stem})) + + def test_multiple_sessions_in_one_file(self): + self.source("first") + self.source("second") + self.assertEqual( + self.manager.get_existing_session_ids(set()), {"first", "second"} + ) + self.assertEqual(self.manager.get_archived_session_count(set()), 0) + + def test_deleted_source_stays_archived(self): + self.source("gone", present=False) + self.assertEqual(self.manager.get_archived_session_count(set()), 1) + self.assertEqual(set(self.manager.get_archived_sessions(set())), {"gone"}) + + def test_one_surviving_copy_suffices(self): + # Same session recorded from two files; one deleted, one still present. + self.source("kept", present=False) + p = self.root / "survivor.jsonl" + p.write_text("{}\n") + fid = self.db.execute( + "INSERT INTO cached_files(project_id,file_path) VALUES (1,?)", (str(p),) + ).lastrowid + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id) VALUES (1,?,?)", + (fid, "kept"), + ) + self.assertEqual(self.manager.get_archived_session_count(set()), 0) + + def test_other_project_cannot_rescue_deleted_session(self): + # A surviving file in another project must not clear this project's orphan. + self.source("same-id", "gone.jsonl", present=False) + self.source("same-id", "present.jsonl", project=2) + self.assertEqual(self.manager.get_archived_session_count(set()), 1) + + def test_hidden_sessions_remain_excluded(self): + self.source("hidden", present=False, hidden=1) + self.assertEqual(self.manager.get_archived_session_count(set()), 0) + self.assertEqual(self.manager.get_archived_sessions(set()), {}) + + def test_uncached_filename_fallback_and_no_input_mutation(self): + # Stems for uncached files survive; the caller's set is not mutated. + names = {"uncached"} + self.source("cached") + self.assertEqual( + self.manager.get_existing_session_ids(names), {"cached", "uncached"} + ) + self.assertEqual(names, {"uncached"}) + + def test_directory_is_not_a_source_file(self): + # A directory sharing the recorded path must not count as a present source. + p = self.source("not-file", present=False) + p.mkdir() + self.assertEqual(self.manager.get_archived_session_count(set()), 1) + + def test_surviving_agent_file_does_not_rescue_deleted_trunk(self): + # Trunk JSONL deleted; a sub-agent file carrying the trunk session id + # survives. The session is archived, not current: trunk discovery + # cannot load it from an agent sub-transcript. + self.source("trunk", "trunk.jsonl", present=False) + self.source("trunk", "subagents/agent-x.jsonl", present=True, sidechain=1, add_session=False) + self.assertNotIn("trunk", self.manager.get_existing_session_ids(set())) + self.assertEqual(self.manager.get_archived_session_count(set()), 1) + self.assertEqual(set(self.manager.get_archived_sessions(set())), {"trunk"}) + + def test_surviving_trunk_with_agent_file_is_current(self): + # Both trunk and its agent file present: the session stays current. + self.source("trunk", "trunk.jsonl", present=True) + self.source("trunk", "subagents/agent-x.jsonl", present=True, sidechain=1, add_session=False) + self.assertIn("trunk", self.manager.get_existing_session_ids(set())) + self.assertEqual(self.manager.get_archived_session_count(set()), 0) + + def test_source_paths_resolve_real_trunk_only(self): + # get_session_source_paths returns the renamed trunk file, never the + # agent sub-transcript, and only files that still exist. + trunk = self.source("renamed", "embedded.jsonl", present=True) + self.source("renamed", "subagents/agent-x.jsonl", present=True, sidechain=1, add_session=False) + self.assertEqual( + self.manager.get_session_source_paths("renamed"), [str(trunk)] + ) + # Including sidechain rows surfaces the agent file too. + self.assertEqual( + len(self.manager.get_session_source_paths("renamed", trunk_only=False)), 2 + ) + + def test_source_paths_omit_deleted_files(self): + self.source("gone", "gone.jsonl", present=False) + self.assertEqual(self.manager.get_session_source_paths("gone"), []) + + def test_sessions_sharing_file_reports_cohabitants(self): + # Two sessions written into one JSONL; each sees the other as sharing. + shared = self.root / "shared.jsonl" + shared.write_text("{}\n") + fid = self.db.execute( + "INSERT INTO cached_files(project_id,file_path) VALUES (1,?)", + (str(shared),), + ).lastrowid + for sid in ("alpha", "beta"): + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id) VALUES (1,?,?)", + (fid, sid), + ) + self.db.execute( + "INSERT INTO sessions(project_id,session_id,hidden) VALUES (1,?,0)", + (sid,), + ) + self.assertEqual( + self.manager.get_sessions_sharing_file(str(shared), "alpha"), {"beta"} + ) + # A file holding a single session reports no cohabitants. + solo = self.source("solo", "solo.jsonl") + self.assertEqual( + self.manager.get_sessions_sharing_file(str(solo), "solo"), set() + ) + + def test_sessions_sharing_file_excludes_hidden_and_agent_rows(self): + shared = self.root / "shared.jsonl" + shared.write_text("{}\n") + fid = self.db.execute( + "INSERT INTO cached_files(project_id,file_path) VALUES (1,?)", + (str(shared),), + ).lastrowid + # Visible cohabitant (trunk row). + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id) VALUES (1,?,?)", + (fid, "visible"), + ) + self.db.execute( + "INSERT INTO sessions(project_id,session_id,hidden) VALUES (1,?,0)", + ("visible",), + ) + # Hidden cohabitant and an agent (sidechain) row must both be ignored. + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id) VALUES (1,?,?)", + (fid, "hidden-one"), + ) + self.db.execute( + "INSERT INTO sessions(project_id,session_id,hidden) VALUES (1,?,1)", + ("hidden-one",), + ) + self.db.execute( + "INSERT INTO messages(project_id,file_id,session_id,_is_sidechain) " + "VALUES (1,?,?,1)", + (fid, "agent-sid"), + ) + self.assertEqual( + self.manager.get_sessions_sharing_file(str(shared), "self"), {"visible"} + ) + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/test/test_tui.py b/test/test_tui.py index e9a4267c..30ef07ff 100644 --- a/test/test_tui.py +++ b/test/test_tui.py @@ -157,6 +157,43 @@ async def test_load_sessions_from_cache(self, temp_project_dir): assert "session-123" in app.sessions assert app.sessions["session-123"].message_count == 2 + @pytest.mark.asyncio + async def test_archiving_last_jsonl_marks_session_archived(self): + """A cached session must be classified as archived, not current, once + the project's last source JSONL is gone, so the empty valid set still + routes it to archived_sessions (Restore available).""" + with tempfile.TemporaryDirectory() as temp_dir: + project_path = Path(temp_dir) # deliberately contains no JSONL files + app = SessionBrowser(project_path) + orphan = { + "orphan-session": SessionCacheData( + session_id="orphan-session", + first_timestamp="2025-01-01T10:00:00Z", + last_timestamp="2025-01-01T10:01:00Z", + message_count=1, + first_user_message="only in cache now", + ) + } + with ( + patch.object( + app.cache_manager, "get_cached_project_data" + ) as mock_cache, + patch.object(app.cache_manager, "get_modified_files", return_value=[]), + patch.object( + app.cache_manager, "get_existing_session_ids", return_value=set() + ), + patch.object( + app.cache_manager, "get_archived_sessions", return_value=orphan + ), + ): + mock_cache.return_value = Mock( + sessions=orphan, working_directories=[str(project_path)] + ) + async with app.run_test() as pilot: + await pilot.pause(0.1) + assert app.sessions == {} + assert "orphan-session" in app.archived_sessions + @pytest.mark.asyncio async def test_load_sessions_with_modified_files(self, temp_project_dir): """Test loading sessions when files have been modified since cache."""