feat: add /session command to list and switch between conversation sessions

  - Add session management functions in metadata.py with proper type annotations
    - list_sessions(): List all sessions with metadata
    - load_session(): Load and switch to a specific session
    - _get_session_info(): Extract session preview information
  - Add /session command (alias: /resume) for interactive session switching
  - Display sessions with preview (first 50 chars), relative time, and message count
  - Support full conversation history restoration when switching sessions
  - Handle edge cases: empty sessions, current session, user cancellation
  - Fix import issues after upstream refactoring (load_agents_md moved to soul.globals)

Author: YangLi-leo

src/kimi_cli/metadata.py

@@ -1,6 +1,7 @@
 import json
 from hashlib import md5
 from pathlib import Path
+from typing import Any, cast
 
 from pydantic import BaseModel, Field
 
@@ -52,3 +53,129 @@ def save_metadata(metadata: Metadata):
     logger.debug("Saving metadata to file: {file}", file=metadata_file)
     with open(metadata_file, "w", encoding="utf-8") as f:
         json.dump(metadata.model_dump(), f, indent=2, ensure_ascii=False)
+
+
+def load_session(work_dir: Path, session_id: str):
+    """Load a session by its id and set it to the current session."""
+    from kimi_cli.session import Session
+
+    logger.debug(
+        "Loading session: {session_id} for work directory: {work_dir}",
+        session_id=session_id,
+        work_dir=work_dir,
+    )
+
+    metadata = load_metadata()
+    work_dir_meta = next((wd for wd in metadata.work_dirs if wd.path == str(work_dir)), None)
+    if work_dir_meta is None:
+        logger.debug("Work directory never been used")
+        return None
+
+    history_file = work_dir_meta.sessions_dir / f"{session_id}.jsonl"
+    if not history_file.exists():
+        logger.warning("Session history file not found: {history_file}", history_file=history_file)
+        return None
+
+    work_dir_meta.last_session_id = session_id
+    save_metadata(metadata)
+
+    logger.info("Loaded session {session_id} and set as current", session_id=session_id)
+
+    return Session(id=session_id, work_dir=work_dir, history_file=history_file)
+
+
+def _get_session_info(session_file: Path) -> dict[str, Any]:
+    """Get session's meta info from the session file."""
+    from datetime import datetime
+
+    info: dict[str, Any] = {
+        "timestamp": datetime.fromtimestamp(session_file.stat().st_mtime),
+        "num_messages": 0,
+        "num_checkpoints": 0,
+        "first_message": None,
+    }
+
+    try:
+        with open(session_file, encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+
+                data = json.loads(line)
+                role = data.get("role")
+
+                if role == "_checkpoint":
+                    info["num_checkpoints"] += 1
+                elif role == "user" or role == "assistant":
+                    info["num_messages"] += 1
+
+                    # Use the first message as preview for the session
+                    if info["first_message"] is None and role == "user":
+                        content = data.get("content")
+                        if not content:
+                            continue
+
+                        text: str | None = None
+                        if isinstance(content, str):
+                            text = content
+                        elif isinstance(content, list):
+                            content_list = cast(list[Any], content)
+                            if len(content_list) > 0:
+                                first_part = content_list[0]
+                                if isinstance(first_part, str):
+                                    text = first_part
+                                elif isinstance(first_part, dict):
+                                    part_dict = cast(dict[str, Any], first_part)
+                                    if "text" in part_dict:
+                                        text = str(part_dict["text"])
+
+                        if text is None:
+                            continue
+
+                        # Skip injected system messages
+                        if text.startswith("<system>") and text.endswith("</system>"):
+                            continue
+                        info["first_message"] = text[:50].strip()
+    except Exception as e:
+        logger.warning(
+            "Failed to get session info from {session_file}: {e}", session_file=session_file, e=e
+        )
+    return info
+
+
+def list_sessions(work_dir: Path) -> list[tuple[str, Path, dict[str, Any]]]:
+    """List all the latest sessions for a work directory.
+
+    Returns:
+        List of tuples (session_id, session_file, info) sorted by timestamp descending.
+    """
+    logger.debug("Listing sessions for work directory: {work_dir}", work_dir=work_dir)
+
+    metadata = load_metadata()
+    work_dir_meta = next((wd for wd in metadata.work_dirs if wd.path == str(work_dir)), None)
+    if work_dir_meta is None:
+        logger.debug("Work directory never been used")
+        return []
+
+    sessions: list[tuple[str, Path, dict[str, Any]]] = []
+    sessions_dir = work_dir_meta.sessions_dir
+
+    # Since the revert to checkpoint will create a new session file,
+    # we need to list the most up-to-date files and sort them by timestamp.
+    for session_file in sessions_dir.glob("*.jsonl"):
+        if "_" in session_file.stem:
+            logger.debug("Skipping backup file: {session_file}", session_file=session_file)
+            continue
+
+        session_id = session_file.stem
+        info = _get_session_info(session_file)
+        info["is_current"] = session_id == work_dir_meta.last_session_id
+
+        sessions.append((session_id, session_file, info))
+
+    sessions.sort(key=lambda x: x[2]["timestamp"], reverse=True)
+
+    logger.debug("Found {num_sessions} sessions", num_sessions=len(sessions))
+
+    return sessions

src/kimi_cli/ui/shell/metacmd.py

@@ -1,10 +1,13 @@
+import getpass
 import tempfile
 import webbrowser
 from collections.abc import Awaitable, Callable, Sequence
+from datetime import datetime, timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING, NamedTuple, overload
 
 from kosong.base.message import Message
+from prompt_toolkit.shortcuts.choice_input import ChoiceInput
 from rich.panel import Panel
 
 import kimi_cli.prompts as prompts
@@ -13,8 +16,10 @@
 from kimi_cli.soul.message import system
 from kimi_cli.soul.runtime import load_agents_md
 from kimi_cli.ui.shell.console import console
+from kimi_cli.ui.shell.liveview import _LeftAlignedMarkdown
 from kimi_cli.utils.changelog import CHANGELOG, format_release_notes
 from kimi_cli.utils.logging import logger
+from kimi_cli.utils.message import message_extract_text
 
 if TYPE_CHECKING:
     from kimi_cli.ui.shell import ShellApp
@@ -255,6 +260,127 @@ async def compact(app: "ShellApp", args: list[str]):
     console.print("[green]✓[/green] Context has been compacted.")
 
 
+@meta_command(aliases=["resume"], kimi_soul_only=True)
+async def session(app: "ShellApp", args: list[str]):
+    """List all conversation sessions and switch to a previous one to continue"""
+    from kimi_cli.metadata import list_sessions
+
+    assert isinstance(app.soul, KimiSoul)
+    work_dir = app.soul._runtime.builtin_args.KIMI_WORK_DIR
+
+    session_list = list_sessions(work_dir)
+
+    if not session_list:
+        console.print("[yellow]No sessions found.[/yellow]")
+        return
+
+    # Build a list of choices for the user to select from
+    choices = []
+    session_map = {}  # Map display string to session_id
+    preview_width = 50  # Fix the width of the preview column
+
+    for _i, (session_id, _path, info) in enumerate(session_list, start=1):
+        # Build a relative time calculation
+        now = datetime.now()
+        time_diff = now - info["timestamp"]
+        if time_diff < timedelta(minutes=5):
+            time_str = "just now"
+        elif time_diff < timedelta(hours=1):
+            minutes = int(time_diff.total_seconds() / 60)
+            time_str = f"{minutes}m ago"
+        elif time_diff < timedelta(days=1):
+            hours = int(time_diff.total_seconds() / 3600)
+            time_str = f"{hours}h ago"
+        elif time_diff < timedelta(days=7):
+            days = time_diff.days
+            time_str = f"{days}d ago"
+        else:
+            time_str = info["timestamp"].strftime("%m-%d")
+
+        preview = info["first_message"] or "empty"
+        if len(preview) > preview_width - 3:
+            preview_display = preview[: preview_width - 3] + "..."
+        else:
+            preview_display = preview.ljust(preview_width)
+
+        marker = "→" if info["is_current"] else " "
+        label = f"{marker} {preview_display}  {time_str} · {info['num_messages']} msgs"
+        choices.append((label, label))
+        session_map[label] = (session_id, info)
+
+    try:
+        result = await ChoiceInput(
+            message="Select a session to switch to(↑↓ navigate, Enter select, Ctrl+C cancel):",
+            options=choices,
+        ).prompt_async()
+    except (EOFError, KeyboardInterrupt):
+        return
+
+    if result is None:
+        return
+
+    # Get session_id and info from the selected label
+    if result not in session_map:
+        console.print("[red]Invalid selection.[/red]")
+        return
+
+    session_id, session_info = session_map[result]
+
+    if session_info["is_current"]:
+        console.print("[yellow]You are already in this session.[/yellow]")
+        return
+
+    from kimi_cli.metadata import load_session
+    from kimi_cli.soul.context import Context
+
+    new_session = load_session(work_dir, session_id)
+    if not new_session:
+        console.print("[red]Failed to load session.[/red]")
+        return
+
+    new_context = Context(file_backend=new_session.history_file)
+    restored = await new_context.restore()
+
+    if not restored or not new_context.history:
+        console.print("[yellow]The session is empty.[/yellow]")
+        return
+
+    # Switch to the new session completely
+    app.soul._context = new_context
+
+    # Update runtime to point to the new session
+    new_runtime = app.soul._runtime._replace(session=new_session)
+    app.soul._runtime = new_runtime
+
+    console.clear()
+    _render_history(list(new_context.history))
+
+    # Confirm successful switch
+    console.print("[green]✓ Session switched successfully[/green]")
+    console.print()
+
+
+def _render_history(history: list[Message]):
+    """Render the history as a table to display"""
+    username = getpass.getuser()
+
+    for msg in history:
+        if msg.role == "user":
+            text = message_extract_text(msg)
+
+            # dmail will add some system messages, skip them
+            if text and not (text.startswith("<system>") and text.endswith("</system>")):
+                console.print(f"[bold]{username}✨[/bold] {text}")
+                console.print()
+
+        elif msg.role == "assistant":
+            text = message_extract_text(msg)
+            if text:
+                md = _LeftAlignedMarkdown(text, justify="left")
+                console.print(md)
+                console.print()
+
+
 from . import (  # noqa: E402
     debug,  # noqa: F401
     setup,  # noqa: F401