Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| de12377272 |
@@ -7,6 +7,8 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
|
||||
|
||||
### Added
|
||||
|
||||
- **Sessions screen hotkey** (#6348): Implemented the CTRL+S SessionsScreen overlay in the TUI, showing active and saved sessions with relative timestamps and management shortcuts.
|
||||
|
||||
- **Git Worktree Sandbox Apply** (#4454): The `plan apply` command now merges
|
||||
LLM-generated changes via `git merge` from an isolated worktree branch
|
||||
instead of flat `shutil.copy2`. Displays spec-aligned Apply Summary
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
from cleveragents.tui.app import CleverAgentsTuiApp
|
||||
from cleveragents.tui.screens import (
|
||||
ActiveSessionEntry,
|
||||
SavedSessionEntry,
|
||||
SessionsScreen,
|
||||
SessionsSnapshot,
|
||||
)
|
||||
|
||||
|
||||
@given("a sessions snapshot with one active and one saved session")
|
||||
def given_sessions_snapshot(context: Any) -> None:
|
||||
now = datetime(2024, 1, 1, 12, 0, 0)
|
||||
active = ActiveSessionEntry(
|
||||
session_id="active-1",
|
||||
title="Session Alpha",
|
||||
persona_name="feature-dev",
|
||||
actor_name="claude-4-sonnet",
|
||||
status="awaiting input",
|
||||
last_prompt_preview="Review authentication module implementation",
|
||||
last_used=now,
|
||||
is_current=True,
|
||||
)
|
||||
saved = SavedSessionEntry(
|
||||
session_id="saved-1",
|
||||
title="debug-session",
|
||||
persona_name="feature-dev",
|
||||
actor_name="claude-4-sonnet",
|
||||
prompt_count=23,
|
||||
last_used=now - timedelta(hours=2),
|
||||
)
|
||||
context.snapshot = SessionsSnapshot(
|
||||
active_sessions=[active],
|
||||
saved_sessions=[saved],
|
||||
active_index=0,
|
||||
)
|
||||
context.snapshot_now = now
|
||||
|
||||
|
||||
@when("I render the sessions screen")
|
||||
def when_render_sessions_screen(context: Any) -> None:
|
||||
snapshot: SessionsSnapshot = context.snapshot
|
||||
now: datetime = context.snapshot_now
|
||||
context.rendered_output = SessionsScreen.render(snapshot, now=now)
|
||||
|
||||
|
||||
@then("the sessions screen output includes the active session summary")
|
||||
def then_active_session_present(context: Any) -> None:
|
||||
rendered: str = context.rendered_output
|
||||
assert "Session Alpha" in rendered, "Active session title missing"
|
||||
assert "awaiting input" in rendered, "Active session status missing"
|
||||
assert "❯" in rendered, "Current session marker missing"
|
||||
|
||||
|
||||
@then("the sessions screen output includes the saved session summary")
|
||||
def then_saved_session_present(context: Any) -> None:
|
||||
rendered: str = context.rendered_output
|
||||
assert "debug-session" in rendered, "Saved session title missing"
|
||||
assert "23 prompts" in rendered, "Prompt count missing from saved session"
|
||||
|
||||
|
||||
@then("the sessions screen output includes the footer shortcuts")
|
||||
def then_footer_present(context: Any) -> None:
|
||||
rendered: str = context.rendered_output
|
||||
assert "ctrl+r Resume saved" in rendered, "Footer shortcuts missing"
|
||||
|
||||
|
||||
@given("the TUI app bindings are available")
|
||||
def given_bindings(context: Any) -> None:
|
||||
context.app_bindings = list(CleverAgentsTuiApp.BINDINGS)
|
||||
|
||||
|
||||
@then("ctrl+s is bound to the sessions action")
|
||||
def then_ctrl_s_binding_present(context: Any) -> None:
|
||||
bindings = context.app_bindings
|
||||
expected = ("ctrl+s", "sessions", "Sessions")
|
||||
assert expected in bindings, "ctrl+s binding missing from TUI app"
|
||||
@@ -0,0 +1,13 @@
|
||||
Feature: Sessions screen overlay
|
||||
Validates the Sessions screen rendering and keybinding wiring.
|
||||
|
||||
Scenario: Sessions screen renders active and saved sessions
|
||||
Given a sessions snapshot with one active and one saved session
|
||||
When I render the sessions screen
|
||||
Then the sessions screen output includes the active session summary
|
||||
And the sessions screen output includes the saved session summary
|
||||
And the sessions screen output includes the footer shortcuts
|
||||
|
||||
Scenario: TUI binds ctrl+s to sessions action
|
||||
Given the TUI app bindings are available
|
||||
Then ctrl+s is bound to the sessions action
|
||||
+152
-6
@@ -4,8 +4,9 @@ from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
import os
|
||||
from datetime import datetime
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Protocol
|
||||
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, Callable
|
||||
|
||||
from cleveragents.tui.first_run import create_default_persona_for_actor, is_first_run
|
||||
from cleveragents.tui.input.modes import InputMode, InputModeRouter
|
||||
@@ -21,8 +22,16 @@ 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
|
||||
from cleveragents.tui.screens import (
|
||||
ActiveSessionEntry,
|
||||
SavedSessionEntry,
|
||||
SessionsScreen,
|
||||
SessionsSnapshot,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from cleveragents.domain.models.core.session import SessionService
|
||||
|
||||
InputSubmittedEvent = Any
|
||||
else:
|
||||
InputSubmittedEvent = Any
|
||||
@@ -33,6 +42,13 @@ _Vertical: type[Any] = object
|
||||
_Header: type[Any] = object
|
||||
_Footer: type[Any] = object
|
||||
_Static: type[Any] = object
|
||||
_DEFAULT_BINDINGS: tuple[tuple[str, str, str], ...] = (
|
||||
("ctrl+q", "quit", "Quit"),
|
||||
("f1", "help", "Help"),
|
||||
("ctrl+t", "cycle_preset", "Cycle Preset"),
|
||||
("ctrl+s", "sessions", "Sessions"),
|
||||
)
|
||||
|
||||
try: # pragma: no branch - import gate for optional dependency
|
||||
_textual_app = importlib.import_module("textual.app")
|
||||
_textual_containers = importlib.import_module("textual.containers")
|
||||
@@ -71,6 +87,8 @@ class _CommandRouter(Protocol):
|
||||
class _FallbackCleverAgentsTuiApp: # pragma: no cover
|
||||
"""Fallback app that raises actionable dependency error."""
|
||||
|
||||
BINDINGS: ClassVar[list[tuple[str, str, str]]] = list(_DEFAULT_BINDINGS)
|
||||
|
||||
def __init__(
|
||||
self, *, command_router: _CommandRouter, persona_state: PersonaState
|
||||
) -> None:
|
||||
@@ -89,21 +107,19 @@ if _TEXTUAL_AVAILABLE:
|
||||
"""Main TUI app."""
|
||||
|
||||
CSS_PATH: ClassVar[str] = "cleveragents.tcss"
|
||||
BINDINGS: ClassVar[list[tuple[str, str, str]]] = [
|
||||
("ctrl+q", "quit", "Quit"),
|
||||
("f1", "help", "Help"),
|
||||
("ctrl+t", "cycle_preset", "Cycle Preset"),
|
||||
]
|
||||
BINDINGS: ClassVar[list[tuple[str, str, str]]] = list(_DEFAULT_BINDINGS)
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
command_router: _CommandRouter,
|
||||
persona_state: PersonaState,
|
||||
session_service_factory: Callable[[], Any] | None = None,
|
||||
) -> None:
|
||||
super().__init__()
|
||||
self._command_router = command_router
|
||||
self._persona_state = persona_state
|
||||
self._session_service_factory = session_service_factory
|
||||
self._session = SessionView(session_id="default", transcript=[])
|
||||
|
||||
def compose(self) -> Any:
|
||||
@@ -114,6 +130,7 @@ if _TEXTUAL_AVAILABLE:
|
||||
yield ReferencePickerOverlay(id="reference-picker")
|
||||
yield SlashCommandOverlay(id="slash-overlay")
|
||||
yield ActorSelectionOverlay(id="actor-selection")
|
||||
yield SessionsScreen(id="sessions-screen")
|
||||
yield PromptInput(
|
||||
placeholder="Type message, /command, or !shell ...", id="prompt"
|
||||
)
|
||||
@@ -132,6 +149,8 @@ if _TEXTUAL_AVAILABLE:
|
||||
slash = self.query_one("#slash-overlay", SlashCommandOverlay)
|
||||
slash.set_commands("", slash_command_specs())
|
||||
actor_overlay = self.query_one("#actor-selection", ActorSelectionOverlay)
|
||||
sessions_screen = self.query_one("#sessions-screen", SessionsScreen)
|
||||
sessions_screen.hide()
|
||||
if first_run:
|
||||
actor_overlay.show()
|
||||
else:
|
||||
@@ -152,6 +171,133 @@ if _TEXTUAL_AVAILABLE:
|
||||
self._persona_state.cycle_preset(self._session.session_id)
|
||||
self._refresh_persona_bar()
|
||||
|
||||
def action_sessions(self) -> None:
|
||||
screen = self.query_one("#sessions-screen", SessionsScreen)
|
||||
if screen.visible:
|
||||
screen.hide()
|
||||
prompt = self.query_one("#prompt", PromptInput)
|
||||
prompt.focus()
|
||||
return
|
||||
snapshot, now = self._collect_sessions_snapshot()
|
||||
screen.show(snapshot, now=now)
|
||||
|
||||
def _collect_sessions_snapshot(self) -> tuple[SessionsSnapshot, datetime]:
|
||||
now = datetime.now()
|
||||
active_entries = self._build_active_sessions(now=now)
|
||||
saved_entries = self._build_saved_sessions(now=now)
|
||||
snapshot = SessionsSnapshot(
|
||||
active_sessions=active_entries,
|
||||
saved_sessions=saved_entries,
|
||||
active_index=0,
|
||||
)
|
||||
return snapshot, now
|
||||
|
||||
def _build_active_sessions(self, *, now: datetime) -> list[ActiveSessionEntry]:
|
||||
persona = self._persona_state.active_persona(self._session.session_id)
|
||||
preview = self._session.transcript[-1] if self._session.transcript else ""
|
||||
entry = ActiveSessionEntry(
|
||||
session_id=self._session.session_id,
|
||||
title=persona.name or self._session.session_id,
|
||||
persona_name=persona.name,
|
||||
actor_name=persona.actor,
|
||||
status="awaiting input",
|
||||
last_prompt_preview=preview,
|
||||
last_used=now,
|
||||
is_current=True,
|
||||
)
|
||||
return [entry]
|
||||
|
||||
def _build_saved_sessions(self, *, now: datetime) -> list[SavedSessionEntry]:
|
||||
service = self._resolve_session_service()
|
||||
if service is None:
|
||||
return []
|
||||
entries: list[SavedSessionEntry] = []
|
||||
try:
|
||||
candidates = service.list() # type: ignore[attr-defined]
|
||||
except Exception:
|
||||
return []
|
||||
for candidate in candidates:
|
||||
entry = self._convert_saved_session(candidate, now=now)
|
||||
if entry is not None:
|
||||
entries.append(entry)
|
||||
return entries
|
||||
|
||||
def _resolve_session_service(self) -> Any | None:
|
||||
factory = getattr(self, "_session_service_factory", None)
|
||||
if factory is None:
|
||||
return None
|
||||
try:
|
||||
return factory()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _convert_saved_session(self, candidate: Any, *, now: datetime) -> SavedSessionEntry | None:
|
||||
session_id = self._attr_or_key(candidate, "session_id")
|
||||
if not session_id:
|
||||
return None
|
||||
title = (
|
||||
self._attr_or_key(candidate, "title")
|
||||
or self._attr_or_key(candidate, "name")
|
||||
or session_id
|
||||
)
|
||||
persona_name = (
|
||||
self._attr_or_key(candidate, "persona_name")
|
||||
or self._metadata_lookup(candidate, "persona_name")
|
||||
or ""
|
||||
)
|
||||
actor_name = (
|
||||
self._attr_or_key(candidate, "actor_name")
|
||||
or self._metadata_lookup(candidate, "actor")
|
||||
or ""
|
||||
)
|
||||
prompt_count_value = (
|
||||
self._attr_or_key(candidate, "prompt_count")
|
||||
or self._metadata_lookup(candidate, "prompt_count")
|
||||
)
|
||||
if prompt_count_value is None:
|
||||
message_count = self._attr_or_key(candidate, "message_count")
|
||||
if isinstance(message_count, int):
|
||||
prompt_count_value = message_count
|
||||
prompt_count = int(prompt_count_value) if isinstance(prompt_count_value, int) else 0
|
||||
last_used_value = (
|
||||
self._attr_or_key(candidate, "last_used")
|
||||
or self._attr_or_key(candidate, "updated_at")
|
||||
)
|
||||
if isinstance(last_used_value, datetime):
|
||||
last_used = last_used_value
|
||||
else:
|
||||
last_used = now
|
||||
return SavedSessionEntry(
|
||||
session_id=session_id,
|
||||
title=title,
|
||||
persona_name=persona_name,
|
||||
actor_name=actor_name,
|
||||
prompt_count=prompt_count,
|
||||
last_used=last_used,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _attr_or_key(candidate: Any, name: str) -> Any | None:
|
||||
if hasattr(candidate, name):
|
||||
return getattr(candidate, name)
|
||||
if isinstance(candidate, dict):
|
||||
value = candidate.get(name)
|
||||
if isinstance(value, (str, int, datetime)):
|
||||
return value
|
||||
return value
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
def _metadata_lookup(candidate: Any, name: str) -> Any | None:
|
||||
metadata = None
|
||||
if hasattr(candidate, "metadata"):
|
||||
metadata = getattr(candidate, "metadata")
|
||||
elif isinstance(candidate, dict):
|
||||
metadata = candidate.get("metadata")
|
||||
if isinstance(metadata, dict):
|
||||
return metadata.get(name)
|
||||
return None
|
||||
|
||||
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)
|
||||
|
||||
@@ -242,6 +242,10 @@ def run_tui(*, headless: bool = False) -> int:
|
||||
print(json.dumps(payload, indent=2))
|
||||
return 0
|
||||
|
||||
app = CleverAgentsTuiApp(command_router=router, persona_state=state)
|
||||
app = CleverAgentsTuiApp(
|
||||
command_router=router,
|
||||
persona_state=state,
|
||||
session_service_factory=lambda: container.session_service(),
|
||||
)
|
||||
app.run()
|
||||
return 0
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
"""Screen overlays for the CleverAgents TUI."""
|
||||
|
||||
from .sessions import SessionsScreen, SessionsSnapshot, ActiveSessionEntry, SavedSessionEntry
|
||||
|
||||
__all__ = [
|
||||
"ActiveSessionEntry",
|
||||
"SavedSessionEntry",
|
||||
"SessionsSnapshot",
|
||||
"SessionsScreen",
|
||||
]
|
||||
@@ -0,0 +1,196 @@
|
||||
"""Sessions screen overlay implementation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
_MAX_PREVIEW = 72
|
||||
|
||||
|
||||
def _load_static_base() -> type[Any]:
|
||||
try:
|
||||
return importlib.import_module("textual.widgets").Static
|
||||
except Exception: # pragma: no cover - textual is optional
|
||||
|
||||
class _FallbackStatic:
|
||||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||||
self._text = ""
|
||||
|
||||
def update(self, text: str) -> None:
|
||||
self._text = text
|
||||
|
||||
def focus(self) -> None: # pragma: no cover - noop in fallback
|
||||
return None
|
||||
|
||||
return _FallbackStatic
|
||||
|
||||
|
||||
_StaticBase = _load_static_base()
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ActiveSessionEntry:
|
||||
"""Renderable data for an active (in-memory) session."""
|
||||
|
||||
session_id: str
|
||||
title: str
|
||||
persona_name: str
|
||||
actor_name: str
|
||||
status: str
|
||||
last_prompt_preview: str
|
||||
last_used: datetime
|
||||
is_current: bool = False
|
||||
|
||||
def normalized_preview(self) -> str:
|
||||
preview = self.last_prompt_preview.strip()
|
||||
if not preview:
|
||||
return "(no prompts yet)"
|
||||
if len(preview) <= _MAX_PREVIEW:
|
||||
return preview
|
||||
return preview[: _MAX_PREVIEW - 1] + "…"
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SavedSessionEntry:
|
||||
"""Renderable data for a persisted session."""
|
||||
|
||||
session_id: str
|
||||
title: str
|
||||
persona_name: str
|
||||
actor_name: str
|
||||
prompt_count: int
|
||||
last_used: datetime
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class SessionsSnapshot:
|
||||
"""Aggregate payload passed to :class:`SessionsScreen`."""
|
||||
|
||||
active_sessions: list[ActiveSessionEntry]
|
||||
saved_sessions: list[SavedSessionEntry]
|
||||
active_index: int = 0
|
||||
|
||||
|
||||
def _relative_time(value: datetime, *, now: datetime | None = None) -> str:
|
||||
"""Render *value* as human-friendly relative timestamp."""
|
||||
|
||||
reference = now or datetime.now(tz=value.tzinfo)
|
||||
delta = reference - value
|
||||
seconds = max(int(delta.total_seconds()), 0)
|
||||
if seconds < 60:
|
||||
return "just now"
|
||||
minutes = seconds // 60
|
||||
if minutes < 60:
|
||||
return f"{minutes}m ago"
|
||||
hours = minutes // 60
|
||||
if hours < 24:
|
||||
return f"{hours}h ago"
|
||||
return value.strftime("%b %d %H:%M")
|
||||
|
||||
|
||||
def _render_active(snapshot: SessionsSnapshot, now: datetime | None) -> list[str]:
|
||||
lines: list[str] = []
|
||||
if not snapshot.active_sessions:
|
||||
lines.append(" (no active sessions)")
|
||||
return lines
|
||||
for index, entry in enumerate(snapshot.active_sessions):
|
||||
marker = "❯" if index == snapshot.active_index or entry.is_current else " "
|
||||
persona = entry.persona_name or "unknown"
|
||||
actor = entry.actor_name or "—"
|
||||
status = entry.status or "idle"
|
||||
timestamp = _relative_time(entry.last_used, now=now)
|
||||
lines.append(
|
||||
f" {marker} {entry.title} • {persona} • {actor} • {status} {timestamp}"
|
||||
)
|
||||
lines.append(f" {entry.normalized_preview()}")
|
||||
lines.append("")
|
||||
if lines:
|
||||
lines.pop()
|
||||
return lines
|
||||
|
||||
|
||||
def _render_saved(snapshot: SessionsSnapshot, now: datetime | None) -> list[str]:
|
||||
lines: list[str] = []
|
||||
if not snapshot.saved_sessions:
|
||||
lines.append(" (no saved sessions)")
|
||||
return lines
|
||||
for entry in snapshot.saved_sessions:
|
||||
persona = entry.persona_name or "unknown"
|
||||
actor = entry.actor_name or "—"
|
||||
prompts = f"{entry.prompt_count} prompts" if entry.prompt_count else "0 prompts"
|
||||
timestamp = _relative_time(entry.last_used, now=now)
|
||||
lines.append(
|
||||
f" {entry.title} • {persona} • {actor} • {prompts} • {timestamp}"
|
||||
)
|
||||
return lines
|
||||
|
||||
|
||||
class SessionsScreen(_StaticBase):
|
||||
"""Textual overlay rendering active and saved sessions."""
|
||||
|
||||
def __init__(self, *args: object, **kwargs: object) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self._snapshot: SessionsSnapshot | None = None
|
||||
self._visible: bool = False
|
||||
|
||||
@property
|
||||
def visible(self) -> bool:
|
||||
return self._visible
|
||||
|
||||
def show(self, snapshot: SessionsSnapshot, *, now: datetime | None = None) -> None:
|
||||
self._snapshot = snapshot
|
||||
self._visible = True
|
||||
self.update(self.render(snapshot, now=now))
|
||||
|
||||
def hide(self) -> None:
|
||||
self._snapshot = None
|
||||
self._visible = False
|
||||
self.update("")
|
||||
|
||||
def toggle(self, snapshot: SessionsSnapshot, *, now: datetime | None = None) -> bool:
|
||||
if self._visible:
|
||||
self.hide()
|
||||
return False
|
||||
self.show(snapshot, now=now)
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def render(snapshot: SessionsSnapshot, *, now: datetime | None = None) -> str:
|
||||
header = "┌─ Sessions ─────────────────────────────────────────────────────────────────────────────┐"
|
||||
footer = "└────────────────────────────────────────────────────────────────────────────────────────┘"
|
||||
separator = "│────────────────────────────────────────────────────────────────────────────────────────│"
|
||||
lines: list[str] = [header]
|
||||
active_lines = _render_active(snapshot, now)
|
||||
if active_lines:
|
||||
for chunk in active_lines:
|
||||
if chunk:
|
||||
lines.append(f"│ {chunk.ljust(88)}│")
|
||||
else:
|
||||
lines.append("│ │")
|
||||
else:
|
||||
lines.append("│ (no active sessions) │")
|
||||
lines.append(separator)
|
||||
lines.append("│ Saved Sessions (from previous runs) │")
|
||||
saved_lines = _render_saved(snapshot, now)
|
||||
if saved_lines:
|
||||
for chunk in saved_lines:
|
||||
lines.append(f"│ {chunk.ljust(88)}│")
|
||||
else:
|
||||
lines.append("│ (no saved sessions) │")
|
||||
lines.append(separator)
|
||||
lines.append(
|
||||
"│ enter Switch │ ctrl+r Resume saved │ ctrl+n New │ d Delete │ r Rename │ esc Back │"
|
||||
)
|
||||
lines.append(footer)
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"ActiveSessionEntry",
|
||||
"SavedSessionEntry",
|
||||
"SessionsSnapshot",
|
||||
"SessionsScreen",
|
||||
]
|
||||
Reference in New Issue
Block a user