commit e2fffefcf261df790eb800b9e692c842472d7ab1 Author: enne2 Date: Wed Aug 19 11:58:37 2026 +0200 Pi Session Launcher: GUI GTK3 per riprendere le sessioni di Pi Agent - elenca sessioni da ~/.pi/agent/sessions/**/*.jsonl (cwd, primo messaggio, nome, ultima modifica) - ricerca/filtro, ordina per recente, nomi in grassetto - doppio clic/Invio: apre Konsole nella cwd della sessione con pi --session - rinomina sessione (append session_info, ultima entry vince come in pi) - install.sh: voce menu KDE diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..faaede6 --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +build/ +dist/ +*.spec +.venv/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..0b045b4 --- /dev/null +++ b/README.md @@ -0,0 +1,47 @@ +# Pi Session Launcher + +Piccola applicazione GUI (GTK3) per **riprendere le sessioni di Pi Agent**. + +Elenco tutte le sessioni salvate in `~/.pi/agent/sessions/` e, selezionandone una: + +- apre un terminale (Konsole) **nella cartella di lavoro** (`cwd`) in cui la sessione era stata avviata; +- riprende la sessione con `pi --session `. + +## Requisiti + +- Python 3 + PyGObject GTK3 (`python3-gi`, `gir1.2-gtk-3.0`) +- Konsole (KDE) — fallback: `xterm` +- `pi` nel PATH + +## Installazione + +```bash +./install.sh # copia la voce nel menu applicazioni KDE (.desktop) +``` + +Avvio manuale: + +```bash +python3 pi-session-launcher.py +``` + +## Uso + +| Azione | Come | +|---|---| +| Selezionare una sessione | clic sulla riga | +| **Riprendere** la sessione | doppio clic, Invio o pulsante **Riprendi sessione** | +| Cercare | digitare nel campo di ricerca (filtra per messaggio e cartella) | +| Aprire la cartella della sessione | pulsante **Apri cartella** (Dolphin) | +| Aggiornare l'elenco | pulsante **Aggiorna** o `Ctrl+R` | + +Le sessioni **con nome** (`/name` in pi) sono mostrate in **grassetto**. + +## Come funziona + +- Ogni sessione è un file JSONL in `~/.pi/agent/sessions/----/_.jsonl`. +- L'header della prima riga contiene il campo `cwd` (cartella di avvio). +- Il primo messaggio utente e il nome (`session_info`) vengono letti dal file. +- Il lancio usa: + `konsole --workdir --hold --separate -e bash -c 'pi --session "$1"; exec bash' _ ` + — `exec bash` mantiene la shell aperta quando pi termina. diff --git a/install.sh b/install.sh new file mode 100755 index 0000000..2d76486 --- /dev/null +++ b/install.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +# install.sh — Pi Session Launcher +# Crea la voce nel menu applicazioni KDE (.desktop) e rende eseguibile lo script. +set -euo pipefail + +DIR="$(cd "$(dirname "$0")" && pwd)" +APP="$DIR/pi-session-launcher.py" +DESKTOP="$HOME/.local/share/applications/pi-session-launcher.desktop" + +# sintassi: ./install.sh [--uninstall] +if [[ "${1:-}" == "--uninstall" ]]; then + rm -f "$DESKTOP" + echo "Rimossa la voce di menu: $DESKTOP" + exit 0 +fi + +chmod +x "$APP" + +mkdir -p "$(dirname "$DESKTOP")" +cat > "$DESKTOP" < + +Requisiti: python3-gi (GTK3), konsole (o xterm come fallback), pi installato. +""" + +import json +import os +import re +import shutil +import subprocess +import sys +import time +from datetime import datetime, timezone + +import gi + +gi.require_version("Gtk", "3.0") +gi.require_version("Gdk", "3.0") +from gi.repository import Gdk, GLib, Gtk, Pango + +APP_TITLE = "Pi Session Launcher" +SESSIONS_DIR = os.path.expanduser("~/.pi/agent/sessions/") +MAX_NAME_SCAN = 16 * 1024 * 1024 # oltre questa dimensione, scandisci solo inizio/fine + +# --------------------------------------------------------------------------- +# Lettura sessioni +# --------------------------------------------------------------------------- + +def _decode(data: bytes) -> str: + return data.decode("utf-8", "replace") + + +def _first_user_message(text: str) -> str: + """Cerca il primo messaggio con role=user nei primi chunk del file.""" + for line in text.splitlines(): + line = line.strip() + if not line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if entry.get("type") != "message": + continue + msg = entry.get("message") or {} + if msg.get("role") != "user": + continue + content = msg.get("content", "") + if isinstance(content, list): + parts = [] + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + parts.append(block.get("text", "")) + return " ".join(parts).strip() + if isinstance(content, str): + return content.strip() + return "" + + +def _session_name(full_text: str) -> str: + """Nome mostrato della sessione: come pi, l'ULTIMA entry session_info vince + (nome vuoto cancella il titolo).""" + name = "" + for line in reversed(full_text.splitlines()): + line = line.strip() + if not line: + continue + if '"type":"session_info"' not in line and '"type": "session_info"' not in line: + continue + try: + entry = json.loads(line) + except json.JSONDecodeError: + continue + if entry.get("type") == "session_info": + return entry.get("name") or "" + return name + + +def parse_session(path: str): + """Legge un file .jsonl di sessione e ne estrae le info utili.""" + try: + stat = os.stat(path) + except OSError: + return None + if stat.st_size == 0: + return None + + info = { + "path": path, + "cwd": "", + "sid": "", + "name": "", + "first": "", + "mtime": stat.st_mtime, + "size": stat.st_size, + } + + try: + with open(path, "rb") as f: + header_raw = f.readline().decode("utf-8", "replace") + header = json.loads(header_raw) + if header.get("type") != "session": + return None + info["cwd"] = header.get("cwd", "") or "" + info["sid"] = header.get("id", "") or "" + + # primo messaggio utente: leggiamo i primi ~64 KB (righe iniziali) + f.seek(0) + head = _decode(f.read(65536)) + info["first"] = _first_user_message(head) + + # nome sessione: scandisci tutto il file (con limite prudente) + if stat.st_size <= MAX_NAME_SCAN: + full = head + _decode(f.read()) + else: + f.seek(max(0, stat.st_size - 1024 * 1024)) + tail = _decode(f.read()) + full = head + tail + info["name"] = _session_name(full) + except (OSError, json.JSONDecodeError, ValueError): + return None + + return info + + +def rename_session(path, new_name): + """Imposta (o cancella, con '') il nome visualizzato di una sessione, + appendendo una entry session_info al file JSONL (stesso meccanismo di pi). + pi usa il nome dell'ultima entry session_info presente nel file.""" + last_id = None + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line: + continue + try: + e = json.loads(line) + except json.JSONDecodeError: + continue + if e.get("type") != "session" and e.get("id"): + last_id = e["id"] + entry = { + "type": "session_info", + "id": os.urandom(4).hex(), + "parentId": last_id, + "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z", + "name": new_name, + } + with open(path, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, ensure_ascii=False, separators=(",", ":")) + "\n") + + +def scan_sessions(): + """Elenca tutte le sessioni, ordinate per ultima modifica (più recente prima).""" + sessions = [] + if not os.path.isdir(SESSIONS_DIR): + return sessions + for root, _dirs, files in os.walk(SESSIONS_DIR): + for fn in files: + if not fn.endswith(".jsonl"): + continue + info = parse_session(os.path.join(root, fn)) + if info: + sessions.append(info) + sessions.sort(key=lambda s: s["mtime"], reverse=True) + return sessions + + +# --------------------------------------------------------------------------- +# Formattazione +# --------------------------------------------------------------------------- + +def fmt_relative(ts: float) -> str: + d = time.time() - ts + if d < 0: + return "adesso" + if d < 60: + return f"{int(d)}s fa" + if d < 3600: + return f"{int(d // 60)}min fa" + if d < 86400: + return f"{int(d // 3600)}h fa" + if d < 172800: + return "ieri" + return f"{int(d // 86400)}g fa" + + +def fmt_absolute(ts: float) -> str: + return time.strftime("%d/%m/%Y %H:%M", time.localtime(ts)) + + +def fmt_size(n: int) -> str: + if n < 1024: + return f"{n} B" + if n < 1024 * 1024: + return f"{n / 1024:.0f} KB" + return f"{n / (1024 * 1024):.1f} MB" + + +def display_text(info: dict, limit: int = 100) -> str: + """Testo principale mostrato per la sessione: nome se presente, altrimenti primo messaggio.""" + if info["name"]: + return info["name"] + first = info["first"].replace("\n", " ") + if len(first) > limit: + first = first[: limit - 1] + "…" + return first or "(sessione senza messaggi)" + + +# --------------------------------------------------------------------------- +# GUI +# --------------------------------------------------------------------------- + +class LauncherWindow(Gtk.ApplicationWindow): + def __init__(self, app): + super().__init__(application=app, title=APP_TITLE) + self.set_default_size(1000, 600) + + # modello: markup_sessione, cwd, rel_time, abs_time, mtime(float), path, size(int), search_text + self.store = Gtk.ListStore(str, str, str, str, float, str, int, str) + self.by_path = {} + + self.search_entry = Gtk.SearchEntry(placeholder_text="Cerca sessione o cartella…") + self.search_entry.connect("search-changed", self._on_search_changed) + + # vista lista + self.filter = self.store.filter_new() + self.filter.set_visible_func(self._filter_visible) + self.tree = Gtk.TreeView(model=self.filter) + self.tree.set_search_column(0) + self.tree.set_headers_clickable(True) + + rend = Gtk.CellRendererText(ellipsize="end") + col = Gtk.TreeViewColumn("Sessione", rend, markup=0) + col.set_expand(True) + col.set_min_width(420) + col.set_sort_column_id(7) + self.tree.append_column(col) + + rend2 = Gtk.CellRendererText(ellipsize="middle") + col2 = Gtk.TreeViewColumn("Cartella", rend2, text=1) + col2.set_expand(True) + col2.set_min_width(280) + col2.set_sort_column_id(1) + self.tree.append_column(col2) + + rend3 = Gtk.CellRendererText() + col3 = Gtk.TreeViewColumn("Modificata", rend3, text=2) + col3.set_sort_column_id(4) + col3.set_alignment(1.0) + self.tree.append_column(col3) + + self.tree.get_selection().set_mode(Gtk.SelectionMode.SINGLE) + self.tree.connect("row-activated", self._on_row_activated) + self.tree.connect("key-press-event", self._on_key_press) + + scrolled = Gtk.ScrolledWindow() + scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC) + scrolled.add(self.tree) + + # barra pulsanti + btn_refresh = Gtk.Button(label="Aggiorna", tooltip_text="Ricarica l'elenco (Ctrl+R)") + btn_refresh.connect("clicked", lambda *_: self.refresh()) + btn_open_dir = Gtk.Button(label="Apri cartella", tooltip_text="Apre la cartella della sessione in Dolphin") + btn_open_dir.connect("clicked", lambda *_: self.open_folder()) + btn_rename = Gtk.Button(label="Rinomina", tooltip_text="Rinomina la sessione selezionata (F2)") + btn_rename.connect("clicked", lambda *_: self.rename()) + btn_launch = Gtk.Button(label="Riprendi sessione", tooltip_text="Apre Konsole nella cartella e riprende la sessione (Invio)") + btn_launch.get_style_context().add_class("suggested-action") + btn_launch.connect("clicked", lambda *_: self.launch()) + btn_quit = Gtk.Button(label="Esci") + btn_quit.connect("clicked", lambda *_: self.close()) + + hbox = Gtk.Box(spacing=8) + hbox.pack_start(btn_refresh, False, False, 0) + hbox.pack_start(btn_open_dir, False, False, 0) + hbox.pack_start(btn_rename, False, False, 0) + hbox.pack_start(Gtk.Box(), True, True, 0) # spaziatore + hbox.pack_start(btn_launch, False, False, 0) + hbox.pack_start(btn_quit, False, False, 0) + + # status bar + self.status = Gtk.Label(label="", xalign=0.0, halign=Gtk.Align.START) + self.status.set_ellipsize(Pango.EllipsizeMode.END) + + box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=6) + box.set_border_width(8) + box.pack_start(self.search_entry, False, False, 0) + box.pack_start(scrolled, True, True, 0) + box.pack_start(hbox, False, False, 0) + sep = Gtk.Separator() + box.pack_start(sep, False, False, 0) + box.pack_start(self.status, False, False, 0) + self.add(box) + + self.refresh() + self.show_all() + self.search_entry.grab_focus() + + # -- popolamento ------------------------------------------------------ + + def refresh(self): + sessions = scan_sessions() + selected_path = self.selected_path() + self.store.clear() + self.by_path.clear() + for s in sessions: + text = display_text(s) + markup = GLib.markup_escape_text(text) + if s["name"]: + markup = f"{markup}" + self.store.append([markup, s["cwd"], fmt_relative(s["mtime"]), + fmt_absolute(s["mtime"]), s["mtime"], s["path"], s["size"], text.lower()]) + self.by_path[s["path"]] = s + if selected_path in self.by_path: + self.select_path(selected_path) + self._update_status() + + def _update_status(self): + n = len(self.store) + self.status.set_text( + f"{n} session{'i' if n != 1 else 'e'} · {SESSIONS_DIR}" + ) + + # -- selezione -------------------------------------------------------- + + def selected_iter(self): + model, it = self.tree.get_selection().get_selected() + return model, it + + def selected_path(self): + model, it = self.selected_iter() + if it is None: + return None + return model[it][5] + + def select_path(self, path): + it = self.store.get_iter_first() + while it is not None: + if self.store[it][5] == path: + self.tree.get_selection().select_iter(it) + self.tree.scroll_to_cell(self.store.get_path(it)) + return + it = self.store.iter_next(it) + + # -- filtri ----------------------------------------------------------- + + def _filter_visible(self, model, it, _data=None): + q = self.search_entry.get_text().strip().lower() + if not q: + return True + return q in model[it][7] or q in model[it][1].lower() + + def _on_search_changed(self, _entry): + self.filter.refilter() + + # -- azioni ----------------------------------------------------------- + + def _on_row_activated(self, _tree, _path, _col): + self.launch() + + def _on_key_press(self, _widget, event): + if event.keyval in (Gdk.KEY_Return, Gdk.KEY_KP_Enter): + self.launch() + return True + if event.keyval == Gdk.KEY_F2: + self.rename() + return True + if event.keyval == Gdk.KEY_r and event.state & Gdk.ModifierType.CONTROL_MASK: + self.refresh() + return True + return False + + def launch(self): + path = self.selected_path() + if not path: + self._flash_status("Seleziona prima una sessione dalla lista.") + return + info = self.by_path.get(path, {}) + cwd = info.get("cwd") or os.path.expanduser("~") + if not os.path.isdir(cwd): + cwd = os.path.expanduser("~") + + konsole = shutil.which("konsole") + if konsole: + # wrapper bash: più affidabile del passaggio diretto di -e a Konsole + # ("exec bash" mantiene la shell aperta quando pi esce) + cmd = [konsole, "--workdir", cwd, "--hold", "--separate", + "-e", "bash", "-c", 'pi --session "$1"; exec bash', "_", path] + else: + term = shutil.which("xterm") + if not term: + self._flash_status("Nessun terminale trovato (installa konsole o xterm).") + return + cmd = [term, "-hold", "-e", "pi", "--session", path] + + try: + subprocess.Popen(cmd, start_new_session=True) + except OSError as e: + self._flash_status(f"Impossibile aprire il terminale: {e}") + return + self._flash_status(f"Avviato: pi --session {os.path.basename(path)} in {cwd}") + + def rename(self): + path = self.selected_path() + if not path: + self._flash_status("Seleziona prima una sessione dalla lista.") + return + info = self.by_path.get(path, {}) + current = info.get("name", "") + + dialog = Gtk.Dialog(title="Rinomina sessione", transient_for=self, modal=True) + dialog.add_buttons(Gtk.STOCK_CANCEL, Gtk.ResponseType.CANCEL, + Gtk.STOCK_OK, Gtk.ResponseType.OK) + entry = Gtk.Entry(text=current) + entry.set_activates_default(True) + lbl = Gtk.Label(label="Nome mostrato (vuoto = cancella il nome):") + lbl.set_xalign(0.0) + box = dialog.get_content_area() + box.set_border_width(12) + box.set_spacing(8) + box.pack_start(lbl, False, False, 0) + box.pack_start(entry, False, False, 0) + dialog.set_default_response(Gtk.ResponseType.OK) + dialog.show_all() + response = dialog.run() + name = entry.get_text().strip() + dialog.destroy() + + if response != Gtk.ResponseType.OK: + return + try: + rename_session(path, name) + except OSError as e: + self._flash_status(f"Errore nella rinomina: {e}") + return + self.refresh() + self._flash_status(f"Sessione rinominata: {name or '(nome cancellato)'}") + + def open_folder(self): + path = self.selected_path() + if not path: + return + info = self.by_path.get(path, {}) + cwd = info.get("cwd") or os.path.expanduser("~") + if not os.path.isdir(cwd): + cwd = os.path.expanduser("~") + dolphin = shutil.which("dolphin") + if dolphin: + subprocess.Popen([dolphin, cwd], start_new_session=True) + + def _flash_status(self, text): + self.status.set_text(text) + GLib.timeout_add(5000, lambda: self._update_status() or False) + + +class LauncherApp(Gtk.Application): + def __init__(self): + super().__init__(application_id="it.enne2.pi-session-launcher") + self.win = None + + def do_activate(self): + if self.win is None: + self.win = LauncherWindow(self) + self.win.present() + + +def main(): + app = LauncherApp() + app.run(sys.argv) + + +if __name__ == "__main__": + main()