Add SessionService tool-window + window-resolution methods

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-10 21:25:39 -04:00
parent 115d0ffc01
commit e588c62fd4
2 changed files with 137 additions and 0 deletions
+53
View File
@@ -1,6 +1,7 @@
import asyncio import asyncio
import contextlib import contextlib
import logging import logging
import shutil
import time import time
from collections.abc import Mapping from collections.abc import Mapping
from dataclasses import dataclass from dataclasses import dataclass
@@ -16,6 +17,7 @@ from hqt.git import worktree
from hqt.git.worktree import WorktreeState from hqt.git.worktree import WorktreeState
from hqt.harnesses.base import HarnessConfigurator, SpawnConfig from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
from hqt.status import window_label from hqt.status import window_label
from hqt.tools import TOOLS
from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.manager import SpawnRequest, TmuxManager
from hqt.tmux.runner import WindowInfo from hqt.tmux.runner import WindowInfo
@@ -386,6 +388,57 @@ class SessionService:
) )
return True 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.
"""
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("Project 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 stop_session(self, session_id: int) -> None: async def stop_session(self, session_id: int) -> None:
with self.factory() as db: with self.factory() as db:
sess = db.get(Session, session_id) sess = db.get(Session, session_id)
+84
View File
@@ -1518,3 +1518,87 @@ async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmu
# The `service` fixture's harness has captures_session_id = False, so the # The `service` fixture's harness has captures_session_id = False, so the
# spawn must NOT be serialized under the capture lock. # spawn must NOT be serialized under the capture lock.
assert observed["locked_during_spawn"] is False assert observed["locked_during_spawn"] is False
# ---------------------------------------------------------------------------
# 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")
def test_session_id_for_window_resolves_and_misses(service, db):
import asyncio
asyncio.run(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")