Trimming Pi- agent essions without losing context

Starred Gists，GitHub Gist，即时分享代码、笔记与代码片段。关注
翻译
#!/usr/bin/env python3
"""
pi-shrink — обрезка Pi-сессий без потери контекста.
Pi при старте читает файл сессии целиком, а в контекст берёт только последний
компакшен и то, что после него. Сессия на сотни мегабайт поэтому ничего не
даёт, но раздувает процесс и делает его главным кандидатом на отстрел
earlyoom'ом.
pi-shrink режет файл ровно по границе последнего компакшена — то есть
выбрасывает только то, что Pi и так не читает. Контекст до и после сверяется
(повторяется алгоритм Pi buildContextEntries), и при расхождении файл не
трогается. Перед обрезкой печатается сводка и запрашивается подтверждение —
отдельного "холостого прогона" нет, отказаться можно ответом на вопрос.
Сессии ищутся в ~/.pi/agent/sessions и в песочницах pi-box (там они лежат в
<проект>/.pi-box/sessions); список песочниц берётся из реестра pi-box.
ЗАПУСК
pi-shrink список сессий (fzf), затем обрезка выбранной
pi-shrink FILE обрезать указанную сессию
pi-shrink -s FILE только статистика, файл не трогается
pi-shrink --list строки для fzf (TSV), самые тяжёлые сверху
ГАРАНТИИ
• режется только по границе последнего компакшена;
• граница ищется по id записи, а не по номеру строки: пустые и битые
строки в файле не сдвигают её;
• контекст до и после сверяется по тому же набору строк, что и запишется;
• бэкап кладётся в /tmp (не рядом с сессией: копия на сотни мегабайт
удваивала бы место в проекте), существующий не затирается;
• если в /tmp места нет — бэкап ложится рядом с файлом;
• запись атомарная (временный файл + rename), права и владелец сохраняются;
• отказ, если сессия занята живым Pi — в том числе из pi-box-песочницы.
ПЕРЕМЕННЫЕ
PI_SESSIONS_ROOT только этот каталог вместо поиска по умолчанию
PI_BOX_HOME реестр песочниц pi-box (по умолчанию ~/.local/share/pi-box)
TMPDIR куда класть бэкапы (по умолчанию /tmp)
FZF_HEIGHT высота меню (по умолчанию ~60%)
NO_COLOR отключить цвета
Счётчики расходов в футере Pi (↑ ↓ R W $) после обрезки сбрасываются: они
считаются по файлу. Контекст и память агента не страдают — это разное.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
DEFAULT_ROOT = os.path.expanduser("~/.pi/agent/sessions")
PI_BOX_HOME = os.path.expanduser(
os.environ.get("PI_BOX_HOME",
os.path.join(os.environ.get("XDG_DATA_HOME",
os.path.expanduser("~/.local/share")),
"pi-box")))
# ── цвета ─────────────────────────────────────────────────────────────────────
# Заполняются в setup_colors(): по умолчанию — только для терминала, потому
# что в конвейере escape-последовательности мешают. Для fzf цвета включаются
# принудительно (fzf запускается с --ansi).
RST = B = DIM = RED = GRN = YEL = CYN = ""
def setup_colors(enabled):
global RST, B, DIM, RED, GRN, YEL, CYN
if enabled:
RST, B, DIM = "\033[0m", "\033[1m", "\033[2m"
RED, GRN, YEL, CYN = "\033[31m", "\033[32m", "\033[33m", "\033[36m"
else:
RST = B = DIM = RED = GRN = YEL = CYN = ""
# ── форматирование ────────────────────────────────────────────────────────────
def human(n):
"""Байты в человекочитаемое."""
for unit, div in (("GB", 1 << 30), ("MB", 1 << 20), ("KB", 1 << 10)):
if n >= div:
return f"{n / div:.1f}{unit}"
return f"{n}B"
def toks(n):
"""Токены: 1.1M / 133k / 42."""
if n >= 1_000_000:
return f"{n / 1_000_000:.1f}M"
if n >= 1_000:
return f"{n / 1_000:.0f}k"
return str(n)
# ── чтение сессии ─────────────────────────────────────────────────────────────
def load(path):
"""Записи сессии. Битые строки пропускаются — как это делает сам Pi."""
out = []
with open(path, encoding="utf-8", errors="replace") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except json.JSONDecodeError:
continue
return out
def raw_lines(path):
with open(path, "rb") as fh:
return fh.readlines()
def load_from_lines(lines):
"""Записи из уже прочитанных строк — с той же фильтрацией, что в load()."""
out = []
for line in lines:
if isinstance(line, bytes):
line = line.decode("utf-8", "replace")
line = line.strip()
if not line:
continue
try:
out.append(json.loads(line))
except json.JSONDecodeError:
continue
return out
def session_cwd(path):
"""Каталог проекта из заголовка сессии."""
for line in open(path, encoding="utf-8", errors="replace"):
try:
e = json.loads(line)
except json.JSONDecodeError:
continue
if e.get("type") == "session":
return e.get("cwd")
break
return None
# ── логика Pi ─────────────────────────────────────────────────────────────────
def context_ids(entries):
"""Повторяет buildContextEntries из Pi: что реально уходит в контекст.
Путь строится от последней записи по parentId; при наличии компакшена
старше него выбрасывается всё, кроме firstKeptEntryId и хвоста. Именно
поэтому обрезка по границе компакшена безопасна: Pi читает ровно это.
"""
by_id = {e["id"]: e for e in entries if e.get("id")}
if not entries:
return []
path, cur = [], entries[-1]
while cur:
path.append(cur)
cur = by_id.get(cur.get("parentId")) if cur.get("parentId") else None
path.reverse()
comp = None
for e in path:
if e.get("type") == "compaction":
comp = e
if not comp:
return [e.get("id") for e in path]
idx = next((i for i, e in enumerate(path) if e.get("id") == comp["id"]), -1)
if idx < 0:
return [e.get("id") for e in path]
out = [comp]
found = False
for i in range(idx):
e = path[i]
if e.get("id") == comp.get("firstKeptEntryId"):
found = True
if found and not (e.get("type") == "message"
and (e.get("message") or {}).get("role") == "system"):
out.append(e)
out.extend(path[idx + 1:])
return [e.get("id") for e in out]
def cut_id(entries):
"""id первой сохраняемой записи = firstKeptEntryId последнего компакшена.
Именно id, а не номер строки: load() пропускает пустые и битые строки,
поэтому индекс в разобранных записях не совпадает с индексом в файле.
Возвращает (id, compaction_entry) или (None, None), если компакшена нет:
без него вся история участвует в контексте, и обрезка означала бы потерю
памяти, а не мусора.
"""
comps = [e for e in entries if e.get("type") == "compaction"]
if not comps:
return None, None
comp = comps[-1]
fk = comp.get("firstKeptEntryId")
if not fk or not any(e.get("id") == fk for e in entries):
return None, None
return fk, comp
def line_index(lines, entry_id):
"""Номер строки в файле, где начинается запись с этим id.
Ищем по "id":"" в разобранном виде, а не подстрокой: битая строка
может содержать тот же текст. Возвращает None, если не нашли.
"""
for i, line in enumerate(lines):
if entry_id.encode() not in line:
continue
try:
if json.loads(line).get("id") == entry_id:
return i
except (json.JSONDecodeError, AttributeError):
continue
return None
def usage_totals(entries):
"""Накопленные расходы — ровно то, что показывает футер Pi."""
t = {"input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0,
"reasoning": 0, "cost": 0.0}
for e in entries:
u = None
if e.get("type") == "usage":
u = e.get("usage")
elif e.get("type") == "message":
m = e.get("message") or {}
if m.get("role") in ("assistant", "toolResult"):
u = m.get("usage")
elif e.get("type") in ("compaction", "branch_summary"):
u = e.get("usage")
if not u:
continue
for k in ("input", "output", "cacheRead", "cacheWrite", "reasoning"):
t[k] += u.get(k, 0) or 0
c = u.get("cost")
if isinstance(c, dict):
t["cost"] += c.get("total", 0) or 0
return t
def last_prompt_tokens(entries):
"""Токенов в последнем запросе — это и есть занятый контекст."""
for e in reversed(entries):
if e.get("type") == "message":
m = e.get("message") or {}
if m.get("role") == "assistant" and m.get("usage"):
u = m["usage"]
n = (u.get("input", 0) or 0) + (u.get("cacheRead", 0) or 0) \
+ (u.get("cacheWrite", 0) or 0)
if n:
return n
return None
def count_compactions(entries):
return sum(1 for e in entries if e.get("type") == "compaction")
def count_images(entries):
n = 0
for e in entries:
if e.get("type") == "message":
c = (e.get("message") or {}).get("content")
if isinstance(c, list):
n += sum(1 for b in c
if isinstance(b, dict) and b.get("type") == "image")
return n
def summarize(path):
"""Всё, что нужно и для списка, и для отчёта."""
st = os.stat(path)
entries = load(path)
header = next((e for e in entries if e.get("type") == "session"), {})
return {
"path": path,
"size": st.st_size,
"mtime": st.st_mtime,
"entries": len(entries),
"compactions": count_compactions(entries),
"images": count_images(entries),
"cwd": header.get("cwd", "?"),
"id": header.get("id", "?"),
"totals": usage_totals(entries),
"prompt": last_prompt_tokens(entries),
}
# ── живые процессы ────────────────────────────────────────────────────────────
def pi_box_project(path):
"""Каталог проекта на хосте, если сессия — из pi-box-песочницы.
Внутри песочницы проект смонтирован в /work, и именно этот путь попадает
в заголовок сессии. На хосте /work не существует, поэтому обычный поиск по
cwd процесс не найдёт никогда. Восстанавливаем проект из пути файла:
<проект>/.pi-box/sessions/<файл>.jsonl.
"""
if session_cwd(path) != "/work":
return None
sessions = os.path.dirname(os.path.realpath(path))
box = os.path.dirname(sessions)
if os.path.basename(box) != ".pi-box":
return None
return os.path.dirname(box)
def pi_box_short(project):
"""Путь проекта с ~ вместо домашнего каталога."""
home = os.path.expanduser("~")
if project.startswith(home + os.sep):
return "~" + project[len(home):]
return project
def box_socket(project, path=None):
"""Сокет tmux той же песочницы — тот же расчёт, что в pi-box.
Uid берём у файла сессии, а не свой: pi-box запускает песочницу под её
владельцем, и сокет лежит в /tmp/pi-box-<его uid>. Запуск pi-shrink под
другим пользователем (например root) иначе не нашёл бы сокет.
"""
import hashlib
uid = os.stat(path).st_uid if path else os.getuid()
h = hashlib.md5(os.path.realpath(project).encode()).hexdigest()[:12]
return f"/tmp/pi-box-{uid}/{h}.sock"
def open_by_pi(path):
"""Pid'ы живых Pi, которые пишут в эту сессию.
По fd проверять бесполезно: Pi открывает файл на каждый append и сразу
закрывает, поэтому в /proc//fd его обычно нет. Опознаём по каталогу
проекта: живой Pi держит сессию в памяти и продолжает в неё писать, а
_persist при flushed=false создаёт файл с флагом "wx" — на уже
существующем это падает с EEXIST.
"""
# Песочница: тем же способом, что и pi-box (сокет + tmux has-session),
# потому что процесса с cwd=/work на хосте не существует.
proj = pi_box_project(path)
if proj:
sock = box_socket(proj, path)
if not os.path.exists(sock):
return []
try:
chk = subprocess.run(["tmux", "-S", sock, "has-session", "-t", "pi-box"],
capture_output=True)
except OSError:
return []
if chk.returncode != 0:
return []
panes = subprocess.run(["tmux", "-S", sock, "list-panes", "-a",
"-F", "#{pane_pid}"],
capture_output=True, text=True)
return [int(x) for x in panes.stdout.split() if x.isdigit()]
cwd = session_cwd(path)
if not cwd:
return []
want = os.path.realpath(cwd)
pids = []
for entry in os.listdir("/proc"):
if not entry.isdigit():
continue
try:
if os.path.realpath(f"/proc/{entry}/cwd") != want:
continue
comm = open(f"/proc/{entry}/comm").read().strip()
except OSError:
continue
if comm == "pi":
pids.append(int(entry))
return pids
def session_roots():
"""Где искать сессии: обычный каталог Pi плюс песочницы pi-box.
У pi-box сессии лежат не в ~/.pi/agent, а в <проект>/.pi-box/sessions,
и их не видно при поиске по одному корню. Реестр песочниц ведёт сам pi-box.
"""
explicit = os.environ.get("PI_SESSIONS_ROOT")
if explicit:
return [explicit]
out = []
if os.path.isdir(DEFAULT_ROOT) or not os.path.exists(DEFAULT_ROOT):
out.append(DEFAULT_ROOT)
boxes = os.path.join(PI_BOX_HOME, "boxes")
try:
with open(boxes) as fh:
for line in fh:
line = line.strip()
if not line:
continue
d = os.path.join(line, ".pi-box", "sessions")
if os.path.isdir(d) and d not in out:
out.append(d)
except OSError:
pass
return out
# ── вывод ─────────────────────────────────────────────────────────────────────
def list_rows(roots, color=False):
"""Строки для fzf: сначала самые тяжёлые. Путь уходит в скрытую колонку.
Первая колонка — то, по чему выбирают глазами; третья — путь, который fzf
отдаёт через --accept-nth. Разбирать текст обратно не нужно, поэтому ANSI
цвета не мешают. Корней может быть несколько: ~/.pi/agent плюс песочницы.
У сессий pi-box в заголовке записан /work, а id всегда "main": по этим
полям песочницы неразличимы. Поэтому показываем имя проекта на хосте.
"""
setup_colors(color)
if isinstance(roots, str):
roots = [roots]
# Обычный Pi кладёт сессии в подкаталог на каждый cwd
# (sessions/--home-kiv-code--/файл.jsonl), а pi-box — плоско в
# <проект>/.pi-box/sessions/. Поэтому оба уровня, но один и тот же файл
# дважды не берём.
files = []
seen = set()
for r in roots:
for pat in ("*/*.jsonl", "*.jsonl"):
for f in Path(r).glob(pat):
rp = f.resolve()
if rp not in seen:
seen.add(rp)
files.append(f)
files = sorted(files, key=lambda p: p.stat().st_size, reverse=True)
rows = []
for f in files:
s = summarize(f)
date = time.strftime("%m-%d %H:%M", time.localtime(s["mtime"]))
left = f"{human(s['size']):>8} {CYN}{date}{RST} {s['entries']:>6} стр"
# У сессий pi-box в заголовке записан /work (туда смонтирован проект),
# а id всегда "main" — у всех песочниц одинаково, глазами не различить.
# Показываем настоящий проект на хосте.
proj = pi_box_project(f)
if proj:
# Цельный путь, а не имя и родитель по отдельности: «agent1 ~/tmp»
# читается как два разных значения, хотя проект — ~/tmp/agent1.
right = f"{B}{pi_box_short(proj)}{RST} {DIM}(pi-box){RST}"
else:
right = f"{DIM}{s['cwd']}{RST} {DIM}{s['id'][:8]}{RST}"
rows.append(f"{left}\t{right}\t{f}")
return rows
def cmd_list(roots, color):
for row in list_rows(roots, color):
print(row)
return 0
def cmd_stats(path):
s = summarize(path)
t = s["totals"]
print(f"{B}{os.path.basename(path)}{RST}")
print(f" размер {YEL}{human(s['size'])}{RST}"
f" строк {s['entries']} компакшенов {s['compactions']}"
f" картинок {s['images']}")
proj = pi_box_project(path)
if proj:
print(f" проект {B}{os.path.basename(proj)}{RST} "
f"{DIM}{pi_box_short(proj)}{RST} {DIM}(pi-box){RST}")
else:
print(f" каталог {DIM}{s['cwd']}{RST}")
print(f" расходы ↑{toks(t['input'])} ↓{toks(t['output'])} "
f"R{toks(t['cacheRead'])} W{toks(t['cacheWrite'])} "
f"{GRN}${t['cost']:.3f}{RST}")
if s["prompt"]:
print(f" контекст {toks(s['prompt'])} токенов в последнем запросе")
return 0
# ── обрезка ───────────────────────────────────────────────────────────────────
def fs_type(path):
"""Тип файловой системы, на которой лежит путь (для предупреждений)."""
target = os.path.realpath(path)
best = None
try:
with open("/proc/mounts", encoding="utf-8", errors="replace") as fh:
for line in fh:
parts = line.split()
if len(parts) < 3:
continue
mp = parts[1].replace("\\040", " ")
if target == mp or target.startswith(mp.rstrip("/") + "/"):
# Берём самую длинную подходящую точку монтирования:
# /tmp может быть смонтирован поверх /.
if best is None or len(mp) > len(best[0]):
best = (mp, parts[2])
except OSError:
pass
return best[1] if best else None
def backup_dir():
"""Каталог для бэкапов: /tmp, чтобы копия не удваивала место в проекте.
Подкаталог на пользователя — иначе в общем /tmp одноимённые сессии из
разных проектов затрут друг друга.
"""
return os.path.join(tempfile.gettempdir(), f"pi-shrink-{os.getuid()}")
def free_bytes(path):
try:
st = os.statvfs(path)
return st.f_bavail * st.f_frsize
except OSError:
return 0
def backup_path(path, size):
"""Куда положить бэкап. Возвращает (путь, где_лежит).
Сначала /tmp: сессия бывает на сотни мегабайт, и копия рядом с ней
удваивала бы место в проекте. Если в /tmp не хватает места — кладём рядом
с файлом: лучше копия в неудобном месте, чем её отсутствие.
"""
d = backup_dir()
try:
os.makedirs(d, mode=0o700, exist_ok=True)
except OSError:
d = None
if d and free_bytes(d) > size * 1.05:
# Имя проекта в имени файла: в /tmp лежат бэкапы всех песочниц, и
# без него не понять, чей это main.jsonl.
proj = pi_box_project(path) or session_cwd(path) or ""
tag = os.path.basename(proj) + "-" if proj else ""
first = os.path.join(d, f"{tag}{os.path.basename(path)}.bak")
where = "tmp"
else:
first = f"{path}.bak"
where = "side"
if not os.path.exists(first):
return first, where
stamp = time.strftime("%Y%m%d-%H%M%S")
return f"{first}.{stamp}", where
def cmd_trim(path):
holders = open_by_pi(path)
if holders:
pids = ", ".join(map(str, holders))
print(f"{RED}сессия занята живым Pi (pid {pids}).{RST}", file=sys.stderr)
print(f"{DIM}Он держит её в памяти и продолжит в неё писать. "
f"Останови процесс, обрежь, потом запусти снова.{RST}",
file=sys.stderr)
return 1
lines = raw_lines(path)
entries = load(path)
if not entries:
print(f"{RED}пустой или нечитаемый файл: {path}{RST}", file=sys.stderr)
return 1
fk, comp = cut_id(entries)
if fk is None:
print(f"{RED}в сессии нет компакшена — обрезать нечего.{RST}",
file=sys.stderr)
print(f"{DIM}Без компакшена вся история участвует в контексте: "
f"обрезка потеряла бы память, а не мусор.{RST}", file=sys.stderr)
return 1
cut = line_index(lines, fk)
if cut is None:
print(f"{RED}не нашёл в файле запись {fk} — файл не тронут.{RST}",
file=sys.stderr)
return 1
keep = [0] + list(range(cut, len(lines)))
# Сверяем контекст по ТОМУ ЖЕ набору строк, который запишем: перечитываем
# сохранённые строки, а не срез entries. Иначе проверка и результат — про
# разные данные, и сдвиг на пустой или битой строке прошёл бы незамеченным.
before = context_ids(entries)
after = context_ids(load_from_lines(lines[i] for i in keep))
if before != after:
print(f"{RED}контекст изменился бы — отказ, файл не тронут.{RST}",
file=sys.stderr)
return 1
size_before = os.path.getsize(path)
size_after = sum(len(lines[i]) for i in keep)
if cut == 0 or size_after == size_before:
print(f"{GRN}✓{RST} нечего обрезать: сессия уже на границе "
f"последнего компакшена ({human(size_before)})")
return 0
drop = len(lines) - len(keep)
# Сводка и подтверждение. Отдельного "холостого прогона" нет: он показывал
# ровно это же, а отказаться можно ответом на вопрос ниже.
print(f"{B}{os.path.basename(path)}{RST}")
proj = pi_box_project(path)
if proj:
print(f" проект {B}{os.path.basename(proj)}{RST} "
f"{DIM}{pi_box_short(proj)}{RST} {DIM}(pi-box){RST}")
else:
print(f" каталог {DIM}{session_cwd(path) or '?'}{RST}")
print(f" файл {YEL}{human(size_before)}{RST} → "
f"{GRN}{human(size_after)}{RST}"
f" {DIM}(−{100 * (1 - size_after / max(size_before, 1)):.0f}%){RST}")
print(f" строк {len(lines)} → {len(keep)} {DIM}выбросится {drop}{RST}")
print(f" контекст {GRN}{len(before)} записей — не изменится{RST}")
print()
if not sys.stdin.isatty():
print(f"{RED}нужен терминал для подтверждения{RST}", file=sys.stderr)
return 1
try:
ans = input(f"{B}Обрезать?{RST} [y/N] ")
except EOFError:
ans = ""
if ans.strip().lower() not in ("y", "yes", "д", "да"):
print(f"{DIM}отменено{RST}")
return 1
# Расходы считаем из уже прочитанных строк: перечитывать бэкап на 700 MB
# ради цифр незачем.
before_totals = usage_totals(entries)
after_totals = usage_totals(load_from_lines(lines[i] for i in keep))
prompt_tokens = last_prompt_tokens(entries)
bak, where = backup_path(path, size_before)
try:
shutil.copy2(path, bak)
except OSError as e:
print(f"{RED}не удалось создать бэкап: {e}{RST}", file=sys.stderr)
print(f"{DIM}Обрезка отменена — без копии рисковать нельзя.{RST}",
file=sys.stderr)
return 1
tmp = f"{path}.tmp.{os.getpid()}"
try:
with open(tmp, "wb") as out:
for i in keep:
out.write(lines[i])
out.flush()
os.fsync(out.fileno())
st = os.stat(path)
try:
os.chown(tmp, st.st_uid, st.st_gid)
except OSError:
pass # не root — файл и так наш
os.chmod(tmp, st.st_mode & 0o7777)
os.replace(tmp, path)
except BaseException:
if os.path.exists(tmp):
os.unlink(tmp)
raise
print(f"{GRN}✓{RST} {B}обрезано{RST} {DIM}{os.path.basename(path)}{RST}")
print(f" {B}расходы{RST} {DIM}были{RST} "
f"↑{toks(before_totals['input'])} ↓{toks(before_totals['output'])}"
f" R{toks(before_totals['cacheRead'])} ${before_totals['cost']:.3f}"
f" {DIM}стали{RST} ↑{toks(after_totals['input'])}"
f" ↓{toks(after_totals['output'])} R{toks(after_totals['cacheRead'])}"
f" ${after_totals['cost']:.3f}")
if prompt_tokens:
print(f" {B}контекст-токены{RST} {toks(prompt_tokens)}"
f" {DIM}(берётся из последнего ответа){RST}")
print()
print(f" {DIM}бэкап:{RST} {CYN}{bak}{RST}")
if where == "tmp":
# /tmp — обычно tmpfs в оперативной памяти: копия не занимает диск,
# но исчезнет при перезагрузке и съедает RAM. Говорим прямо, иначе
# на неё понадеются как на обычный файл.
fstype = fs_type(bak)
if fstype == "tmpfs":
print(f" {YEL}эфемерный{RST}{DIM}: /tmp в памяти — пропадёт при "
f"перезагрузке и занимает RAM.{RST}")
print(f" {DIM}нужен надолго:{RST} mv {B}\"{bak}\" "
f"~/куда-нибудь{RST}")
else:
print(f" {DIM}в /tmp; скопируй отсюда, если нужен надолго.{RST}")
print(f" {DIM}вернуть:{RST} mv {B}\"{bak}\" \"{path}\"{RST}")
return 0
# ── меню (fzf) ────────────────────────────────────────────────────────────────
def run_fzf(rows, prompt, header, expect=None):
"""Запускает fzf. Возвращает (нажатая_клавиша, выбор) или (None, None).
fzf рисует себя в /dev/tty, поэтому stdout остаётся чистым — в него
попадает только результат.
"""
args = ["fzf", "--ansi", f"--height={os.environ.get('FZF_HEIGHT', '~60%')}",
"--reverse", "--border", "--no-multi", f"--prompt={prompt}",
f"--header={header}", "--delimiter=\t", "--with-nth=1,2",
"--accept-nth=3"]
if expect:
args.append(f"--expect={expect}")
try:
proc = subprocess.run(args, input="\n".join(rows) + "\n",
capture_output=True, text=True)
except FileNotFoundError:
return None, None
if proc.returncode != 0:
return None, None # Esc или нет совпадений
out = proc.stdout.splitlines()
if not expect:
return "", (out[0] if out else None)
key = out[0] if out else ""
sel = out[1] if len(out) > 1 else ""
return key, (sel or None)
# ── точка входа ───────────────────────────────────────────────────────────────
def main(argv):
mode = "menu"
file = None
roots = session_roots()
i = 1
while i < len(argv):
a = argv[i]
if a in ("-h", "--help"):
print(__doc__)
return 0
elif a == "--color":
pass # цвета в конвейере (для fzf --ansi)
elif a == "--list":
mode = "list"
elif a in ("-s", "--stats"):
mode = "stats"
elif a.startswith("-"):
print(f"{RED}Неизвестный флаг: {a}{RST}", file=sys.stderr)
return 2
else:
file = a
i += 1
color = sys.stdout.isatty() and "NO_COLOR" not in os.environ
want_color = color or "--color" in argv
setup_colors(want_color or mode == "menu")
if mode == "list":
return cmd_list(roots, want_color)
# Без файла: меню со всеми сессиями (в том числе из песочниц).
if not file:
rows = list_rows(roots, color=True)
if not rows:
print(f"{YEL}Сессий не найдено{RST}", file=sys.stderr)
return 0
_, file = run_fzf(rows, "сессия> ", "Enter — выбрать · Esc — выход")
if not file:
return 0
setup_colors(color)
if mode == "menu":
return cmd_trim(file)
if mode == "stats":
return cmd_stats(file)
return cmd_trim(file)
if __name__ == "__main__":
try:
sys.exit(main(sys.argv))
except KeyboardInterrupt:
sys.exit(130)
添加评论
点赞收藏
点踩分享查看原文
评论
?
参与讨论