diff --git a/devloop/.claude-plugin/plugin.json b/devloop/.claude-plugin/plugin.json index 7114595..04c1df0 100644 --- a/devloop/.claude-plugin/plugin.json +++ b/devloop/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "devloop", - "version": "0.2.2", + "version": "0.2.3", "description": "Dev-loop workflow, native-first rebuild: git/PR (GitHub + GitLab) + cwd-aware enter + Board-managed context delivery + lint/test gates. Built on native Claude Code events (CwdChanged / PostCompact / FileChanged / monitors).", "author": { "name": "qiankunli", diff --git a/devloop/.codex-plugin/plugin.json b/devloop/.codex-plugin/plugin.json index b08adac..c700c0b 100644 --- a/devloop/.codex-plugin/plugin.json +++ b/devloop/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "devloop", - "version": "0.2.2", + "version": "0.2.3", "description": "Dev-loop workflow: git/PR (GitHub + GitLab) + Board-managed workspace/repo context delivery + lint/test gates + execution-level hard intercepts for Claude Code and Codex.", "author": { "name": "qiankunli", diff --git a/devloop/AGENTS.md b/devloop/AGENTS.md index 638bb86..809d021 100644 --- a/devloop/AGENTS.md +++ b/devloop/AGENTS.md @@ -72,7 +72,8 @@ devloop/ │ ├── posttool_git_refresh.py # PostToolUse：git 状态命令后刷新 branch 段 │ ├── sessionend_release.py # SessionEnd：释放本 session 的 owner 锁（正常退出路径） │ └── pretool_*.py # 命令/编辑硬拦截（guard harness；含 owner 锁与裸 worktree add 拦截） -├── scripts/ # 工作流驱动 adapter：enter / commit_flow + smart_* / pr / release / lint/test/review / init_* +├── ui/board/ # Board UI adapter：稳定状态 projection + Claude status line / Codex tmux sidecar +├── scripts/ # 工作流驱动 adapter：enter / commit_flow + smart_* / pr / release / lint/test/review / init_* / Claude Board setup ├── monitors/monitors.json # ★PR-sweep 后台轮询（替代 hook 心跳 scheduler） ├── commands/ # slash：enter / gcam / gcamp / gcampr（lint/test 归 skill，gate 自动触发） ├── skills/ # git-ops / gcam / gcamp / gcampr / fix-lint / run-test diff --git a/devloop/README.md b/devloop/README.md index 8f9810c..8b7bc32 100644 --- a/devloop/README.md +++ b/devloop/README.md @@ -14,7 +14,7 @@ devloop 通过两个控制杠杆落实这条生命周期：**Board 上下文投 - **PR/MR 生命周期入口**：`/enter` 选择 repo/branch；多 session 并发时，`--worktree ` 将 branch 以隔离 checkout 形态展开，并统一处理基线、依赖准备和安全清理。`/gcam`、`/gcamp`、`/gcampr` 依次覆盖 commit、push 与创建/复用 PR/MR；已有 PR/MR 的冲突处理走可恢复的 `smart_rebase.sh start/continue/finish`，以 rebase 前保存的远端 SHA 做精确 `force-with-lease`。新工作从目标分支建立干净基线，最终 merge 始终留给人。 - **component 感知**（多代码目录仓）：一个 git 仓可能有多个自带工具链、可独立 lint/test 的目录——`server/` + `cli/`、`packages/*`、`cmd/*`。devloop 按**本次改动**决定跑哪些：改了 `cli/**` 就只跑 `cli` 的 lint/test，不静默回落仓根或 `server/`；改动跨多个 component 就都跑；clean tree 从仓根发起时枚举**全部** component（绝不替你猜一个）。验证戳也按 component 记——「A 过 B 挂」不会被记成整仓已验。术语见 [`CONCEPTS.md`](./CONCEPTS.md)。 -- **Board 与 PR/MR 感知**：workspace/repo 的 `.devloop/` 保存当前 branch、工作区、近期 PR/MR、验证和 session 归属等结构化运行态；hook 与 monitor 持续刷新事实，Board 按相关性选择紧凑内容并决定 session/turn 投递。状态与投递游标都自动排除在 git 提交之外。 +- **Board 与 PR/MR 感知**：workspace/repo 的 `.devloop/` 保存当前 branch、工作区、近期 PR/MR、验证和 session 归属等结构化运行态；hook 与 monitor 持续刷新事实，Board 按相关性选择紧凑内容并决定 session/turn 投递。Claude 可用原生 status line 常驻展示工作上下文与健康状态；Codex 在 tmux 中使用三行 sidecar。状态与投递游标都自动排除在 git 提交之外。 - **硬拦截**（PreToolUse deny）：保护分支 commit/push、`git add -A`、直接 `git worktree add`、过期分支（PR 已 merged/closed）改文件、别的 session 占用的 checkout 上切分支或改文件（引导 worktree）、工作区根跑子项目命令、裸 `pytest`、uv 项目 `pip install`、编辑 `requirements.txt`、`lifecycle.pre_commit` 含 lint 时 lint 过期的裸 `git commit` gate。 - **自动进项目**：`cd` 进子项目时（`CwdChanged`）自动刷新上下文、浮现 AGENTS.md References，无需手动 `/enter`。 - **生命周期 hook**：`pre_commit / post_commit / pre_mr / post_mr` 四相位可挂 hook，挂哪相位由 config 决定；两类——**inline 门禁**（失败挡 commit/MR）与 **signal hook**（advisory、后台跑、不挡）。当前内置三个：`lint`、`test`（门禁），`review`（signal——后台跑 [ocr](https://github.com/alibaba/open-code-review) 审全量改动、结果回流会话、有开放 MR 时发评论）。机制见 [`docs/lifecycle-hooks.md`](./docs/lifecycle-hooks.md)；code-review 细节见 [`docs/code-review.md`](./docs/code-review.md)。 @@ -28,6 +28,7 @@ devloop 通过两个控制杠杆落实这条生命周期：**Board 上下文投 | `/gcam ""` | 只 commit | | `/gcamp ""` | commit + push | | `/gcampr "" [--branch ]` | commit + push + 建/复用 PR/MR | +| `/devloop:setup-board` | 为 Claude Code 安装原生 Board status line | lint / test 无独立 slash 命令：正常由 gcam* 的 `pre_commit` gate 自动触发；手动跑走 fix-lint / run-test skill（自然语言"修下 lint"/"跑下测试"）。两条路径**共用同一套 component 选择**，所以 gate 替你跑的和你手动跑的永远是同一批 component；每次执行会自述本轮选了哪些 component、为什么（`changed files under: cli` / `clean tree, all components: …`），选错一眼可见，不用等错的测试跑完再猜。 @@ -52,11 +53,25 @@ bash /scripts/smart_rebase.sh finish --repo 运行时要求：**Python 3.10+**。devloop launcher 会从 PATH 自动选择首个满足版本的 `python3`、 `python` 或带版本号的 `python3.x`；需要固定解释器时设置 `DEVLOOP_PYTHON`。 +Claude Board HUD 使用 Claude Code 原生 status line，不依赖 tmux。安装 plugin 后运行一次 +`/devloop:setup-board`；安装器会备份 `~/.claude/settings.json`，发现已有非 devloop status line +时不会覆盖，只有得到明确确认后才允许替换。 + +Codex 暂无自定义 status line provider，当前 sidecar 依赖 +[tmux](https://github.com/tmux/tmux)。devloop 会自动检测：未安装 tmux 或 CLI 不在 tmux +会话中时不启动 sidecar，其余能力不受影响。macOS 如需 Codex HUD，从 tmux 内启动： + +```bash +brew install tmux +tmux +# 然后在 tmux 会话中启动 codex +``` ``` # Claude Code 内 /plugin marketplace add https://github.com/qiankunli/devloop.git /plugin install devloop@devloop +/devloop:setup-board ``` Codex： @@ -109,6 +124,9 @@ devloop 对外部的依赖（连哪个 forge、用什么 token）+ 工作区注 // 这里一般留空；也可用 init_workspace.py 显式补充。 "workspaces": [], + // Board HUD 总开关：Claude 原生 status line 与 Codex tmux sidecar 共用。 + "board": { "hud": { "enabled": true } }, + // 代码评审平台，按 repo 的 origin host 索引。PR/MR 创建与状态注入需要 token； // 没有匹配 token 时相关功能静默跳过，其余照常。provider 由 host 推断，type 可覆写。 "forges": { diff --git a/devloop/commands/setup-board.md b/devloop/commands/setup-board.md new file mode 100644 index 0000000..e5ed305 --- /dev/null +++ b/devloop/commands/setup-board.md @@ -0,0 +1,18 @@ +--- +description: Configure Claude Code's native devloop Board status line +--- + +Configure the native Claude Code Board status line using the plugin's installer: + +```bash +"${CLAUDE_PLUGIN_ROOT}/scripts/python" "${CLAUDE_PLUGIN_ROOT}/scripts/setup_claude_board.py" \ + --plugin-root "${CLAUDE_PLUGIN_ROOT}" +``` + +If the installer exits with `CONFLICT`, an existing non-devloop `statusLine` was found. +Ask the user whether to replace it. Only after explicit confirmation, rerun the same +command with `--replace`. Never edit or overwrite the existing status line manually. + +On success, tell the user Claude reloads settings automatically and the Board should +appear after the next interaction. The installer prints a backup path when it changes +an existing settings file. diff --git a/devloop/config/config.example.json b/devloop/config/config.example.json index 90c0845..31582ca 100644 --- a/devloop/config/config.example.json +++ b/devloop/config/config.example.json @@ -10,6 +10,8 @@ "worktree": { "keep_recent": 5 }, + "board": { "hud": { "enabled": true } }, + "lifecycle": { "default": { "pre_commit": [], "post_commit": [], "pre_mr": [], "post_mr": [] }, "repos": { diff --git a/devloop/docs/board.md b/devloop/docs/board.md index ffa4a2b..24fe467 100644 --- a/devloop/docs/board.md +++ b/devloop/docs/board.md @@ -31,8 +31,11 @@ focus，把事实投影为共享 `Board`，再得到相关 `BoardView`。每次 已经变化的部分。 UI 读取 `BoardRuntime.snapshot()` 得到 JSON-ready 的结构化 view；读取不经过 prompt -renderer，也不改变 delivery receipt。下一阶段的 Board UI 只消费这个 read surface， -不直接拼接各状态源。 +renderer，也不改变 delivery receipt。Board HUD 的两个 adapter 消费同一个 snapshot： +Claude 原生 `statusLine.command` 每次从官方 stdin 取得当前 `cwd + session_id`，展示 +focus 与健康状态两行；Codex 在 tmux 中启动时，`SessionStart` 创建固定三行只读 +sidecar，前两行相同，第三行展示 watcher 观察到的最近一次 snapshot 变化。HUD 不直接 +拼接各状态源，也不参与 prompt receipt。 投递游标按 session 存在 `.devloop/board/sessions/`。`PostCompact` 会让状态条目 在下一轮重放；已经消费的 event 不会因压缩再次触发。只有 UI channel 的条目不经过 @@ -47,8 +50,8 @@ Board 只投递当前工作所需信息。当前分支的 PR 会进入 branch token。每个条目独立去重，某个事实变化不会捎带未变化的整块上下文。 相关性属于 `BoardView`，投递节奏属于 `DeliveryPolicy`；事实生产者不选择 channel、 -prompt scope 或重放行为。Requirement 当前仅以兼容卡片加入 Board，其独立 provider -与领域抽象留到 Board UI 完成后再推进。 +prompt scope 或重放行为。Requirement 当前仅以兼容卡片加入 Board；其独立 provider +与领域抽象留到 requirement 域推进时处理，不以某个 UI adapter 是否完成为前置。 ### 投递状态不是业务事实 @@ -62,3 +65,17 @@ validation 等事实仍由原 owner 的 segment 提供。 branch、dirty、validation 等描述“现在在哪里”，compaction 后必须重放；review 结果和待办提醒描述“发生过什么 / 请做什么”，按身份限次投递，避免 agent 重复处理同一件事。 + +### HUD 是展示面，不是第二条状态总线 + +HUD 固定保留工作上下文与当前健康状态；失败或阻塞事实始终留在健康状态行，不能只作为 +会被覆盖的实时消息出现。Codex sidecar 的第三行由常驻 watcher 比较前后两帧 Board item +revision 得出，仅在进程内保留最新一条，不另建事件 ledger；HUD 重启后从 “watching +Board”重新开始。Claude 每次 status line 刷新都会启动新进程，因此只渲染两条稳定状态， +不为凑齐第三行额外持久化一套 UI 事件游标。 + +Claude 原生 status line 通过一次性 `/devloop:setup-board` 配置；安装器备份 settings， +发现其它 status line 时必须经用户确认才替换。tmux 只补 Codex 尚未开放给 plugin 的 +自定义展示位置：HUD pane 以 CLI session 与 leader pane 标识，重复 SessionStart 复用同一 +pane，leader 回到 shell 后自动退出。非 tmux Codex 会话安静降级，不影响 Board 的 prompt +投递；两端均可通过 `board.hud.enabled=false` 显式关闭。 diff --git a/devloop/domain/board/model.py b/devloop/domain/board/model.py index 88a13aa..499ed8e 100644 --- a/devloop/domain/board/model.py +++ b/devloop/domain/board/model.py @@ -173,9 +173,10 @@ def signature(self) -> str: return hashlib.sha1(raw.encode("utf-8")).hexdigest() def to_dict(self) -> dict[str, object]: - """Return the presentation-neutral shape consumed by the future Board UI.""" + """Return the presentation-neutral shape consumed by Board UI adapters.""" return { "id": self.id, + "revision": self.signature, "type": self.type.value, "kind": self.kind.value, "scope": _json_ready(self.scope), @@ -228,7 +229,7 @@ def to_dict(self) -> dict[str, object]: @dataclass(frozen=True) class Board: - """Shared structured facts for prompt delivery today and collaborative UI later.""" + """Shared structured facts for prompt delivery and UI projections.""" root: str items: tuple[BoardItem, ...] diff --git a/devloop/hooks/board_hud_start.py b/devloop/hooks/board_hud_start.py new file mode 100644 index 0000000..e9e1917 --- /dev/null +++ b/devloop/hooks/board_hud_start.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""SessionStart side effect: best-effort automatic Board HUD inside tmux.""" +from __future__ import annotations + +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from hooks import hook_io # noqa: E402 +from ui.board.tmux import ensure_hud_pane # noqa: E402 + + +def handle(inp: hook_io.HookInput) -> None: + ensure_hud_pane(inp.cwd, inp.session_id) + + +if __name__ == "__main__": + raise SystemExit(hook_io.observe(handle)) diff --git a/devloop/hooks/hooks.codex.json b/devloop/hooks/hooks.codex.json index c940054..9ffc4ad 100644 --- a/devloop/hooks/hooks.codex.json +++ b/devloop/hooks/hooks.codex.json @@ -34,6 +34,12 @@ "command": "\"${CLAUDE_PLUGIN_ROOT}/scripts/python\" \"${CLAUDE_PLUGIN_ROOT}/hooks/sessionstart_codex_init.py\"", "timeout": 10, "statusMessage": "Initializing devloop session" + }, + { + "type": "command", + "command": "\"${CLAUDE_PLUGIN_ROOT}/scripts/python\" \"${CLAUDE_PLUGIN_ROOT}/hooks/board_hud_start.py\"", + "timeout": 5, + "statusMessage": "Starting Board HUD" } ] } diff --git a/devloop/lib/config.py b/devloop/lib/config.py index 04c1d97..ba30e62 100644 --- a/devloop/lib/config.py +++ b/devloop/lib/config.py @@ -67,6 +67,9 @@ # load() 的距离分层覆盖——repo 的 .devloop/config.json 写一份 worktree.keep_recent 即覆盖全局。 # 任何规范入口建 worktree 时都消费它（worktree.create_or_reuse）。 "worktree": {"keep_recent": 5}, + # Board HUD is a read-only display surface. It starts automatically only when the + # current CLI already runs inside tmux; false is an explicit opt-out. + "board": {"hud": {"enabled": True}}, } _LOCAL_NAME = ".devloop" @@ -98,7 +101,7 @@ def plugin_root() -> Path: env_root = os.environ.get("PLUGIN_ROOT") or os.environ.get("CLAUDE_PLUGIN_ROOT") if env_root: return Path(env_root) - return Path(__file__).resolve().parent.parent.parent + return Path(__file__).resolve().parent.parent # ── read / write ───────────────────────────────────────────────────────────── @@ -201,6 +204,13 @@ def worktree(repo_dir: str | Path | None = None) -> dict: return load(repo_dir).get("worktree") or {} +def board_hud(repo_dir: str | Path | None = None) -> dict: + """Resolved Board HUD settings shared by native and sidecar UI adapters.""" + board = load(repo_dir).get("board") or {} + hud = board.get("hud") if isinstance(board, dict) else None + return hud if isinstance(hud, dict) else {} + + def notify(repo_dir: str | Path | None = None) -> dict: """已解析的 `notify` section。`notify should-arm` 决议器读 `channels`(bool):本会话是否已有 常驻 `channel all` 覆盖——true → should-arm 报 skip(交给 channel、不 arm waiter),默认 false diff --git a/devloop/scripts/board_hud.py b/devloop/scripts/board_hud.py new file mode 100644 index 0000000..5418583 --- /dev/null +++ b/devloop/scripts/board_hud.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Render Board state for native status lines and the tmux sidecar.""" +from __future__ import annotations + +import argparse +import json +import os +import shutil +import signal +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from domain.board import BoardRuntime # noqa: E402 +from lib import config # noqa: E402 +from ui.board.hud import ( # noqa: E402 + HudPulseTracker, + frame_from_snapshot, + render_frame, + render_statusline, +) +from ui.board.tmux import LEADER_ENV, pane_command # noqa: E402 + + +def _runtime(cwd: str, session_id: str) -> BoardRuntime | None: + return BoardRuntime.resolve(cwd, session_id) + + +def _snapshot(cwd: str, session_id: str) -> dict: + runtime = _runtime(cwd, session_id) + return runtime.snapshot() if runtime else {"root": cwd, "focus": None, "items": []} + + +def _watch_text(cwd: str, session_id: str, tracker: HudPulseTracker) -> str | None: + """Keep the last visible frame when a transient Board read is unavailable.""" + try: + snapshot = _snapshot(cwd, session_id) + frame = frame_from_snapshot(snapshot, tracker) + return render_frame(frame, shutil.get_terminal_size((120, 3)).columns, True) + except (OSError, ValueError): + return None + + +def _shell_commands(environ: dict[str, str] | None = None) -> set[str]: + environ = os.environ if environ is None else environ + commands = {"bash", "dash", "fish", "sh", "zsh"} + configured = Path(environ.get("SHELL", "")).name + if configured: + commands.add(configured) + return commands + + +def _args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Render devloop's Board HUD") + parser.add_argument("--watch", action="store_true") + parser.add_argument("--json", action="store_true") + parser.add_argument("--claude-statusline", action="store_true") + parser.add_argument("--cwd", default=str(Path.cwd())) + parser.add_argument("--session-id", default=os.environ.get("DEVLOOP_HUD_SESSION", "")) + parser.add_argument("--leader-pane", default=os.environ.get(LEADER_ENV, "")) + return parser.parse_args() + + +def main() -> int: + args = _args() + if args.claude_statusline: + try: + payload = json.loads(sys.stdin.read() or "{}") + except json.JSONDecodeError: + return 0 + workspace = payload.get("workspace") if isinstance(payload, dict) else None + cwd = ( + workspace.get("current_dir") + if isinstance(workspace, dict) and workspace.get("current_dir") + else payload.get("cwd") if isinstance(payload, dict) else None + ) or args.cwd + session_id = ( + str(payload.get("session_id") or args.session_id) + if isinstance(payload, dict) + else args.session_id + ) + if not config.board_hud(cwd).get("enabled", True): + return 0 + runtime = _runtime(cwd, session_id) + if runtime is None: + return 0 + columns = os.environ.get("COLUMNS", "") + width = int(columns) if columns.isdigit() else shutil.get_terminal_size((120, 2)).columns + frame = frame_from_snapshot(runtime.snapshot()) + print(render_statusline(frame, max(1, width - 4), not os.environ.get("NO_COLOR"))) + return 0 + if args.json: + print(json.dumps(_snapshot(args.cwd, args.session_id), indent=2, ensure_ascii=False)) + return 0 + if not args.watch: + frame = frame_from_snapshot(_snapshot(args.cwd, args.session_id)) + print(render_frame(frame, shutil.get_terminal_size((120, 3)).columns, sys.stdout.isatty())) + return 0 + + stopped = False + + def stop(_signum=None, _frame=None): + nonlocal stopped + stopped = True + + signal.signal(signal.SIGINT, stop) + signal.signal(signal.SIGTERM, stop) + tracker = HudPulseTracker() + inactive_leader_ticks = 0 + shell_commands = _shell_commands() + sys.stdout.write("\x1b[?25l\x1b[2J\x1b[H") + sys.stdout.flush() + try: + while not stopped: + if args.leader_pane: + leader_command = pane_command(args.leader_pane) + if leader_command is None: + break + inactive_leader_ticks = ( + inactive_leader_ticks + 1 + if Path(leader_command).name in shell_commands + else 0 + ) + if inactive_leader_ticks >= 3: + break + text = _watch_text(args.cwd, args.session_id, tracker) + if text is not None: + lines = "\n".join("\x1b[2K" + line for line in text.splitlines()) + sys.stdout.write("\x1b[H" + lines + "\x1b[J") + sys.stdout.flush() + time.sleep(1) + finally: + sys.stdout.write("\x1b[?25h") + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/devloop/scripts/setup_claude_board.py b/devloop/scripts/setup_claude_board.py new file mode 100644 index 0000000..5db4289 --- /dev/null +++ b/devloop/scripts/setup_claude_board.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python3 +"""Install devloop's command-backed Board status line into Claude settings.""" +from __future__ import annotations + +import argparse +import json +import os +import shlex +import shutil +import sys +import tempfile +from datetime import datetime +from pathlib import Path + + +LAUNCHER_NAME = "board_statusline.py" + + +def _claude_dir(value: str | None = None) -> Path: + return Path(value or os.environ.get("CLAUDE_CONFIG_DIR") or Path.home() / ".claude").expanduser() + + +def _load_settings(path: Path) -> dict: + if not path.exists() or not path.read_text(encoding="utf-8").strip(): + return {} + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ValueError(f"Claude settings must be a JSON object: {path}") + return value + + +def _atomic_json(path: Path, value: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2, ensure_ascii=False) + handle.write("\n") + os.replace(name, path) + except BaseException: + Path(name).unlink(missing_ok=True) + raise + + +def _launcher_source(claude_dir: Path, plugin_root: Path) -> str: + """A stable user-level launcher that follows marketplace version updates.""" + return f'''#!/usr/bin/env python3 +from __future__ import annotations + +import os +from pathlib import Path + +CLAUDE_DIR = Path({str(claude_dir)!r}) +FALLBACK_ROOT = Path({str(plugin_root)!r}) + + +def version_key(path: Path) -> tuple[int, ...]: + parts = path.name.split(".") + return tuple(int(part) for part in parts) if parts and all(part.isdigit() for part in parts) else () + + +def plugin_root() -> Path: + base = Path(os.environ.get("CLAUDE_CONFIG_DIR") or CLAUDE_DIR) + candidates = [ + path for path in base.glob("plugins/cache/*/devloop/*") + if version_key(path) and (path / "scripts/board_hud.py").is_file() + ] + if candidates: + return max(candidates, key=lambda path: (version_key(path), path.stat().st_mtime)) + return FALLBACK_ROOT + + +root = plugin_root() +launcher = root / "scripts/python" +script = root / "scripts/board_hud.py" +if launcher.is_file() and script.is_file(): + os.execv(str(launcher), [str(launcher), str(script), "--claude-statusline"]) +''' + + +def install(claude_dir: Path, plugin_root: Path, replace: bool = False) -> tuple[Path, Path | None]: + settings_path = claude_dir / "settings.json" + settings = _load_settings(settings_path) + existing = settings.get("statusLine") + existing_command = existing.get("command", "") if isinstance(existing, dict) else "" + + launcher_path = claude_dir / "plugins" / "devloop" / LAUNCHER_NAME + command = f"{shlex.quote(sys.executable)} {shlex.quote(str(launcher_path))}" + owned = LAUNCHER_NAME in existing_command + if existing_command and not owned and not replace: + raise FileExistsError( + "Claude already has a non-devloop statusLine; rerun with --replace only after user confirmation" + ) + + backup = None + if settings_path.exists(): + stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f") + backup = settings_path.with_name(f"{settings_path.name}.bak.{stamp}") + shutil.copy2(settings_path, backup) + + launcher_path.parent.mkdir(parents=True, exist_ok=True) + launcher_path.write_text(_launcher_source(claude_dir, plugin_root), encoding="utf-8") + launcher_path.chmod(0o700) + settings["statusLine"] = { + "type": "command", + "command": command, + "refreshInterval": 2, + } + _atomic_json(settings_path, settings) + return settings_path, backup + + +def _args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Configure Claude's native devloop Board status line") + parser.add_argument("--claude-config-dir") + parser.add_argument("--plugin-root", default=str(Path(__file__).resolve().parent.parent)) + parser.add_argument("--replace", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = _args() + try: + settings, backup = install( + _claude_dir(args.claude_config_dir), + Path(args.plugin_root).expanduser().resolve(), + args.replace, + ) + except FileExistsError as exc: + print(f"CONFLICT: {exc}", file=sys.stderr) + return 2 + except (OSError, ValueError) as exc: + print(f"ERROR: unable to configure Claude Board status line: {exc}", file=sys.stderr) + return 1 + print(f"Configured Claude Board status line in {settings}") + if backup: + print(f"Backup: {backup}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/devloop/tests/run_all.py b/devloop/tests/run_all.py index f14daea..30600f5 100644 --- a/devloop/tests/run_all.py +++ b/devloop/tests/run_all.py @@ -13,7 +13,19 @@ from _testkit import run_all # noqa: E402 (bootstrap first) -MODULES = ["test_cmdparse", "test_forge", "test_git_ops", "test_rebase", "test_notify", "test_board", "test_context", "test_guards", "test_lifecycle", "test_review"] +MODULES = [ + "test_cmdparse", + "test_forge", + "test_git_ops", + "test_rebase", + "test_notify", + "test_board", + "test_board_hud", + "test_context", + "test_guards", + "test_lifecycle", + "test_review", +] def main() -> int: diff --git a/devloop/tests/test_board.py b/devloop/tests/test_board.py index 24d633b..531d9f6 100644 --- a/devloop/tests/test_board.py +++ b/devloop/tests/test_board.py @@ -129,6 +129,7 @@ def test_board_snapshot_is_structured_and_does_not_consume_prompt_delivery(): assert snapshot["focus"]["repo_root"] == root identity = next(item for item in snapshot["items"] if item["type"] == "repo.identity") assert identity["kind"] == "state" + assert identity["revision"] assert identity["payload"]["branch"] == "feat/board" assert "text" not in identity diff --git a/devloop/tests/test_board_hud.py b/devloop/tests/test_board_hud.py new file mode 100644 index 0000000..3d7573c --- /dev/null +++ b/devloop/tests/test_board_hud.py @@ -0,0 +1,300 @@ +#!/usr/bin/env python3 +"""Board HUD fixed-line projection, pulse replacement, and tmux lifecycle.""" +from __future__ import annotations + +import json +import subprocess +import tempfile +from pathlib import Path + +from _testkit import _hook_input, _load_hook, _load_script, run_main # noqa: E402 + + +def _item(item_type: str, payload: dict, revision: str) -> dict: + return { + "id": f"/repo:{item_type}", + "revision": revision, + "type": item_type, + "kind": "state", + "scope": {"workspace_root": "/ws", "repo_root": "/repo"}, + "payload": payload, + } + + +def _snapshot(*items: dict) -> dict: + return { + "root": "/ws", + "focus": {"workspace_root": "/ws", "repo_root": "/repo"}, + "items": list(items), + } + + +def _identity(revision: str = "i1", **changes) -> dict: + payload = { + "code_dir": "/repo/devloop", + "branch": "feat/board-ui", + "ahead": 1, + "behind": 0, + "modified_count": 2, + "untracked_count": 0, + "pr_label": "PR #110", + "pr_state": "open", + } + payload.update(changes) + return _item("repo.identity", payload, revision) + + +def test_hud_renders_three_semantic_lines(): + from ui.board.hud import HudPulseTracker, frame_from_snapshot, render_frame + + snapshot = _snapshot( + _identity(), + _item( + "requirement.current", + {"source": "requirement.current", "text": "Requirement: Board UI"}, + "r1", + ), + _item( + "repo.validation", + {"components": [{"component": ".", "lint_at": 1, "test_at": 2}]}, + "v1", + ), + _item( + "repo.review", + {"status": "success", "reviewed_sha": "abc", "findings": 2, "failed_files": 0}, + "rv1", + ), + ) + rendered = render_frame( + frame_from_snapshot(snapshot, HudPulseTracker()), + width=120, + color=False, + ) + lines = rendered.splitlines() + assert len(lines) == 3 + assert "req:Board UI" in lines[0] and "repo:devloop" in lines[0] + assert "feat/board-ui" in lines[1] and "PR #110:open" in lines[1] + assert "validation:✓" in lines[1] and "review:2 findings" in lines[1] + assert lines[2] == "watching Board" + + +def test_hud_pulse_is_latest_change_and_critical_state_stays_stable(): + from ui.board.hud import HudPulseTracker, frame_from_snapshot, render_frame + + tracker = HudPulseTracker() + base = _snapshot( + _identity(modified_count=0), + _item("repo.validation", {"components": []}, "v1"), + ) + frame_from_snapshot(base, tracker, now=1) + + dirty = _snapshot( + _identity("i2", modified_count=3), + _item("repo.validation", {"components": []}, "v1"), + ) + first = render_frame(frame_from_snapshot(dirty, tracker, now=2), color=False) + assert "working tree changed · 3 files" in first.splitlines()[2] + + blocked = _snapshot( + _identity("i2", modified_count=3), + _item("repo.validation", {"components": []}, "v1"), + _item("repo.pr-blocked", {"label": "PR #110", "readiness": "ci_blocked"}, "b1"), + ) + second = render_frame(frame_from_snapshot(blocked, tracker, now=3), color=False) + lines = second.splitlines() + assert "BLOCKED:ci_blocked" in lines[1] + assert "merge blocked · ci_blocked" in lines[2] + + review = _snapshot( + _identity("i2", modified_count=3), + _item("repo.validation", {"components": []}, "v1"), + _item("repo.pr-blocked", {"label": "PR #110", "readiness": "ci_blocked"}, "b1"), + _item( + "repo.review", + {"status": "success", "reviewed_sha": "abc", "findings": 2, "failed_files": 0}, + "rv1", + ), + ) + third = render_frame(frame_from_snapshot(review, tracker, now=4), color=False) + lines = third.splitlines() + assert "BLOCKED:ci_blocked" in lines[1] + assert "review success · 2 findings" in lines[2] + + +def test_hud_never_wraps_and_sanitizes_dynamic_text(): + from ui.board.hud import frame_from_snapshot, render_frame + + snapshot = _snapshot( + _identity(branch="feat/very-long\x1b[31m-branch-name"), + _item( + "requirement.current", + {"text": "Requirement: 一个非常长的 requirement title that cannot fit"}, + "r1", + ), + ) + lines = render_frame(frame_from_snapshot(snapshot), width=24, color=False).splitlines() + assert len(lines) == 3 + assert all(len(line) <= 24 for line in lines) + assert all("\x1b" not in line for line in lines) + + +def test_native_statusline_keeps_stable_slots_and_never_drops_blocker(): + from ui.board.hud import frame_from_snapshot, render_statusline + + snapshot = _snapshot( + _identity(modified_count=2), + _item("repo.pr-blocked", {"readiness": "ci_blocked"}, "b1"), + _item("repo.validation", {"components": []}, "v1"), + ) + lines = render_statusline(frame_from_snapshot(snapshot), width=30, color=False).splitlines() + assert len(lines) == 2 + assert "BLOCKED:ci_blocked" in lines[1] + assert "feat/board-ui" not in lines[1] + + +def test_watch_frame_survives_transient_board_read_error(): + script = _load_script("board_hud") + script._snapshot = lambda *_: (_ for _ in ()).throw(ValueError("partial state")) + + assert script._watch_text("/repo", "session-a", script.HudPulseTracker()) is None + + +def test_watch_recognizes_the_users_configured_shell(): + script = _load_script("board_hud") + assert "nu" in script._shell_commands({"SHELL": "/opt/homebrew/bin/nu"}) + + +class _FakeTmux: + def __init__(self, panes: str = ""): + self.panes = panes + self.calls: list[list[str]] = [] + + def __call__(self, args: list[str]) -> subprocess.CompletedProcess[str]: + self.calls.append(args) + if args[0] == "display-message": + return subprocess.CompletedProcess(args, 0, "80\n", "") + if args[0] == "list-panes": + return subprocess.CompletedProcess(args, 0, self.panes, "") + if args[0] == "split-window": + return subprocess.CompletedProcess(args, 0, "%9\n", "") + return subprocess.CompletedProcess(args, 0, "", "") + + +def test_tmux_hud_creates_then_reuses_owned_three_line_pane(): + from lib import config + from ui.board import tmux + + assert (config.plugin_root() / "scripts/board_hud.py").is_file() + env = {"TMUX": "/tmp/tmux", "TMUX_PANE": "%1"} + + def installed(_: str) -> str: + return "/usr/bin/tmux" + + fake = _FakeTmux() + assert tmux.ensure_hud_pane( + "/repo", "session-a", env=env, run_tmux=fake, find_executable=installed, + ) == "created" + split = next(call for call in fake.calls if call[0] == "split-window") + assert split[split.index("-l") + 1] == "3" + command = split[-1] + assert "DEVLOOP_HUD_OWNER=1" in command + assert "DEVLOOP_HUD_SESSION=session-a" in command + assert "--leader-pane %1" in command + + existing = ( + "%8\texec env DEVLOOP_HUD_OWNER=1 DEVLOOP_HUD_SESSION=session-a " + "DEVLOOP_HUD_LEADER_PANE=%1 python board_hud.py --watch\n" + ) + reused = _FakeTmux(existing) + assert tmux.ensure_hud_pane( + "/repo", "session-a", env=env, run_tmux=reused, find_executable=installed, + ) == "reused" + assert any(call[:3] == ["resize-pane", "-t", "%8"] for call in reused.calls) + assert not any(call[0] == "split-window" for call in reused.calls) + + stale = _FakeTmux( + "%7\texec env DEVLOOP_HUD_OWNER=1 DEVLOOP_HUD_SESSION=old-session " + "DEVLOOP_HUD_LEADER_PANE=%1 python board_hud.py --watch\n" + ) + assert tmux.ensure_hud_pane( + "/repo", "session-a", env=env, run_tmux=stale, find_executable=installed, + ) == "created" + assert ["kill-pane", "-t", "%7"] in stale.calls + + +def test_tmux_hud_is_disabled_when_tmux_is_not_installed(): + from ui.board import tmux + + fake = _FakeTmux() + result = tmux.ensure_hud_pane( + "/repo", + "session-a", + env={"TMUX": "/tmp/tmux", "TMUX_PANE": "%1"}, + run_tmux=fake, + find_executable=lambda _: None, + ) + assert result == "skipped_tmux_unavailable" + assert fake.calls == [] + + +def test_sessionstart_hud_hook_is_best_effort_observer(): + hook = _load_hook("board_hud_start") + seen = [] + original = hook.ensure_hud_pane + try: + hook.ensure_hud_pane = lambda cwd, session_id: seen.append((cwd, session_id)) + assert hook.handle(_hook_input("", {"cwd": "/ws", "session_id": "s1"})) is None + finally: + hook.ensure_hud_pane = original + assert seen == [("/ws", "s1")] + + +def test_claude_uses_native_statusline_while_codex_keeps_sidecar_hook(): + root = Path(__file__).resolve().parent.parent + claude = json.loads((root / "hooks/hooks.json").read_text(encoding="utf-8")) + codex = json.loads((root / "hooks/hooks.codex.json").read_text(encoding="utf-8")) + + def commands(value: dict) -> list[str]: + return [ + hook["command"] + for groups in value["hooks"].values() + for group in groups + for hook in group.get("hooks", []) + ] + + assert not any("board_hud_start.py" in command for command in commands(claude)) + assert any("board_hud_start.py" in command for command in commands(codex)) + + +def test_claude_statusline_setup_preserves_settings_and_refuses_foreign_owner(): + setup = _load_script("setup_claude_board") + plugin_root = Path(__file__).resolve().parent.parent + with tempfile.TemporaryDirectory() as tmp: + claude_dir = Path(tmp) / ".claude" + claude_dir.mkdir() + settings_path = claude_dir / "settings.json" + settings_path.write_text(json.dumps({"theme": "dark"}), encoding="utf-8") + + saved, backup = setup.install(claude_dir, plugin_root) + value = json.loads(saved.read_text(encoding="utf-8")) + assert value["theme"] == "dark" + assert value["statusLine"]["type"] == "command" + assert value["statusLine"]["refreshInterval"] == 2 + assert setup.LAUNCHER_NAME in value["statusLine"]["command"] + assert backup is not None and backup.is_file() + launcher = claude_dir / "plugins/devloop" / setup.LAUNCHER_NAME + assert "plugins/cache/*/devloop/*" in launcher.read_text(encoding="utf-8") + + foreign = {"theme": "dark", "statusLine": {"type": "command", "command": "other-hud"}} + settings_path.write_text(json.dumps(foreign), encoding="utf-8") + try: + setup.install(claude_dir, plugin_root) + except FileExistsError: + pass + else: + raise AssertionError("foreign status line must require explicit replacement") + assert json.loads(settings_path.read_text(encoding="utf-8")) == foreign + + +if __name__ == "__main__": + run_main(globals()) diff --git a/devloop/ui/__init__.py b/devloop/ui/__init__.py new file mode 100644 index 0000000..c31aa4d --- /dev/null +++ b/devloop/ui/__init__.py @@ -0,0 +1 @@ +"""Human-facing adapters over devloop's domain read models.""" diff --git a/devloop/ui/board/__init__.py b/devloop/ui/board/__init__.py new file mode 100644 index 0000000..c87da58 --- /dev/null +++ b/devloop/ui/board/__init__.py @@ -0,0 +1 @@ +"""Board terminal UI adapters.""" diff --git a/devloop/ui/board/hud.py b/devloop/ui/board/hud.py new file mode 100644 index 0000000..d58eb3b --- /dev/null +++ b/devloop/ui/board/hud.py @@ -0,0 +1,340 @@ +"""Fixed three-line terminal projection over a presentation-neutral Board snapshot.""" +from __future__ import annotations + +import re +import unicodedata +from dataclasses import dataclass +from datetime import datetime +from enum import Enum +from pathlib import Path +from typing import Iterable + + +class HudTone(str, Enum): + NORMAL = "normal" + DIM = "dim" + TITLE = "title" + INFO = "info" + SUCCESS = "success" + WARNING = "warning" + ERROR = "error" + + +@dataclass(frozen=True) +class HudSegment: + text: str + tone: HudTone = HudTone.NORMAL + priority: int = 5 + + +@dataclass(frozen=True) +class HudPulse: + text: str + tone: HudTone = HudTone.DIM + occurred_at: float | None = None + + +@dataclass(frozen=True) +class HudFrame: + """Three semantic slots: context, current health, and latest transient pulse.""" + + context: tuple[HudSegment, ...] + health: tuple[HudSegment, ...] + pulse: HudPulse + + +_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f-\x9f]") +_ANSI_SGR_RE = re.compile(r"\x1b\[[0-9;]*m") +_ANSI = { + HudTone.DIM: "\x1b[2m", + HudTone.TITLE: "\x1b[1m", + HudTone.INFO: "\x1b[36m", + HudTone.SUCCESS: "\x1b[32m", + HudTone.WARNING: "\x1b[33m", + HudTone.ERROR: "\x1b[31m", +} +_RESET = "\x1b[0m" + + +def _clean(value: object) -> str: + return _CONTROL_RE.sub("", _ANSI_SGR_RE.sub("", str(value or ""))).strip() + + +def _items(snapshot: dict) -> dict[str, dict]: + rows = snapshot.get("items") if isinstance(snapshot, dict) else None + if not isinstance(rows, list): + return {} + return { + str(item.get("type")): item + for item in rows + if isinstance(item, dict) and item.get("type") + } + + +def _payload(item: dict | None) -> dict: + payload = item.get("payload") if isinstance(item, dict) else None + return payload if isinstance(payload, dict) else {} + + +def _context_segments(snapshot: dict, items: dict[str, dict]) -> tuple[HudSegment, ...]: + out = [HudSegment("[Board]", HudTone.TITLE, 0)] + workspace = _payload(items.get("workspace")) + workspace_root = _clean(workspace.get("root") or snapshot.get("root")) + if workspace_root: + out.append(HudSegment(f"ws:{Path(workspace_root).name}", HudTone.DIM, 3)) + requirement = _clean(_payload(items.get("requirement.current")).get("text")) + if requirement: + requirement = re.sub(r"^Requirement:\s*", "", requirement) + out.append(HudSegment(f"req:{requirement}", HudTone.NORMAL, 1)) + + identity = _payload(items.get("repo.identity")) + code_dir = _clean(identity.get("code_dir")) + if code_dir: + out.append(HudSegment(f"repo:{Path(code_dir).name}", HudTone.INFO, 0)) + elif not workspace_root: + root = _clean(snapshot.get("root")) + out.append(HudSegment(f"workspace:{Path(root).name}" if root else "no workspace", HudTone.DIM, 2)) + return tuple(out) + + +def _validation_segment(item: dict | None) -> HudSegment: + components = _payload(item).get("components") + if not isinstance(components, list) or not components: + return HudSegment("validation:never", HudTone.DIM, 2) + lint_ok = all(isinstance(row, dict) and row.get("lint_at") for row in components) + test_ok = all(isinstance(row, dict) and row.get("test_at") for row in components) + if lint_ok and test_ok: + return HudSegment("validation:✓", HudTone.SUCCESS, 1) + parts = [] + if lint_ok: + parts.append("lint✓") + if test_ok: + parts.append("test✓") + return HudSegment("validation:" + ("/".join(parts) or "pending"), HudTone.WARNING, 1) + + +def _health_segments(items: dict[str, dict]) -> tuple[HudSegment, ...]: + out: list[HudSegment] = [] + identity = _payload(items.get("repo.identity")) + if not identity: + subprojects = _payload(items.get("workspace")).get("subprojects") + count = len(subprojects) if isinstance(subprojects, list) else 0 + return ( + HudSegment("focus:workspace", HudTone.DIM, 0), + HudSegment(f"repos:{count}", HudTone.NORMAL, 1), + ) + branch = _clean(identity.get("branch")) + if branch: + out.append(HudSegment(branch, HudTone.INFO, 0)) + modified = int(identity.get("modified_count") or 0) + untracked = int(identity.get("untracked_count") or 0) + dirty = modified + untracked + out.append(HudSegment(f"dirty:{dirty}" if dirty else "clean", HudTone.WARNING if dirty else HudTone.SUCCESS, 2)) + ahead = int(identity.get("ahead") or 0) + behind = int(identity.get("behind") or 0) + if ahead or behind: + out.append(HudSegment(f"↑{ahead} ↓{behind}", HudTone.DIM, 4)) + pr_label = _clean(identity.get("pr_label")) + pr_state = _clean(identity.get("pr_state")) + if pr_label: + out.append(HudSegment(f"{pr_label}:{pr_state or '?'}", HudTone.NORMAL, 1)) + + blocked = _payload(items.get("repo.pr-blocked")) + if blocked: + # A narrow status line may drop ordinary context, but never the fact that + # the current PR is blocked (docs/board.md: stable health beats decoration). + out.append(HudSegment(f"BLOCKED:{_clean(blocked.get('readiness'))}", HudTone.ERROR, -1)) + out.append(_validation_segment(items.get("repo.validation"))) + + review = _payload(items.get("repo.review")) + if review: + status = _clean(review.get("status")) or "?" + findings = int(review.get("findings") or 0) + if status in {"error", "stale"}: + tone = HudTone.ERROR + elif status == "running": + tone = HudTone.INFO + elif findings: + tone = HudTone.WARNING + else: + tone = HudTone.SUCCESS + detail = f"{findings} findings" if findings else status + out.append(HudSegment(f"review:{detail}", tone, 1)) + pending = int(_payload(items.get("repo.review-label")).get("pending") or 0) + if pending: + out.append(HudSegment(f"labels:{pending}", HudTone.WARNING, 3)) + return tuple(out) + + +def _revision(item: dict) -> str: + revision = item.get("revision") + return str(revision) if revision else repr(item.get("payload")) + + +class HudPulseTracker: + """Remember the latest visible change in-process; newer Board revisions replace it.""" + + def __init__(self): + self._items: dict[str, dict] | None = None + self.current = HudPulse("watching Board", HudTone.DIM) + + def observe(self, items: dict[str, dict], now: float | None = None) -> HudPulse: + if self._items is None: + self._items = items + return self.current + + changed: list[tuple[int, HudPulse]] = [] + for item_type in set(self._items) | set(items): + before = self._items.get(item_type) + after = items.get(item_type) + if before is not None and after is not None and _revision(before) == _revision(after): + continue + pulse = _pulse_for_change(item_type, before, after, now) + if pulse: + changed.append((_pulse_priority(item_type), pulse)) + self._items = items + if changed: + self.current = min(changed, key=lambda row: row[0])[1] + return self.current + + +def _pulse_priority(item_type: str) -> int: + return { + "repo.pr-blocked": 0, + "repo.review": 1, + "repo.validation": 2, + "repo.identity": 3, + "requirement.current": 4, + "repo.review-label": 5, + }.get(item_type, 9) + + +def _pulse_for_change( + item_type: str, + before: dict | None, + after: dict | None, + now: float | None, +) -> HudPulse | None: + old, new = _payload(before), _payload(after) + at = datetime.now().timestamp() if now is None else now + if item_type == "repo.pr-blocked": + return HudPulse( + f"merge blocked · {_clean(new.get('readiness'))}" if after else "merge blocker cleared", + HudTone.ERROR if after else HudTone.SUCCESS, + at, + ) + if item_type == "repo.review": + if not after: + return HudPulse("review state cleared", HudTone.DIM, at) + findings = int(new.get("findings") or 0) + status = _clean(new.get("status")) or "updated" + text = f"review {status} · {findings} findings" if findings else f"review {status}" + tone = HudTone.ERROR if status in {"error", "stale"} else HudTone.WARNING if findings else HudTone.SUCCESS + return HudPulse(text, tone, at) + if item_type == "repo.validation": + segment = _validation_segment(after) + return HudPulse(segment.text.replace(":", " ", 1), segment.tone, at) + if item_type == "repo.identity": + old_branch, new_branch = _clean(old.get("branch")), _clean(new.get("branch")) + if old_branch != new_branch: + return HudPulse(f"switched {old_branch or '?'} → {new_branch or '?'}", HudTone.INFO, at) + old_dirty = int(old.get("modified_count") or 0) + int(old.get("untracked_count") or 0) + new_dirty = int(new.get("modified_count") or 0) + int(new.get("untracked_count") or 0) + if old_dirty != new_dirty: + return HudPulse(f"working tree changed · {new_dirty} files", HudTone.WARNING if new_dirty else HudTone.SUCCESS, at) + if old.get("pr_state") != new.get("pr_state"): + return HudPulse(f"{_clean(new.get('pr_label'))} became {_clean(new.get('pr_state'))}", HudTone.INFO, at) + return HudPulse("repository state updated", HudTone.DIM, at) + if item_type == "requirement.current": + if not after: + return HudPulse("requirement closed", HudTone.SUCCESS, at) + text = re.sub(r"^Requirement:\s*", "", _clean(new.get("text"))) + return HudPulse(f"requirement focused · {text}", HudTone.INFO, at) + if item_type == "repo.review-label": + return HudPulse(f"review labels pending · {int(new.get('pending') or 0)}", HudTone.WARNING, at) + return None + + +def frame_from_snapshot( + snapshot: dict, + tracker: HudPulseTracker | None = None, + now: float | None = None, +) -> HudFrame: + items = _items(snapshot) + pulse = tracker.observe(items, now) if tracker else HudPulse("watching Board", HudTone.DIM) + return HudFrame( + context=_context_segments(snapshot, items), + health=_health_segments(items), + pulse=pulse, + ) + + +def _display_width(text: str) -> int: + return sum( + 0 if unicodedata.combining(char) else 2 if unicodedata.east_asian_width(char) in {"F", "W"} else 1 + for char in text + ) + + +def _truncate(text: str, width: int) -> str: + if _display_width(text) <= width: + return text + if width <= 1: + return "…"[:width] + out = "" + for char in text: + if _display_width(out + char) > width - 1: + break + out += char + return out.rstrip() + "…" + + +def _fit_segments(segments: Iterable[HudSegment], width: int) -> list[HudSegment]: + selected = [segment for segment in segments if _clean(segment.text)] + width = max(1, width) + while len(selected) > 1 and _display_width(" · ".join(_clean(item.text) for item in selected)) > width: + worst = max(range(len(selected)), key=lambda index: (selected[index].priority, index)) + selected.pop(worst) + if not selected: + return [HudSegment("")] + plain = " · ".join(_clean(item.text) for item in selected) + if _display_width(plain) <= width: + return selected + first = selected[0] + return [HudSegment(_truncate(_clean(first.text), width), first.tone, first.priority)] + + +def _paint(segment: HudSegment, color: bool) -> str: + text = _clean(segment.text) + code = _ANSI.get(segment.tone) if color else None + return f"{code}{text}{_RESET}" if code else text + + +def _render_segments(segments: Iterable[HudSegment], width: int, color: bool) -> str: + return " · ".join(_paint(item, color) for item in _fit_segments(segments, width)) + + +def render_frame(frame: HudFrame, width: int = 120, color: bool = True) -> str: + """Render exactly three lines; semantic slots never wrap into one another.""" + pulse_prefix = "" + if frame.pulse.occurred_at is not None: + pulse_prefix = datetime.fromtimestamp(frame.pulse.occurred_at).strftime("%H:%M:%S ") + pulse = HudSegment(pulse_prefix + frame.pulse.text, frame.pulse.tone, 0) + return "\n".join(( + _render_segments(frame.context, width, color), + _render_segments(frame.health, width, color), + _render_segments((pulse,), width, color), + )) + + +def render_statusline(frame: HudFrame, width: int = 120, color: bool = True) -> str: + """Render stable Board state for a command-backed status line. + + Claude starts a fresh process for each refresh, so the tmux watcher's + in-memory pulse has no honest equivalent here. Keep the native surface to + the two durable slots instead of persisting a second event ledger. + """ + return "\n".join(( + _render_segments(frame.context, width, color), + _render_segments(frame.health, width, color), + )) diff --git a/devloop/ui/board/tmux.py b/devloop/ui/board/tmux.py new file mode 100644 index 0000000..b8872a3 --- /dev/null +++ b/devloop/ui/board/tmux.py @@ -0,0 +1,152 @@ +"""Small tmux lifecycle adapter for the automatic Board HUD sidecar.""" +from __future__ import annotations + +import os +import re +import shlex +import shutil +import subprocess +from pathlib import Path +from typing import Callable + +from lib import config + + +HUD_HEIGHT = 3 +MIN_WINDOW_HEIGHT = 45 +OWNER_ENV = "DEVLOOP_HUD_OWNER" +SESSION_ENV = "DEVLOOP_HUD_SESSION" +LEADER_ENV = "DEVLOOP_HUD_LEADER_PANE" + +RunTmux = Callable[[list[str]], subprocess.CompletedProcess[str]] +FindExecutable = Callable[[str], str | None] + + +def _default_run(args: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["tmux", *args], + capture_output=True, + text=True, + timeout=1, + check=False, + ) + + +def _pane_id(value: str | None) -> str | None: + value = (value or "").strip() + return value if re.fullmatch(r"%\d+", value) else None + + +def _watch_command(plugin_root: Path, cwd: str, session_id: str, leader: str) -> str: + launcher = plugin_root / "scripts" / "python" + script = plugin_root / "scripts" / "board_hud.py" + env = { + OWNER_ENV: "1", + SESSION_ENV: session_id, + LEADER_ENV: leader, + } + assignments = " ".join(f"{key}={shlex.quote(value)}" for key, value in env.items()) + argv = " ".join(shlex.quote(str(value)) for value in ( + launcher, + script, + "--watch", + "--cwd", + cwd, + "--session-id", + session_id, + "--leader-pane", + leader, + )) + return f"exec env {assignments} {argv}" + + +def _owned_panes(output: str, session_id: str | None, leader: str) -> list[str]: + wanted = [f"{OWNER_ENV}=1", f"{LEADER_ENV}={shlex.quote(leader)}"] + if session_id is not None: + wanted.append(f"{SESSION_ENV}={shlex.quote(session_id)}") + found: list[str] = [] + for line in output.splitlines(): + pane, _, command = line.partition("\t") + pane = _pane_id(pane) or "" + if pane and all(marker in command for marker in wanted): + found.append(pane) + return found + + +def ensure_hud_pane( + cwd: str, + session_id: str, + *, + env: dict[str, str] | None = None, + run_tmux: RunTmux = _default_run, + find_executable: FindExecutable = shutil.which, +) -> str: + """Ensure one three-line HUD for this CLI pane; all failures degrade to a no-op.""" + env = dict(os.environ if env is None else env) + if find_executable("tmux") is None: + return "skipped_tmux_unavailable" + leader = _pane_id(env.get("TMUX_PANE")) + if not env.get("TMUX") or not leader: + return "skipped_not_tmux" + if not config.board_hud(cwd).get("enabled", True): + return "skipped_disabled" + + try: + height_result = run_tmux(["display-message", "-p", "-t", leader, "#{window_height}"]) + try: + window_height = int((height_result.stdout or "").strip()) + except ValueError: + window_height = 0 + if 0 < window_height < MIN_WINDOW_HEIGHT: + return "skipped_window_too_small" + + panes_result = run_tmux([ + "list-panes", "-t", leader, "-F", "#{pane_id}\t#{pane_start_command}", + ]) + pane_output = panes_result.stdout or "" + panes = _owned_panes(pane_output, session_id, leader) + if panes: + keeper, *duplicates = panes + run_tmux(["resize-pane", "-t", keeper, "-y", str(HUD_HEIGHT)]) + for pane in duplicates: + run_tmux(["kill-pane", "-t", pane]) + return "reused" + + # A new CLI session may reuse the same shell/tmux pane after the previous CLI exits. + # Replace its stale HUD before creating this session's owner, rather than stacking panes. + for pane in _owned_panes(pane_output, None, leader): + run_tmux(["kill-pane", "-t", pane]) + + root = config.plugin_root() + command = _watch_command(root, cwd, session_id, leader) + created = run_tmux([ + "split-window", "-v", "-l", str(HUD_HEIGHT), "-d", "-P", "-F", "#{pane_id}", + "-t", leader, "-c", cwd, command, + ]) + return "created" if _pane_id(created.stdout) else "failed" + except (OSError, subprocess.SubprocessError): + return "failed" + + +def pane_exists(pane_id: str, run_tmux: RunTmux = _default_run) -> bool: + pane = _pane_id(pane_id) + if not pane: + return False + try: + result = run_tmux(["display-message", "-p", "-t", pane, "#{pane_id}"]) + except (OSError, subprocess.SubprocessError): + return False + return result.returncode == 0 and _pane_id(result.stdout) == pane + + +def pane_command(pane_id: str, run_tmux: RunTmux = _default_run) -> str | None: + """Foreground command for a pane, or None once the leader can no longer be observed.""" + pane = _pane_id(pane_id) + if not pane: + return None + try: + result = run_tmux(["display-message", "-p", "-t", pane, "#{pane_current_command}"]) + except (OSError, subprocess.SubprocessError): + return None + command = (result.stdout or "").strip() + return command if result.returncode == 0 and command else None