1052 lines
39 KiB
Python
Executable File
1052 lines
39 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
claude-sessions — Session lifecycle manager for Claude Code tmux sessions.
|
|
|
|
Commands:
|
|
(no args) Interactive menu (new session: folder browser, unique topic, profile)
|
|
list Show live Claude tmux sessions and resumable registry entries
|
|
recover [--days N] Detect sessions active before a restart
|
|
spawn <project> Spawn a new session (wraps claude-tmux); --headless for scripted use
|
|
resume <session-id> Resume a past session by conversation ID
|
|
register Record a new session in the registry (called by claude-tmux)
|
|
gc [--days N] Prune stale entries and backfill session IDs
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# === Terminal colours ===
|
|
RED = "\033[0;31m"
|
|
GREEN = "\033[0;32m"
|
|
YELLOW = "\033[0;33m"
|
|
CYAN = "\033[0;36m"
|
|
BOLD = "\033[1m"
|
|
DIM = "\033[2m"
|
|
NC = "\033[0m"
|
|
|
|
|
|
# === Paths ===
|
|
HOME = Path.home()
|
|
CLAUDE_ROOT = HOME / "dev" / "claude"
|
|
CONFIG_FILE = HOME / ".config" / "claude" / "profiles.yaml"
|
|
REGISTRY_FILE = HOME / ".local" / "share" / "claude-sessions" / "registry.json"
|
|
HOSTNAME = subprocess.run(["hostname", "-s"], capture_output=True, text=True).stdout.strip()
|
|
|
|
PATH_SCAN_ROOTS = [
|
|
(HOME / "dev" / "claude", 2),
|
|
(HOME / "dev" / "claude" / "octopus", 1),
|
|
(HOME / "dev" / "claude" / "gamedev", 1),
|
|
(HOME / "dev" / "claude" / "worktrees", 1),
|
|
(HOME, 1),
|
|
]
|
|
|
|
DEFAULT_PROFILE_NAME = "oreillyit-anthropic" # same default as claude-tmux
|
|
|
|
DEFAULT_PROFILES = {
|
|
"oreillyit-anthropic": {
|
|
"display_name": "O'Reilly IT",
|
|
"claude_dir": str(HOME / ".claude-oreillyit-anthropic"),
|
|
"default_mode": "quick",
|
|
},
|
|
"default": {
|
|
"display_name": "Personal",
|
|
"claude_dir": str(HOME / ".claude"),
|
|
"default_mode": "quick",
|
|
},
|
|
}
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────────────────────────
|
|
|
|
def warn(msg: str) -> None:
|
|
print(f"{YELLOW}Warning:{NC} {msg}", file=sys.stderr)
|
|
|
|
def die(msg: str) -> None:
|
|
print(f"{RED}Error:{NC} {msg}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
# ── Profiles ────────────────────────────────────────────────────────────────
|
|
|
|
def load_profiles() -> dict:
|
|
if CONFIG_FILE.exists():
|
|
try:
|
|
import yaml
|
|
with CONFIG_FILE.open() as f:
|
|
config = yaml.safe_load(f)
|
|
if config and "profiles" in config:
|
|
profiles = config["profiles"]
|
|
for p in profiles.values():
|
|
if "claude_dir" in p:
|
|
p["claude_dir"] = str(Path(p["claude_dir"]).expanduser())
|
|
return profiles
|
|
except Exception as e:
|
|
warn(f"Could not load {CONFIG_FILE}: {e}")
|
|
return DEFAULT_PROFILES
|
|
|
|
|
|
def all_profile_dirs(profiles: dict) -> list[tuple[str, Path]]:
|
|
result = []
|
|
for name, cfg in profiles.items():
|
|
d = Path(cfg.get("claude_dir", "")).expanduser()
|
|
if d.is_dir():
|
|
result.append((name, d))
|
|
for d in HOME.iterdir():
|
|
if d.is_dir() and d.name.startswith(".claude-") and not d.name.endswith((".json", ".backup")):
|
|
pname = d.name[len(".claude-"):]
|
|
if not any(n == pname for n, _ in result):
|
|
result.append((pname, d))
|
|
return result
|
|
|
|
|
|
# ── Registry ────────────────────────────────────────────────────────────────
|
|
|
|
def load_registry() -> dict:
|
|
if REGISTRY_FILE.exists():
|
|
try:
|
|
return json.loads(REGISTRY_FILE.read_text())
|
|
except Exception:
|
|
pass
|
|
return {"sessions": []}
|
|
|
|
|
|
def save_registry(registry: dict) -> None:
|
|
REGISTRY_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
REGISTRY_FILE.write_text(json.dumps(registry, indent=2))
|
|
|
|
|
|
# ── Path hash utilities ──────────────────────────────────────────────────────
|
|
|
|
def path_to_hash(path: str | Path) -> str:
|
|
return str(path).replace("/", "-")
|
|
|
|
|
|
def build_path_map() -> dict[str, str]:
|
|
result: dict[str, str] = {}
|
|
|
|
def scan(p: Path, depth: int) -> None:
|
|
if not p.is_dir() or depth < 0:
|
|
return
|
|
result[path_to_hash(p)] = str(p)
|
|
if depth > 0:
|
|
try:
|
|
for child in p.iterdir():
|
|
if child.is_dir() and not child.name.startswith("."):
|
|
scan(child, depth - 1)
|
|
except PermissionError:
|
|
pass
|
|
|
|
for root, depth in PATH_SCAN_ROOTS:
|
|
scan(root, depth)
|
|
return result
|
|
|
|
|
|
# ── tmux utilities ───────────────────────────────────────────────────────────
|
|
|
|
def list_tmux_sessions() -> set[str]:
|
|
try:
|
|
r = subprocess.run(["tmux", "ls", "-F", "#{session_name}"], capture_output=True, text=True)
|
|
if r.returncode == 0:
|
|
return set(r.stdout.strip().splitlines())
|
|
except Exception:
|
|
pass
|
|
return set()
|
|
|
|
|
|
def title_case(s: str) -> str:
|
|
return " ".join(w.capitalize() for w in s.replace("-", " ").replace("_", " ").split())
|
|
|
|
|
|
def live_session_paths() -> dict[str, str]:
|
|
"""Map live tmux session name -> directory the session was created in."""
|
|
try:
|
|
r = subprocess.run(["tmux", "ls", "-F", "#{session_name}\t#{session_path}"],
|
|
capture_output=True, text=True)
|
|
except Exception:
|
|
return {}
|
|
if r.returncode != 0:
|
|
return {}
|
|
result = {}
|
|
for line in r.stdout.splitlines():
|
|
name, _, path = line.partition("\t")
|
|
result[name] = path
|
|
return result
|
|
|
|
|
|
def short_path(path: str) -> str:
|
|
"""Abbreviate $HOME to ~ for display."""
|
|
if not path:
|
|
return "?"
|
|
home = str(HOME)
|
|
if path == home:
|
|
return "~"
|
|
if path.startswith(home + "/"):
|
|
return "~" + path[len(home):]
|
|
return path
|
|
|
|
|
|
def resolve_project_dir(p: str) -> Optional[Path]:
|
|
"""Same lookup order as claude-tmux: literal path, then under CLAUDE_ROOT, then CLAUDE_ROOT/projects."""
|
|
for cand in (Path(p).expanduser(), CLAUDE_ROOT / p, CLAUDE_ROOT / "projects" / p):
|
|
if cand.is_dir():
|
|
return cand.resolve()
|
|
return None
|
|
|
|
|
|
def topic_in_use(topic: str, live: set[str]) -> bool:
|
|
# tmux rewrites '.' and ':' in session names to '_', so compare the stored form.
|
|
return topic in live or topic.replace(".", "_").replace(":", "_") in live
|
|
|
|
|
|
# ── Folder picker ────────────────────────────────────────────────────────────
|
|
|
|
def pick_folder_gui(start: Path) -> Optional[Path]:
|
|
"""GTK folder chooser via zenity (has a Create Folder button). None if cancelled or unavailable."""
|
|
if not (os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")):
|
|
return None
|
|
if subprocess.run(["which", "zenity"], capture_output=True).returncode != 0:
|
|
return None
|
|
r = subprocess.run(
|
|
["zenity", "--file-selection", "--directory",
|
|
"--title=Select a folder for the new Claude session",
|
|
f"--filename={start}/"],
|
|
capture_output=True, text=True,
|
|
)
|
|
if r.returncode == 0 and r.stdout.strip():
|
|
return Path(r.stdout.strip())
|
|
return None
|
|
|
|
|
|
def pick_folder_tui(start: Path, page_size: int = 15) -> Optional[Path]:
|
|
"""Terminal folder browser with new-folder support. Returns None if cancelled."""
|
|
cur = start.resolve() if start.is_dir() else HOME
|
|
flt = ""
|
|
page = 0
|
|
show_hidden = False
|
|
while True:
|
|
try:
|
|
subdirs = sorted(
|
|
(d for d in cur.iterdir()
|
|
if d.is_dir() and (show_hidden or not d.name.startswith("."))),
|
|
key=lambda d: d.name.lower(),
|
|
)
|
|
except PermissionError:
|
|
subdirs = []
|
|
warn(f"Cannot read {cur}")
|
|
if flt:
|
|
subdirs = [d for d in subdirs if flt.lower() in d.name.lower()]
|
|
pages = max(1, -(-len(subdirs) // page_size))
|
|
page = min(page, pages - 1)
|
|
shown = subdirs[page * page_size:(page + 1) * page_size]
|
|
|
|
os.system("clear 2>/dev/null || true")
|
|
print(f"\n{BOLD}Select folder{NC} {CYAN}{short_path(str(cur))}{NC}")
|
|
print("─" * 50)
|
|
if flt:
|
|
print(f" {DIM}filter: {flt!r}{NC}")
|
|
if not shown:
|
|
print(f" {DIM}(no subfolders){NC}")
|
|
for i, d in enumerate(shown, 1):
|
|
print(f" {CYAN}{i:>2}){NC} {d.name}/")
|
|
if pages > 1:
|
|
print(f" {DIM}page {page + 1}/{pages}{NC}")
|
|
print(f"\n {DIM}number = open · Enter or s = select this folder · u = up · n = new folder{NC}")
|
|
print(f" {DIM}/text = filter · > < = page · h = hidden ({'on' if show_hidden else 'off'}) · ~ or /path = jump · q = cancel{NC}")
|
|
|
|
try:
|
|
raw = input(" > ").strip()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
return None
|
|
|
|
if raw in ("", "s"):
|
|
return cur
|
|
elif raw == "q":
|
|
return None
|
|
elif raw == "u":
|
|
cur, flt, page = cur.parent, "", 0
|
|
elif raw == ">":
|
|
page = min(page + 1, pages - 1)
|
|
elif raw == "<":
|
|
page = max(page - 1, 0)
|
|
elif raw == "h":
|
|
show_hidden = not show_hidden
|
|
page = 0
|
|
elif raw == "n":
|
|
try:
|
|
name = input(" New folder name: ").strip()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
continue
|
|
if not name:
|
|
continue
|
|
if "/" in name or name in (".", ".."):
|
|
print(f" {YELLOW}Name must not contain '/' or be '.'/'..'{NC}")
|
|
time.sleep(1.2)
|
|
continue
|
|
target = cur / name
|
|
if target.exists():
|
|
print(f" {YELLOW}{name} already exists{NC}")
|
|
time.sleep(1.2)
|
|
continue
|
|
try:
|
|
target.mkdir()
|
|
except OSError as e:
|
|
print(f" {YELLOW}Could not create folder: {e}{NC}")
|
|
time.sleep(1.5)
|
|
continue
|
|
cur, flt, page = target, "", 0
|
|
elif raw.startswith("/") and len(raw) > 1 and not Path(raw).is_dir():
|
|
flt, page = raw[1:], 0
|
|
elif raw.startswith(("/", "~")):
|
|
dest = Path(raw).expanduser()
|
|
if dest.is_dir():
|
|
cur, flt, page = dest.resolve(), "", 0
|
|
else:
|
|
print(f" {YELLOW}Not a folder: {raw}{NC}")
|
|
time.sleep(1.2)
|
|
elif raw.isdigit() and 1 <= int(raw) <= len(shown):
|
|
cur, flt, page = shown[int(raw) - 1], "", 0
|
|
else:
|
|
print(f" {YELLOW}Unknown choice: {raw!r}{NC}")
|
|
time.sleep(1)
|
|
|
|
|
|
def pick_folder(start: Path) -> Optional[Path]:
|
|
return pick_folder_gui(start) or pick_folder_tui(start)
|
|
|
|
|
|
def prompt_topic(folder: Path, live: set[str]) -> Optional[str]:
|
|
"""Topic defaults to the folder name; if that is taken, the user must supply a unique one."""
|
|
default = title_case(folder.name)
|
|
taken = topic_in_use(default, live)
|
|
if taken:
|
|
print(f" {YELLOW}Topic '{default}' is already in use — enter a unique topic.{NC}")
|
|
while True:
|
|
try:
|
|
if taken:
|
|
topic = input(" Topic: ").strip()
|
|
else:
|
|
topic = input(f" Topic [{BOLD}{default}{NC}]: ").strip() or default
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
return None
|
|
if not topic:
|
|
print(f" {YELLOW}A topic is required.{NC}")
|
|
elif topic_in_use(topic, live):
|
|
print(f" {YELLOW}Topic '{topic}' is already in use — enter a different one.{NC}")
|
|
elif any(c in topic for c in ".:"):
|
|
print(f" {YELLOW}Topic must not contain '.' or ':' (tmux rewrites them).{NC}")
|
|
else:
|
|
return topic
|
|
|
|
|
|
def discover_profile_choices(profiles_cfg: dict) -> list[tuple[str, str]]:
|
|
"""Profiles from ~/.claude-[company]-[profile] dirs as (name, "Company (profile)").
|
|
|
|
A dir counts as a profile if it has a projects/ folder and a profile part
|
|
(~/.claude-[company] alone is skipped). Company display names
|
|
come from profiles.yaml display_name (parenthetical stripped) when known,
|
|
otherwise the Title-Cased directory segment. ~/.claude is the "default" profile.
|
|
"""
|
|
import re
|
|
company_names: dict[str, str] = {}
|
|
for name, cfg in profiles_cfg.items():
|
|
company = name.split("-", 1)[0]
|
|
disp = re.sub(r"\s*\(.*?\)\s*$", "", cfg.get("display_name", "")).strip()
|
|
if disp:
|
|
company_names.setdefault(company, disp)
|
|
|
|
choices = []
|
|
for d in sorted(HOME.iterdir()):
|
|
if not (d.is_dir() and d.name.startswith(".claude-") and (d / "projects").is_dir()):
|
|
continue
|
|
name = d.name[len(".claude-"):]
|
|
company, _, variant = name.partition("-")
|
|
if not variant:
|
|
continue
|
|
label = f"{company_names.get(company) or title_case(company)} ({variant})"
|
|
choices.append((name, label))
|
|
if (HOME / ".claude" / "projects").is_dir():
|
|
choices.append(("default", "Personal (main)"))
|
|
return choices
|
|
|
|
|
|
def prompt_profile(profiles_cfg: dict) -> Optional[str]:
|
|
"""Numbered profile picker. Enter accepts the default. Returns the profile name, or None to cancel."""
|
|
choices = discover_profile_choices(profiles_cfg)
|
|
if not choices:
|
|
return DEFAULT_PROFILE_NAME
|
|
names = [n for n, _ in choices]
|
|
default = DEFAULT_PROFILE_NAME if DEFAULT_PROFILE_NAME in names else names[0]
|
|
print(f"\n {BOLD}Profile:{NC}")
|
|
for i, (n, label) in enumerate(choices, 1):
|
|
mark = f" {DIM}(default){NC}" if n == default else ""
|
|
print(f" {CYAN}{i:>2}){NC} {label}{mark}")
|
|
while True:
|
|
try:
|
|
raw = input(f" Profile [{BOLD}{names.index(default) + 1}{NC}]: ").strip()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
return None
|
|
if not raw:
|
|
return default
|
|
if raw.isdigit() and 1 <= int(raw) <= len(names):
|
|
return names[int(raw) - 1]
|
|
print(f" {YELLOW}Enter a number from 1 to {len(names)}.{NC}")
|
|
|
|
|
|
# ── Session file parsing ─────────────────────────────────────────────────────
|
|
|
|
def read_jsonl_header(path: Path, max_lines: int = 30) -> dict:
|
|
session_id = None
|
|
start_time = None
|
|
try:
|
|
with path.open(errors="replace") as f:
|
|
for i, line in enumerate(f):
|
|
if i >= max_lines:
|
|
break
|
|
try:
|
|
obj = json.loads(line.strip())
|
|
if not session_id and "sessionId" in obj:
|
|
session_id = obj["sessionId"]
|
|
if not start_time and "timestamp" in obj:
|
|
start_time = obj["timestamp"]
|
|
if session_id and start_time:
|
|
break
|
|
except json.JSONDecodeError:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
return {"session_id": session_id or path.stem, "start_time": start_time}
|
|
|
|
|
|
def scan_profile_sessions(profile_name: str, claude_dir: Path,
|
|
path_map: dict[str, str], cutoff: float) -> list[dict]:
|
|
projects_dir = claude_dir / "projects"
|
|
if not projects_dir.is_dir():
|
|
return []
|
|
|
|
records = []
|
|
for proj_dir in projects_dir.iterdir():
|
|
if not proj_dir.is_dir():
|
|
continue
|
|
project_hash = proj_dir.name
|
|
real_path = path_map.get(project_hash)
|
|
|
|
jsonl_files = sorted(
|
|
proj_dir.glob("*.jsonl"),
|
|
key=lambda f: f.stat().st_mtime,
|
|
reverse=True,
|
|
)
|
|
for jf in jsonl_files:
|
|
mtime = jf.stat().st_mtime
|
|
if mtime < cutoff:
|
|
break
|
|
meta = read_jsonl_header(jf)
|
|
records.append({
|
|
"session_id": meta["session_id"],
|
|
"profile": profile_name,
|
|
"project_hash": project_hash,
|
|
"project_dir": real_path,
|
|
"start_time": meta["start_time"],
|
|
"last_activity": datetime.fromtimestamp(mtime).strftime("%b %d %H:%M"),
|
|
"mtime": mtime,
|
|
})
|
|
return records
|
|
|
|
|
|
# ── Registry backfill ────────────────────────────────────────────────────────
|
|
|
|
def backfill_session_ids(registry: dict, profiles: dict) -> int:
|
|
"""For registry entries with no session_id, find the matching .jsonl by scan."""
|
|
updated = 0
|
|
for entry in registry.get("sessions", []):
|
|
if entry.get("session_id"):
|
|
continue
|
|
profile_name = entry.get("profile", "")
|
|
profile_cfg = profiles.get(profile_name, {})
|
|
claude_dir = Path(profile_cfg.get("claude_dir", "")).expanduser()
|
|
if not claude_dir.is_dir():
|
|
continue
|
|
project_dir = entry.get("project_dir", "")
|
|
if not project_dir:
|
|
continue
|
|
proj_hash = path_to_hash(project_dir)
|
|
proj_path = claude_dir / "projects" / proj_hash
|
|
if not proj_path.is_dir():
|
|
continue
|
|
|
|
spawned_at = entry.get("spawned_at", "")
|
|
spawn_ts = 0.0
|
|
if spawned_at:
|
|
try:
|
|
spawn_ts = datetime.fromisoformat(spawned_at).timestamp()
|
|
except ValueError:
|
|
pass
|
|
|
|
best: Optional[Path] = None
|
|
best_mtime = 0.0
|
|
for jf in proj_path.glob("*.jsonl"):
|
|
mtime = jf.stat().st_mtime
|
|
if mtime > spawn_ts and mtime > best_mtime:
|
|
best, best_mtime = jf, mtime
|
|
|
|
if best:
|
|
meta = read_jsonl_header(best)
|
|
entry["session_id"] = meta["session_id"]
|
|
entry["last_activity"] = datetime.fromtimestamp(best_mtime).isoformat()
|
|
updated += 1
|
|
return updated
|
|
|
|
|
|
# ── Spawning ─────────────────────────────────────────────────────────────────
|
|
|
|
def resolve_claude_profile_bin() -> str:
|
|
r = subprocess.run(["which", "claude-profile"], capture_output=True, text=True)
|
|
if r.returncode == 0 and r.stdout.strip():
|
|
return r.stdout.strip()
|
|
die("claude-profile not found in PATH")
|
|
|
|
|
|
def spawn_session(project_dir: str, profile: str, mode: str,
|
|
resume_id: Optional[str] = None,
|
|
name_override: Optional[str] = None,
|
|
attach: bool = False) -> bool:
|
|
proj_path = Path(project_dir)
|
|
if not proj_path.is_dir():
|
|
warn(f"Project directory not found: {project_dir}")
|
|
return False
|
|
|
|
project_basename = proj_path.name
|
|
base_name = name_override or title_case(project_basename)
|
|
if resume_id:
|
|
base_name = f"{base_name} (resume)"
|
|
|
|
live = list_tmux_sessions()
|
|
name = base_name
|
|
n = 2
|
|
while name in live:
|
|
name = f"{base_name} #{n}"
|
|
n += 1
|
|
|
|
rc_flag = f'--remote-control "{name}"'
|
|
resume_flag = f"--resume {resume_id}" if resume_id else ""
|
|
extra = " ".join(filter(None, [rc_flag, resume_flag]))
|
|
|
|
claude_profile_bin = resolve_claude_profile_bin()
|
|
launch_cmd = (
|
|
f"{claude_profile_bin} {profile} --mode {mode} "
|
|
f"--project {project_basename} -- {extra}"
|
|
)
|
|
|
|
r = subprocess.run(
|
|
["tmux", "new-session", "-d", "-s", name, "-c", str(proj_path)],
|
|
capture_output=True,
|
|
)
|
|
if r.returncode != 0:
|
|
warn(f"tmux new-session failed: {r.stderr.decode().strip()}")
|
|
return False
|
|
|
|
subprocess.run(["tmux", "send-keys", "-t", name, "-l", launch_cmd])
|
|
subprocess.run(["tmux", "send-keys", "-t", name, "Enter"])
|
|
|
|
print(f"{GREEN}Spawned{NC} {BOLD}{name}{NC}")
|
|
print(f" dir: {project_dir}")
|
|
print(f" profile: {profile} / mode: {mode}")
|
|
if resume_id:
|
|
print(f" resume: {resume_id[:8]}…")
|
|
print(f" attach: tmux attach -t \"{name}\"")
|
|
|
|
if attach:
|
|
if os.environ.get("TMUX"):
|
|
subprocess.run(["tmux", "switch-client", "-t", f"={name}"])
|
|
else:
|
|
os.execvp("tmux", ["tmux", "attach", "-t", f"={name}"])
|
|
|
|
return True
|
|
|
|
|
|
def find_claude_md_dir(start: Path) -> Optional[Path]:
|
|
"""Nearest directory at or above start (stopping at $HOME) containing a CLAUDE.md."""
|
|
for d in [start, *start.parents]:
|
|
if d == HOME.parent or d == Path("/"):
|
|
break
|
|
if (d / "CLAUDE.md").is_file():
|
|
return d
|
|
if d == HOME:
|
|
break
|
|
return None
|
|
|
|
|
|
# ── Commands ─────────────────────────────────────────────────────────────────
|
|
|
|
def cmd_list(args) -> None:
|
|
live = list_tmux_sessions()
|
|
|
|
if live:
|
|
print(f"\n{BOLD}Live tmux sessions:{NC}")
|
|
for s in sorted(live):
|
|
print(f" {GREEN}●{NC} {s}")
|
|
else:
|
|
print(f"\n{DIM}No active tmux sessions.{NC}")
|
|
|
|
registry = load_registry()
|
|
resumable = [s for s in registry.get("sessions", []) if s.get("tmux_name") not in live]
|
|
|
|
if resumable:
|
|
print(f"\n{BOLD}Resumable (registry, not currently live):{NC}")
|
|
fmt = " {:<4} {:<32} {:<24} {:<10} {}"
|
|
print(fmt.format("", "Project", "Profile", "Session ID", "Last active"))
|
|
print(fmt.format("", "-" * 32, "-" * 24, "-" * 10, "-" * 15))
|
|
for i, s in enumerate(resumable, 1):
|
|
proj = Path(s.get("project_dir", "")).name or s.get("project", "?")
|
|
sid = (s.get("session_id") or "")[:8] or "pending…"
|
|
ts = (s.get("last_activity") or s.get("spawned_at") or "?")[:16]
|
|
profile = s.get("profile", "?")
|
|
print(fmt.format(f"{i})", proj, profile, sid, ts))
|
|
else:
|
|
print(f"\n{DIM}Registry is empty — run 'spawn' to start tracking sessions.{NC}")
|
|
|
|
print()
|
|
|
|
|
|
def cmd_recover(args) -> None:
|
|
profiles = load_profiles()
|
|
path_map = build_path_map()
|
|
cutoff = time.time() - args.days * 86400
|
|
|
|
all_sessions: list[dict] = []
|
|
for profile_name, claude_dir in all_profile_dirs(profiles):
|
|
all_sessions.extend(scan_profile_sessions(profile_name, claude_dir, path_map, cutoff))
|
|
|
|
all_sessions.sort(key=lambda s: s["mtime"], reverse=True)
|
|
|
|
if not all_sessions:
|
|
print(f"No sessions found in the last {args.days} days.")
|
|
return
|
|
|
|
print(f"\n{BOLD}Sessions active in the last {args.days} day(s):{NC}\n")
|
|
print(f" {'#':<4} {'Project':<32} {'Profile':<24} {'Session ID':<10} {'Last Active':<16} Note")
|
|
print(f" {'-'*4} {'-'*32} {'-'*24} {'-'*10} {'-'*16} {'-'*20}")
|
|
|
|
for i, s in enumerate(all_sessions, 1):
|
|
if s["project_dir"]:
|
|
proj_raw = Path(s["project_dir"]).name
|
|
proj = proj_raw[:30] + ("…" if len(proj_raw) > 30 else "")
|
|
dim_proj = ""
|
|
else:
|
|
h = s["project_hash"]
|
|
for prefix in ("-home-paul-", "-root-", "-home-"):
|
|
if h.startswith(prefix):
|
|
h = h[len(prefix):]
|
|
break
|
|
parts = h.rsplit("-", 4)
|
|
hint = "/".join(parts[-2:]) if len(parts) >= 2 else h
|
|
proj = hint[:30]
|
|
dim_proj = DIM
|
|
|
|
sid = s["session_id"][:8]
|
|
ts = s["last_activity"]
|
|
profile_raw = s["profile"]
|
|
profile = profile_raw[:22] + ("…" if len(profile_raw) > 22 else "")
|
|
note = "" if s["project_dir"] else f"{YELLOW}path?{NC}"
|
|
proj_padded = f"{proj:<32}"
|
|
print(f" {i:<4} {dim_proj}{proj_padded}{NC if dim_proj else ''} {profile:<24} {sid:<10} {ts:<16} {note}")
|
|
|
|
print()
|
|
|
|
if not sys.stdin.isatty():
|
|
return
|
|
|
|
try:
|
|
raw = input("Resume which sessions? (e.g. 1,3 or 'all', Enter to skip): ").strip()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
return
|
|
|
|
if not raw:
|
|
return
|
|
|
|
if raw.lower() == "all":
|
|
selected = [s for s in all_sessions if s["project_dir"]]
|
|
else:
|
|
try:
|
|
indices = [int(x.strip()) - 1 for x in raw.replace(" ", "").split(",")]
|
|
selected = [all_sessions[i] for i in indices if 0 <= i < len(all_sessions)]
|
|
except ValueError:
|
|
die("Invalid selection — expected numbers like 1,3,5 or 'all'")
|
|
return
|
|
|
|
if not selected:
|
|
return
|
|
|
|
profiles_cfg = load_profiles()
|
|
print()
|
|
# Land in the session only when resuming exactly one; several stay detached.
|
|
attach = len(selected) == 1
|
|
for s in selected:
|
|
if not s["project_dir"]:
|
|
warn(f"Skipping: project directory not found for {s['project_hash']}")
|
|
continue
|
|
profile_cfg = profiles_cfg.get(s["profile"], {})
|
|
mode = getattr(args, "mode", None) or profile_cfg.get("default_mode", "quick")
|
|
spawn_session(project_dir=s["project_dir"], profile=s["profile"],
|
|
mode=mode, resume_id=s["session_id"], attach=attach)
|
|
print()
|
|
|
|
|
|
def cmd_spawn(args) -> None:
|
|
claude_tmux = subprocess.run(["which", "claude-tmux"], capture_output=True, text=True).stdout.strip()
|
|
if not claude_tmux:
|
|
die("claude-tmux not found in PATH")
|
|
project = args.project
|
|
name = args.name
|
|
|
|
if getattr(args, "headless", False):
|
|
# Non-interactive: no prompts, no attach. Errors exit non-zero instead of asking.
|
|
folder = resolve_project_dir(project)
|
|
if folder is None and args.create:
|
|
folder = Path(project).expanduser()
|
|
if not folder.is_absolute():
|
|
die(f"--create needs an absolute path (got {project!r})")
|
|
try:
|
|
folder.mkdir(parents=True)
|
|
except OSError as e:
|
|
die(f"Could not create {folder}: {e}")
|
|
folder = folder.resolve()
|
|
if folder is None:
|
|
die(f"Folder not found: {project} (use --create with an absolute path to make it)")
|
|
name = name or title_case(folder.name)
|
|
if topic_in_use(name, list_tmux_sessions()):
|
|
print(f"{RED}Error:{NC} topic '{name}' is already in use — pass a unique --name", file=sys.stderr)
|
|
sys.exit(2)
|
|
if any(c in name for c in ".:"):
|
|
die("Topic must not contain '.' or ':' (tmux rewrites them)")
|
|
project = str(folder)
|
|
args.attach = False
|
|
|
|
cmd = [claude_tmux, project]
|
|
if args.mode:
|
|
cmd += ["--mode", args.mode]
|
|
if args.profile:
|
|
cmd += ["--profile", args.profile]
|
|
if name:
|
|
cmd += ["--name", name]
|
|
if args.attach:
|
|
cmd += ["--attach"]
|
|
os.execvp(claude_tmux, cmd)
|
|
|
|
|
|
def cmd_resume(args) -> None:
|
|
session_id = args.session_id
|
|
registry = load_registry()
|
|
entry = next(
|
|
(s for s in registry.get("sessions", [])
|
|
if (s.get("session_id") or "").startswith(session_id)),
|
|
None,
|
|
)
|
|
|
|
if entry:
|
|
project_dir = entry.get("project_dir", "")
|
|
profile = args.profile or entry.get("profile", "oreillyit-anthropic")
|
|
mode = args.mode or entry.get("mode", "quick")
|
|
else:
|
|
if not args.project:
|
|
die(f"Session {session_id!r} not in registry.\nProvide --project (and optionally --profile, --mode).")
|
|
profile = args.profile or "oreillyit-anthropic"
|
|
profiles_cfg = load_profiles()
|
|
mode = args.mode or profiles_cfg.get(profile, {}).get("default_mode", "quick")
|
|
path_map = build_path_map()
|
|
project_dir = next((p for p in path_map.values() if Path(p).name == args.project), None)
|
|
if not project_dir:
|
|
die(f"Project directory not found for '{args.project}'")
|
|
|
|
spawn_session(project_dir=project_dir, profile=profile, mode=mode,
|
|
resume_id=session_id, attach=args.attach)
|
|
|
|
|
|
def cmd_register(args) -> None:
|
|
"""Called by claude-tmux after spawning a session."""
|
|
registry = load_registry()
|
|
entry = {
|
|
"tmux_name": args.tmux_name,
|
|
"project": Path(args.project_dir).name,
|
|
"project_dir": args.project_dir,
|
|
"profile": args.profile,
|
|
"mode": args.mode,
|
|
"machine": HOSTNAME,
|
|
"spawned_at": datetime.now(timezone.utc).isoformat(),
|
|
"session_id": None,
|
|
"last_activity": None,
|
|
}
|
|
# Remove any prior entry for this tmux session name
|
|
registry["sessions"] = [
|
|
s for s in registry.get("sessions", [])
|
|
if s.get("tmux_name") != args.tmux_name
|
|
]
|
|
registry["sessions"].append(entry)
|
|
save_registry(registry)
|
|
|
|
|
|
def cmd_gc(args) -> None:
|
|
registry = load_registry()
|
|
profiles = load_profiles()
|
|
live = list_tmux_sessions()
|
|
cutoff = time.time() - args.days * 86400
|
|
|
|
# Backfill session IDs first
|
|
backfilled = backfill_session_ids(registry, profiles)
|
|
if backfilled:
|
|
print(f"Backfilled {backfilled} session ID(s).")
|
|
|
|
before = len(registry.get("sessions", []))
|
|
|
|
def keep(s: dict) -> bool:
|
|
if s.get("tmux_name") in live:
|
|
return True
|
|
spawned = s.get("spawned_at", "")
|
|
if spawned:
|
|
try:
|
|
ts = datetime.fromisoformat(spawned).timestamp()
|
|
return ts > cutoff
|
|
except ValueError:
|
|
pass
|
|
return False
|
|
|
|
registry["sessions"] = [s for s in registry.get("sessions", []) if keep(s)]
|
|
after = len(registry["sessions"])
|
|
save_registry(registry)
|
|
print(f"Pruned {before - after} stale entries ({after} remaining).")
|
|
|
|
|
|
# ── Interactive menu ─────────────────────────────────────────────────────────
|
|
|
|
def cmd_menu(_args=None) -> None:
|
|
"""Interactive session picker. Shown when claude-sessions is invoked with no args."""
|
|
profiles_cfg = load_profiles()
|
|
|
|
while True:
|
|
live = sorted(list_tmux_sessions())
|
|
registry = load_registry()
|
|
resumable = [
|
|
s for s in registry.get("sessions", [])
|
|
if s.get("tmux_name") not in live and s.get("session_id")
|
|
]
|
|
|
|
# Build numbered options
|
|
options: list[tuple[str, dict]] = [] # (kind, data)
|
|
for name in live:
|
|
options.append(("live", {"tmux_name": name}))
|
|
for entry in resumable:
|
|
options.append(("resumable", entry))
|
|
|
|
# ── Print menu ──
|
|
os.system("clear 2>/dev/null || true")
|
|
print(f"\n{BOLD}Claude Sessions{NC}")
|
|
print("─" * 50)
|
|
|
|
if not options:
|
|
print(f"\n {DIM}No sessions.{NC}")
|
|
else:
|
|
reg_dirs = {e.get("tmux_name"): e.get("project_dir", "") for e in registry.get("sessions", [])}
|
|
tmux_dirs = live_session_paths()
|
|
resumable_names = {
|
|
id(d): Path(d.get("project_dir", "")).name or d.get("project", "?")
|
|
for k, d in options if k == "resumable"
|
|
}
|
|
name_w = min(36, max([len(d["tmux_name"]) for k, d in options if k == "live"]
|
|
+ [len(n) for n in resumable_names.values()] + [7]))
|
|
|
|
if live:
|
|
print(f"\n {BOLD}Live:{NC}")
|
|
print(f" {DIM}{'Session':<{name_w}} Location{NC}")
|
|
for idx, (kind, data) in enumerate(options, 1):
|
|
if kind != "live":
|
|
continue
|
|
n = data["tmux_name"]
|
|
loc = short_path(reg_dirs.get(n) or tmux_dirs.get(n, ""))
|
|
print(f" {CYAN}{idx:>2}){NC} {GREEN}●{NC} {n:<{name_w}} {DIM}{loc}{NC}")
|
|
|
|
if resumable:
|
|
print(f"\n {BOLD}Resumable:{NC}")
|
|
print(f" {DIM}{'Session':<{name_w}} {'Location':<40} Details{NC}")
|
|
for idx, (kind, data) in enumerate(options, 1):
|
|
if kind != "resumable":
|
|
continue
|
|
proj = resumable_names[id(data)]
|
|
sid = (data.get("session_id") or "")[:8]
|
|
ts = (data.get("last_activity") or data.get("spawned_at") or "?")[:16]
|
|
profile = data.get("profile", "?")
|
|
loc = short_path(data.get("project_dir", ""))
|
|
print(f" {CYAN}{idx:>2}){NC} {DIM}○{NC} {proj:<{name_w}} "
|
|
f"{loc:<40} {DIM}{profile} · {sid} · {ts}{NC}")
|
|
|
|
print(f"\n {CYAN} r){NC} Recover sessions from restart (--days 7)")
|
|
print(f" {CYAN} n){NC} New session")
|
|
print(f" {CYAN} q){NC} Quit")
|
|
print()
|
|
|
|
try:
|
|
raw = input(" Choice: ").strip().lower()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
break
|
|
|
|
if raw == "q":
|
|
break
|
|
|
|
elif raw == "r":
|
|
import types
|
|
a = types.SimpleNamespace(days=7, mode=None, attach=False)
|
|
cmd_recover(a)
|
|
input("\nPress Enter to return to menu…")
|
|
|
|
elif raw == "n":
|
|
default_dir = find_claude_md_dir(Path.cwd())
|
|
try:
|
|
if default_dir:
|
|
print(f" Found CLAUDE.md in {BOLD}{default_dir}{NC}")
|
|
ans = input(
|
|
f" Folder [{BOLD}{short_path(str(default_dir))}{NC}] "
|
|
f"(Enter = use this, b = browse, or type a project/path): "
|
|
).strip() or str(default_dir)
|
|
else:
|
|
ans = input(" Folder (b = browse, or type a project/path): ").strip()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
continue
|
|
if not ans:
|
|
continue
|
|
if ans.lower() == "b":
|
|
folder = pick_folder(default_dir or Path.cwd())
|
|
if folder is None:
|
|
continue
|
|
print(f" Folder: {BOLD}{short_path(str(folder))}{NC}")
|
|
else:
|
|
folder = resolve_project_dir(ans)
|
|
if folder is None:
|
|
print(f" {YELLOW}Folder not found: {ans}{NC}")
|
|
time.sleep(1.5)
|
|
continue
|
|
topic = prompt_topic(folder, list_tmux_sessions())
|
|
if topic is None:
|
|
continue
|
|
profile = prompt_profile(profiles_cfg)
|
|
if profile is None:
|
|
continue
|
|
import types
|
|
a = types.SimpleNamespace(project=str(folder), mode=None, profile=profile, name=topic, attach=True)
|
|
print()
|
|
cmd_spawn(a) # exec's claude-tmux --attach; replaces this process
|
|
|
|
elif raw.isdigit():
|
|
idx = int(raw) - 1
|
|
if idx < 0 or idx >= len(options):
|
|
print(f" {YELLOW}Invalid choice{NC}")
|
|
time.sleep(1)
|
|
continue
|
|
kind, data = options[idx]
|
|
|
|
if kind == "live":
|
|
name = data["tmux_name"]
|
|
print(f"\n Attaching to {BOLD}{name}{NC}…")
|
|
if os.environ.get("TMUX"):
|
|
subprocess.run(["tmux", "switch-client", "-t", f"={name}"])
|
|
else:
|
|
os.execvp("tmux", ["tmux", "attach", "-t", f"={name}"])
|
|
break # exec'd — this line only reached on switch-client
|
|
|
|
elif kind == "resumable":
|
|
project_dir = data.get("project_dir", "")
|
|
profile = data.get("profile", "oreillyit-anthropic")
|
|
mode = profiles_cfg.get(profile, {}).get("default_mode", "quick")
|
|
session_id = data.get("session_id", "")
|
|
if not project_dir:
|
|
print(f" {YELLOW}Project directory unknown — cannot resume{NC}")
|
|
time.sleep(1)
|
|
continue
|
|
print()
|
|
spawn_session(project_dir=project_dir, profile=profile,
|
|
mode=mode, resume_id=session_id, attach=True)
|
|
else:
|
|
print(f" {YELLOW}Unknown choice: {raw!r}{NC}")
|
|
time.sleep(1)
|
|
|
|
|
|
# ── Entrypoint ───────────────────────────────────────────────────────────────
|
|
|
|
def main() -> None:
|
|
p = argparse.ArgumentParser(
|
|
prog="claude-sessions",
|
|
description="Session lifecycle manager for Claude Code tmux sessions",
|
|
add_help=True,
|
|
)
|
|
sub = p.add_subparsers(dest="command", metavar="COMMAND")
|
|
|
|
sub.add_parser("list", help="Show live and resumable sessions")
|
|
sub.add_parser("menu", help="Interactive session menu (default when no args)")
|
|
|
|
rp = sub.add_parser("recover", help="Detect sessions active before a restart")
|
|
rp.add_argument("--days", type=int, default=7, help="Look back N days (default: 7)")
|
|
rp.add_argument("--mode", default=None)
|
|
rp.add_argument("--attach", action="store_true")
|
|
|
|
sp = sub.add_parser("spawn", help="Spawn a new session (wraps claude-tmux)")
|
|
sp.add_argument("project", help="Project name or path")
|
|
sp.add_argument("--mode", default=None)
|
|
sp.add_argument("--profile", default=None)
|
|
sp.add_argument("--name", default=None)
|
|
sp.add_argument("--attach", action="store_true")
|
|
sp.add_argument("--headless", action="store_true",
|
|
help="Non-interactive: never prompt or attach; fail (exit 2) if the topic is in use")
|
|
sp.add_argument("--create", action="store_true",
|
|
help="With --headless: create the folder (absolute path) if it does not exist")
|
|
|
|
rsp = sub.add_parser("resume", help="Resume a past conversation by session ID")
|
|
rsp.add_argument("session_id", help="Session UUID or prefix")
|
|
rsp.add_argument("--project", default=None)
|
|
rsp.add_argument("--profile", default=None)
|
|
rsp.add_argument("--mode", default=None)
|
|
rsp.add_argument("--attach", action="store_true")
|
|
|
|
reg = sub.add_parser("register", help="Record a new session (called by claude-tmux)")
|
|
reg.add_argument("--project-dir", required=True)
|
|
reg.add_argument("--profile", required=True)
|
|
reg.add_argument("--mode", required=True)
|
|
reg.add_argument("--tmux-name", required=True)
|
|
|
|
gp = sub.add_parser("gc", help="Prune stale entries and backfill session IDs")
|
|
gp.add_argument("--days", type=int, default=30)
|
|
|
|
args = p.parse_args()
|
|
|
|
# No args + interactive TTY → menu
|
|
if not args.command:
|
|
if sys.stdin.isatty() and sys.stdout.isatty():
|
|
cmd_menu()
|
|
else:
|
|
p.print_help()
|
|
sys.exit(0)
|
|
|
|
{
|
|
"list": cmd_list,
|
|
"menu": cmd_menu,
|
|
"recover": cmd_recover,
|
|
"spawn": cmd_spawn,
|
|
"resume": cmd_resume,
|
|
"register": cmd_register,
|
|
"gc": cmd_gc,
|
|
}[args.command](args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|