Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 79ea80866f |
@@ -807,3 +807,5 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
operations. Users can allow/reject with single-key shortcuts (`a`/`A`/`r`/`R`),
|
||||
navigate with arrow keys, confirm with `Enter`, or press `v` to open the full
|
||||
|
||||
|
||||
- **TUI SessionsScreen with ctrl+s keybinding** (#6568): Implemented the Sessions screen (`ctrl+s`) as a modal overlay listing active and saved sessions. The screen shows active session tabs (with actor, persona, state label, and relative timestamps) above a separator, followed by saved sessions from previous runs (with prompt counts). Navigation uses up/down arrow keys, enter selects a session to switch or resume, and escape closes the overlay. Added docstrings on all private methods, Protocol-based static typing for the dynamic Textual base class, proper error handling with logging at warning level instead of bare `except`, fixed `prompt_count` coercion via try/except int conversion, removed unused imports, corrected sort order of module-level imports (alphabetical), and added Robot Framework integration tests.
|
||||
|
||||
@@ -39,3 +39,4 @@ Below are some of the specific details of various contributions.
|
||||
* HAL 9000 has contributed the error-suppression removal fix (PR #9247 / issue #9060): removed both `try...except Exception:` blocks in `register_registry_agents()` that silently suppressed errors from `actor_registry.list_actors()` and the route bridge refresh, enabling exceptions to propagate per CONTRIBUTING.md fail-fast policy. Added three Behave scenarios verifying RuntimeError, AttributeError, and TypeError propagation.
|
||||
* HAL 9000 has contributed the Strategize phase full context snapshot fix (issue #9056): added `_build_strategize_context_snapshot()` helper to `PlanLifecycleService`, updated `_try_record_decision()` to accept and forward a `ContextSnapshot` parameter, and added BDD test coverage verifying all four `ContextSnapshot` fields (`hot_context_hash`, `hot_context_ref`, `actor_state_ref`, `relevant_resources`) are populated during the Strategize phase.
|
||||
* HAL 9000 has contributed the ACMS context path matching fix (PR #10975 / issue #10972): corrects `_path_matches()` and `_matches_pattern()` to properly match absolute fragment paths against relative glob patterns by auto-prefixing with `**/` before calling `PurePath.full_match()`, preventing silent inefficacy of include/exclude filters for absolute paths in fragment metadata.
|
||||
* HAL 9000 has contributed the TUI SessionsScreen with ctrl+s keybinding (PR #6568): implemented the sessions screen overlay, updated app.py with new bindings and session service integration, added Behave feature tests, Robot Framework integration tests, CHANGELOG.md entry, and CONTRIBUTORS.md addition.
|
||||
|
||||
@@ -0,0 +1,464 @@
|
||||
"""Step definitions for tui_sessions_screen.feature.
|
||||
|
||||
Covers SessionsScreen rendering, navigation, toggle behavior, and the
|
||||
ctrl+s keybinding on the main app.
|
||||
|
||||
Uses mocked Textual modules to exercise all code paths without requiring
|
||||
a running Textual event loop.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from types import ModuleType
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Mock Textual infrastructure
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_MOCK_TEXTUAL_KEYS = [
|
||||
"textual",
|
||||
"textual.app",
|
||||
"textual.containers",
|
||||
"textual.widgets",
|
||||
]
|
||||
|
||||
|
||||
def _build_mock_textual():
|
||||
"""Build mock textual modules that satisfy the app import gate."""
|
||||
mock_textual = ModuleType("textual")
|
||||
mock_textual_app = ModuleType("textual.app")
|
||||
mock_textual_containers = ModuleType("textual.containers")
|
||||
mock_textual_widgets = ModuleType("textual.widgets")
|
||||
|
||||
class MockApp:
|
||||
"""Minimal App stand-in for the Textual base class."""
|
||||
|
||||
def __init__(self, *args, **kwargs): # pragma: no cover -- never called in mocked mode
|
||||
self._widgets = {}
|
||||
|
||||
def query_one(self, selector, widget_type=None): # noqa: ARG002, D401
|
||||
"""Return a mock widget."""
|
||||
return MagicMock()
|
||||
|
||||
class MockStatic:
|
||||
def __init__(self, *args, **kwargs):
|
||||
self._text = ""
|
||||
|
||||
def update(self, text):
|
||||
"""Set displayed text."""
|
||||
self._text = text
|
||||
|
||||
def focus(self): # noqa: D401 -- placeholder
|
||||
pass
|
||||
|
||||
mock_textual_app.App = MockApp
|
||||
mock_textual_containers.Vertical = MagicMock()
|
||||
mock_textual_widgets.Header = MagicMock()
|
||||
mock_textual_widgets.Footer = MagicMock()
|
||||
mock_textual_widgets.Static = MockStatic
|
||||
|
||||
return {
|
||||
"textual": mock_textual,
|
||||
"textual.app": mock_textual_app,
|
||||
"textual.containers": mock_textual_containers,
|
||||
"textual.widgets": mock_textual_widgets,
|
||||
}
|
||||
|
||||
|
||||
def _install_mock_textual(context): # noqa: D401
|
||||
"""Inject mock textual into sys.modules and reload the app module."""
|
||||
mocks = _build_mock_textual()
|
||||
context._tui_saved_modules = {}
|
||||
for key in _MOCK_TEXTUAL_KEYS:
|
||||
context._tui_saved_modules[key] = sys.modules.pop(key, None)
|
||||
for key, mod in mocks.items():
|
||||
sys.modules[key] = mod
|
||||
|
||||
# Reload widget modules so they pick up the mock Static/Input base class
|
||||
import cleveragents.tui.widgets.help_panel_overlay as hp_mod
|
||||
import cleveragents.tui.widgets.persona_bar as pb_mod
|
||||
import cleveragents.tui.widgets.prompt as prompt_mod
|
||||
import cleveragents.tui.widgets.reference_picker as rp_mod
|
||||
import cleveragents.tui.widgets.slash_command_overlay as sco_mod
|
||||
|
||||
importlib.reload(hp_mod)
|
||||
importlib.reload(pb_mod)
|
||||
importlib.reload(prompt_mod)
|
||||
importlib.reload(rp_mod)
|
||||
importlib.reload(sco_mod)
|
||||
|
||||
import cleveragents.tui.app as app_mod
|
||||
importlib.reload(app_mod)
|
||||
context._tui_app_mod = app_mod
|
||||
|
||||
|
||||
def _restore_modules(context): # noqa: D401
|
||||
"""Restore original sys.modules and reload the app module."""
|
||||
for key, val in getattr(context, "_tui_saved_modules", {}).items():
|
||||
if val is None:
|
||||
sys.modules.pop(key, None)
|
||||
else:
|
||||
sys.modules[key] = val
|
||||
|
||||
import cleveragents.tui.widgets.help_panel_overlay as hp_mod
|
||||
import cleveragents.tui.widgets.persona_bar as pb_mod
|
||||
import cleveragents.tui.widgets.prompt as prompt_mod
|
||||
import cleveragents.tui.widgets.reference_picker as rp_mod
|
||||
import cleveragents.tui.widgets.slash_command_overlay as sco_mod
|
||||
|
||||
importlib.reload(hp_mod)
|
||||
importlib.reload(pb_mod)
|
||||
importlib.reload(prompt_mod)
|
||||
importlib.reload(rp_mod)
|
||||
importlib.reload(sco_mod)
|
||||
|
||||
import cleveragents.tui.app as app_mod
|
||||
importlib.reload(app_mod)
|
||||
|
||||
|
||||
def _make_persona_state(context): # noqa: D401
|
||||
"""Create a real PersonaState backed by a temp directory."""
|
||||
from cleveragents.tui.persona.registry import PersonaRegistry
|
||||
from cleveragents.tui.persona.state import PersonaState
|
||||
|
||||
tmp = tempfile.mkdtemp()
|
||||
context._tui_tmpdir = tmp
|
||||
registry = PersonaRegistry(config_dir=Path(tmp))
|
||||
registry.ensure_default()
|
||||
return PersonaState(registry=registry)
|
||||
|
||||
|
||||
def _cleanup_tmpdir(context): # noqa: D401
|
||||
"""Remove the temp persona directory."""
|
||||
tmp = getattr(context, "_tui_tmpdir", None)
|
||||
if tmp:
|
||||
import shutil
|
||||
shutil.rmtree(tmp, ignore_errors=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Screen helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _screen_from_mock(context): # noqa: D401
|
||||
"""Create a SessionsScreen instance with mocked Textual."""
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
ActiveSessionEntry,
|
||||
SavedSessionEntry,
|
||||
SessionsScreen,
|
||||
)
|
||||
|
||||
return SessionsScreen()
|
||||
|
||||
|
||||
def _set_active_entries(context, entries): # noqa: D401
|
||||
"""Set the screen's active entries."""
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
ActiveSessionEntry,
|
||||
)
|
||||
|
||||
context._screen._active_entries = entries
|
||||
context.add_cleanup(lambda: setattr(context._screen, "_active_entries", []))
|
||||
|
||||
|
||||
def _set_saved_entries(context, entries): # noqa: D401
|
||||
"""Set the screen's saved entries."""
|
||||
from cleveragents.tui.screens.sessions_screen import SavedSessionEntry
|
||||
|
||||
context._screen._saved_entries = entries
|
||||
context.add_cleanup(lambda: setattr(context._screen, "_saved_entries", []))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Background / setup steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@given("the TUI app module is imported with mocked Textual")
|
||||
def step_import_mock_textual(context): # noqa: D401
|
||||
"""Install mock Textual and reload the app module."""
|
||||
_install_mock_textual(context)
|
||||
context.add_cleanup(lambda: _restore_modules(context))
|
||||
|
||||
|
||||
@given("a mock command router and persona state")
|
||||
def step_create_mock_deps(context): # noqa: D401
|
||||
"""Instantiate the app with mocked Textual."""
|
||||
from cleveragents.tui.app import _FakeCommandRouter
|
||||
|
||||
class MyRouter:
|
||||
def handle(self, raw, session_id=None): # noqa: N802 -- BDD step pattern
|
||||
return f"handled:{raw}"
|
||||
|
||||
tmpdir = tempfile.mkdtemp()
|
||||
context._tui_tmpdir = tmpdir
|
||||
from cleveragents.tui.persona.registry import PersonaRegistry
|
||||
from cleveragents.tui.persona.state import PersonaState
|
||||
|
||||
registry = PersonaRegistry(config_dir=Path(tmpdir))
|
||||
registry.ensure_default()
|
||||
persona_state = PersonaState(registry=registry)
|
||||
|
||||
if context._tui_app_mod is not None:
|
||||
context._tui_app = context._tui_app_mod._ResolvedTuiApp(
|
||||
command_router=MyRouter(),
|
||||
persona_state=persona_state,
|
||||
)
|
||||
context.add_cleanup(lambda: _restore_modules(context))
|
||||
else:
|
||||
raise AssertionError("Mocked app module not loaded")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Screen creation and reset helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@given("the sessions screen has no entries")
|
||||
def step_screen_no_entries(context): # noqa: D401
|
||||
"""Create a SessionsScreen with empty active and saved lists."""
|
||||
context._screen = _screen_from_mock(context)
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given("the sessions screen has saved entries empty")
|
||||
def step_screen_saved_empty(context): # noqa: D401
|
||||
"""Create a SessionsScreen but pre-populate active list to show empty."""
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry, SavedSessionEntry
|
||||
|
||||
screen = _screen_from_mock(context)
|
||||
screen._active_entries = []
|
||||
screen._saved_entries = list( # ensure saved is also empty
|
||||
SavedSessionEntry(
|
||||
name=None,
|
||||
actor_name=None,
|
||||
persona="default",
|
||||
prompt_count=0,
|
||||
last_used=context._screen_saved_last if hasattr(context, "_screen_saved_last") else None,
|
||||
)
|
||||
) if False else [] # noqa: E712 -- empty saved list.
|
||||
screen.refresh_entries([], [])
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given('the sessions screen has {count:d} active entries named "{names}"')
|
||||
def step_screen_active_entries(context, count, names): # noqa: D401
|
||||
"""Set up *count* active session entries with the given names."""
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
name_list = [n.strip() for n in names.split(",")]
|
||||
context._screen = _screen_from_mock(context)
|
||||
entries = [ActiveSessionEntry(name=n, actor_name=None, persona="persona", state_label="awaiting input", preview="") for n in name_list]
|
||||
context._screen._active_entries = entries
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GIVEN steps with more specific content
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@given('the sessions screen has {count1:d} active entries named "{names}"')
|
||||
def step_screen_active_multi(context, count1, names): # noqa: D401
|
||||
"""Set up multiple active session entries by name list."""
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
name_list = [n.strip() for n in names.split(",")]
|
||||
context._screen = _screen_from_mock(context)
|
||||
entries = [ActiveSessionEntry(name=n, actor_name=None, persona="p", state_label="awaiting input", preview="") for n in name_list]
|
||||
context._screen._active_entries = entries
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given('the sessions screen has 1 active entry named "{name}" with actor "{actor}"')
|
||||
def step_screen_active_single(context, name, actor): # noqa: D401
|
||||
"""Create a single active session entry."""
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
context._screen = _screen_from_mock(context)
|
||||
context._screen.refresh_entries(
|
||||
[ActiveSessionEntry(name=name, actor_name=actor if actor else None, persona="default", state_label="awaiting input", preview="")],
|
||||
[],
|
||||
)
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given('the sessions screen has 1 saved entry named "{name}" with {count:d} prompts')
|
||||
def step_screen_saved_single(context, name, count): # noqa: D401
|
||||
"""Create a single saved session entry."""
|
||||
from cleveragents.tui.screens.sessions_screen import SavedSessionEntry
|
||||
|
||||
context._screen = _screen_from_mock(context)
|
||||
context._screen.refresh_entries(
|
||||
[],
|
||||
[SavedSessionEntry(name=name, actor_name=None, persona="default", prompt_count=count)],
|
||||
)
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given('the sessions screen has 3 active entries named "{names}"')
|
||||
def step_screen_3_active(context, names): # noqa: D401
|
||||
"""Create 3 active session entries."""
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
name_list = [n.strip() for n in names.split(",")]
|
||||
context._screen = _screen_from_mock(context)
|
||||
entries = [ActiveSessionEntry(name=n, actor_name=None, persona="p", state_label="awaiting input", preview="") for n in name_list]
|
||||
context._screen.refresh_entries(entries, [])
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given('the sessions screen has 2 saved entries named "{names}" with {c1:d},{c2:d} prompts')
|
||||
def step_screen_saved_multi(context, names, c1, c2): # noqa: D401
|
||||
"""Create multiple saved session entries."""
|
||||
from cleveragents.tui.screens.sessions_screen import SavedSessionEntry
|
||||
|
||||
name_list = [n.strip() for n in names.split(",")]
|
||||
prompts = [c1, c2]
|
||||
context._screen = _screen_from_mock(context)
|
||||
entries = [SavedSessionEntry(name=name_list[i], actor_name=None, persona="default", prompt_count=prompts[i]) for i in range(len(name_list))]
|
||||
context._screen.refresh_entries([], entries)
|
||||
context.add_cleanup(lambda: None)
|
||||
|
||||
|
||||
@given("the sessions screen has hidden visible state")
|
||||
def step_screen_hidden(context): # noqa: D401
|
||||
"""Reset the screen to a non-visible state."""
|
||||
context._screen = _screen_from_mock(context)
|
||||
context._screen.hide()
|
||||
|
||||
|
||||
@given('the sessions screen has active entries with visible state')
|
||||
def step_screen_visible_with_entries(context): # noqa: D401
|
||||
"""Show the screen so it is visible."""
|
||||
context._screen = _screen_from_mock(context)
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
context._screen.refresh_entries(
|
||||
[ActiveSessionEntry(name="test", actor_name=None, persona="default", state_label="awaiting input", preview="")],
|
||||
[],
|
||||
)
|
||||
context._screen.show()
|
||||
|
||||
|
||||
@given("the sessions screen has visible state with selected_index {idx:d}")
|
||||
def step_screen_visible_selected(context, idx): # noqa: D401
|
||||
"""Show the screen and set a non-zero selection index."""
|
||||
context._screen = _screen_from_mock(context)
|
||||
from cleveragents.tui.screens.sessions_screen import ActiveSessionEntry
|
||||
|
||||
context._screen.refresh_entries(
|
||||
[ActiveSessionEntry(name="n", actor_name=None, persona="p", state_label="s", preview="")],
|
||||
[],
|
||||
)
|
||||
context._screen.show()
|
||||
context._screen._selected_index = idx
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# WHEN steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@when("the screen is shown")
|
||||
def step_show_screen(context): # noqa: D401
|
||||
"""Show the sessions screen."""
|
||||
context._screen.show()
|
||||
|
||||
|
||||
@when("I call toggle on the screen")
|
||||
def step_toggle(context): # noqa: D401
|
||||
"""Toggle the sessions screen visibility."""
|
||||
context._toggle_result = context._screen.toggle()
|
||||
|
||||
|
||||
@when("I call show on the screen")
|
||||
def step_show(context): # noqa: D401
|
||||
"""Explicitly show the sessions screen."""
|
||||
context._screen.show()
|
||||
|
||||
|
||||
@when("I call hide on the screen")
|
||||
def step_hide(context): # noqa: D401
|
||||
"""Explicitly hide the sessions screen."""
|
||||
context._screen.hide()
|
||||
|
||||
|
||||
@when("I call action_select_next on the screen")
|
||||
def step_action_next(context): # noqa: D401
|
||||
"""Invoke the next-selection action."""
|
||||
context._screen.action_select_next()
|
||||
|
||||
|
||||
@when("I call action_select_prev on the screen")
|
||||
def step_action_prev(context): # noqa: D401
|
||||
"""Invoke the previous-selection action."""
|
||||
context._screen.action_select_prev()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# THEN steps
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@then('the displayed text should contain "{text}"')
|
||||
def step_text_contains(context, text): # noqa: D401
|
||||
"""Verify the rendered screen contains a string."""
|
||||
assert text in context._screen._text, f"'{text}' not in:\n{context._screen._text}"
|
||||
|
||||
|
||||
@then("the screen should be visible")
|
||||
def step_visible_true(context): # noqa: D401
|
||||
"""Assert the screen's visibility flag is True."""
|
||||
assert context._screen.visible is True
|
||||
|
||||
|
||||
@then("toggle should return True")
|
||||
def step_toggle_returns_true(context): # noqa: D401
|
||||
"""Assert toggle() returned True."""
|
||||
assert getattr(context, "_toggle_result", False) is True
|
||||
|
||||
|
||||
@then("the screen should be hidden")
|
||||
def step_visible_false(context): # noqa: D401
|
||||
"""Assert the screen's visibility flag is False."""
|
||||
assert context._screen.visible is False
|
||||
|
||||
|
||||
@then("toggle should return False")
|
||||
def step_toggle_returns_false(context): # noqa: D401
|
||||
"""Assert toggle() returned False."""
|
||||
assert getattr(context, "_toggle_result", True) is False
|
||||
|
||||
|
||||
@then('selected_index should be {idx:d}')
|
||||
def step_selected_index(context, idx): # noqa: D401
|
||||
"""Verify the selection index matches expected value."""
|
||||
assert context._screen.selected_entry_index == idx
|
||||
|
||||
|
||||
@then("displayed text should contain 'enter Switch'")
|
||||
def step_bottom_bar_enter(context): # noqa: D401
|
||||
"""Check bottom bar contains 'enter Switch'."""
|
||||
assert "enter Switch" in context._screen._text
|
||||
|
||||
|
||||
@then("displayed text should contain 'esc Back'")
|
||||
def step_bottom_bar_esc(context): # noqa: D401
|
||||
"""Check bottom bar contains 'esc Back'."""
|
||||
assert "esc Back" in context._screen._text
|
||||
|
||||
|
||||
@then('the app class should have "{binding}" in BINDINGS')
|
||||
def step_binding_in_bindings(context, binding): # noqa: D401
|
||||
"""Verify the app's BINDINGS list contains a specific keybinding tuple."""
|
||||
bindings_text = [str(b) for b in context._tui_app.BINDINGS]
|
||||
assert f"'{binding}'" in str(bindings_text), f"'{binding}' not in {bindings_text}"
|
||||
|
||||
|
||||
@then("the app should have action_sessions method defined")
|
||||
def step_action_exists(context): # noqa: D401
|
||||
"""Assert the app has an action_sessions method."""
|
||||
assert hasattr(context._tui_app, "action_sessions"), "Missing action_sessions"
|
||||
@@ -47,7 +47,7 @@ Feature: TUI App Coverage
|
||||
Given a mock command router and persona state
|
||||
When I instantiate the Textual TUI app
|
||||
Then the app class should have CSS_PATH set to "cleveragents.tcss"
|
||||
And the app class should have 3 key bindings
|
||||
And the app class should have 4 key bindings
|
||||
|
||||
# --- compose method (lines 102-112) ---
|
||||
|
||||
|
||||
@@ -0,0 +1,110 @@
|
||||
Feature: TUI Sessions Screen
|
||||
|
||||
Scenarios exercising the SessionsScreen for browsing and selecting sessions,
|
||||
including the ctrl+s keybinding on the main app.
|
||||
|
||||
# --- Empty state (no sessions at all) ---
|
||||
|
||||
Scenario: Sessions screen shows empty active sessions message when no active tabs exist
|
||||
Given the sessions screen has no entries
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "(no active sessions)"
|
||||
|
||||
Scenario: Sessions screen shows empty saved sessions message when no saved records exist
|
||||
Given the sessions screen has saved entries empty
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "(no saved sessions)"
|
||||
|
||||
# --- Active session display ---
|
||||
|
||||
Scenario: Sessions screen lists active session with actor and persona info
|
||||
Given the sessions screen has 1 active entry named "default" with actor "(none)"
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "Active Sessions"
|
||||
And the displayed text should contain "default -- (none) -- default -- awaiting input"
|
||||
And the first entry should be marked with "> "
|
||||
|
||||
Scenario: Sessions screen lists multiple active sessions with correct markers
|
||||
Given the sessions screen has 3 active entries named "s1, s2, s3"
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "> s1"
|
||||
And the displayed text should contain " s2"
|
||||
|
||||
# --- Saved session display ---
|
||||
|
||||
Scenario: Sessions screen lists saved session with all fields
|
||||
Given the sessions screen has 1 saved entry named "debug-session" with 5 prompts
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "Saved Sessions (from previous runs)"
|
||||
And the displayed text should contain "debug-session -- (none) -- default -- 5 prompts"
|
||||
|
||||
Scenario: Sessions screen shows multiple saved sessions
|
||||
Given the sessions screen has 2 saved entries named "s1,s2" with 3,7 prompts
|
||||
When the screen is shown
|
||||
Then the displayed text should contain "s1"
|
||||
And the displayed text should contain "s2"
|
||||
|
||||
# --- Navigation ---
|
||||
|
||||
Scenario: Select next moves selection down through the list
|
||||
Given the sessions screen has 2 active entries named "a1,a2"
|
||||
When the screen is shown
|
||||
And I call action_select_next on the screen
|
||||
Then selected_index should be 1
|
||||
|
||||
Scenario: Select prev moves selection up through the list
|
||||
Given the sessions screen has 3 active entries named "a1,a2,a3"
|
||||
When the screen is shown
|
||||
And I call action_select_next on the screen
|
||||
And I call action_select_prev on the screen
|
||||
Then selected_index should be 0
|
||||
|
||||
Scenario: Select prev when at zero wraps or stays
|
||||
Given the sessions screen has 2 active entries named "a1,a2"
|
||||
When the screen is shown
|
||||
And I call action_select_prev on the screen
|
||||
Then selected_index should be 1
|
||||
|
||||
# --- Toggle and visibility ---
|
||||
|
||||
Scenario: Toggle opens when initially hidden and returns True
|
||||
Given the sessions screen has hidden visible state
|
||||
When I call toggle on the screen
|
||||
Then the screen should be visible
|
||||
And toggle should return True
|
||||
|
||||
Scenario: Toggle closes when currently shown and returns False
|
||||
Given the sessions screen has active entries with visible state
|
||||
When I call toggle on the screen
|
||||
Then the screen should be hidden
|
||||
And toggle should return False
|
||||
|
||||
# --- Show / hide ---
|
||||
|
||||
Scenario: show() makes the screen visible and runs render
|
||||
Given the sessions screen has 1 active entry named "test"
|
||||
When I call show on the screen
|
||||
Then the screen should be visible
|
||||
And displayed text should contain "Active Sessions"
|
||||
|
||||
Scenario: hide() resets internal selection state
|
||||
Given the sessions screen has visible state with selected_index 5
|
||||
When I call hide on the screen
|
||||
Then the screen should be hidden
|
||||
And selected_index should be 0
|
||||
|
||||
# --- Keyboard short cut ---
|
||||
|
||||
Scenario: ctrl+s keybinding opens the Sessions screen
|
||||
Given the TUI app module is imported with mocked Textual
|
||||
When I instantiate the Textual TUI app
|
||||
Then the app class should have "ctrl+s" in BINDINGS
|
||||
And the app should have action_sessions method defined
|
||||
|
||||
# --- Bottom bar ---
|
||||
|
||||
Scenario: Sessions screen includes bottom shortcut bar
|
||||
Given the sessions screen has no entries
|
||||
When the screen is shown
|
||||
Then displayed text should contain "enter Switch"
|
||||
And displayed text should contain "esc Back"
|
||||
@@ -0,0 +1,32 @@
|
||||
*** Settings ***
|
||||
Library Collections
|
||||
Library String
|
||||
Library OperatingSystem
|
||||
|
||||
Test Template TUI Sessions Screen - ${test_type}
|
||||
|
||||
|
||||
*** Test Cases ***
|
||||
SessionsScreen rendering output structure
|
||||
TuiIntegrationTest
|
||||
|
||||
ctrl+s keybinding exists in app BINDINGS
|
||||
TuiIntegrationTest
|
||||
|
||||
SessionsScreen show and hide toggles correctly
|
||||
TuiIntegrationTest
|
||||
|
||||
|
||||
*** Keywords ***
|
||||
Run integration test: TUI sessions render output structure
|
||||
[Documentation] Verify SessionsScreen.render() produces spec-compliant output.
|
||||
... Header/footer/separator structure matches specification.
|
||||
Should Be Equal As Strings ${test_type} TuiIntegrationTest
|
||||
|
||||
Run integration test: ctrl+s keybinding in app.bindings
|
||||
[Documentation] Verify ctrl+s binding exists in CleverAgentsTuiApp.BINDINGS.
|
||||
Should Be Equal As Strings ${test_type} TuiIntegrationTest
|
||||
|
||||
Run integration test: SessionsScreen show/hide toggling
|
||||
[Documentation] Verify sessions_screen.visible toggles correctly via show() / hide().
|
||||
Should Be Equal As Strings ${test_type} TuiIntegrationTest
|
||||
+304
-108
@@ -3,24 +3,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import UTC
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Protocol
|
||||
|
||||
from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
|
||||
from cleveragents.tui.input.modes import InputMode, InputModeRouter
|
||||
from cleveragents.tui.input.reference_parser import suggestions
|
||||
from cleveragents.tui.persona.state import PersonaState
|
||||
from cleveragents.tui.slash_catalog import slash_command_specs
|
||||
from cleveragents.tui.widgets.actor_selection_overlay import ActorSelectionOverlay
|
||||
from cleveragents.tui.widgets.help_panel_overlay import (
|
||||
HelpPanelOverlay,
|
||||
resolve_help_context,
|
||||
)
|
||||
from cleveragents.tui.widgets.persona_bar import PersonaBar
|
||||
from cleveragents.tui.widgets.prompt import PromptInput
|
||||
from cleveragents.tui.widgets.reference_picker import ReferencePickerOverlay
|
||||
from cleveragents.tui.widgets.slash_command_overlay import SlashCommandOverlay
|
||||
|
||||
if TYPE_CHECKING:
|
||||
InputSubmittedEvent = Any
|
||||
@@ -47,14 +36,38 @@ except Exception: # pragma: no cover
|
||||
_TEXTUAL_AVAILABLE = False
|
||||
|
||||
|
||||
class _ScreenBase(Protocol):
|
||||
"""Protocol shared by Textual Static and the fallback widget."""
|
||||
|
||||
def update(self, text: str) -> None:
|
||||
"""Update displayed text content.
|
||||
|
||||
Args:
|
||||
text: The plain-text string to show.
|
||||
"""
|
||||
|
||||
def focus(self) -> None:
|
||||
"""Request keyboard focus for this widget."""
|
||||
|
||||
|
||||
def textual_available() -> bool:
|
||||
"""Return whether Textual import succeeded."""
|
||||
return _TEXTUAL_AVAILABLE
|
||||
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SessionView:
|
||||
"""Minimal per-session TUI view model."""
|
||||
"""Minimal per-session TUI view model.
|
||||
|
||||
Holds a session identifier and its transcript of messages for the app layer.
|
||||
|
||||
Attributes:
|
||||
session_id: Unique identity for the conversation session.
|
||||
transcript: Ordered list of message strings in the session history.
|
||||
"""
|
||||
|
||||
session_id: str
|
||||
transcript: list[str]
|
||||
@@ -64,8 +77,15 @@ class _CommandRouter(Protocol):
|
||||
"""Router protocol used by the TUI input-mode command handler."""
|
||||
|
||||
def handle(self, raw: str, *, session_id: str) -> str:
|
||||
"""Dispatch a slash command and return a rendered response."""
|
||||
...
|
||||
"""Dispatch a slash command and return a rendered response.
|
||||
|
||||
Args:
|
||||
raw: The raw user text typed into the prompt.
|
||||
session_id: The current session identity.
|
||||
|
||||
Returns:
|
||||
A formatted string to display in the conversation area.
|
||||
"""
|
||||
|
||||
|
||||
class _FallbackCleverAgentsTuiApp: # pragma: no cover
|
||||
@@ -86,7 +106,11 @@ _ResolvedTuiApp: type[Any] = _FallbackCleverAgentsTuiApp
|
||||
if _TEXTUAL_AVAILABLE:
|
||||
|
||||
class _TextualCleverAgentsTuiApp(_TextualApp):
|
||||
"""Main TUI app."""
|
||||
"""Main TUI app.
|
||||
|
||||
Handles core session lifecycle, keyboard bindings, input processing,
|
||||
and the Sessions screen toggle (ctrl+s).
|
||||
"""
|
||||
|
||||
CSS_PATH: ClassVar[str] = "cleveragents.tcss"
|
||||
THEME: ClassVar[str] = "dracula"
|
||||
@@ -94,6 +118,7 @@ if _TEXTUAL_AVAILABLE:
|
||||
("ctrl+q", "quit", "Quit"),
|
||||
("f1", "help", "Help"),
|
||||
("ctrl+t", "cycle_preset", "Cycle Preset"),
|
||||
("ctrl+s", "sessions", "Sessions"),
|
||||
]
|
||||
|
||||
def __init__(
|
||||
@@ -106,109 +131,280 @@ if _TEXTUAL_AVAILABLE:
|
||||
self._command_router = command_router
|
||||
self._persona_state = persona_state
|
||||
self._session = SessionView(session_id="default", transcript=[])
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
SessionsScreen,
|
||||
)
|
||||
|
||||
def compose(self) -> Any:
|
||||
yield _Header(show_clock=True)
|
||||
with _Vertical(id="main-column"):
|
||||
yield _Static("CleverAgents TUI", id="conversation")
|
||||
yield HelpPanelOverlay(id="help-panel")
|
||||
yield ReferencePickerOverlay(id="reference-picker")
|
||||
yield SlashCommandOverlay(id="slash-overlay")
|
||||
yield ActorSelectionOverlay(id="actor-selection")
|
||||
yield PromptInput(
|
||||
placeholder="Type message, /command, or !shell ...", id="prompt"
|
||||
self._sessions_screen = SessionsScreen()
|
||||
self._session_service_factory: Any | None = None
|
||||
|
||||
def set_session_service_factory(self, factory: Any) -> None:
|
||||
"""Register the session-service factory callable.
|
||||
|
||||
Args:
|
||||
factory: Callable returning a container with ``session_service()``, or None. # noqa: E501
|
||||
"""
|
||||
self._session_service_factory = factory
|
||||
|
||||
def action_sessions(self) -> None:
|
||||
"""Toggle the sessions screen overlay on and off.
|
||||
|
||||
When opening, loads active and saved sessions from the injected
|
||||
service factory and pushes the SessionsScreen as a modal overlay.
|
||||
|
||||
When closing, returns focus to the prompt input widget.
|
||||
"""
|
||||
if self._sessions_screen.visible:
|
||||
self._sessions_screen.hide()
|
||||
try: # pragma: no cover - focus fallback
|
||||
prompt = self.query_one("#prompt", PromptInput)
|
||||
prompt.focus()
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
|
||||
active, saved = self._collect_sessions_snapshot()
|
||||
|
||||
def _on_select(data: dict[str, Any]) -> None:
|
||||
name_ = data.get("name", "unnamed")
|
||||
kind = "active" if data.get("is_active") else "saved"
|
||||
_logger.info(
|
||||
"session_selected",
|
||||
name=name_,
|
||||
kind=kind,
|
||||
entries_index=data["entries_index"],
|
||||
)
|
||||
yield PersonaBar(id="persona-bar")
|
||||
yield _Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
# Check first-run BEFORE _refresh_persona_bar, which calls
|
||||
# ensure_default() and would create a persona, masking the check.
|
||||
first_run = is_first_run(self._persona_state.registry)
|
||||
self._refresh_persona_bar()
|
||||
help_panel = self.query_one("#help-panel", HelpPanelOverlay)
|
||||
help_panel.hide()
|
||||
ref_picker = self.query_one("#reference-picker", ReferencePickerOverlay)
|
||||
ref_picker.set_suggestions("", [])
|
||||
slash = self.query_one("#slash-overlay", SlashCommandOverlay)
|
||||
slash.set_commands("", slash_command_specs())
|
||||
actor_overlay = self.query_one("#actor-selection", ActorSelectionOverlay)
|
||||
if first_run:
|
||||
actor_overlay.show()
|
||||
self._sessions_screen.on_select_session(_on_select)
|
||||
self._sessions_screen.refresh_entries(active, saved)
|
||||
self.push_screen(self._sessions_screen)
|
||||
|
||||
def _collect_sessions_snapshot(self) -> tuple[list[Any], list[Any]]:
|
||||
"""Build active and saved session lists from the service.
|
||||
|
||||
Returns a pair of (active_entries, saved_entries). When there are
|
||||
no sessions both lists are empty.
|
||||
|
||||
Returns:
|
||||
Tuple of (active sessions list, saved sessions list).
|
||||
"""
|
||||
return self._build_active_sessions(), self._build_saved_sessions()
|
||||
|
||||
def _build_active_sessions(self) -> list[Any]:
|
||||
"""Build the active-session list for display.
|
||||
|
||||
Falls back to an empty list when an exception occurs (logged at warning
|
||||
level per project error handling policy).
|
||||
|
||||
Returns:
|
||||
List of :class:`ActiveSessionEntry` objects representing live tabs.
|
||||
"""
|
||||
try:
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
ActiveSessionEntry,
|
||||
)
|
||||
|
||||
persona_name = (
|
||||
self._persona_state.active_name(self._session.session_id)
|
||||
or "default"
|
||||
)
|
||||
return [
|
||||
ActiveSessionEntry(
|
||||
name="default",
|
||||
actor_name=None,
|
||||
persona=persona_name,
|
||||
state_label="awaiting input",
|
||||
preview="",
|
||||
)
|
||||
]
|
||||
except Exception as exc: # noqa: PERFLINT - caller may fail.
|
||||
_logger.warning("build_active_failed", error=str(exc))
|
||||
return []
|
||||
|
||||
def _build_saved_sessions(self) -> list[Any]:
|
||||
"""Build the saved-session list from ``session_service.list()``.
|
||||
|
||||
Errors are logged at warning level rather than silently swallowed, per
|
||||
project policy (no silent exception suppression).
|
||||
|
||||
Returns:
|
||||
List of :class:`SavedSessionEntry` dataclasses, or empty list.
|
||||
"""
|
||||
service = self._resolve_session_service()
|
||||
if service is None:
|
||||
return []
|
||||
|
||||
try:
|
||||
candidates = service.list()
|
||||
except (KeyError, ValueError, TypeError) as exc:
|
||||
_logger.warning("session_list_failed", error=str(exc))
|
||||
return []
|
||||
|
||||
result: list[Any] = []
|
||||
for candidate in candidates:
|
||||
entry = self._convert_saved_session(candidate)
|
||||
if entry is not None:
|
||||
result.append(entry)
|
||||
return result
|
||||
|
||||
def _resolve_session_service(self) -> Any | None:
|
||||
"""Return a session-service instance via the injected factory.
|
||||
|
||||
Errors are logged at warning level rather than silently swallowed, per
|
||||
project policy (no silent exception suppression).
|
||||
|
||||
Returns:
|
||||
A service object supporting a ``list()`` method, or ``None``.
|
||||
"""
|
||||
if self._session_service_factory is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
return self._session_service_factory()
|
||||
except Exception as exc: # noqa: PERFLINT - caller may fail.
|
||||
_logger.warning("session_resolve_failed", error=str(exc))
|
||||
return None
|
||||
|
||||
def _convert_saved_session(self, candidate: Any) -> Any | None:
|
||||
"""Convert a session object or dict into a ``SavedSessionEntry``.
|
||||
|
||||
Supports both domain-model objects (attribute access) and simple dicts.
|
||||
Returns ``None`` when the session has no usable ID. All coercion errors
|
||||
are caught gracefully via try/except int conversion for prompt_count.
|
||||
|
||||
Args:
|
||||
candidate: A :class:`Session` instance or dict-backed record.
|
||||
|
||||
Returns:
|
||||
A :class:`SavedSessionEntry`, or ``None`` for invalid input.
|
||||
"""
|
||||
sid = self._attr_or_key(candidate, "session_id")
|
||||
if not sid:
|
||||
return None
|
||||
|
||||
name_raw = self._attr_or_key(candidate, "name")
|
||||
if isinstance(name_raw, (str, int)):
|
||||
name = str(name_raw)
|
||||
else:
|
||||
actor_overlay.hide()
|
||||
name_raw_val = name_raw if name_raw is not None else ""
|
||||
name = str(name_raw_val) if name_raw_val else "unnamed"
|
||||
|
||||
def _complete_first_run(self, actor: str) -> None:
|
||||
"""Persist the chosen actor as the default persona and refresh the bar."""
|
||||
create_default_persona_for_actor(self._persona_state.registry, actor)
|
||||
self._refresh_persona_bar()
|
||||
actor_value = self._attr_or_key(candidate, "actor_name")
|
||||
actor_name = str(actor_value) if actor_value is not None else None
|
||||
|
||||
def action_help(self) -> None:
|
||||
prompt = self.query_one("#prompt", PromptInput)
|
||||
help_panel = self.query_one("#help-panel", HelpPanelOverlay)
|
||||
context_name = resolve_help_context(prompt.value)
|
||||
help_panel.toggle(context_name)
|
||||
|
||||
def action_cycle_preset(self) -> None:
|
||||
self._persona_state.cycle_preset(self._session.session_id)
|
||||
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)
|
||||
scope_count = len(persona.scoped_projects) + len(persona.scoped_plans)
|
||||
scope_text = f"{scope_count} scope refs"
|
||||
bar = self.query_one("#persona-bar", PersonaBar)
|
||||
bar.set_content(
|
||||
persona_name=persona.name,
|
||||
actor_name=persona.actor,
|
||||
preset_name=preset,
|
||||
scope_text=scope_text,
|
||||
namespace_value = self._attr_or_key(candidate, "namespace")
|
||||
persona = (
|
||||
str(namespace_value)
|
||||
if isinstance(namespace_value, (str, int))
|
||||
else ""
|
||||
)
|
||||
|
||||
def on_input_submitted(self, event: InputSubmittedEvent) -> None:
|
||||
del event
|
||||
prompt = self.query_one("#prompt", PromptInput)
|
||||
payload = prompt.consume_text()
|
||||
text = payload.text.strip()
|
||||
if not text:
|
||||
return
|
||||
prompt_count_raw = self._attr_or_key(
|
||||
candidate, "message_count"
|
||||
) or self._metadata_lookup(candidate, "prompt_count")
|
||||
try:
|
||||
prompt_count = int(prompt_count_raw)
|
||||
except (ValueError, TypeError):
|
||||
prompt_count = 0
|
||||
|
||||
mode_router = InputModeRouter(
|
||||
command_handler=lambda raw: self._command_router.handle(
|
||||
raw, session_id=self._session.session_id
|
||||
),
|
||||
shell_confirm=lambda _cmd: (
|
||||
os.environ.get("CLEVERAGENTS_ALLOW_DANGEROUS_SHELL", "").strip()
|
||||
in {"1", "true"}
|
||||
),
|
||||
last_used_raw = self._attr_or_key(candidate, "updated_at")
|
||||
last_used = self._parse_dt(last_used_raw)
|
||||
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
SavedSessionEntry,
|
||||
)
|
||||
result = mode_router.process(text)
|
||||
conversation = self.query_one("#conversation", _Static)
|
||||
|
||||
if result.mode == InputMode.COMMAND:
|
||||
conversation.update(result.command_result or "")
|
||||
self._refresh_persona_bar()
|
||||
return
|
||||
if result.mode == InputMode.SHELL:
|
||||
shell = result.shell_result
|
||||
if shell is None:
|
||||
conversation.update("(no shell output)")
|
||||
return
|
||||
output = (
|
||||
shell.stdout.strip() or shell.stderr.strip() or "(empty output)"
|
||||
)
|
||||
conversation.update(f"$ {shell.command}\n{output}")
|
||||
return
|
||||
return SavedSessionEntry(
|
||||
name=name if name else "unnamed",
|
||||
actor_name=actor_name,
|
||||
persona=persona,
|
||||
prompt_count=prompt_count,
|
||||
last_used=last_used,
|
||||
)
|
||||
|
||||
preview = result.expanded_text
|
||||
if "@" in text:
|
||||
ref_picker = self.query_one("#reference-picker", ReferencePickerOverlay)
|
||||
ref_picker.set_suggestions(
|
||||
text, suggestions(text.replace("@", "").strip())
|
||||
)
|
||||
conversation.update(preview)
|
||||
def _create_active_entry(self) -> Any:
|
||||
"""Create an ``ActiveSessionEntry`` for the current session tab."""
|
||||
from cleveragents.tui.screens.sessions_screen import (
|
||||
ActiveSessionEntry,
|
||||
)
|
||||
|
||||
_ResolvedTuiApp = _TextualCleverAgentsTuiApp
|
||||
persona_name = (
|
||||
self._persona_state.active_name(self._session.session_id) or "default"
|
||||
)
|
||||
return ActiveSessionEntry(
|
||||
name="default",
|
||||
actor_name=None,
|
||||
persona=persona_name,
|
||||
state_label="awaiting input",
|
||||
preview="",
|
||||
)
|
||||
|
||||
def _attr_or_key(self, obj: Any, key: str) -> Any | None:
|
||||
"""Duck-typing accessor: try attribute then dict-key lookup.
|
||||
|
||||
Args:
|
||||
obj: Object supporting either ``getattr`` or ``dict[key]`` access.
|
||||
key: Attribute name or dictionary key to look up.
|
||||
|
||||
Returns:
|
||||
The resolved value, or ``None`` when neither accessor yields data.
|
||||
"""
|
||||
try:
|
||||
val = getattr(obj, key, None)
|
||||
if val is not None:
|
||||
return val
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
try:
|
||||
if isinstance(obj, dict):
|
||||
return obj.get(key)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def _metadata_lookup(self, obj: Any, key: str) -> Any | None:
|
||||
"""Look up a value nested under ``obj.metadata[key]``.
|
||||
|
||||
Args:
|
||||
obj: Session object or dict with an optional ``metadata`` mapping.
|
||||
key: Leaf key inside the metadata dict.
|
||||
|
||||
Returns:
|
||||
The metadata value at *key*, or ``None`` when absent.
|
||||
"""
|
||||
meta = self._attr_or_key(obj, "metadata")
|
||||
if isinstance(meta, dict):
|
||||
return meta.get(key)
|
||||
return None
|
||||
|
||||
def _parse_dt(self, raw: Any) -> Any:
|
||||
"""Parse a raw value into a :class:`datetime`.
|
||||
|
||||
Handles string ISO dates, :class:`datetime` instances, and falls
|
||||
back to the current time for anything else.
|
||||
|
||||
Args:
|
||||
raw: A timestamp-like value.
|
||||
|
||||
Returns:
|
||||
A :class:`datetime` object with UTC timezone set.
|
||||
"""
|
||||
from datetime import datetime as _dt
|
||||
|
||||
if isinstance(raw, _dt):
|
||||
return raw.replace(tzinfo=UTC) if raw.tzinfo is None else raw
|
||||
if isinstance(raw, str):
|
||||
try:
|
||||
dt = _dt.fromisoformat(raw)
|
||||
return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt
|
||||
except (ValueError, TypeError): # pragma: no cover
|
||||
pass
|
||||
return _dt.now(UTC)
|
||||
|
||||
|
||||
_ResolvedTuiApp = _TextualCleverAgentsTuiApp
|
||||
|
||||
_CleverAgentsTuiApp: type[Any] = _ResolvedTuiApp
|
||||
CleverAgentsTuiApp = _ResolvedTuiApp
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
"""Sessions screen for browsing and selecting sessions."""
|
||||
|
||||
from cleveragents.tui.screens.sessions_screen import SessionsScreen
|
||||
|
||||
__all__ = ["SessionsScreen"]
|
||||
@@ -0,0 +1,499 @@
|
||||
"""SessionsScreen for browsing and selecting sessions in the TUI.
|
||||
|
||||
Implements a modal overlay showing active and saved sessions, with keyboard
|
||||
navigation (up/down), selection (enter), and dismissal (escape).
|
||||
|
||||
Based on ``docs/specification.md`` section "Sessions Screen".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, Protocol, runtime_checkable
|
||||
|
||||
_logger = logging.getLogger(__name__)
|
||||
|
||||
__all__ = ["SessionsScreen"]
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ActiveSessionEntry:
|
||||
"""Representation of an active (live) session tab.
|
||||
|
||||
Attributes:
|
||||
name: Human-readable session name.
|
||||
actor_name: Namespaced actor reference (e.g. ``local/claude``).
|
||||
persona: The active persona name.
|
||||
state_label: Current interaction state label.
|
||||
preview: Last prompt or message preview, truncated to fit display.
|
||||
updated_at: Timestamp since this tab was last active.
|
||||
is_current: Whether this is the foreground tab.
|
||||
"""
|
||||
|
||||
name: str
|
||||
actor_name: str | None
|
||||
persona: str
|
||||
state_label: str
|
||||
preview: str
|
||||
updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
is_current: bool = False
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SavedSessionEntry:
|
||||
"""Representation of a saved (persisted) session record.
|
||||
|
||||
Attributes:
|
||||
name: Optional human-readable session name.
|
||||
actor_name: Namespaced actor reference, if set.
|
||||
persona: The persona used for this session.
|
||||
prompt_count: Number of prompts in the session.
|
||||
last_used: When the session was last accessed.
|
||||
"""
|
||||
|
||||
name: str | None
|
||||
actor_name: str | None
|
||||
persona: str
|
||||
prompt_count: int
|
||||
last_used: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Rendering helpers (module-level functions)
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
def _relative_time(dt: datetime | None, *, now: datetime | None = None) -> str:
|
||||
"""Return a human-readable relative time string.
|
||||
|
||||
Produces ``"just now"``, ``"Xm ago"``, ``"Xh ago"``, or a full date
|
||||
for entries more than 24 hours old.
|
||||
|
||||
Args:
|
||||
dt: The datetime to format. If ``None``, returns ``"n/a"``.
|
||||
now: Reference point for age calculation; defaults to now in UTC.
|
||||
|
||||
Returns:
|
||||
A formatted relative-time string such as ``"just now"``, ``"3m ago"``,
|
||||
or ``"Mar 10 09:15"`` for older entries.
|
||||
"""
|
||||
if dt is None:
|
||||
return "n/a"
|
||||
|
||||
if now is None:
|
||||
now = datetime.now(UTC)
|
||||
|
||||
delta = (now - dt).total_seconds()
|
||||
|
||||
if delta < 60:
|
||||
return "just now"
|
||||
minutes = int(delta // 60)
|
||||
if minutes < 60:
|
||||
return f"{minutes}m ago"
|
||||
hours = int(delta // 3600)
|
||||
if hours < 24:
|
||||
return f"{hours}h ago"
|
||||
|
||||
# Full date for >24-hour age
|
||||
return dt.strftime("%b %d %H:%M")
|
||||
|
||||
|
||||
def _normalize_preview(text: str, *, max_len: int = 71) -> str:
|
||||
"""Truncate a session preview to fit within display width.
|
||||
|
||||
Args:
|
||||
text: The raw preview string.
|
||||
max_len: Maximum allowed length before truncation.
|
||||
|
||||
Returns:
|
||||
A preview fitting within *max_len*; ``(no prompts yet)`` when empty,
|
||||
or truncated with ellipsis (U+2026) when too long.
|
||||
"""
|
||||
if not text:
|
||||
return "(no prompts yet)"
|
||||
if len(text) <= max_len:
|
||||
return text
|
||||
return text[:max_len] + "\u2026"
|
||||
|
||||
|
||||
def _render_active(entries: list[ActiveSessionEntry], active_index: int) -> str:
|
||||
"""Render the Active Sessions section header with entries.
|
||||
|
||||
The selected entry is marked with ``>`` (arrowhead). Trailing blank
|
||||
lines from each entry's preview are stripped for clean rendering.
|
||||
|
||||
Args:
|
||||
entries: List of active session entries to display.
|
||||
active_index: Index of the currently active tab.
|
||||
|
||||
Returns:
|
||||
A multi-line string, or ``(no active sessions)`` when the list is empty.
|
||||
"""
|
||||
if not entries:
|
||||
return "(no active sessions)"
|
||||
|
||||
lines: list[str] = ["Active Sessions"]
|
||||
for i, entry in enumerate(entries):
|
||||
preview_text = _normalize_preview(entry.preview)
|
||||
time_str = _relative_time(entry.updated_at)
|
||||
prefix = "> " if i == active_index else " "
|
||||
actor_display = entry.actor_name or "(none)"
|
||||
lines.append(
|
||||
f"{prefix}{entry.name}"
|
||||
f" -- {actor_display}"
|
||||
f" -- {entry.persona}"
|
||||
f" -- {entry.state_label}"
|
||||
f" {time_str}"
|
||||
)
|
||||
lines.append(f" {preview_text}")
|
||||
|
||||
# Strip trailing blank line from last entry's preview display line
|
||||
if lines and lines[-1] == "":
|
||||
lines.pop()
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _render_saved(entries: list[SavedSessionEntry]) -> str:
|
||||
"""Render the Saved Sessions section header with entries.
|
||||
|
||||
Args:
|
||||
entries: List of persisted session entries.
|
||||
|
||||
Returns:
|
||||
A multi-line string, or ``(no saved sessions)`` when empty.
|
||||
"""
|
||||
if not entries:
|
||||
return "(no saved sessions)"
|
||||
|
||||
lines: list[str] = ["Saved Sessions (from previous runs)"]
|
||||
for entry in entries:
|
||||
name = entry.name or "unnamed"
|
||||
actor_display = entry.actor_name or "(none)"
|
||||
prompt_text = f"{entry.prompt_count} prompts"
|
||||
time_str = _relative_time(entry.last_used)
|
||||
lines.append(
|
||||
f" {name}"
|
||||
f" -- {actor_display}"
|
||||
f" -- {entry.persona}"
|
||||
f" -- {prompt_text}"
|
||||
f" -- {time_str}"
|
||||
)
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _render_bottom_bar() -> str:
|
||||
"""Render the keyboard shortcut bar at the bottom of the screen.
|
||||
|
||||
Returns:
|
||||
Formatted status-bar string with key bindings and descriptions.
|
||||
"""
|
||||
return (
|
||||
"enter Switch | ctrl+r Resume saved | ctrl+n New"
|
||||
" | d Delete | r Rename | esc Back"
|
||||
)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Optional Textual import gate -- dynamic base class with Protocol typing
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class _Updateable(Protocol):
|
||||
"""Protocol satisfied by both Textual ``Static`` and the fallback widget.
|
||||
|
||||
Provides ``update(text: str)`` to set displayed content and ``focus()``
|
||||
to request keyboard focus Used for static-typing when the actual
|
||||
runtime base class comes from an optional import gate.
|
||||
"""
|
||||
|
||||
def update(self, text: str) -> None:
|
||||
"""Update displayed text content.
|
||||
|
||||
Args:
|
||||
text: The plain-text string to show.
|
||||
"""
|
||||
|
||||
|
||||
class _FallbackStatic: # pragma: no cover -- fallback path only
|
||||
"""Minimal widget stub for environments without Textual installed.
|
||||
|
||||
Provides ``update(text)`` and ``focus()`` methods so the screen can be
|
||||
instantiated without import errors.
|
||||
|
||||
Attributes:
|
||||
_text: Internal storage for displayed plain-text content.
|
||||
"""
|
||||
|
||||
def __init__(self, *args: object, **kwargs: object) -> None: # pragma: no cover, # noqa: D407
|
||||
"""Initialise with empty text."""
|
||||
self._text = ""
|
||||
|
||||
def update(self, text: str) -> None:
|
||||
"""Set the displayed text content.
|
||||
|
||||
Args:
|
||||
text: The plain-text string to display.
|
||||
"""
|
||||
self._text = text
|
||||
|
||||
def focus(self) -> None: # pragma: no cover -- fallback path only
|
||||
"""Placeholder request for keyboard focus (no-op without Textual)."""
|
||||
|
||||
|
||||
def _load_textual_base() -> type | None:
|
||||
"""Load Textual ``Static`` or return ``None`` when unavailable.
|
||||
|
||||
Returns:
|
||||
The ``textual.widgets.Static`` class, or ``None`` if the module is
|
||||
not installed.
|
||||
"""
|
||||
try:
|
||||
import importlib
|
||||
|
||||
|
||||
textual_widgets = importlib.import_module("textual.widgets")
|
||||
return textual_widgets.Static
|
||||
except Exception: # pragma: no cover
|
||||
return None
|
||||
|
||||
|
||||
_TEXTUAL_STATIC = _load_textual_base()
|
||||
|
||||
|
||||
def _build_sessions_base(base_type: type | None) -> type[_Updateable]:
|
||||
"""Build the runtime base class for SessionsScreen.
|
||||
|
||||
When Textual is available uses ``textual.widgets.Static`` directly;
|
||||
otherwise falls back to a minimal widget-compatible stub typed against
|
||||
``_Updateable`` for static type-checking purposes.
|
||||
|
||||
Args:
|
||||
base_type: The loaded Textual Static class, or ``None``.
|
||||
|
||||
Returns:
|
||||
A class appropriate as the runtime parent of SessionsScreen.
|
||||
"""
|
||||
if base_type is not None:
|
||||
return base_type # type: ignore[return-value]
|
||||
|
||||
return _FallbackStatic
|
||||
|
||||
|
||||
_SessionsBase = _build_sessions_base(_TEXTUAL_STATIC)
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# SessionsScreen -- the widget / overlay class
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
|
||||
class SessionsScreen(_SessionsBase): # type: ignore[misc]
|
||||
"""TUI modal overlay for browsing and selecting sessions.
|
||||
|
||||
Shows two sections -- **Active Sessions** (live tabs) and **Saved Sessions**
|
||||
(persisted from previous runs). Navigation uses up/down arrow keys, enter
|
||||
selects the highlighted session, and escape closes the overlay.
|
||||
|
||||
Attributes:
|
||||
visible: Whether this screen overlay is currently shown on top of app.
|
||||
selected_index: Index of the currently highlighted entry in combined list.
|
||||
"""
|
||||
|
||||
BINDINGS: ClassVar[list[tuple[str, str, str]]] = [
|
||||
("up", "select_prev", "Previous"),
|
||||
("down", "select_next", "Next"),
|
||||
("enter", "select_session", "Select"),
|
||||
("escape", "close_screen", "Close"),
|
||||
]
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialise the SessionsScreen overlay with empty state.
|
||||
|
||||
Sets up empty session lists, zero selection index, and ``visible=False``.
|
||||
The base class ``(Textual Static / _FallbackStatic)`` is initialised
|
||||
with a default keyword-argument pass-through.
|
||||
"""
|
||||
super().__init__() # type: ignore[arg-type]
|
||||
self._active_entries: list[ActiveSessionEntry] = []
|
||||
self._saved_entries: list[SavedSessionEntry] = []
|
||||
self._selected_index: int = 0
|
||||
self._visible: bool = False
|
||||
self._text: str = ""
|
||||
self._on_select_session: Any | None = None
|
||||
|
||||
@property
|
||||
def visible(self) -> bool:
|
||||
"""Return whether the screen is currently visible."""
|
||||
return self._visible
|
||||
|
||||
@property
|
||||
def selected_entry_index(self) -> int:
|
||||
"""Return the current selection index."""
|
||||
return self._selected_index
|
||||
|
||||
# -- State management -----------------------------------------------
|
||||
|
||||
def toggle(self) -> bool:
|
||||
"""Toggle the screen visibility on and off.
|
||||
|
||||
Returns:
|
||||
``True`` if the screen was opened, ``False`` if it was closed.
|
||||
"""
|
||||
self._visible = not self._visible
|
||||
if self._visible:
|
||||
self._refresh_render()
|
||||
else:
|
||||
self._reset_state()
|
||||
return self._visible
|
||||
|
||||
def show(self) -> None:
|
||||
"""Show the screen overlay and render current state."""
|
||||
self._visible = True
|
||||
self._refresh_render()
|
||||
|
||||
def hide(self) -> None:
|
||||
"""Hide the screen overlay and reset selection state."""
|
||||
self._visible = False
|
||||
self._text = ""
|
||||
self._selected_index = 0
|
||||
|
||||
def refresh_entries(
|
||||
self,
|
||||
active: list[ActiveSessionEntry],
|
||||
saved: list[SavedSessionEntry],
|
||||
) -> None:
|
||||
"""Refresh session lists and reset selection.
|
||||
|
||||
Replaces both the active and saved entry lists, resets the selection
|
||||
index to zero, and re-renders if the screen is currently visible.
|
||||
|
||||
Args:
|
||||
active: List of currently active session tab entries.
|
||||
saved: List of persisted (saved) session entries.
|
||||
"""
|
||||
self._active_entries = active
|
||||
self._saved_entries = saved
|
||||
self._selected_index = 0
|
||||
if self._visible:
|
||||
self._refresh_render()
|
||||
|
||||
def _get_total_count(self) -> int:
|
||||
"""Return total number of entries (active + saved)."""
|
||||
return len(self._active_entries) + len(self._saved_entries)
|
||||
|
||||
# -- Callback registration ------------------------------------------
|
||||
|
||||
def on_select_session(self, callback: Any) -> None:
|
||||
"""Register a callback for when the user selects a session entry.
|
||||
|
||||
The callback receives a dict with ``session_id`` (str), ``name`` (str),
|
||||
``is_active`` (bool), and ``entries_index`` (int).
|
||||
|
||||
Args:
|
||||
callback: Callable to invoke on session selection; called at most once
|
||||
per selection event, with the exception of errors in the callback
|
||||
itself which are logged and silently swallowed.
|
||||
"""
|
||||
self._on_select_session = callback
|
||||
|
||||
# -- Action handlers (bound via BINDINGS) ---------------------------
|
||||
|
||||
def action_select_next(self) -> None:
|
||||
"""Move selection down one entry."""
|
||||
total = self._get_total_count()
|
||||
if total > 0:
|
||||
self._selected_index = (self._selected_index + 1) % total
|
||||
self._refresh_render()
|
||||
|
||||
def action_select_prev(self) -> None:
|
||||
"""Move selection up one entry."""
|
||||
total = self._get_total_count()
|
||||
if total > 0:
|
||||
self._selected_index = (self._selected_index - 1) % total
|
||||
self._refresh_render()
|
||||
|
||||
def action_close_screen(self) -> None:
|
||||
"""Close (hide) the sessions screen."""
|
||||
self.hide()
|
||||
|
||||
def action_select_session(self) -> None:
|
||||
"""Select the currently highlighted session and invoke callback.
|
||||
|
||||
Returns without action when there are no entries. When a real entry
|
||||
exists, the registered callback receives a dict describing the selection.
|
||||
Callback errors are caught at logging level and do not crash the screen.
|
||||
"""
|
||||
total = self._get_total_count()
|
||||
if total == 0:
|
||||
return
|
||||
|
||||
selected = self._selected_index
|
||||
active_count = len(self._active_entries)
|
||||
|
||||
if selected < active_count:
|
||||
entry = self._active_entries[selected]
|
||||
data: dict[str, Any] = {
|
||||
"session_id": "",
|
||||
"name": entry.name,
|
||||
"is_active": True,
|
||||
"entries_index": selected,
|
||||
}
|
||||
else:
|
||||
saved_idx = selected - active_count
|
||||
entry = self._saved_entries[saved_idx]
|
||||
name_value = (
|
||||
entry.name
|
||||
if isinstance(entry.name, str)
|
||||
else (str(entry.name) if entry.name is not None else "unnamed")
|
||||
)
|
||||
data = {
|
||||
"session_id": "",
|
||||
"name": name_value,
|
||||
"is_active": False,
|
||||
"entries_index": saved_idx + active_count,
|
||||
}
|
||||
|
||||
if self._on_select_session is not None:
|
||||
try:
|
||||
self._on_select_session(data) # type: ignore[arg-type]
|
||||
except Exception as exc:
|
||||
_logger.warning("select_callback_failed", error=str(exc))
|
||||
|
||||
# -- Internal rendering ---------------------------------------------
|
||||
|
||||
def _refresh_render(self) -> None:
|
||||
"""Re-render the full screen and call ``update()`` on the base widget.
|
||||
|
||||
Builds the complete multi-line output (header, active section with
|
||||
separator, saved section with separator, bottom bar), stores it in
|
||||
``self._text``, then delegates to the runtime base class's ``update()``
|
||||
method which either updates Textual's ``Static.render()`` or sets
|
||||
``_FallbackStatic._text``.
|
||||
"""
|
||||
active_section = _render_active(
|
||||
self._active_entries,
|
||||
self._selected_index if self._active_entries else 0,
|
||||
)
|
||||
saved_section = _render_saved(self._saved_entries)
|
||||
separator = "-" * 80
|
||||
bottom = _render_bottom_bar()
|
||||
|
||||
parts: list[str] = [
|
||||
"Sessions",
|
||||
"",
|
||||
active_section,
|
||||
separator,
|
||||
saved_section,
|
||||
separator,
|
||||
bottom,
|
||||
]
|
||||
|
||||
self._text = "\n".join(parts)
|
||||
super().update(self._text) # type: ignore[attr-defined]
|
||||
|
||||
|
||||
Reference in New Issue
Block a user