feat(tui): unify session history and context

nanobot/channels/websocket/tests/test_websocket_channel.py

@@ -4747,6 +4747,36 @@ def test_handle_webui_thread_get_returns_json(tmp_path, monkeypatch) -> None:
     assert body["has_pending_tool_calls"] is False
 
 
+@pytest.mark.asyncio
+async def test_handle_session_context_get_reads_detached_session() -> None:
+    from urllib.parse import quote
+
+    from websockets.datastructures import Headers
+    from websockets.http11 import Request
+
+    from nanobot.session import Session
+
+    session = Session(
+        key="websocket:context-route",
+        messages=[{"role": "user", "content": "hello"}],
+    )
+    manager = MagicMock()
+    manager.read_session_snapshot.return_value = session
+    gateway = _basic_handler(MagicMock(), session_manager=manager)
+    gateway.tokens.api_tokens["tok"] = time.monotonic() + 300.0
+    encoded = quote(session.key, safe="")
+    request = Request(
+        f"/api/sessions/{encoded}/context",
+        Headers([("Authorization", "Bearer tok")]),
+    )
+
+    response = await gateway.http._handle_session_context_get(request, encoded)
+
+    assert response.status_code == 200
+    assert json.loads(response.body.decode())["replay_messages"] == 1
+    manager.read_session_snapshot.assert_called_once_with(session.key)
+
+
 def test_handle_webui_thread_get_reports_registered_turn_as_pending(
     tmp_path,
     monkeypatch,

nanobot/cli/agent.py

@@ -48,7 +48,7 @@ console = Console()
 
 def agent(
     message: str | None = typer.Option(None, "--message", "-m", help="Message to send to the agent"),
-    session_id: str = typer.Option("cli:direct", "--session", "-s", help="Session ID"),
+    session_id: str | None = typer.Option(None, "--session", "-s", help="Session ID"),
     workspace: str | None = typer.Option(None, "--workspace", "-w", help="Workspace directory"),
     config: str | None = typer.Option(None, "--config", "-c", help="Path to config file"),
     markdown: bool = typer.Option(
@@ -111,6 +111,8 @@ def agent(
                 raise typer.Exit(exit_code)
             return
 
+    session_id = session_id or "cli:direct"
+
     try:
         provider = make_provider(runtime_config)
     except ValueError as exc:

nanobot/cli/tui_launcher.py

@@ -48,7 +48,7 @@ def launch_tui(
     *,
     config_path: Path,
     workspace_override: str | None,
-    session_id: str,
+    session_id: str | None,
     theme: str,
 ) -> int:
     """Run the native TUI, owning a local gateway only when one is not running."""
@@ -78,7 +78,9 @@ def launch_tui(
                 "NANOBOT_TUI_THEME": theme,
             }
         )
-        chat_id = _websocket_chat_id(session_id)
+        state_path = config_path.parent / "tui" / "state.json"
+        env["NANOBOT_TUI_STATE_PATH"] = str(state_path)
+        chat_id = _initial_tui_chat_id(session_id, state_path)
         if chat_id:
             env["NANOBOT_TUI_CHAT_ID"] = chat_id
         else:
@@ -328,3 +330,28 @@ def _websocket_chat_id(session_id: str) -> str | None:
     if session_id == "cli:direct":
         return "tui-direct"
     return session_id.split(":", 1)[-1] or None
+
+
+def _initial_tui_chat_id(session_id: str | None, state_path: Path) -> str | None:
+    """Resume the default TUI, while keeping an explicit selector authoritative."""
+    if session_id is not None:
+        return _websocket_chat_id(session_id)
+    return _read_tui_chat_id(state_path) or _websocket_chat_id("cli:direct")
+
+
+def _read_tui_chat_id(path: Path) -> str | None:
+    """Read the last attached chat without making launch depend on optional state."""
+    try:
+        raw_payload: Any = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(raw_payload, dict):
+        return None
+    payload = cast(dict[str, Any], raw_payload)
+    value = payload.get("chat_id")
+    if not isinstance(value, str):
+        return None
+    value = value.strip()
+    if not value or len(value) > 256 or any(character in value for character in "\r\n"):
+        return None
+    return value

nanobot/session/manager.py

@@ -1800,6 +1800,10 @@ class SessionManager:
         """Read a session without populating the cache."""
         return cast(dict[str, Any] | None, self._store.read(key))
 
+    def read_session_snapshot(self, key: str) -> Session | None:
+        """Load a detached session snapshot without populating the runtime cache."""
+        return self._store.load(key)
+
     def read_session_metadata(self, key: str) -> dict[str, Any] | None:
         """Read session metadata without loading the transcript."""
         return cast(dict[str, Any] | None, self._store.read_metadata(key))

nanobot/webui/session_context.py

@@ -0,0 +1,51 @@
+"""Read-only projection of the session material available to the agent."""
+
+from __future__ import annotations
+
+from typing import Any, cast
+
+from nanobot.session.manager import Session
+from nanobot.utils.helpers import estimate_message_tokens, truncate_text
+
+_SUMMARY_PREVIEW_CHARS = 4_000
+
+
+def session_context_payload(session: Session) -> dict[str, Any]:
+    """Return an explainable view of session replay without building a model prompt.
+
+    The final prompt also contains workspace instructions, memory, skills, and a
+    model-specific token budget.  This projection deliberately reports only the
+    session-owned part: archived summary plus the replayable raw suffix.
+    """
+    replay = session.get_history(max_messages=0, include_runtime_context=False)
+    raw_summary = session.metadata.get("_last_summary")
+    summary = ""
+    summary_preview = ""
+    summary_at: str | None = None
+    if isinstance(raw_summary, dict):
+        summary_data = cast(dict[str, object], raw_summary)
+        text = summary_data.get("text")
+        last_active = summary_data.get("last_active")
+        if isinstance(text, str):
+            summary = text.strip()
+            summary_preview = truncate_text(summary, _SUMMARY_PREVIEW_CHARS)
+        if isinstance(last_active, str):
+            summary_at = last_active
+
+    replay_tokens = sum(estimate_message_tokens(message) for message in replay)
+    summary_tokens = (
+        estimate_message_tokens({"role": "system", "content": summary}) if summary else 0
+    )
+
+    return {
+        "schema_version": 1,
+        "session_key": session.key,
+        "total_messages": len(session.messages),
+        "archived_messages": min(session.last_consolidated, len(session.messages)),
+        "replay_messages": len(replay),
+        "estimated_replay_tokens": replay_tokens,
+        "estimated_summary_tokens": summary_tokens,
+        "estimated_session_tokens": replay_tokens + summary_tokens,
+        "archived_summary": summary_preview or None,
+        "archived_summary_at": summary_at,
+    }

nanobot/webui/ws_http.py

@@ -97,6 +97,7 @@ from nanobot.webui.session_automations import (
     session_automation_jobs,
     session_automations_payload,
 )
+from nanobot.webui.session_context import session_context_payload
 from nanobot.webui.session_list_index import (
     WEBUI_SESSION_INDEX_INTERNAL_FIELDS,
     indexed_workspace_scope,
@@ -678,6 +679,10 @@ class GatewayHTTPHandler:
         if m:
             return self._handle_webui_thread_get(request, m.group(1))
 
+        m = re.match(r"^/api/sessions/([^/]+)/context$", got)
+        if m:
+            return await self._handle_session_context_get(request, m.group(1))
+
         m = re.match(r"^/api/sessions/([^/]+)/file-preview$", got)
         if m:
             return self._handle_file_preview(request, m.group(1))
@@ -692,6 +697,24 @@ class GatewayHTTPHandler:
 
         return None
 
+    async def _handle_session_context_get(self, request: WsRequest, key: str) -> Response:
+        if not self.check_api_token(request):
+            return _http_error(401, "Unauthorized")
+        decoded_key = _decode_api_key(key)
+        if decoded_key is None:
+            return _http_error(400, "invalid session key")
+        if not _is_websocket_channel_session_key(decoded_key):
+            return _http_error(404, "session not found")
+        if self.session_manager is None:
+            return _http_error(503, "session manager unavailable")
+        session = await asyncio.to_thread(
+            self.session_manager.read_session_snapshot,
+            decoded_key,
+        )
+        if session is None:
+            return _http_error(404, "session not found")
+        return _http_json_response(session_context_payload(session))
+
     async def _handle_sessions_list(self, request: WsRequest) -> Response:
         if not self.check_api_token(request):
             return _http_error(401, "Unauthorized")