clean cmux tasks

cmux-clean-idle.py

#!/usr/bin/env python3
"""
cmux のワークスペースを走査して、Claude が動いていない & コマンドが
受付待ち（裸のシェルプロンプト）になっているだけのワークスペースを
列挙／一括停止する。

判定:
  - read-screen の最終 N 行を読む
  - 画面に Claude マーカー (🚀 claude / ⏵⏵ auto mode / esc to interrupt 等)
    があれば keep
  - 最後の非空行が裸のプロンプト (❯ / % / $ / # / >) のみなら IDLE
  - それ以外 (ログが流れている / コマンド実行中) は keep

使い方:
  python3 cmux-clean-idle.py                       # dry-run (候補表示のみ)
  python3 cmux-clean-idle.py --close               # 候補を停止
  python3 cmux-clean-idle.py --include-dead        # 読めない (?) も候補に含める
  python3 cmux-clean-idle.py --include-dead --close
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys

CMUX = "/usr/local/bin/cmux"
SCREEN_LINES = "30"

CLAUDE_MARKERS = re.compile(
    r"🚀\s*claude"
    r"|⏵⏵\s*auto mode"
    r"|claude --resume"
    r"|esc to interrupt"
    r"|Opus 4|Sonnet 4|Haiku 4"
    r"|Try \"/help\""
    r"|⎿\s*ctrl\+c"
)
PROMPT_RE = re.compile(r"^\s*[❯%$#>]\s*$")


def cmux(*args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run([CMUX, *args], capture_output=True, text=True)


def all_workspaces() -> list[tuple[str, str, list[str]]]:
    """Return [(ws_ref, title, [surface_refs in selected order])]."""
    r = cmux("tree", "--all", "--json")
    if r.returncode != 0:
        sys.exit(f"cmux tree failed: {r.stderr}")
    data = json.loads(r.stdout)
    out: list[tuple[str, str, list[str]]] = []
    for w in data.get("windows", []):
        for ws in w.get("workspaces", []):
            surfaces: list[tuple[bool, bool, str]] = []
            for pane in ws.get("panes", []):
                pane_focused = bool(pane.get("focused"))
                for s in pane.get("surfaces", []):
                    if s.get("type") != "terminal":
                        continue
                    surfaces.append(
                        (
                            pane_focused and bool(s.get("selected_in_pane")),
                            bool(s.get("selected_in_pane")),
                            s["ref"],
                        )
                    )
            # focused-and-selected first, then selected-in-pane, then rest
            surfaces.sort(key=lambda t: (not t[0], not t[1]))
            refs = [r for _, _, r in surfaces]
            out.append((ws["ref"], ws.get("title", "") or "", refs))
    return out


def read_screen(ws_ref: str, surface_refs: list[str]) -> str | None:
    # try workspace first (works for single-surface workspaces)
    r = cmux("read-screen", "--workspace", ws_ref, "--lines", SCREEN_LINES)
    if r.returncode == 0:
        return r.stdout
    # fall back to explicit surface refs
    for sref in surface_refs:
        r = cmux("read-screen", "--surface", sref, "--lines", SCREEN_LINES)
        if r.returncode == 0:
            return r.stdout
    return None


def classify(ws_ref: str, surface_refs: list[str]) -> tuple[str, str]:
    """Returns (status, reason). status in {'idle', 'keep', 'unknown'}."""
    screen = read_screen(ws_ref, surface_refs)
    if screen is None:
        return ("unknown", "unreadable")
    if CLAUDE_MARKERS.search(screen):
        return ("keep", "claude")
    lines = [ln.rstrip() for ln in screen.splitlines() if ln.strip()]
    if not lines:
        return ("unknown", "empty")
    last = lines[-1]
    if PROMPT_RE.match(last):
        return ("idle", "shell-prompt")
    return ("keep", "active")


def main() -> int:
    args = sys.argv[1:]
    do_close = "--close" in args
    include_dead = "--include-dead" in args
    current = os.environ.get("CMUX_WORKSPACE_ID", "")

    workspaces = all_workspaces()
    if not workspaces:
        print("no workspaces found")
        return 0

    rows: list[tuple[str, str, str, str]] = []
    candidates: list[tuple[str, str]] = []
    for ref, title, surface_refs in workspaces:
        status, reason = classify(ref, surface_refs)
        is_current = bool(current) and current == ref
        if is_current:
            status, reason = ("keep", "current")
        rows.append((status, reason, ref, title))
        if status == "idle":
            candidates.append((ref, title))
        elif status == "unknown" and include_dead:
            candidates.append((ref, title))

    width = max((len(r[2]) for r in rows), default=12)
    for status, reason, ref, title in rows:
        tag = {"idle": "IDLE", "keep": "keep", "unknown": "?   "}[status]
        print(f"  {tag:5s} {reason:14s} {ref:<{width}}  {title[:80]}")

    print()
    print(f"idle候補: {len(candidates)} 件")
    if not candidates:
        return 0
    if not do_close:
        print("実際に停止するには --close を付けてください")
        return 0

    for ref, title in candidates:
        print(f"close {ref}  {title[:60]}")
        r = cmux("close-workspace", "--workspace", ref)
        if r.returncode != 0:
            print(f"  failed: {r.stderr.strip()}", file=sys.stderr)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

config.toml

[tasks."cmux:clean-idle"]
description = "cmux: Claude が動いていない & シェルプロンプトで止まっているワークスペースを列挙 (dry-run)"
run = """
python3 ~/.config/mise/scripts/cmux-clean-idle.py
"""

[tasks."cmux:clean-idle:close"]
description = "cmux: 上記の idle ワークスペースを実際に停止する"
run = """
python3 ~/.config/mise/scripts/cmux-clean-idle.py --close
"""

[tasks."cmux:clean-idle:close-all"]
description = "cmux: idle に加えて 読み取り不能 (廃) ワークスペースもまとめて停止する"
run = """
python3 ~/.config/mise/scripts/cmux-clean-idle.py --include-dead --close
"""