fix(session): resume latest session when last_session_id is missing

src/kimi_cli/session.py

@@ -286,7 +286,14 @@ async def list_all(cls) -> builtins.list[Session]:
 
     @staticmethod
     async def continue_(work_dir: KaosPath) -> Session | None:
-        """Get the last session for a work directory."""
+        """Get the last session for a work directory.
+
+        Prefers the explicitly recorded ``last_session_id``. Falls back to the
+        most recently updated non-empty session on disk when that id is missing
+        or no longer resolves: ``last_session_id`` is only persisted on a clean
+        (SUCCESS) exit, so it can be lost after a Ctrl-C/crash even though the
+        conversation history still exists on disk. See #2222.
+        """
         work_dir = work_dir.canonical()
         logger.debug("Continuing session for work directory: {work_dir}", work_dir=work_dir)
 
@@ -295,15 +302,32 @@ async def continue_(work_dir: KaosPath) -> Session | None:
         if work_dir_meta is None:
             logger.debug("Work directory never been used")
             return None
-        if work_dir_meta.last_session_id is None:
-            logger.debug("Work directory never had a session")
-            return None
 
-        logger.debug(
-            "Found last session for work directory: {session_id}",
-            session_id=work_dir_meta.last_session_id,
-        )
-        return await Session.find(work_dir, work_dir_meta.last_session_id)
+        if work_dir_meta.last_session_id is not None:
+            session = await Session.find(work_dir, work_dir_meta.last_session_id)
+            if session is not None:
+                logger.debug(
+                    "Found last session for work directory: {session_id}",
+                    session_id=work_dir_meta.last_session_id,
+                )
+                return session
+            logger.debug(
+                "Recorded last session {session_id} no longer exists, falling back",
+                session_id=work_dir_meta.last_session_id,
+            )
+
+        # Fall back to the most recently updated non-empty session. Session.list
+        # is already sorted by updated_at (desc) and filters out empty sessions.
+        sessions = await Session.list(work_dir)
+        if sessions:
+            logger.debug(
+                "Continuing most recent session for work directory: {session_id}",
+                session_id=sessions[0].id,
+            )
+            return sessions[0]
+
+        logger.debug("Work directory has no resumable session")
+        return None
 
 
 def _migrate_session_context_file(work_dir_meta: WorkDirMeta, session_id: str) -> None:

tests/core/test_session.py

@@ -115,6 +115,39 @@ async def test_continue_without_last_returns_none(isolated_share_dir: Path, work
     assert result is None
 
 
+async def test_continue_falls_back_to_latest_when_last_missing(
+    isolated_share_dir: Path, work_dir: KaosPath
+):
+    # last_session_id is only persisted on a clean exit, so after a Ctrl-C/crash
+    # it can be missing even though the conversation history still exists. In
+    # that case --continue should still resume the most recent session. See #2222.
+    older = await Session.create(work_dir)
+    newer = await Session.create(work_dir)
+    _write_context_message(older.context_file, "older conversation")
+    _write_context_message(newer.context_file, "newer conversation")
+    _write_wire_turn(older.dir, "older session")
+    _write_wire_turn(newer.dir, "newer session")
+
+    now = time.time()
+    os.utime(older.context_file, (now - 10, now - 10))
+    os.utime(newer.context_file, (now, now))
+
+    result = await Session.continue_(work_dir)
+    assert result is not None
+    assert result.id == newer.id
+
+
+async def test_continue_fallback_ignores_empty_sessions(
+    isolated_share_dir: Path, work_dir: KaosPath
+):
+    # An empty session (e.g. created then abandoned) must not be resumable.
+    empty = await Session.create(work_dir)
+    _write_wire_metadata(empty.dir)
+
+    result = await Session.continue_(work_dir)
+    assert result is None
+
+
 async def test_list_ignores_empty_sessions(isolated_share_dir: Path, work_dir: KaosPath):
     empty = await Session.create(work_dir)
     populated = await Session.create(work_dir)