2 * 1024 * 1024:
lines = lines[1:] # первая строка обрезана
except OSError:
return []
out = []
for line in lines:
try:
r = json.loads(line)
except ValueError:
continue
if not isinstance(r, dict) or not isinstance(r.get("message"), dict):
continue
if (skip_sidechain and r.get("isSidechain")) or r.get("isMeta") or r.get("isCompactSummary"):
continue
msg = r["message"]
if r.get("type") == "user":
text = _user_text(msg.get("content"))
if text:
out.append({"key": "u:" + str(r.get("uuid") or len(out)), "role": "user",
"text": text, "ts": r.get("timestamp")})
elif r.get("type") == "assistant":
c = msg.get("content")
parts = [c] if isinstance(c, str) else [
b.get("text") or "" for b in (c or []) if isinstance(b, dict) and b.get("type") == "text"]
text = "\n\n".join(p.strip() for p in parts if p and p.strip())
if not text:
continue
key = "a:" + str(msg.get("id") or r.get("uuid"))
if out and out[-1]["key"] == key: # одно сообщение модели записано несколькими строками
out[-1]["text"] += "\n\n" + text
else:
out.append({"key": key, "role": "assistant", "text": text, "ts": r.get("timestamp")})
return out[-limit:]
def md_html(text: str) -> str:
"""Минимальный markdown: блоки кода, `код`, **жирный**, заголовки. Остальное — как есть."""
if len(text) > MAX_MESSAGE_CHARS:
text = text[:MAX_MESSAGE_CHARS] + "\n\n… (обрезано)"
out = []
for i, part in enumerate(re.split(r"```[^\n]*\n(.*?)(?:```|$)", text, flags=re.S)):
if i % 2:
out.append(f"{html.escape(part.rstrip())}")
continue
if i and part.startswith("\n"):
part = part[1:] # перевод строки сразу после блока кода дал бы пустую строку
h = html.escape(part)
h = re.sub(r"`([^`\n]+)`", r"\1", h)
h = re.sub(r"\*\*(.+?)\*\*", r"\1", h)
h = re.sub(r"(?m)^#{1,6}\s+(.+)$", r'\1', h)
h = re.sub(r"\[([^\]]+)\]\((https?://[^)\s]+)\)", r"\1", h)
out.append(h)
return "".join(out).strip()
def _fmt_ts(ts):
if not ts:
return ""
try:
from datetime import datetime
return datetime.fromisoformat(ts.replace("Z", "+00:00")).astimezone().strftime("%H:%M")
except ValueError:
return ""
def message_payload(sid, aid, after=""):
"""Сообщения после `after` (ключ последнего показанного). Без after — только последнее."""
s = next((x for x in state["tree"] if x["id"] == sid), None)
if s is None:
return {"error": "Сессия больше не активна", "messages": []}
if aid:
a = next((x for x in s["agents"] if x["id"] == aid), None)
if a is None:
return {"error": "Агент больше не показывается", "uuid": s["uuid"], "messages": []}
msgs = recent_messages(a.get("transcript"))
title, who = a["desc"] or a["type"], {"user": "Задание", "assistant": "Агент"}
else:
msgs = recent_messages(s.get("transcript"), skip_sidechain=True)
title, who = s["name"], {"user": "Вы", "assistant": "Claude"}
keys = [m["key"] for m in msgs]
if after in keys:
i = keys.index(after)
msgs = msgs[i:] # включая сам after — у модели он мог дописаться
elif not after:
msgs = msgs[-1:]
return {"title": title, "uuid": s["uuid"], "max_chars": HISTORY_MAX_CHARS,
"error": "" if msgs or after else "Сообщений пока нет",
"messages": [{"key": m["key"], "role": m["role"], "who": who[m["role"]],
"time": _fmt_ts(m.get("ts")), "chars": len(m["text"]),
"html": md_html(m["text"])} for m in msgs]}
# ---------- отрисовка ----------
STATUS = {"working": "green", "waiting": "yellow", "idle": "dim",
"running": "green", "done": "dim"}
def elapsed(t0, t1=None):
sec = int((t1 or time.time()) - t0)
return f"{sec // 60}:{sec % 60:02d}" if sec < 3600 else f"{sec // 3600} ч {sec % 3600 // 60} м"
def wt_html(g, uuid) -> str:
if not g:
return ""
label = html.escape(g["name"]) + (f' · {html.escape(g["branch"])}' if g["branch"] else "")
return (f'⎇ {label}')
def render_agents(agents, parent, uuid="", parent_root=None):
"""Дерево вложенными блоками: отступ и тонкая линия слева вместо псевдографики."""
kids = [a for a in agents if a.get("parent") == parent]
if not kids:
return ""
out = ['']
for a in kids:
color = STATUS.get(a["status"], "dim")
model = html.escape(a["model_label"])
model_html = f'
{model}' if model else ""
desc = html.escape(a["desc"]) if a["desc"] else ""
atype = html.escape(a["type"])
took = elapsed(a["start"], a["end"]) if a.get("start") else ""
g = a.get("git")
show = {"always": True, "never": False}.get(AGENT_DIR, bool(g) and g["root"] != parent_root)
wt = wt_html(g, uuid) if g and show else ""
meta = " · ".join(x for x in (took, tokens_html(a.get("tokens")), wt) if x)
out.append(f"""
●
{desc or atype}{model_html}
{f'
{meta}
' if meta else ''}
{render_agents(agents, a["id"], uuid, (g or {}).get("root") or parent_root)}
""")
out.append("
{html.escape(state["config_error"])}
' + _render_tree()
return _render_tree()
def _render_tree() -> str:
if state["error"]:
return f'{html.escape(state["error"])}
'
if not state["tree"]:
return 'Нет активных сессий Claude Code
'
parts = []
for s in state["tree"]:
color = STATUS.get(s["status"], "dim")
model = html.escape(s["model_label"])
meta = " · ".join(x for x in (tokens_html(s.get("tokens")), wt_html(s.get("git"), s["uuid"])) if x)
parts.append(f"""
●
{html.escape(s['name'])}
{f'{model}' if model else ''}
{f'
{meta}
' if meta else ''}
{render_agents(s['agents'], None, s['uuid'], (s.get('git') or {}).get('root'))}
""")
return "".join(parts)
# Фон панели выбирается на клиенте (⚙ внизу) и хранится в localStorage этого браузера/web view;
# текст и приглушённый текст подбираются под яркость фона, цвета статусов не меняются.
BG_PRESETS = ("#002b36", "#1e1e1e", "#000000", "#2b2b3a", "#fdf6e3")
LIGHT_BG_TEXT = "#3b4a50" # текст на светлом фоне; на тёмном — fg темы по умолчанию
BG_STORAGE_KEY = "claude-panel-bg"
CLIENT_COLORS_JS = """function panelColors(bg) {
if (typeof bg !== 'string' || !/^#[0-9a-f]{6}$/i.test(bg)) return null;
bg = bg.toLowerCase();
const rgb = h => [1, 3, 5].map(i => parseInt(h.slice(i, i + 2), 16));
const lin = c => (c /= 255) <= 0.03928 ? c / 12.92 : Math.pow((c + 0.055) / 1.055, 2.4);
const lum = h => { const [r, g, b] = rgb(h).map(lin); return 0.2126 * r + 0.7152 * g + 0.0722 * b; };
const ratio = (a, c) => (Math.max(a, c) + 0.05) / (Math.min(a, c) + 0.05); // контраст по WCAG
const L = lum(bg); // текст — тот из двух, что контрастнее фону
const fg = ratio(lum('__LIGHT__'), L) > ratio(lum('__DARK__'), L) ? '__LIGHT__' : '__DARK__';
const f = rgb(fg), k = rgb(bg); // dim — как у темы: 60% текста, 40% фона
const dim = '#' + f.map((c, i) => Math.round(c * 0.6 + k[i] * 0.4).toString(16).padStart(2, '0')).join('');
return {bg, fg, dim};
}""".replace("__LIGHT__", LIGHT_BG_TEXT).replace("__DARK__", DEFAULT_THEME["fg"])
def render_bgctl() -> str:
swatches = "".join(f''
for c in BG_PRESETS)
return (f'')
def render_page() -> str:
t = state["theme"]
return f"""
{render_tree()}
{render_limits()}
{render_bgctl()}
"""
# ---------- лимиты подписки и баланс ----------
USAGE_URL = "https://api.anthropic.com/api/oauth/usage"
KIND_TITLES = {"session": "Сессия · 5 ч", "weekly_all": "Неделя · все модели"}
CURRENCY_SIGNS = {"EUR": "€", "USD": "$", "GBP": "£"}
SESSION_KEYCHAIN_SERVICE = "claude-ai-sessionKey" # запасной способ для баланса
BROWSER_UA = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/129.0 Safari/537.36")
def keychain(service: str):
try:
out = subprocess.run(["security", "find-generic-password", "-s", service, "-w"],
capture_output=True, text=True, timeout=10).stdout.strip()
return out or None
except Exception:
return None
def get_token() -> str:
"""OAuth-токен Claude Code: на macOS — из Keychain, иначе из ~/.claude/.credentials.json."""
raw = keychain("Claude Code-credentials")
if raw:
try:
return json.loads(raw)["claudeAiOauth"]["accessToken"]
except (ValueError, KeyError):
pass
with open(os.path.expanduser("~/.claude/.credentials.json")) as f:
return json.load(f)["claudeAiOauth"]["accessToken"]
def _get_json(url: str, headers: dict) -> dict:
with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=15) as r:
return json.load(r)
def _oauth_headers(token):
return {"Authorization": f"Bearer {token}", "anthropic-beta": "oauth-2025-04-20",
"Content-Type": "application/json", "User-Agent": "iterm2-claude-panel/1.0"}
def fetch_usage() -> dict:
return _get_json(USAGE_URL, _oauth_headers(get_token()))
def fetch_balance():
"""Пробует способы по очереди и запоминает сработавший. Возвращает (json, способ)."""
global ORG_UUID
token = get_token()
if not ORG_UUID:
try:
with open(os.path.expanduser("~/.claude.json"), encoding="utf-8") as f:
ORG_UUID = (json.load(f).get("oauthAccount") or {}).get("organizationUuid") or ""
except (OSError, ValueError):
pass
if not ORG_UUID:
ORG_UUID = _get_json("https://api.anthropic.com/api/oauth/profile",
_oauth_headers(token))["organization"]["uuid"]
path = f"/api/organizations/{ORG_UUID}/prepaid/credits"
oauth = _oauth_headers(token)
attempts = [("oauth-api", f"https://api.anthropic.com/api/oauth{path[4:]}", oauth),
("oauth-claude.ai", f"https://claude.ai{path}", {**oauth, "User-Agent": BROWSER_UA})]
session = keychain(SESSION_KEYCHAIN_SERVICE)
if session:
attempts.append(("sessionKey", f"https://claude.ai{path}", {
"Cookie": f"sessionKey={session}", "User-Agent": BROWSER_UA, "Accept": "application/json"}))
if state["balance_via"]: # сначала то, что уже срабатывало
attempts.sort(key=lambda a: a[0] != state["balance_via"])
last = None
for name, url, headers in attempts:
try:
return _get_json(url, headers), name
except Exception as e:
last = e
raise last
async def usage_poller():
loop = asyncio.get_running_loop()
while True:
try:
state["usage"] = await loop.run_in_executor(None, fetch_usage)
state["usage_error"], state["usage_updated"] = None, time.time()
except urllib.error.HTTPError as e:
state["usage_error"] = ("Токен истёк — запустите claude, он обновит токен"
if e.code == 401 else f"Лимиты: API ответил HTTP {e.code}")
except FileNotFoundError:
state["usage_error"] = "Не найден токен Claude Code — выполните claude login"
except Exception as e:
state["usage_error"] = f"Лимиты: {e}"
await asyncio.sleep(USAGE_REFRESH_SEC)
async def balance_poller():
loop = asyncio.get_running_loop()
while True:
if SHOW_BALANCE:
try:
state["balance"], state["balance_via"] = await loop.run_in_executor(None, fetch_balance)
state["balance_error"] = None
except urllib.error.HTTPError as e:
state["balance_error"] = (f"нет доступа (HTTP {e.code}) — нужен sessionKey"
if e.code in (401, 403) else f"HTTP {e.code}")
except Exception as e:
state["balance_error"] = str(e)
await asyncio.sleep(BALANCE_REFRESH_SEC)
def parse_limits(data: dict):
"""[(подпись, used%, resets_at, severity)] из массива `limits` (его рисует Desktop);
для старого формата ответа — из five_hour / seven_day."""
rows = []
for lim in data.get("limits") or []:
scope = lim.get("scope") or {}
model = (scope.get("model") or {}).get("display_name")
surface = scope.get("surface")
if isinstance(surface, dict):
surface = surface.get("display_name")
scoped = model or surface
label = KIND_TITLES.get(lim.get("kind", "")) or (f"Неделя · {scoped}" if scoped else lim.get("kind", ""))
rows.append((label, float(lim.get("percent") or 0), lim.get("resets_at"), lim.get("severity")))
if rows:
return rows
for key, label in (("five_hour", "Сессия · 5 ч"), ("seven_day", "Неделя · все модели")):
w = data.get(key)
if isinstance(w, dict) and w.get("utilization") is not None:
rows.append((label, float(w["utilization"]), w.get("resets_at"), None))
return rows
def parse_balance(b):
"""(баланс, валюта, пополнено в активных траншах, дата последнего пополнения)."""
if not isinstance(b, dict):
return None
m = (b.get("balance") or {}).get("money")
if m:
amount, cur = m["amount_minor"] / 10 ** m.get("exponent", 2), m.get("currency", "")
elif b.get("amount") is not None:
amount, cur = b["amount"] / 100, b.get("currency", "")
else:
return None
granted, last = 0.0, None
for t in (b.get("tranches") or []) + (b.get("promo_tranches") or []):
g = (t.get("granted") or {}).get("money")
if g:
granted += g["amount_minor"] / 10 ** g.get("exponent", 2)
if t.get("granted_at") and (last is None or t["granted_at"] > last):
last = t["granted_at"]
return amount, cur, granted or None, last
def fmt_reset(value) -> str:
if not value:
return ""
try:
dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except ValueError:
return ""
sec = int((dt - datetime.now(timezone.utc)).total_seconds())
if sec <= 0:
return "сбрасывается"
if sec >= 86400:
return dt.astimezone().strftime("%a %H:%M")
h, sec = divmod(sec, 3600)
m = sec // 60
return " ".join(([f"{h} ч"] if h else []) + ([f"{m} мин"] if m else [])) or "<1 мин"
def money(v: float, cur: str) -> str:
return f"{CURRENCY_SIGNS.get(cur, cur + ' ')}{v:,.2f}".replace(",", " ")
def render_limits() -> str:
data = state["usage"] or {}
parts = []
if state["usage_error"]:
parts.append(f'{html.escape(state["usage_error"])}
')
for label, used, resets, severity in parse_limits(data):
used = max(0.0, min(100.0, used))
lvl = ("crit" if severity == "critical" or (severity is None and used >= 90) else
"warn" if severity == "warning" or (severity is None and used >= 75) else "")
reset = fmt_reset(resets)
parts.append(f"""
{html.escape(label)}
{used:.0f}%
{f'
сброс {html.escape(reset)}
' if reset else ''}
""")
if SHOW_BREAKDOWN:
rows = (data.get("seven_day_breakdown") or {}).get("rows") or []
items = [f"{html.escape(r.get('display_name') or r.get('key', ''))} {r.get('percent', 0)}%"
for r in rows if r.get("percent")]
if items:
parts.append(f'{" · ".join(items)}
')
if SHOW_BALANCE:
bal = parse_balance(state["balance"])
if bal:
amount, cur, granted, last = bal
sub = ([f"пополнено {money(granted, cur)}"] if granted else []) + (
[datetime.fromisoformat(last.replace("Z", "+00:00")).astimezone().strftime("%d.%m")]
if last else [])
parts.append(f"""
Баланс
{money(amount, cur)}
{f'
{html.escape(" · ".join(sub))}
' if sub else ''}
""")
elif state["balance_error"]:
parts.append(f'Баланс'
f'—
'
f'{html.escape(state["balance_error"])}
Лимиты: загрузка…
')
if state["usage_updated"]:
upd = datetime.fromtimestamp(state["usage_updated"]).strftime("%H:%M:%S")
stale = state["usage_error"] is not None # последнее обновление не удалось — показаны старые цифры
parts.append(f'')
return "".join(parts)
# ---------- действия по клику (через tmux) ----------
def known_roots():
roots = set()
def walk(g):
if g:
roots.add(g["root"])
for s in state["tree"]:
walk(s.get("git"))
for a in s["agents"]:
walk(a.get("git"))
return roots
_last_open = {} # path -> time последнего открытия (защита от двойных запросов)
async def open_worktree(uuid, path):
"""Делит панель tmux сессии по горизонтали и сразу запускает в новой панели файловый менеджер.
Команда передаётся tmux при создании панели, а не вводом текста в оболочку:
так не возникает гонки «текст отправлен раньше, чем оболочка готова»,
и панель закрывается сама, когда выходите из lf. Сессии вне tmux не поддерживаются."""
if path not in known_roots() or not os.path.isdir(path):
return # открываем только папки, которые панель сама показала
if not (uuid and uuid.startswith("tmux:") and uuid in known_targets()):
return
now = time.time()
if now - _last_open.get(path, 0) < 3:
return # повторный клик/запрос — уже открываем
_last_open[path] = now
state["open_log"] = (state.get("open_log") or [])[-9:] + [
{"t": time.strftime("%H:%M:%S"), "uuid": uuid, "path": path}]
sock, pane = _tmux_target(uuid)
cmd = f"exec {FILE_MANAGER}"
await asyncio.get_running_loop().run_in_executor(
None, lambda: _tmux(sock, "split-window", "-h", "-t", pane, "-c", path, cmd))
def known_targets():
return {s["uuid"] for s in state["tree"] if s.get("uuid")}
def _tmux_target(target):
sock, _, pane = target[5:].partition("|")
return sock, pane
def _tmux(sock, *args):
tmux = _which("tmux")
if tmux:
subprocess.run([tmux, "-S", sock, *args], capture_output=True, timeout=5)
async def activate(uuid):
if uuid not in known_targets() or not uuid.startswith("tmux:"):
return # только сессии в tmux, которые панель сама показала
sock, pane = _tmux_target(uuid)
loop = asyncio.get_running_loop()
for args in (("switch-client", "-t", pane), ("select-window", "-t", pane),
("select-pane", "-t", pane)):
await loop.run_in_executor(None, lambda a=args: _tmux(sock, *a))
async def refresher():
last_cleanup = 0
while True:
try:
load_config()
loop = asyncio.get_running_loop()
sessions = await loop.run_in_executor(None, load_sessions)
alive_now = [x for x in sessions if not x["ended"]]
procs = await loop.run_in_executor(None, running_processes, [x["pid"] for x in alive_now])
tmux = await loop.run_in_executor(
None, tmux_panes, {x["tmux_socket"] for x in alive_now if x["tmux_socket"]})
state["tree"] = build_tree(sessions, None, procs, tmux)
state["error"] = None
if time.time() - last_cleanup > 3600:
cleanup_logs()
last_cleanup = time.time()
except Exception as e:
state["error"] = f"Ошибка: {e}"
await asyncio.sleep(POLL_SEC)
def cleanup_logs():
for path in glob.glob(os.path.join(LOG_DIR, "*.jsonl")):
try:
if time.time() - os.path.getmtime(path) > KEEP_LOGS_SEC:
os.remove(path)
except OSError:
pass
# ---------- HTTP ----------
async def handle(reader, writer):
try:
request = (await reader.read(2048)).decode("latin-1")
target = request.split(" ", 2)[1] if request.count(" ") >= 2 else "/"
url = urllib.parse.urlparse(target)
headers = {}
for line in request.split("\r\n")[1:]:
name, sep, value = line.partition(":")
if sep:
headers[name.strip().lower()] = value.strip()
port = writer.get_extra_info("sockname")[1]
# защита от DNS rebinding и чужих страниц в браузере: только свой адрес в Host,
# а действия (/activate, /open-wt) — ещё и не из чужого сайта
if (headers.get("host", "").lower() not in (f"127.0.0.1:{port}", f"localhost:{port}")
or (url.path in ("/activate", "/open-wt")
and headers.get("sec-fetch-site", "").lower() == "cross-site")):
writer.write(b"HTTP/1.1 403 Forbidden\r\nConnection: close\r\nContent-Length: 0\r\n\r\n")
await writer.drain()
return
ctype = "text/html; charset=utf-8"
if url.path == "/limits":
body = render_limits()
elif url.path == "/fragment":
body = render_tree()
elif url.path == "/open-wt":
q = urllib.parse.parse_qs(url.query)
await open_worktree(q.get("id", [""])[0], q.get("path", [""])[0])
body = "ok"
elif url.path == "/message":
q = urllib.parse.parse_qs(url.query)
body = json.dumps(message_payload(q.get("sid", [""])[0], q.get("aid", [""])[0],
q.get("after", [""])[0]), ensure_ascii=False)
ctype = "application/json; charset=utf-8"
elif url.path == "/activate":
uuid = urllib.parse.parse_qs(url.query).get("id", [""])[0]
await activate(uuid)
body = "ok"
elif url.path == "/raw":
body = json.dumps({"log_dir": LOG_DIR, "open_log": state.get("open_log"),
"theme": state["theme"],
"usage_error": state["usage_error"], "balance_via": state["balance_via"],
"balance_error": state["balance_error"],
"log_files": sorted(os.path.basename(p) for p in
glob.glob(os.path.join(LOG_DIR, "*.jsonl"))),
"error": state["error"], "debug": state.get("debug"),
"tree": state["tree"]},
ensure_ascii=False, indent=2, default=str)
ctype = "application/json; charset=utf-8"
else:
body = render_page()
data = body.encode("utf-8")
writer.write(
f"HTTP/1.1 200 OK\r\nContent-Type: {ctype}\r\nCache-Control: no-store\r\n"
f"Connection: close\r\nContent-Length: {len(data)}\r\n\r\n".encode() + data
)
await writer.drain()
finally:
writer.close()
# ---------- запуск ----------
# Два режима одного файла: служба launchd (--headless) держит порт и всё считает,
# скрипт AutoLaunch в iTerm2 только показывает её страницу в Toolbelt.
SERVICE_LABEL = "com.andrey.claude-agents" # метка LaunchAgent и идентификатор web view
async def serve():
"""Сервер панели без iTerm2: страница, дерево сессий, лимиты и баланс. Работает до отмены."""
global _server
load_config()
_server = await asyncio.start_server(handle, "127.0.0.1", PORT)
tasks = [asyncio.create_task(c) for c in (refresher(), usage_poller(), balance_poller())]
try:
async with _server:
await _server.serve_forever()
finally:
for t in tasks:
t.cancel()
def port_alive(port, host="127.0.0.1", timeout=0.5) -> bool:
try:
with socket.create_connection((host, port), timeout=timeout):
return True
except OSError:
return False
def kickstart_service():
"""Просит launchd запустить службу панели, если она не запущена."""
try:
subprocess.run(["launchctl", "kickstart", f"gui/{os.getuid()}/{SERVICE_LABEL}"],
capture_output=True, timeout=10)
except Exception:
pass
async def register_toolbelt(connection):
"""Скрипт AutoLaunch: регистрирует web view со страницей службы; свой сервер не поднимает."""
load_config()
if not port_alive(PORT):
kickstart_service()
for _ in range(20): # до ~5 с ждём, пока служба займёт порт, иначе web view покажет ошибку
if port_alive(PORT):
break
await asyncio.sleep(0.25)
await iterm2.tool.async_register_web_view_tool(
connection,
display_name="Claude",
identifier=SERVICE_LABEL,
reveal_if_already_registered=False,
url=f"http://127.0.0.1:{PORT}/",
)
def run(argv):
if "--headless" in argv:
asyncio.run(serve())
elif iterm2 is not None:
iterm2.run_forever(register_toolbelt)
else:
raise SystemExit("нужен iTerm2 или флаг --headless")
if __name__ == "__main__":
run(sys.argv[1:])