Merge branch 'tool-windows'

Add tool windows: tool registry, styled aux tmux windows
(TmuxRunner.new_aux_window / TmuxManager.open_aux_window),
SessionService tool-window + clone-for-window methods, and
hqt tool / hqt palette CLI subcommands (fzf tool launcher).

# Conflicts:
#	src/hqt/cli.py
#	src/hqt/tmux/manager.py
#	tests/test_cli.py
#	tests/test_sessions.py
This commit is contained in:
2026-06-11 09:43:37 -04:00
9 changed files with 639 additions and 7 deletions
+109
View File
@@ -1,5 +1,6 @@
import logging
import os
import shlex
import shutil
import subprocess
import sys
@@ -128,6 +129,114 @@ def doctor():
click.echo("bubblewrap: ✗ not found — sandboxed sessions unavailable")
def _build_session_service():
"""Wire a SessionService the same way HqtApp.on_mount does (for CLI use).
Mirrors on_mount's DB + tmux wiring, including seeding harness rows, but
deliberately omits apply_theme — a one-off CLI call must not re-theme the
session's status bar.
"""
from hqt.config import get_settings
from hqt.db.engine import ensure_db, get_engine, get_session_factory
from hqt.harnesses.registry import discover_harnesses, ensure_harnesses_in_db
from hqt.sessions.service import SessionService
from hqt.tmux.manager import TmuxManager
from hqt.tmux.runner import TmuxRunner
settings = get_settings()
ensure_db(settings)
factory = get_session_factory(get_engine(settings))
with factory() as db:
ensure_harnesses_in_db(db)
runner = TmuxRunner(settings.tmux_path, settings.tui_session_name)
return SessionService(factory, TmuxManager(runner), discover_harnesses())
@main.command(name="tool")
@click.argument("tool")
@click.argument("window")
def tool_cmd(tool, window):
"""Run TOOL for the session in tmux WINDOW.
TOOL is nvim/lazygit/shell (opens a styled tool window) or "clone" (a fresh
harness with the same project + model). WINDOW is the tmux window name (the
hqt-<id> key, e.g. from #{window_name}).
"""
import asyncio
from hqt.errors import ServiceError
svc = _build_session_service()
try:
if tool == "clone":
asyncio.run(svc.clone_session_for_window(window))
else:
asyncio.run(svc.open_tool_window_for_window(window, tool))
except ServiceError as err:
click.echo(str(err), err=True)
raise SystemExit(1)
PALETTE_ENTRIES = ["nvim", "lazygit", "shell", "clone"]
# fzf colors matching the Catppuccin Frappé status bar / the Alt+o switcher.
_PALETTE_FZF_COLORS = (
"bg:#292c3c,bg+:#414559,fg:#c6d0f5,fg+:#c6d0f5,hl:#ef9f76,hl+:#ef9f76,"
"pointer:#ef9f76,prompt:#8caaee,info:#838ba7,border:#838ba7"
)
def _palette_popup_command(window: str) -> str:
"""Shell pipeline for the fzf popup; `window` is baked in as a literal.
display-popup does NOT format-expand its command, so the window name must be
concrete here (run-shell already expanded #{window_name} before `hqt palette`
ran). The selected entry runs `hqt tool <choice> <window>`.
"""
entries = "\\n".join(PALETTE_ENTRIES) + "\\n"
return (
f"printf '{entries}' | "
f"fzf --reverse --no-info --prompt='tool ' --pointer='' "
f"--color='{_PALETTE_FZF_COLORS}' | "
f"xargs -r -I{{}} hqt tool {{}} {shlex.quote(window)}"
)
@main.command(name="palette")
@click.argument("window")
def palette_cmd(window):
"""Pop an fzf tool palette for the session in tmux WINDOW (bound to M-p)."""
from hqt.config import get_settings
settings = get_settings()
svc = _build_session_service()
if svc.session_id_for_window(window) is None:
subprocess.run(
[
settings.tmux_path,
"display-message",
"hqt: open the tool palette from a harness window",
]
)
return
subprocess.run(
[
settings.tmux_path,
"display-popup",
"-E",
"-w",
"40%",
"-h",
"30%",
"-T",
" open tool ",
"-S",
"fg=#838ba7",
_palette_popup_command(window),
]
)
@main.command(name="list")
def list_cmd():
"""List projects and sessions."""
+81
View File
@@ -2,6 +2,7 @@ import asyncio
import contextlib
import json
import logging
import shutil
import time
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
@@ -19,6 +20,7 @@ from hqt.git.worktree import WorktreeState
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.sandbox import Bind, SandboxPolicy
from hqt.status import window_label
from hqt.tools import TOOLS
from hqt.tmux.manager import SpawnRequest, TmuxManager
from hqt.tmux.runner import WindowInfo
@@ -489,6 +491,85 @@ class SessionService:
)
return True, ""
def session_id_for_window(self, window_name: str) -> int | None:
"""Resolve a tmux window name to its active hqt session id, or None.
None means the window is not an hqt session window (a tool window, the
TUI home window, or an unrelated tmux window).
"""
with self.factory() as db:
sess = (
db.query(Session)
.filter_by(tmux_session_name=window_name, archived=False)
.first()
)
return sess.id if sess else None
async def open_tool_window(self, session_id: int, tool: str) -> str | None:
"""Open a tool (nvim/lazygit/shell) in a new styled window for a session.
Spawns at the next free index in the session's project directory and
switches to it. The window is NOT a tracked session. Raises ServiceError
for an unknown tool, a missing binary, or a missing session/project.
Returns the new window_id, or None if the tmux spawn fails.
"""
log.info("Opening tool window: session=%s tool=%s", session_id, tool)
spec = TOOLS.get(tool)
if spec is None:
raise ServiceError(f"Unknown tool '{tool}'")
if spec.command and shutil.which(spec.command[0]) is None:
raise ServiceError(f"{spec.command[0]} not found on PATH")
with self.factory() as db:
sess = db.get(Session, session_id)
if sess is None:
raise ServiceError("Session not found")
project = db.get(Project, sess.project_id)
if project is None:
raise ServiceError(f"Project {sess.project_id} no longer exists")
cwd = project.path
label = f"{spec.label} · {project.name}"
return await self.tmux.open_aux_window(spec.label, cwd, spec.command, label)
async def open_tool_window_for_window(
self, window_name: str, tool: str
) -> str | None:
"""Open a tool window for the session identified by its tmux window name.
Used by the `hqt tool` CLI (the tmux binding passes #{window_name}).
Raises ServiceError if the name is not an active hqt session window.
"""
session_id = self.session_id_for_window(window_name)
if session_id is None:
raise ServiceError(f"{window_name!r} is not an hqt session window")
return await self.open_tool_window(session_id, tool)
async def clone_session_for_window(
self, window_name: str
) -> "CreateSessionResult":
"""Open a fresh harness session cloning the one in `window_name`.
Same project, harness, and model as the source session, but a brand-new
conversation (a new hqt-<id> window at the next index). Raises
ServiceError if `window_name` is not an active hqt session window — so
invoking clone from a tool window (nvim/shell) or the TUI is a clean
no-op.
"""
with self.factory() as db:
sess = (
db.query(Session)
.options(selectinload(Session.harness))
.filter_by(tmux_session_name=window_name, archived=False)
.first()
)
if sess is None:
raise ServiceError(f"{window_name!r} is not an hqt session window")
project_id = sess.project_id
harness_name = sess.harness.name
model = sess.model
return await self.create_session(
project_id, harness_name, nickname=None, model=model
)
async def stop_session(self, session_id: int) -> None:
with self.factory() as db:
sess = db.get(Session, session_id)
+9
View File
@@ -128,3 +128,12 @@ class TmuxManager:
async def reset_window_indexes(self, home_window: str) -> bool:
"""Compact tmux window indexes while keeping the home window at 0."""
return await self.runner.reset_window_indexes(home_window)
async def open_aux_window(
self, name: str, cwd: str, command: list[str], label: str
) -> str | None:
"""Open a styled tool window (non-session) and switch to it.
Returns the new window_id, or None on failure.
"""
return await self.runner.new_aux_window(name, cwd, command, label)
+72
View File
@@ -1,5 +1,6 @@
import asyncio
import logging
import shlex
from dataclasses import dataclass
log = logging.getLogger(__name__)
@@ -422,6 +423,77 @@ class TmuxRunner:
return window_id
async def new_aux_window(
self, name: str, cwd: str, command: list[str], label: str
) -> str | None:
"""Create an auxiliary (non-session) tool window and switch to it.
Appends at the next free index, then styles by window_id — not name — so
duplicate names (tool windows are spawned fresh every time) stay
unambiguous. Deliberately does NOT set remain-on-exit: the window closes
when the tool exits (a quit shell closes its window too). The window is
never tracked by hqt; it lives purely as a tmux window.
Returns the window_id, or None on failure (the half-created window is
cleaned up).
"""
idx = await self._next_window_index()
args = [
"new-window",
"-t",
f"{self.session_name}:{idx}",
"-n",
name,
"-c",
cwd,
"-P",
"-F",
"#{window_id}",
]
if command:
args.append(shlex.join(command))
rc, stdout, err = await self._exec(*args)
if rc != 0:
log.error("new-window (aux) failed: %s", err)
return None
window_id = stdout.strip()
# Style by window_id: automatic-rename + allow-rename off (nvim/lazygit
# emit OSC title sequences that would otherwise scramble the label), the
# @hqt_label, then the Frappé per-window theme — one atomic invocation
# (";" argv separators). Mirrors new_window's window-option setup.
rc, _, err = await self._exec(
"set-option",
"-w",
"-t",
window_id,
"automatic-rename",
"off",
";",
"set-option",
"-w",
"-t",
window_id,
"allow-rename",
"off",
";",
"set-option",
"-w",
"-t",
window_id,
"@hqt_label",
label,
";",
*_window_theme_args(window_id),
)
if rc != 0:
log.error("set-option (aux window) failed for %s: %s", name, err)
await self._exec("kill-window", "-t", window_id)
return None
await self._exec("select-window", "-t", window_id)
return window_id
async def has_window(self, window_name: str) -> bool:
"""Check if a window with this name exists."""
rc, stdout, _ = await self._exec(
+28
View File
@@ -0,0 +1,28 @@
"""Tools that can be opened in their own tmux window for a session's project.
A tool window is NOT an hqt session: hqt spawns and styles it, then forgets it.
It lives purely as a tmux window until the tool exits. (``clone`` is handled
separately in the service — it creates a real session, not an aux window.)
"""
from dataclasses import dataclass
@dataclass(frozen=True)
class Tool:
"""How to open a tool in its own window.
``label`` is the base ``@hqt_label`` text (the project name is appended per
spawn). ``command`` is the argv to run; an empty list means "use tmux's
default shell" (a plain interactive shell).
"""
label: str
command: list[str]
TOOLS: dict[str, Tool] = {
"nvim": Tool(label="nvim", command=["nvim"]),
"lazygit": Tool(label="lazygit", command=["lazygit"]),
"shell": Tool(label="shell", command=[]),
}
+111 -7
View File
@@ -2,9 +2,8 @@ from unittest.mock import patch
from click.testing import CliRunner
from hqt import cli as cli_module
from hqt import sandbox
from hqt.cli import main
from hqt import cli, sandbox
from hqt.config import Settings
def _which(found: set[str]):
@@ -16,10 +15,10 @@ def test_doctor_reports_bwrap_present():
# tmux is discovered via cli.shutil.which; bwrap availability comes from the
# shared sandbox.is_available() helper (the single source of truth).
with (
patch.object(cli_module.shutil, "which", side_effect=_which({"tmux", "bwrap"})),
patch.object(cli.shutil, "which", side_effect=_which({"tmux", "bwrap"})),
patch.object(sandbox, "is_available", return_value=True),
):
result = CliRunner().invoke(main, ["doctor"])
result = CliRunner().invoke(cli.main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✓" in result.output
@@ -27,9 +26,114 @@ def test_doctor_reports_bwrap_present():
def test_doctor_reports_bwrap_missing():
"""Test that doctor reports bubblewrap as missing when not present."""
with (
patch.object(cli_module.shutil, "which", side_effect=_which({"tmux"})),
patch.object(cli.shutil, "which", side_effect=_which({"tmux"})),
patch.object(sandbox, "is_available", return_value=False),
):
result = CliRunner().invoke(main, ["doctor"])
result = CliRunner().invoke(cli.main, ["doctor"])
assert result.exit_code == 0
assert "bubblewrap: ✗" in result.output
def test_tool_cmd_opens_tool_window(monkeypatch, tmp_path):
monkeypatch.setattr(
"hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")
)
calls = {}
async def fake_for_window(self, window, tool):
calls["tool"] = (window, tool)
return "@9"
monkeypatch.setattr(
"hqt.sessions.service.SessionService.open_tool_window_for_window",
fake_for_window,
)
result = CliRunner().invoke(cli.main, ["tool", "lazygit", "hqt-5"])
assert result.exit_code == 0, result.output
assert calls["tool"] == ("hqt-5", "lazygit")
def test_tool_cmd_clone_dispatches_to_clone(monkeypatch, tmp_path):
monkeypatch.setattr(
"hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")
)
calls = {}
async def fake_clone(self, window):
calls["clone"] = window
return "RESULT"
monkeypatch.setattr(
"hqt.sessions.service.SessionService.clone_session_for_window", fake_clone
)
result = CliRunner().invoke(cli.main, ["tool", "clone", "hqt-5"])
assert result.exit_code == 0, result.output
assert calls["clone"] == "hqt-5"
def test_tool_cmd_reports_service_error(monkeypatch, tmp_path):
from hqt.errors import ServiceError
monkeypatch.setattr(
"hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")
)
async def boom(self, window, tool):
raise ServiceError("'hqt-5' is not an hqt session window")
monkeypatch.setattr(
"hqt.sessions.service.SessionService.open_tool_window_for_window", boom
)
result = CliRunner().invoke(cli.main, ["tool", "nvim", "hqt-5"])
assert result.exit_code == 1
assert "not an hqt session window" in result.output
def test_palette_cmd_shows_popup_for_session_window(monkeypatch, tmp_path):
monkeypatch.setattr(
"hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")
)
monkeypatch.setattr(
"hqt.sessions.service.SessionService.session_id_for_window",
lambda self, window: 5,
)
runs = []
monkeypatch.setattr(cli.subprocess, "run", lambda argv, **kw: runs.append(argv))
result = CliRunner().invoke(cli.main, ["palette", "hqt-5"])
assert result.exit_code == 0, result.output
assert len(runs) == 1
argv = runs[0]
assert "display-popup" in argv
popup_cmd = argv[-1]
assert "fzf" in popup_cmd
assert "nvim" in popup_cmd and "clone" in popup_cmd
# the window is baked into the command for `hqt tool {} <window>`
assert "hqt-5" in popup_cmd
def test_palette_cmd_hints_for_non_session_window(monkeypatch, tmp_path):
monkeypatch.setattr(
"hqt.config.get_settings", lambda: Settings(db_path=tmp_path / "t.db")
)
monkeypatch.setattr(
"hqt.sessions.service.SessionService.session_id_for_window",
lambda self, window: None,
)
runs = []
monkeypatch.setattr(cli.subprocess, "run", lambda argv, **kw: runs.append(argv))
result = CliRunner().invoke(cli.main, ["palette", "nvim"])
assert result.exit_code == 0, result.output
assert len(runs) == 1
argv = runs[0]
assert "display-message" in argv
assert "display-popup" not in argv
+118
View File
@@ -2010,3 +2010,121 @@ async def test_sandboxed_worktree_binds_repo_git_dir(
)
binds = mock_wrap.call_args.args[3]
assert Bind(Path("/tmp/myproj/.git"), writable=True) in binds
# ---------------------------------------------------------------------------
# Task 4: SessionService tool-window methods
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_open_tool_window_spawns_styled_window(service, db, tmux, monkeypatch):
await service.create_session(project_id=1, harness_name="claude-code")
monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
tmux.open_aux_window = AsyncMock(return_value="@5")
wid = await service.open_tool_window(1, "lazygit")
assert wid == "@5"
tmux.open_aux_window.assert_awaited_once_with(
"lazygit", "/tmp/myproj", ["lazygit"], "lazygit · myproj"
)
@pytest.mark.asyncio
async def test_open_tool_window_shell_skips_which_check(service, db, tmux, monkeypatch):
await service.create_session(project_id=1, harness_name="claude-code")
monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: None)
tmux.open_aux_window = AsyncMock(return_value="@3")
wid = await service.open_tool_window(1, "shell")
assert wid == "@3"
tmux.open_aux_window.assert_awaited_once_with(
"shell", "/tmp/myproj", [], "shell · myproj"
)
@pytest.mark.asyncio
async def test_open_tool_window_unknown_tool_raises(service):
with pytest.raises(ServiceError):
await service.open_tool_window(1, "emacs")
@pytest.mark.asyncio
async def test_open_tool_window_missing_binary_raises(service, db, monkeypatch):
await service.create_session(project_id=1, harness_name="claude-code")
monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: None)
with pytest.raises(ServiceError):
await service.open_tool_window(1, "lazygit")
@pytest.mark.asyncio
async def test_open_tool_window_unknown_session_raises(service, monkeypatch):
monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
with pytest.raises(ServiceError):
await service.open_tool_window(999, "nvim")
@pytest.mark.asyncio
async def test_session_id_for_window_resolves_and_misses(service, db):
await service.create_session(project_id=1, harness_name="claude-code")
assert service.session_id_for_window("hqt-1") == 1
assert service.session_id_for_window("not-an-hqt-window") is None
@pytest.mark.asyncio
async def test_open_tool_window_for_window_resolves_by_name(
service, db, tmux, monkeypatch
):
await service.create_session(project_id=1, harness_name="claude-code")
monkeypatch.setattr("hqt.sessions.service.shutil.which", lambda b: "/usr/bin/" + b)
tmux.open_aux_window = AsyncMock(return_value="@2")
wid = await service.open_tool_window_for_window("hqt-1", "nvim")
assert wid == "@2"
tmux.open_aux_window.assert_awaited_once_with(
"nvim", "/tmp/myproj", ["nvim"], "nvim · myproj"
)
@pytest.mark.asyncio
async def test_open_tool_window_for_window_unknown_window_raises(service):
with pytest.raises(ServiceError):
await service.open_tool_window_for_window("not-an-hqt-window", "nvim")
# ---------------------------------------------------------------------------
# Task 5: clone_session_for_window
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_clone_session_for_window_reuses_project_harness_model(
service, db, monkeypatch
):
await service.create_session(
project_id=1, harness_name="claude-code", nickname="orig", model="opus"
)
captured = {}
async def fake_create(
project_id, harness_name, nickname=None, model=None, worktree_branch=None
):
captured["args"] = (project_id, harness_name, nickname, model)
return "SENTINEL"
monkeypatch.setattr(service, "create_session", fake_create)
result = await service.clone_session_for_window("hqt-1")
assert result == "SENTINEL"
# Same project + harness + model; a fresh sibling, so no nickname.
assert captured["args"] == (1, "claude-code", None, "opus")
@pytest.mark.asyncio
async def test_clone_session_for_window_unknown_window_raises(service):
with pytest.raises(ServiceError):
await service.clone_session_for_window("not-an-hqt-window")
+87
View File
@@ -1208,3 +1208,90 @@ async def test_poll_info_empty_names(runner):
result = await mgr.poll_info([])
assert result == {}
assert runner._exec.call_count == 1
# ---------------------------------------------------------------------------
# Task 2: new_aux_window — styled tool windows (never tracked)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_new_aux_window_spawns_styles_and_selects(runner):
runner._exec.side_effect = [
(0, "0\n", ""), # _next_window_index: indices [0] -> next index 1
(0, "@7\n", ""), # new-window -P -F '#{window_id}'
(0, "", ""), # set-option (automatic-rename + allow-rename + @hqt_label + theme)
(0, "", ""), # select-window
]
wid = await runner.new_aux_window("lazygit", "/proj", ["lazygit"], "lazygit · proj")
assert wid == "@7"
calls = runner._exec.call_args_list
# new-window: next free index, name, cwd, print window_id, then the command.
assert calls[1].args == (
"new-window", "-t", "hqt-main:1", "-n", "lazygit",
"-c", "/proj", "-P", "-F", "#{window_id}", "lazygit",
)
# post-creation options target the window_id and NEVER set remain-on-exit.
assert calls[2].args[:6] == (
"set-option", "-w", "-t", "@7", "automatic-rename", "off",
)
# allow-rename off too, so nvim/lazygit OSC titles can't scramble the label.
assert "allow-rename" in calls[2].args
assert "@hqt_label" in calls[2].args
assert "lazygit · proj" in calls[2].args
assert "remain-on-exit" not in calls[2].args
# switch to it.
assert calls[3].args == ("select-window", "-t", "@7")
@pytest.mark.asyncio
async def test_new_aux_window_shell_omits_command_arg(runner):
runner._exec.side_effect = [
(0, "2\n", ""), # indices [2] -> next index 3
(0, "@9\n", ""),
(0, "", ""),
(0, "", ""),
]
wid = await runner.new_aux_window("shell", "/proj", [], "shell · proj")
assert wid == "@9"
calls = runner._exec.call_args_list
assert calls[1].args == (
"new-window", "-t", "hqt-main:3", "-n", "shell",
"-c", "/proj", "-P", "-F", "#{window_id}",
)
@pytest.mark.asyncio
async def test_new_aux_window_returns_none_when_new_window_fails(runner):
runner._exec.side_effect = [
(0, "0\n", ""), # next index
(1, "", "boom"), # new-window fails
]
wid = await runner.new_aux_window("nvim", "/p", ["nvim"], "nvim · p")
assert wid is None
@pytest.mark.asyncio
async def test_new_aux_window_kills_window_when_set_option_fails(runner):
runner._exec.side_effect = [
(0, "0\n", ""), # _next_window_index
(0, "@7\n", ""), # new-window -> window_id @7
(1, "", "nope"), # set-option fails
(0, "", ""), # kill-window cleanup
]
wid = await runner.new_aux_window("nvim", "/p", ["nvim"], "nvim · p")
assert wid is None
# the half-built window must be cleaned up by id.
assert runner._exec.call_args_list[-1].args == ("kill-window", "-t", "@7")
@pytest.mark.asyncio
async def test_manager_open_aux_window_delegates(runner):
from unittest.mock import AsyncMock
runner.new_aux_window = AsyncMock(return_value="@4")
mgr = TmuxManager(runner)
wid = await mgr.open_aux_window("nvim", "/p", ["nvim"], "nvim · p")
assert wid == "@4"
runner.new_aux_window.assert_awaited_once_with("nvim", "/p", ["nvim"], "nvim · p")
+24
View File
@@ -0,0 +1,24 @@
from hqt.tools import TOOLS, Tool
def test_registry_has_the_three_tools():
assert set(TOOLS) == {"nvim", "lazygit", "shell"}
def test_each_entry_is_a_tool():
assert all(isinstance(t, Tool) for t in TOOLS.values())
def test_nvim_and_lazygit_have_commands():
assert TOOLS["nvim"].command == ["nvim"]
assert TOOLS["lazygit"].command == ["lazygit"]
def test_shell_has_empty_command_meaning_default_shell():
assert TOOLS["shell"].command == []
def test_labels_are_the_bare_tool_names():
assert TOOLS["nvim"].label == "nvim"
assert TOOLS["lazygit"].label == "lazygit"
assert TOOLS["shell"].label == "shell"