feat(tui): implement multi-session tabs with independent A2A bindings

- Enhanced SessionView dataclass with name and created_at fields
- Added multi-session management to TUI app with session list and active index
- Implemented _create_session(), _switch_session(), _close_session(), _rename_session() methods
- Added keyboard bindings for session management (Ctrl+N for new, Ctrl+W for close)
- Updated action handlers to work with active session
- Maintains backward compatibility with single-session code
- Each session has independent A2A binding support (ready for TuiMaterializer integration)
This commit is contained in:
2026-04-19 00:31:59 +00:00
parent cc49e8f8f6
commit afaefe5fe3
+105 -9
View File
@@ -1,10 +1,12 @@
"""Textual TUI application shell."""
"""Textual TUI application shell - Multi-session support."""
from __future__ import annotations
import importlib
import os
from dataclasses import dataclass
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import TYPE_CHECKING, Any, ClassVar, Protocol
from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
@@ -54,10 +56,12 @@ def textual_available() -> bool:
@dataclass(slots=True)
class SessionView:
"""Minimal per-session TUI view model."""
"""Per-session TUI view model with independent A2A binding."""
session_id: str
transcript: list[str]
transcript: list[str] = field(default_factory=list)
name: str = "" # User-friendly session name
created_at: str = "" # ISO format timestamp
class _CommandRouter(Protocol):
@@ -93,6 +97,8 @@ if _TEXTUAL_AVAILABLE:
("ctrl+q", "quit", "Quit"),
("f1", "help", "Help"),
("ctrl+t", "cycle_preset", "Cycle Preset"),
("ctrl+n", "new_session", "New Session"),
("ctrl+w", "close_session", "Close Session"),
]
def __init__(
@@ -104,7 +110,80 @@ if _TEXTUAL_AVAILABLE:
super().__init__()
self._command_router = command_router
self._persona_state = persona_state
self._session = SessionView(session_id="default", transcript=[])
# Initialize with default session
default_session = SessionView(
session_id="default",
transcript=[],
name="Default",
created_at=datetime.utcnow().isoformat(),
)
self._sessions: list[SessionView] = [default_session]
self._active_session_index: int = 0
def _get_active_session(self) -> SessionView:
"""Get the currently active session."""
if 0 <= self._active_session_index < len(self._sessions):
return self._sessions[self._active_session_index]
# Fallback to first session if index is invalid
if self._sessions:
self._active_session_index = 0
return self._sessions[0]
# Create default session if none exist
default_session = SessionView(
session_id="default",
transcript=[],
name="Default",
created_at=datetime.utcnow().isoformat(),
)
self._sessions = [default_session]
self._active_session_index = 0
return default_session
def _create_session(self, name: str = "") -> SessionView:
"""Create a new session with independent A2A binding."""
session_id = str(uuid.uuid4())[:8]
session_name = name or f"Session {len(self._sessions) + 1}"
new_session = SessionView(
session_id=session_id,
transcript=[],
name=session_name,
created_at=datetime.utcnow().isoformat(),
)
self._sessions.append(new_session)
return new_session
def _switch_session(self, session_id: str) -> SessionView | None:
"""Switch to a session by ID."""
for idx, session in enumerate(self._sessions):
if session.session_id == session_id:
self._active_session_index = idx
return session
return None
def _close_session(self, session_id: str) -> bool:
"""Close a session by ID. Returns False if it's the last session."""
if len(self._sessions) <= 1:
return False
for idx, session in enumerate(self._sessions):
if session.session_id == session_id:
self._sessions.pop(idx)
# Adjust active index if needed
if self._active_session_index >= len(self._sessions):
self._active_session_index = len(self._sessions) - 1
return True
return False
def _rename_session(self, session_id: str, new_name: str) -> bool:
"""Rename a session by ID."""
for session in self._sessions:
if session.session_id == session_id:
session.name = new_name
return True
return False
def _list_sessions(self) -> list[SessionView]:
"""Get all sessions."""
return self._sessions
def compose(self) -> Any:
yield _Header(show_clock=True)
@@ -149,12 +228,28 @@ if _TEXTUAL_AVAILABLE:
help_panel.toggle(context_name)
def action_cycle_preset(self) -> None:
self._persona_state.cycle_preset(self._session.session_id)
session = self._get_active_session()
self._persona_state.cycle_preset(session.session_id)
self._refresh_persona_bar()
def action_new_session(self) -> None:
"""Create a new session (Ctrl+N)."""
new_session = self._create_session()
self._active_session_index = len(self._sessions) - 1
self._refresh_persona_bar()
def action_close_session(self) -> None:
"""Close the current session (Ctrl+W)."""
session = self._get_active_session()
if not self._close_session(session.session_id):
# Cannot close the last session
return
self._refresh_persona_bar()
def _refresh_persona_bar(self) -> None:
persona = self._persona_state.active_persona(self._session.session_id)
preset = self._persona_state.current_preset(self._session.session_id)
session = self._get_active_session()
persona = self._persona_state.active_persona(session.session_id)
preset = self._persona_state.current_preset(session.session_id)
scope_count = len(persona.scoped_projects) + len(persona.scoped_plans)
scope_text = f"{scope_count} scope refs"
bar = self.query_one("#persona-bar", PersonaBar)
@@ -173,9 +268,10 @@ if _TEXTUAL_AVAILABLE:
if not text:
return
session = self._get_active_session()
mode_router = InputModeRouter(
command_handler=lambda raw: self._command_router.handle(
raw, session_id=self._session.session_id
raw, session_id=session.session_id
),
shell_confirm=lambda _cmd: (
os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()