tui_gateway+cli: session.list filters + session.peek + bare --resume picker sentinel

hermes_cli/_parser.py

@@ -145,8 +145,16 @@ def build_top_level_parser():
         "--resume",
         "-r",
         metavar="SESSION",
+        # nargs="?" + const=True: bare `--resume` parses to the sentinel True,
+        # which `hermes --tui` turns into the session picker
+        # (HERMES_TUI_RESUME=picker). `--resume <id|title>` is unchanged.
+        nargs="?",
+        const=True,
         default=None,
-        help="Resume a previous session by ID or title",
+        help=(
+            "Resume a previous session by ID or title. With --tui, bare "
+            "--resume (no argument) opens the session picker."
+        ),
     )
     parser.add_argument(
         "--continue",
@@ -294,8 +302,14 @@ def build_top_level_parser():
         "--resume",
         "-r",
         metavar="SESSION_ID",
+        # Same bare-flag picker sentinel as the top-level --resume.
+        nargs="?",
+        const=True,
         default=argparse.SUPPRESS,
-        help="Resume a previous session by ID (shown on exit)",
+        help=(
+            "Resume a previous session by ID (shown on exit). With --tui, "
+            "bare --resume opens the session picker."
+        ),
     )
     chat_parser.add_argument(
         "--continue",

hermes_cli/main.py

@@ -2000,7 +2000,8 @@ def _resolve_tui_heap_mb(default_mb: int = 8192) -> int:
 
 
 def _launch_tui(
-    resume_session_id: Optional[str] = None,
+    # str session id, the bare-`--resume` picker sentinel True, or None.
+    resume_session_id: "Optional[str | bool]" = None,
     tui_dev: bool = False,
     model: Optional[str] = None,
     provider: Optional[str] = None,
@@ -2019,6 +2020,14 @@ def _launch_tui(
     """Replace current process with the TUI."""
     tui_dir = PROJECT_ROOT / "ui-tui"
 
+    # Bare `--resume` arrives as the argparse sentinel True: open the TUI
+    # resume picker instead of resuming a specific session id. Normalize it
+    # here so everything downstream (exit summary, env forwarding) keeps
+    # seeing either a real session id string or None.
+    resume_picker = resume_session_id is True
+    if resume_picker:
+        resume_session_id = None
+
     import tempfile
 
     env = os.environ.copy()
@@ -2125,7 +2134,11 @@ def _launch_tui(
     # resolved for this invocation; direct `node ui-tui/dist/entry.js` users can
     # still set HERMES_TUI_RESUME themselves.
     env.pop("HERMES_TUI_RESUME", None)
-    if resume_session_id:
+    if resume_picker:
+        # Bare --resume: tell the TUI to open the resume picker before any
+        # session.create (create is lazy, so nothing is wasted).
+        env["HERMES_TUI_RESUME"] = "picker"
+    elif resume_session_id:
         env["HERMES_TUI_RESUME"] = resume_session_id
 
     argv, cwd = _make_tui_argv(tui_dir, tui_dev)
@@ -2234,6 +2247,18 @@ def cmd_chat(args):
     """Run interactive chat CLI."""
     use_tui = _resolve_use_tui(args)
 
+    # Bare `--resume` (argparse sentinel True) opens the TUI resume picker —
+    # `_launch_tui` translates it to HERMES_TUI_RESUME=picker. The classic
+    # REPL has no picker overlay, so point at the equivalents instead of
+    # silently resuming something the user didn't choose.
+    if getattr(args, "resume", None) is True and not use_tui:
+        print("Bare --resume opens the session picker, which requires the TUI.")
+        print(
+            "Use 'hermes --tui --resume', 'hermes --resume <id|title>', "
+            "'hermes -c', or 'hermes sessions browse'."
+        )
+        sys.exit(2)
+
     # Resolve --continue into --resume with the latest session or by name
     continue_val = getattr(args, "continue_last", None)
     if continue_val and not getattr(args, "resume", None):
@@ -2259,9 +2284,10 @@ def cmd_chat(args):
                 print(f"No previous {kind} session found to continue.")
                 sys.exit(1)
 
-    # Resolve --resume by title if it's not a direct session ID
+    # Resolve --resume by title if it's not a direct session ID. The bare
+    # picker sentinel (True) is not a name — leave it for _launch_tui.
     resume_val = getattr(args, "resume", None)
-    if resume_val:
+    if resume_val and resume_val is not True:
         resolved = _resolve_session_by_name_or_id(resume_val)
         if resolved:
             args.resume = resolved

tests/hermes_cli/test_tui_resume_flow.py

@@ -118,6 +118,82 @@ def test_cmd_chat_tui_resume_resolves_title_before_launch(monkeypatch, main_mod)
     assert captured["resume"] == "20260409_000000_aa11bb"
 
 
+def test_bare_resume_parses_to_picker_sentinel():
+    from hermes_cli._parser import build_top_level_parser
+
+    parser, _subparsers, _chat_parser = build_top_level_parser()
+
+    args = parser.parse_args(["--tui", "--resume"])
+    assert args.resume is True
+
+    args = parser.parse_args(["--resume", "abc123"])
+    assert args.resume == "abc123"
+
+    args = parser.parse_args(["chat", "--tui", "--resume"])
+    assert args.resume is True
+
+
+def test_cmd_chat_tui_bare_resume_skips_resolution_and_launches_picker(
+    monkeypatch, main_mod
+):
+    captured = {}
+
+    def fake_launch(resume_session_id=None, **kwargs):
+        captured["resume"] = resume_session_id
+        raise SystemExit(0)
+
+    def boom(_val):
+        raise AssertionError("bare --resume must not hit name/id resolution")
+
+    monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", boom)
+    monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
+
+    with pytest.raises(SystemExit):
+        main_mod.cmd_chat(_args(resume=True))
+
+    assert captured["resume"] is True
+
+
+def test_cmd_chat_bare_resume_without_tui_exits_with_guidance(
+    monkeypatch, capsys, main_mod
+):
+    monkeypatch.setattr(main_mod, "_resolve_use_tui", lambda args: False)
+    monkeypatch.setattr(
+        main_mod,
+        "_launch_tui",
+        lambda *a, **kw: pytest.fail("must not launch the TUI"),
+    )
+
+    with pytest.raises(SystemExit) as exc:
+        main_mod.cmd_chat(_args(tui=False, resume=True))
+
+    assert exc.value.code == 2
+    out = capsys.readouterr().out
+    assert "requires the TUI" in out
+    assert "hermes --tui --resume" in out
+
+
+def test_launch_tui_sets_picker_env_for_bare_resume(monkeypatch, main_mod):
+    captured = {}
+
+    monkeypatch.setenv("HERMES_TUI_RESUME", "stale-missing-session")
+    monkeypatch.setattr(
+        main_mod,
+        "_make_tui_argv",
+        lambda tui_dir, tui_dev: (["node", "dist/entry.js"], Path(".")),
+    )
+    monkeypatch.setattr(
+        main_mod.subprocess,
+        "call",
+        lambda argv, cwd=None, env=None: captured.update({"env": env}) or 1,
+    )
+
+    with pytest.raises(SystemExit):
+        main_mod._launch_tui(resume_session_id=True)
+
+    assert captured["env"]["HERMES_TUI_RESUME"] == "picker"
+
+
 def test_cmd_chat_tui_passes_model_and_provider(monkeypatch, main_mod):
     captured = {}
 

tests/test_tui_gateway_server.py

@@ -4816,6 +4816,351 @@ def test_session_most_recent_handles_db_unavailable(monkeypatch):
     assert resp["result"]["session_id"] is None
 
 
+# ── session.list (resume-picker filters + widened projection) ───────
+
+
+def _picker_row(sid, source, started, **extra):
+    row = {
+        "id": sid,
+        "source": source,
+        "title": f"title-{sid}",
+        "preview": f"preview-{sid}",
+        "started_at": started,
+        "last_active": started,
+        "message_count": 3,
+        "ended_at": None,
+        "cwd": None,
+        "model": None,
+    }
+    row.update(extra)
+    return row
+
+
+class _PickerDB:
+    """list_sessions_rich stand-in honouring the kwargs session.list maps."""
+
+    def __init__(self, rows):
+        self.rows = rows
+        self.calls = []
+
+    def list_sessions_rich(
+        self,
+        *,
+        source=None,
+        limit=20,
+        offset=0,
+        order_by_last_active=False,
+        id_query=None,
+    ):
+        self.calls.append(
+            {
+                "source": source,
+                "limit": limit,
+                "offset": offset,
+                "order_by_last_active": order_by_last_active,
+                "id_query": id_query,
+            }
+        )
+        rows = [dict(r) for r in self.rows]
+        if source:
+            rows = [r for r in rows if r.get("source") == source]
+        if id_query:
+            needle = id_query.strip().lower()
+            rows = [r for r in rows if needle in r["id"].lower()]
+        if order_by_last_active:
+            rows.sort(key=lambda r: r.get("last_active") or 0, reverse=True)
+        else:
+            rows.sort(key=lambda r: r.get("started_at") or 0, reverse=True)
+        return rows[offset : offset + limit]
+
+
+def _picker_db():
+    return _PickerDB(
+        [
+            _picker_row(
+                "tui-1",
+                "tui",
+                100,
+                cwd="/home/u/proj",
+                model="nous/hermes-4",
+                ended_at=150.0,
+            ),
+            _picker_row("cli-1", "cli", 90),
+            _picker_row("cron-1", "cron", 80),
+            _picker_row("cron-2", "cron", 70),
+            _picker_row("tg-1", "telegram", 60),
+            _picker_row("tool-1", "tool", 50),
+        ]
+    )
+
+
+def test_session_list_no_params_keeps_legacy_behavior_with_widened_rows(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request({"id": "1", "method": "session.list", "params": {}})
+
+    rows = resp["result"]["sessions"]
+    # Legacy semantics: started_at DESC, `tool` denied, one DB fetch.
+    assert [r["id"] for r in rows] == ["tui-1", "cli-1", "cron-1", "cron-2", "tg-1"]
+    assert db.calls == [
+        {
+            "source": None,
+            "limit": 400,
+            "offset": 0,
+            "order_by_last_active": False,
+            "id_query": None,
+        }
+    ]
+    # Widened projection, None-safe for rows missing the new columns.
+    first, second = rows[0], rows[1]
+    assert first["cwd"] == "/home/u/proj"
+    assert first["model"] == "nous/hermes-4"
+    assert first["ended_at"] == 150.0
+    assert first["last_active"] == 100
+    assert second["cwd"] is None
+    assert second["model"] is None
+    assert second["ended_at"] is None
+    assert second["last_active"] == 90
+    # Legacy keys are still present and unchanged.
+    assert second["title"] == "title-cli-1"
+    assert second["preview"] == "preview-cli-1"
+    assert second["message_count"] == 3
+    assert second["source"] == "cli"
+
+
+def test_session_list_single_source_passes_through_to_db(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.list", "params": {"sources": ["tui"]}}
+    )
+
+    assert [r["id"] for r in resp["result"]["sessions"]] == ["tui-1"]
+    assert db.calls[0]["source"] == "tui"  # pushed into SQL, not Python-filtered
+
+
+def test_session_list_multi_source_filters_gateway_side(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request(
+        {
+            "id": "1",
+            "method": "session.list",
+            "params": {"sources": ["cron", "telegram"]},
+        }
+    )
+
+    assert [r["id"] for r in resp["result"]["sessions"]] == [
+        "cron-1",
+        "cron-2",
+        "tg-1",
+    ]
+    assert db.calls[0]["source"] is None  # multi-source: DB scan + gateway filter
+
+
+def test_session_list_sources_tool_stays_denied(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.list", "params": {"sources": ["tool"]}}
+    )
+
+    assert resp["result"]["sessions"] == []
+
+
+def test_session_list_query_maps_to_id_query_on_last_active_path(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.list", "params": {"query": "cron"}}
+    )
+
+    assert [r["id"] for r in resp["result"]["sessions"]] == ["cron-1", "cron-2"]
+    assert db.calls[0]["id_query"] == "cron"
+    assert db.calls[0]["order_by_last_active"] is True
+
+
+def test_session_list_offset_limit_paginate(monkeypatch):
+    db = _picker_db()
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    def page(offset, limit):
+        resp = server.handle_request(
+            {
+                "id": "1",
+                "method": "session.list",
+                "params": {
+                    "sources": ["tui", "cli", "cron", "telegram"],
+                    "offset": offset,
+                    "limit": limit,
+                },
+            }
+        )
+        return [r["id"] for r in resp["result"]["sessions"]]
+
+    assert page(0, 2) == ["tui-1", "cli-1"]
+    assert page(2, 2) == ["cron-1", "cron-2"]
+    assert page(4, 2) == ["tg-1"]
+    assert page(6, 2) == []
+
+
+# ── session.peek (resume-picker Space preview) ───────────────────────
+
+
+class _PeekDB:
+    def __init__(self, session, messages):
+        self.session = session
+        self.messages = messages
+
+    def get_session(self, session_id):
+        if self.session and session_id == self.session["id"]:
+            return dict(self.session)
+        return None
+
+    def get_messages(self, session_id):
+        return [dict(m) for m in self.messages]
+
+
+def _peek_db():
+    session = {
+        "id": "sess-1",
+        "title": "picker demo",
+        "source": "tui",
+        "model": "nous/hermes-4",
+        "cwd": "/home/u/proj",
+        "started_at": 100.0,
+        "ended_at": 400.0,
+        "end_reason": "tui_shutdown",
+        "message_count": 6,
+        "actual_cost_usd": None,
+        "estimated_cost_usd": 0.42,
+    }
+    messages = [
+        {"id": 1, "role": "system", "content": "sys prompt", "timestamp": 100.0},
+        {"id": 2, "role": "user", "content": "first prompt", "timestamp": 110.0},
+        {"id": 3, "role": "assistant", "content": "first answer", "timestamp": 120.0},
+        {"id": 4, "role": "tool", "content": "tool output", "timestamp": 130.0},
+        {"id": 5, "role": "user", "content": "second prompt", "timestamp": 140.0},
+        {"id": 6, "role": "assistant", "content": "final answer", "timestamp": 150.0},
+    ]
+    return _PeekDB(session, messages)
+
+
+def test_session_peek_returns_metadata_and_head_tail(monkeypatch):
+    monkeypatch.setattr(server, "_get_db", _peek_db)
+
+    resp = server.handle_request(
+        {
+            "id": "1",
+            "method": "session.peek",
+            "params": {"session_id": "sess-1", "head": 1, "tail": 2},
+        }
+    )
+
+    result = resp["result"]
+    meta = result["session"]
+    assert meta == {
+        "id": "sess-1",
+        "title": "picker demo",
+        "source": "tui",
+        "model": "nous/hermes-4",
+        "cwd": "/home/u/proj",
+        "started_at": 100.0,
+        "ended_at": 400.0,
+        "end_reason": "tui_shutdown",
+        "message_count": 6,
+        "last_active": 150.0,  # last message timestamp, not ended_at
+        "cost_usd": 0.42,  # estimated fallback when actual is None
+    }
+    # Only displayable (user/assistant) messages, system/tool rows skipped.
+    assert [(m["role"], m["content"]) for m in result["head"]] == [
+        ("user", "first prompt")
+    ]
+    assert [(m["role"], m["content"]) for m in result["tail"]] == [
+        ("user", "second prompt"),
+        ("assistant", "final answer"),
+    ]
+    assert result["total_messages"] == 4
+    assert all(m["truncated"] is False for m in result["head"] + result["tail"])
+
+
+def test_session_peek_head_tail_never_overlap(monkeypatch):
+    db = _peek_db()
+    db.messages = db.messages[:3]  # system + user + assistant
+    monkeypatch.setattr(server, "_get_db", lambda: db)
+
+    resp = server.handle_request(
+        {
+            "id": "1",
+            "method": "session.peek",
+            "params": {"session_id": "sess-1", "head": 2, "tail": 2},
+        }
+    )
+
+    result = resp["result"]
+    head_ids = [m["id"] for m in result["head"]]
+    tail_ids = [m["id"] for m in result["tail"]]
+    assert head_ids == [2, 3]
+    assert tail_ids == []  # both displayable rows already consumed by head
+    assert result["total_messages"] == 2
+
+
+def test_session_peek_does_not_build_an_agent(monkeypatch):
+    spawned = []
+    monkeypatch.setattr(server, "_get_db", _peek_db)
+    monkeypatch.setattr(
+        server, "_make_agent", lambda *a, **kw: spawned.append("make") or None
+    )
+    monkeypatch.setattr(
+        server, "_start_agent_build", lambda *a, **kw: spawned.append("build")
+    )
+    before_sessions = dict(server._sessions)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.peek", "params": {"session_id": "sess-1"}}
+    )
+
+    assert "result" in resp
+    assert spawned == []
+    assert server._sessions == before_sessions  # no live session registered
+
+
+def test_session_peek_unknown_id_is_clean_error(monkeypatch):
+    monkeypatch.setattr(server, "_get_db", _peek_db)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.peek", "params": {"session_id": "nope"}}
+    )
+
+    assert resp["error"]["code"] == 4007
+    assert resp["error"]["message"] == "session not found"
+
+
+def test_session_peek_requires_session_id(monkeypatch):
+    monkeypatch.setattr(server, "_get_db", _peek_db)
+
+    resp = server.handle_request({"id": "1", "method": "session.peek", "params": {}})
+
+    assert resp["error"]["code"] == 4006
+
+
+def test_session_peek_db_unavailable(monkeypatch):
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+    monkeypatch.setattr(server, "_db_error", "locked")
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.peek", "params": {"session_id": "sess-1"}}
+    )
+
+    assert resp["error"]["code"] == 5046
+    assert "state.db unavailable" in resp["error"]["message"]
+
+
 # ── browser.manage ───────────────────────────────────────────────────
 
 

tui_gateway/server.py

@@ -3296,6 +3296,23 @@ def _(rid, params: dict) -> dict:
 
 @method("session.list")
 def _(rid, params: dict) -> dict:
+    """List stored sessions for the resume picker / sidebar.
+
+    Optional params — omitting all of them keeps the legacy behaviour
+    (most-recently-started first, ``tool`` rows denied, default limit 200):
+
+      - ``sources``: list of source tags to include (e.g. ``["cli", "tui"]``).
+        Powers the picker's tab strip. The ``tool`` deny-list still applies
+        on top. A single-element list is pushed into SQL
+        (``list_sessions_rich(source=...)``); multi-element lists are
+        filtered gateway-side over a bounded scan (the DB layer only takes a
+        single ``source`` string).
+      - ``query``: case-insensitive *session-id* substring filter, mapped to
+        ``list_sessions_rich(id_query=...)``. The DB applies ``id_query``
+        only on its order-by-last-active path, so query results are ordered
+        by most recent activity. Title/preview search stays client-side.
+      - ``offset`` / ``limit``: pagination over the filtered list.
+    """
     db = _get_db()
     if db is None:
         return _db_unavailable_error(rid, code=5006)
@@ -3310,16 +3327,67 @@ def _(rid, params: dict) -> dict:
         # platform is added or a user names their own source.
         deny = frozenset({"tool"})
 
-        limit = int(params.get("limit", 200) or 200)
-        # Over-fetch modestly so per-source filtering doesn't leave us
-        # short; the compression-tip projection in ``list_sessions_rich``
-        # can also merge rows.
-        fetch_limit = max(limit * 2, 200)
-        rows = [
-            s
-            for s in db.list_sessions_rich(source=None, limit=fetch_limit)
-            if (s.get("source") or "").strip().lower() not in deny
-        ][:limit]
+        try:
+            limit = max(1, int(params.get("limit", 200) or 200))
+        except (TypeError, ValueError):
+            limit = 200
+        try:
+            offset = max(0, int(params.get("offset", 0) or 0))
+        except (TypeError, ValueError):
+            offset = 0
+        query = str(params.get("query") or "").strip()
+        raw_sources = params.get("sources")
+        sources: list = []
+        if isinstance(raw_sources, (list, tuple)):
+            sources = [
+                str(s).strip().lower() for s in raw_sources if str(s).strip()
+            ]
+
+        if not sources and not query and offset == 0:
+            # Legacy path (no filter params) — byte-for-byte today's
+            # behaviour. Over-fetch modestly so per-source filtering doesn't
+            # leave us short; the compression-tip projection in
+            # ``list_sessions_rich`` can also merge rows.
+            fetch_limit = max(limit * 2, 200)
+            rows = [
+                s
+                for s in db.list_sessions_rich(source=None, limit=fetch_limit)
+                if (s.get("source") or "").strip().lower() not in deny
+            ][:limit]
+        else:
+            # Filtered/paginated path. Single source pushes into SQL; the
+            # deny-list and multi-source filter run gateway-side, so keep
+            # scanning DB pages until the requested window is full (bounded
+            # by a generous safety cap so a pathological DB can't pin us).
+            source_arg = sources[0] if len(sources) == 1 else None
+            allowed = frozenset(sources) if sources else None
+            wanted = offset + limit
+
+            def _eligible(row: dict) -> bool:
+                src = (row.get("source") or "").strip().lower()
+                if src in deny:
+                    return False
+                return allowed is None or src in allowed
+
+            collected: list = []
+            db_offset = 0
+            page = max(wanted * 2, 200)
+            while len(collected) < wanted and db_offset < 10_000:
+                batch = db.list_sessions_rich(
+                    source=source_arg,
+                    limit=page,
+                    offset=db_offset,
+                    # ``id_query`` only applies on the order-by-last-active
+                    # path; keep legacy started_at ordering when unfiltered.
+                    order_by_last_active=bool(query),
+                    id_query=query or None,
+                )
+                collected.extend(r for r in batch if _eligible(r))
+                if len(batch) < page:
+                    break
+                db_offset += page
+            rows = collected[offset : offset + limit]
+
         return _ok(
             rid,
             {
@@ -3331,6 +3399,14 @@ def _(rid, params: dict) -> dict:
                         "started_at": s.get("started_at") or 0,
                         "message_count": s.get("message_count") or 0,
                         "source": s.get("source") or "",
+                        # Picker row metadata (None-safe; older rows may
+                        # predate these columns).
+                        "cwd": s.get("cwd") or None,
+                        "last_active": s.get("last_active")
+                        or s.get("started_at")
+                        or 0,
+                        "ended_at": s.get("ended_at"),
+                        "model": s.get("model") or None,
                     }
                     for s in rows
                 ]
@@ -3340,6 +3416,112 @@ def _(rid, params: dict) -> dict:
         return _err(rid, 5006, str(e))
 
 
+@method("session.peek")
+def _(rid, params: dict) -> dict:
+    """DB-only preview of a stored session for the resume picker.
+
+    ``{session_id, head?, tail?}`` → session metadata plus the first ``head``
+    and last ``tail`` displayable messages (``user``/``assistant`` rows with
+    non-empty text content; tool spam and empty tool-call carriers are
+    skipped). Purely a read: no agent is constructed, no live session state
+    is created or switched — this powers the picker's Space preview, which
+    must stay cheap while the user scrolls.
+
+    Response shape::
+
+        {
+          "session": {id, title, source, model, cwd, started_at, ended_at,
+                      end_reason, message_count, last_active, cost_usd},
+          "head": [{id, role, content, truncated, timestamp}, ...],
+          "tail": [{...}],            # never overlaps head
+          "total_messages": <int>     # displayable (user/assistant) count
+        }
+    """
+    target = str(params.get("session_id") or "").strip()
+    if not target:
+        return _err(rid, 4006, "session_id required")
+    try:
+        head = max(0, int(params.get("head", 2) or 0))
+    except (TypeError, ValueError):
+        head = 2
+    try:
+        tail = max(0, int(params.get("tail", 2) or 0))
+    except (TypeError, ValueError):
+        tail = 2
+    db = _get_db()
+    if db is None:
+        return _db_unavailable_error(rid, code=5046)
+    try:
+        row = db.get_session(target)
+        if not row:
+            return _err(rid, 4007, "session not found")
+
+        msgs = db.get_messages(target)
+        last_active = (
+            (msgs[-1].get("timestamp") if msgs else None)
+            or row.get("started_at")
+            or 0
+        )
+
+        def _displayable(m: dict) -> bool:
+            if (m.get("role") or "") not in ("user", "assistant"):
+                return False
+            content = m.get("content")
+            if isinstance(content, str):
+                return bool(content.strip())
+            return bool(content)  # multimodal parts list
+
+        def _peek_msg(m: dict) -> dict:
+            content = m.get("content")
+            if not isinstance(content, str):
+                try:
+                    content = json.dumps(content, ensure_ascii=False)
+                except (TypeError, ValueError):
+                    content = str(content)
+            content = content or ""
+            return {
+                "id": m.get("id"),
+                "role": m.get("role") or "",
+                "content": content[:2000],
+                "truncated": len(content) > 2000,
+                "timestamp": m.get("timestamp"),
+            }
+
+        display = [m for m in msgs if _displayable(m)]
+        head_msgs = display[:head] if head else []
+        # Slice the remainder so head and tail never overlap, even when
+        # head + tail >= len(display).
+        tail_msgs = display[head:][-tail:] if tail else []
+
+        cost = row.get("actual_cost_usd")
+        if cost is None:
+            cost = row.get("estimated_cost_usd")
+
+        return _ok(
+            rid,
+            {
+                "session": {
+                    "id": row["id"],
+                    "title": row.get("title") or "",
+                    "source": row.get("source") or "",
+                    "model": row.get("model") or None,
+                    "cwd": row.get("cwd") or None,
+                    "started_at": row.get("started_at") or 0,
+                    "ended_at": row.get("ended_at"),
+                    "end_reason": row.get("end_reason"),
+                    "message_count": row.get("message_count") or 0,
+                    "last_active": last_active,
+                    "cost_usd": cost,
+                },
+                "head": [_peek_msg(m) for m in head_msgs],
+                "tail": [_peek_msg(m) for m in tail_msgs],
+                "total_messages": len(display),
+            },
+        )
+    except Exception as e:
+        return _err(rid, 5046, str(e))
+
+
 @method("session.most_recent")
 def _(rid, params: dict) -> dict:
     """Return the most recent human-facing session id, or ``None``.