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 <file>
- rinomina sessione (append session_info, ultima entry vince come in pi)
- install.sh: voce menu KDE
This commit is contained in:
2026-08-19 11:58:37 +02:00
commit e2fffefcf2
4 changed files with 571 additions and 0 deletions
+6
View File
@@ -0,0 +1,6 @@
__pycache__/
*.pyc
build/
dist/
*.spec
.venv/
+47
View File
@@ -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 <file sessione>`.
## 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/--<percorso>--/<timestamp>_<uuid>.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 <cwd> --hold --separate -e bash -c 'pi --session "$1"; exec bash' _ <file>`
`exec bash` mantiene la shell aperta quando pi termina.
Executable
+35
View File
@@ -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" <<EOF
[Desktop Entry]
Type=Application
Name=Pi Session Launcher
GenericName=Riprendi sessioni di Pi Agent
Comment=Elenca le sessioni di Pi Agent e le riprende nella loro cartella
Exec=python3 $APP
Icon=utilities-terminal
Terminal=false
Categories=Development;
Keywords=pi;agent;session;resume;
EOF
echo "✓ Installato: $DESKTOP"
echo " Cerca 'Pi Session Launcher' nel menu applicazioni KDE."
echo " (Per rimuovere: $0 --uninstall)"
+483
View File
@@ -0,0 +1,483 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Pi Session Launcher
===================
GUI (GTK3) che elenca tutte le sessioni salvate di Pi Agent
(~/.pi/agent/sessions/**/*.jsonl) e, selezionandone una:
* apre un terminale (Konsole) nella cartella di lavoro della sessione (cwd)
* riprende la sessione con: pi --session <file di sessione>
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"<b>{markup}</b>"
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()