showing and selectig sessions is working well enough

This commit is contained in:
Joe Fleming
2026-03-07 14:17:58 -07:00
parent ad7030ae38
commit 71afa51f09

339
app.py
View File

@@ -5,13 +5,12 @@ A Textual-based TUI for managing OpenCode sessions.
from textual.app import App, ComposeResult from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical from textual.containers import Container, Horizontal, Vertical
from textual.widgets import Header, Footer, Static, Input, Button, Select, Tree, DataTable, Label from textual.widgets import Header, Footer, Input, Button, Select, Tree, DataTable, Label, Static
from textual.screen import ModalScreen from textual.screen import ModalScreen
from textual import events
from typing import List, Optional from typing import List, Optional
from database import Database, Session, Project from database import Database, Session
from datetime import datetime from datetime import datetime
from pathlib import Path import os
class MoveCopyDialog(ModalScreen): class MoveCopyDialog(ModalScreen):
@@ -29,39 +28,46 @@ class MoveCopyDialog(ModalScreen):
self.is_copy = is_copy self.is_copy = is_copy
self.target_project: Optional[str] = None self.target_project: Optional[str] = None
self.target_workspace: Optional[str] = None self.target_workspace: Optional[str] = None
self.sql_preview: List[str] = []
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
# Get currently selected project from tree to exclude from destination list
current_project_id = self.app.filter_project_id if hasattr(self.app, 'filter_project_id') else None
project_options = [
(os.path.basename(p.worktree) if p.worktree else (p.name or p.id), p.id)
for p in self.db.projects.values()
if p.id != current_project_id # Exclude current project
]
if not project_options:
project_options = [("-- Same project --", "")]
with Container(id="dialog"): with Container(id="dialog"):
yield Label(f"{'Copy' if self.is_copy else 'Move'} Sessions", id="dialog-title") yield Label(f"{'📋 Copy' if self.is_copy else '➡️ Move'} {len(self.sessions)} Session(s)", id="dialog-title")
yield Label("Target Project:") yield Label("Destination Project:", id="preview-label")
yield Select( yield Select(
[("-- Select Project --", "")] + [ [("-- Select --", "")] + project_options,
(f"{p.name or p.worktree} ({p.id})", p.id)
for p in self.db.projects.values()
],
id="project-select", id="project-select",
prompt="Select a project" prompt="Select a project"
) )
yield Label("Target Workspace (optional):") yield Label("Workspace (optional):")
yield Select( yield Select(
[("-- No workspace --", "")] + [ [("-- None --", "")] + [
(f"{w.name or w.branch or w.id}", w.id) (f"{w.name or w.branch or w.id}", w.id)
for w in self.db.workspaces.values() for w in self.db.workspaces.values()
], ],
id="workspace-select", id="workspace-select",
prompt="Select a workspace" prompt="Select workspace"
) )
yield Label("Preview:", id="preview-label")
yield Static("", id="sql-preview") yield Static("", id="sql-preview")
with Horizontal(): with Horizontal(id="dialog-buttons"):
yield Button("Cancel", variant="error", id="cancel-btn") yield Button("Cancel", variant="default", id="cancel-btn")
yield Button("Confirm", variant="primary", id="confirm-btn", disabled=True) yield Button("Confirm", variant="primary", id="confirm-btn", disabled=True)
def on_mount(self) -> None: def on_mount(self) -> None:
self.query_one("#dialog").border_title = f"{'Copy' if self.is_copy else 'Move'} Sessions" self.query_one("#dialog").border_title = f"{'Copy' if self.is_copy else 'Move'} Sessions"
def on_select_changed(self, event: Select.Changed) -> None: def on_select_changed(self, event):
if event.select.id == "project-select": if event.select.id == "project-select":
self.target_project = event.value if event.value else None self.target_project = event.value if event.value else None
elif event.select.id == "workspace-select": elif event.select.id == "workspace-select":
@@ -71,42 +77,38 @@ class MoveCopyDialog(ModalScreen):
confirm_btn = self.query_one("#confirm-btn") confirm_btn = self.query_one("#confirm-btn")
confirm_btn.disabled = not (self.target_project is not None) confirm_btn.disabled = not (self.target_project is not None)
def update_preview(self) -> None: def update_preview(self):
if not self.target_project: if not self.target_project:
self.sql_preview = [] preview = self.query_one("#sql-preview")
self.query_one("#sql-preview").update("Select a project to see preview") preview.update("Select a project to see preview")
return return
target_proj = self.db.projects.get(self.target_project) target_proj = self.db.projects.get(self.target_project)
target_ws = self.db.workspaces.get(self.target_workspace) if self.target_workspace else None target_ws = self.db.workspaces.get(self.target_workspace) if self.target_workspace else None
preview_lines = [ lines = ["", f"Source: {len(self.sessions)} session(s)"]
f"Source: {len(self.sessions)} session(s)", if target_proj:
f"Destination Project: {target_proj.name or target_proj.worktree if target_proj else 'Unknown'}", lines.append(f"Destination Project: {target_proj.name or target_proj.worktree}")
]
if target_ws: if target_ws:
preview_lines.append(f"Destination Workspace: {target_ws.name or target_ws.branch or target_ws.id}") lines.append(f"Destination Workspace: {target_ws.name or target_ws.branch or target_ws.id}")
else: else:
preview_lines.append("Destination Workspace: (keep current or none)") lines.append("Destination Workspace: (keep current or none)")
lines.append("")
preview_lines.append("") lines.append("SQL to execute:")
preview_lines.append("SQL statements:")
for sess in self.sessions: for sess in self.sessions:
if self.is_copy: if self.is_copy:
preview_lines.append(f" COPY session {sess.id[:8]}... → new ID (with all related data)") lines.append(f" COPY session {sess.id[:12]}...")
else: else:
if target_ws: if target_ws:
preview_lines.append(f" UPDATE session SET project_id = {self.target_project}, workspace_id = {self.target_workspace} WHERE id = {sess.id[:16]}...;") lines.append(f" UPDATE session SET project={self.target_project[:12]}, workspace={self.target_workspace[:12] if self.target_workspace else 'NULL'} WHERE id={sess.id[:12]}...")
else: else:
preview_lines.append(f" UPDATE session SET project_id = {self.target_project} WHERE id = {sess.id[:16]}...;") lines.append(f" UPDATE session SET project={self.target_project[:12]} WHERE id={sess.id[:12]}...")
if self.is_copy: if self.is_copy:
preview_lines.append("\nNote: Copy creates new sessions with all associated messages, parts, todos, and shares.") lines.append("\nNote: Copy creates new sessions with all related data.")
self.sql_preview = preview_lines self.query_one("#sql-preview").update("\n".join(lines))
self.query_one("#sql-preview").update("\n".join(preview_lines))
def action_cancel(self) -> None: def action_cancel(self) -> None:
self.app.pop_screen() self.app.pop_screen()
@@ -132,8 +134,10 @@ class MoveCopyDialog(ModalScreen):
if success: if success:
self.app.notify(f"{'Copied' if self.is_copy else 'Moved'} {len(self.sessions)} session(s) successfully!", severity="information") self.app.notify(f"{'Copied' if self.is_copy else 'Moved'} {len(self.sessions)} session(s) successfully!", severity="information")
self.app.pop_screen() self.app.pop_screen()
# Trigger refresh in main app # Get reference to main app and refresh
self.app.query_one(SessionMoverApp).refresh_sessions() app = self.app
if hasattr(app, 'refresh_sessions'):
app.refresh_sessions()
else: else:
self.app.notify(f"Operation failed: {sql}", severity="error") self.app.notify(f"Operation failed: {sql}", severity="error")
@@ -141,7 +145,7 @@ class MoveCopyDialog(ModalScreen):
class BackupDialog(ModalScreen): class BackupDialog(ModalScreen):
"""Dialog showing backup creation.""" """Dialog showing backup creation."""
def __init__(self, backup_path: Path): def __init__(self, backup_path):
super().__init__() super().__init__()
self.backup_path = backup_path self.backup_path = backup_path
@@ -162,13 +166,17 @@ class SessionMoverApp(App):
CSS = """ CSS = """
Screen { Screen {
layout: horizontal; layout: vertical;
}
#main-hsplit {
height: 1fr;
} }
#project-panel { #project-panel {
width: 30%; width: 30%;
border-right: solid $primary; border-right: solid $primary;
padding: 1; padding: 0 1;
} }
#session-panel { #session-panel {
@@ -185,9 +193,13 @@ class SessionMoverApp(App):
padding: 0 1; padding: 0 1;
} }
#filter-bar {
height: auto;
padding: 0 0 1 0;
}
#filter-input { #filter-input {
width: 100%; width: 1fr;
margin-bottom: 1;
} }
#sessions-table { #sessions-table {
@@ -198,27 +210,29 @@ class SessionMoverApp(App):
border: solid $primary-lighten-2; border: solid $primary-lighten-2;
} }
#action-bar { /* Footer will be at bottom automatically */
#current-selection {
height: auto;
padding: 1;
background: $accent;
color: $text;
}
#selection-count {
height: auto; height: auto;
padding: 1; padding: 1;
background: $surface; background: $surface;
}
Button {
margin-right: 1;
}
#hotkeys {
color: $text-muted; color: $text-muted;
} }
/* Dialog styles */ /* Dialog styles */
#dialog, #backup-dialog { #dialog, #backup-dialog {
width: 60; width: 50;
height: auto; height: auto;
border: thick $primary;
background: $surface; background: $surface;
padding: 2; padding: 1;
border: solid $primary;
} }
#dialog-title, #backup-title { #dialog-title, #backup-title {
@@ -227,57 +241,65 @@ class SessionMoverApp(App):
margin-bottom: 1; margin-bottom: 1;
} }
#dialog-select {
margin-bottom: 1;
}
#sql-preview { #sql-preview {
height: 10; height: 8;
border: solid $primary-lighten-3; border: solid $primary-darken-1;
padding: 1; padding: 1;
overflow: auto; margin: 1 0;
} }
#preview-label { #preview-label {
margin-top: 1;
text-style: bold; text-style: bold;
} }
.hidden { #dialog-buttons {
display: none; height: auto;
align: center middle;
}
#dialog-buttons Button {
margin: 0 1;
min-width: 15;
} }
""" """
BINDINGS = [ BINDINGS = [
("q", "quit", "Quit"), ("q", "quit", "Quit"),
("b", "backup", "Backup DB"), ("b", "backup", "Backup"),
("m", "move", "Move"), ("m", "move", "Move"),
("c", "copy", "Copy"), ("c", "copy", "Copy"),
("a", "toggle_archived", "Show Archived"), ("a", "toggle_archived", "Archived"),
("r", "refresh", "Refresh"), ("r", "refresh", "Refresh"),
("escape", "cancel_selection", "Deselect"), ("space", "toggle_selection", "Select"),
("escape", "clear_selection", "Clear"),
] ]
def __init__(self, db_path: str = "opencode.db"): def __init__(self, db_path: str = "opencode.db"):
super().__init__() super().__init__()
self.db = Database(db_path) self.db = Database(db_path)
self.selected_sessions: List[Session] = [] self.selected_session_ids: set = set() # Track by ID
self.show_archived = False self.show_archived = False
self.current_filter = "" self.current_filter = ""
self.backup_made = False self.backup_made = False
self.filter_project_id = None # Start with no project selected
self.filter_workspace_id = None
def compose(self) -> ComposeResult: def compose(self) -> ComposeResult:
yield Header() yield Header()
with Horizontal(): with Horizontal(id="main-hsplit"):
with Vertical(id="project-panel"): with Vertical(id="project-panel"):
yield Static("Projects & Workspaces", id="projects-title")
yield Tree("Projects", id="project-tree") yield Tree("Projects", id="project-tree")
with Vertical(id="session-panel"): with Vertical(id="session-panel"):
with Horizontal(): yield Static("Select a project from the left", id="current-selection")
yield Input(placeholder="Filter sessions...", id="filter-input") with Horizontal(id="filter-bar"):
yield Input(placeholder="🔍 Filter...", id="filter-input")
yield Button("Clear", id="filter-clear", variant="default") yield Button("Clear", id="filter-clear", variant="default")
yield DataTable(id="sessions-table") yield DataTable(id="sessions-table")
with Container(id="action-bar"): yield Static("0 selected", id="selection-count")
yield Static(
"Hotkeys: [b] Backup [m] Move [c] Copy [a] Archive [r] Refresh [q] Quit",
id="hotkeys"
)
yield Footer() yield Footer()
def on_mount(self) -> None: def on_mount(self) -> None:
@@ -290,128 +312,177 @@ class SessionMoverApp(App):
tree.root.expand() tree.root.expand()
self.refresh_project_tree() self.refresh_project_tree()
# Setup sessions table # Setup sessions table (but don't load data yet)
table = self.query_one("#sessions-table") table = self.query_one("#sessions-table")
table.zebra_stripes = True table.zebra_stripes = True
table.cursor_type = "row" table.cursor_type = "row"
table.add_columns("ID", "Title", "Project", "Workspace", "Msgs", "Todos", "Created", "Archived") table.add_columns(" ", "ID", "Title", "Project", "Workspace", "Msgs", "Created")
self.refresh_sessions()
# Don't load sessions until a project is selected
status = self.query_one("#current-selection")
status.update("Select a project from the left")
def on_data_table_row_selected(self, event):
"""Handle row selection in either table."""
# Use click to toggle selection in sessions table only
pass # Handled by click handler
def refresh_project_tree(self) -> None: def refresh_project_tree(self) -> None:
"""Update the project/workspace tree.""" """Update the project tree."""
tree = self.query_one("#project-tree") tree = self.query_one("#project-tree")
tree.clear() tree.clear()
root = tree.root root = tree.root
root.data = None root.data = None
root.expand()
for proj_name, ws_names, has_workspaces in self.db.get_project_tree(): # Flat list of projects
proj_node = root.add(proj_name, {"type": "project", "name": proj_name}) for proj in self.db.projects.values():
proj_node.expand() proj_name = os.path.basename(proj.worktree) if proj.worktree else (proj.name or "Unknown")
root.add(proj_name, {"type": "project", "id": proj.id, "label": proj_name, "full_path": proj.worktree})
if has_workspaces:
for ws_name in ws_names:
proj_node.add_leaf(ws_name, {"type": "workspace", "name": ws_name})
def refresh_sessions(self) -> None: def refresh_sessions(self) -> None:
"""Update the sessions table.""" """Update the sessions table."""
table = self.query_one("#sessions-table") table = self.query_one("#sessions-table")
# Save cursor position before clearing
cursor_row = table.cursor_row
cursor_column = table.cursor_column
table.clear(columns=True) table.clear(columns=True)
table.add_columns("ID", "Title", "Project", "Workspace", "Msgs", "Todos", "Created", "Archived") table.add_columns(" ", "ID", "Title", "Project", "Workspace", "Msgs", "Created")
sessions = self.db.get_sessions( sessions = self.db.get_sessions(
include_archived=self.show_archived, include_archived=self.show_archived,
project_id=self.filter_project_id,
workspace_id=self.filter_workspace_id,
search=self.current_filter search=self.current_filter
) )
for sess in sessions: for sess in sessions:
created_dt = datetime.fromtimestamp(sess.time_created / 1000).strftime("%Y-%m-%d %H:%M") if sess.time_created else "N/A" created_dt = datetime.fromtimestamp(sess.time_created / 1000).strftime("%m-%d %H:%M") if sess.time_created else "N/A"
archived_str = "🗓️" if sess.time_archived else ""
# Check if selected by ID
marker = "" if sess.id in self.selected_session_ids else ""
table.add_row( table.add_row(
marker,
sess.id[:8], sess.id[:8],
sess.title[:50], sess.title,
sess.project_name or sess.project_id[:8], sess.project_name[:20] if sess.project_name and len(sess.project_name) > 20 else (sess.project_name or sess.project_id[:8]),
sess.workspace_name or "", sess.workspace_name[:15] if sess.workspace_name and len(sess.workspace_name) > 15 else (sess.workspace_name or ""),
str(sess.message_count), str(sess.message_count),
str(sess.todo_count),
created_dt, created_dt,
archived_str, key=sess.id # Full ID as key
key=sess.id
) )
self.notify(f"Loaded {len(sessions)} sessions", severity="information") # Restore cursor position if valid
if cursor_row is not None and cursor_row < len(sessions):
table.move_cursor(row=cursor_row, column=cursor_column)
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: # Update selection count
"""Handle row selection.""" count = len(self.selected_session_ids)
session_id = event.row_key count_widget = self.query_one("#selection-count")
if session_id:
session = self.db.get_session(session_id)
if session:
if session in self.selected_sessions:
self.selected_sessions.remove(session)
else:
self.selected_sessions.append(session)
self.update_status()
def update_status(self) -> None:
"""Update footer/status with selection count."""
count = len(self.selected_sessions)
if count > 0: if count > 0:
self.notify(f"{count} session(s) selected", severity="information") count_widget.update(f"{count} session(s) selected")
else:
count_widget.update("0 selected")
def on_input_changed(self, event: Input.Changed) -> None: def on_tree_node_highlighted(self, event):
"""Handle filter input.""" """Handle project tree selection."""
if event.input.id == "filter-input": tree = self.query_one("#project-tree")
self.current_filter = event.value node = tree.cursor_node
self.refresh_sessions()
if node is None:
return
node_data = getattr(node, 'data', None)
if node_data is None:
return
node_type = node_data.get("type")
if node_type != "project":
return
status = self.query_one("#current-selection")
proj_id = node_data.get("id")
proj = self.db.projects.get(proj_id)
self.filter_project_id = proj_id
self.filter_workspace_id = None
if proj:
status.update(f"📁 {proj.worktree}")
# Clear selection when changing projects
if self.selected_session_ids:
count = len(self.selected_session_ids)
self.selected_session_ids.clear()
self.notify(f"Selection cleared ({count} session(s)))", severity="information")
self.refresh_sessions()
def on_button_pressed(self, event: Button.Pressed) -> None: def action_toggle_selection(self) -> None:
"""Handle button clicks.""" """Toggle selection of current row."""
if event.button.id == "filter-clear": table = self.query_one("#sessions-table")
self.query_one("#filter-input").value = "" row_key = table.cursor_row
self.current_filter = "" if row_key is not None:
self.refresh_sessions() sessions = self.db.get_sessions(
include_archived=self.show_archived,
project_id=self.filter_project_id,
workspace_id=self.filter_workspace_id,
search=self.current_filter
)
if 0 <= row_key < len(sessions):
session_id = sessions[row_key].id
if session_id in self.selected_session_ids:
self.selected_session_ids.discard(session_id)
else:
self.selected_session_ids.add(session_id)
self.refresh_sessions()
def action_backup(self) -> None: def action_clear_selection(self) -> None:
"""Create database backup.""" """Clear all selections."""
backup_path = self.db.create_backup() self.selected_session_ids.clear()
self.push_screen(BackupDialog(backup_path)) self.refresh_sessions()
def action_move(self) -> None: def action_move(self) -> None:
"""Open move dialog.""" """Open move dialog."""
if not self.selected_sessions: if not self.selected_session_ids:
self.notify("No sessions selected", severity="warning") self.notify("No sessions selected", severity="warning")
return return
# Ensure backup exists before write operation # Get session objects for the dialog
sessions = [self.db.get_session(sid) for sid in self.selected_session_ids]
sessions = [s for s in sessions if s]
if not self.backup_made: if not self.backup_made:
self.db.create_backup() self.db.create_backup()
self.backup_made = True self.backup_made = True
self.notify("Backup created", severity="information") self.notify("Backup created", severity="information")
self.push_screen(MoveCopyDialog(self.db, self.selected_sessions, is_copy=False)) self.push_screen(MoveCopyDialog(self.db, sessions, is_copy=False))
def action_copy(self) -> None: def action_copy(self) -> None:
"""Open copy dialog.""" """Open copy dialog."""
if not self.selected_sessions: if not self.selected_session_ids:
self.notify("No sessions selected", severity="warning") self.notify("No sessions selected", severity="warning")
return return
# Ensure backup exists before write operation # Get session objects for the dialog
sessions = [self.db.get_session(sid) for sid in self.selected_session_ids]
sessions = [s for s in sessions if s]
if not self.backup_made: if not self.backup_made:
self.db.create_backup() self.db.create_backup()
self.backup_made = True self.backup_made = True
self.notify("Backup created", severity="information") self.notify("Backup created", severity="information")
self.push_screen(MoveCopyDialog(self.db, self.selected_sessions, is_copy=True)) self.push_screen(MoveCopyDialog(self.db, sessions, is_copy=True))
def action_toggle_archived(self) -> None: def action_toggle_archived(self) -> None:
"""Toggle archived sessions visibility.""" """Toggle archived sessions visibility."""
self.show_archived = not self.show_archived self.show_archived = not self.show_archived
self.refresh_sessions() self.refresh_sessions()
status = "shown" if self.show_archived else "hidden"
self.notify(f"Archived sessions {status}", severity="information")
def action_refresh(self) -> None: def action_refresh(self) -> None:
"""Reload all data from database.""" """Reload all data from database."""