diff --git a/pyproject.toml b/pyproject.toml index 7f1829c..cd3f4a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,3 +33,10 @@ required_after_code_or_test_changes = [ "uv run ruff check src tests", "uv run ty check", ] + +[[tool.ty.overrides]] +include = ["tests/**"] + +[tool.ty.overrides.rules] +unresolved-attribute = "ignore" +invalid-assignment = "ignore" diff --git a/src/hqt/cli.py b/src/hqt/cli.py index e9efa0a..91ddefc 100644 --- a/src/hqt/cli.py +++ b/src/hqt/cli.py @@ -10,7 +10,7 @@ log = logging.getLogger(__name__) @click.group(invoke_without_command=True) -@click.option('--inside-tmux', is_flag=True, hidden=True) +@click.option("--inside-tmux", is_flag=True, hidden=True) @click.pass_context def main(ctx, inside_tmux): """HQ Terminal - orchestrate AI coding harness sessions via tmux.""" @@ -22,19 +22,22 @@ def launch(inside_tmux: bool = False): """Bootstrap into tmux and launch TUI.""" from hqt.config import get_settings from hqt.db.engine import ensure_db + settings = get_settings() ensure_db(settings) tui_name = settings.tui_session_name win_name = settings.tui_window_name tui_cmd = f"HQT_INSIDE=1 {sys.executable} -m hqt" - if inside_tmux or os.environ.get('HQT_INSIDE'): + if inside_tmux or os.environ.get("HQT_INSIDE"): # We ARE the TUI process inside tmux — run the app from hqt.logging import setup_logging + setup_logging() log.info("Starting hqt TUI") try: from hqt.tui.app import HqtApp + app = HqtApp() app.run() except Exception: @@ -46,27 +49,57 @@ def launch(inside_tmux: bool = False): # Check if the hqt session already exists rc = subprocess.call( [settings.tmux_path, "has-session", "-t", tui_name], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, ) if rc == 0: # Session exists — ensure TUI window is present result = subprocess.run( - [settings.tmux_path, "list-windows", "-t", tui_name, "-F", "#{window_name}"], - capture_output=True, text=True, + [ + settings.tmux_path, + "list-windows", + "-t", + tui_name, + "-F", + "#{window_name}", + ], + capture_output=True, + text=True, ) windows = result.stdout.strip().splitlines() if win_name not in windows: # Recreate TUI window at the end subprocess.call( - [settings.tmux_path, "new-window", "-t", tui_name, "-n", win_name, tui_cmd], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + [ + settings.tmux_path, + "new-window", + "-t", + tui_name, + "-n", + win_name, + tui_cmd, + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, ) - os.execvp(settings.tmux_path, [settings.tmux_path, "attach", "-t", f"{tui_name}:={win_name}"]) + os.execvp( + settings.tmux_path, + [settings.tmux_path, "attach", "-t", f"{tui_name}:={win_name}"], + ) else: # Create new session with TUI as first window - os.execvp(settings.tmux_path, [ - settings.tmux_path, "new-session", "-s", tui_name, "-n", win_name, tui_cmd, - ]) + os.execvp( + settings.tmux_path, + [ + settings.tmux_path, + "new-session", + "-s", + tui_name, + "-n", + win_name, + tui_cmd, + ], + ) @main.command() @@ -74,7 +107,8 @@ def doctor(): """Check system requirements.""" import shutil from hqt.harnesses.registry import discover_harnesses - tmux = shutil.which('tmux') + + tmux = shutil.which("tmux") if tmux: click.echo(f"tmux: ✓ {tmux}") else: @@ -87,12 +121,13 @@ def doctor(): click.echo("No harnesses found on PATH") -@main.command(name='list') +@main.command(name="list") def list_cmd(): """List projects and sessions.""" from hqt.config import get_settings from hqt.db.engine import get_engine, get_session_factory, ensure_db from hqt.projects.service import ProjectService + settings = get_settings() ensure_db(settings) engine = get_engine(settings) diff --git a/src/hqt/config.py b/src/hqt/config.py index 6aacd7b..15e90d5 100644 --- a/src/hqt/config.py +++ b/src/hqt/config.py @@ -1,9 +1,12 @@ from dataclasses import dataclass, field from pathlib import Path + @dataclass class Settings: - db_path: Path = field(default_factory=lambda: Path.home() / ".local/share/hqt/hqt.db") + db_path: Path = field( + default_factory=lambda: Path.home() / ".local/share/hqt/hqt.db" + ) config_dir: Path = field(default_factory=lambda: Path.home() / ".config/hqt") skills_dir: Path = field(default_factory=lambda: Path.home() / ".config/hqt/skills") tmux_path: str = "tmux" @@ -17,8 +20,10 @@ class Settings: # truncated session name (the trailing "-" there is tmux's last-window flag). tui_window_name: str = "⌂ HQT" + _settings: Settings | None = None + def get_settings() -> Settings: global _settings if _settings is None: diff --git a/src/hqt/db/engine.py b/src/hqt/db/engine.py index 0940911..55e9866 100644 --- a/src/hqt/db/engine.py +++ b/src/hqt/db/engine.py @@ -4,13 +4,16 @@ from sqlalchemy.orm import sessionmaker from hqt.config import Settings from hqt.db.models import Base + def get_engine(settings: Settings) -> Engine: settings.db_path.parent.mkdir(parents=True, exist_ok=True) return create_engine(f"sqlite:///{settings.db_path}") + def get_session_factory(engine: Engine) -> sessionmaker: return sessionmaker(bind=engine) + def ensure_db(settings: Settings) -> None: settings.db_path.parent.mkdir(parents=True, exist_ok=True) engine = get_engine(settings) diff --git a/src/hqt/db/models.py b/src/hqt/db/models.py index d8dc1e5..db9c03a 100644 --- a/src/hqt/db/models.py +++ b/src/hqt/db/models.py @@ -2,9 +2,11 @@ from datetime import datetime from sqlalchemy import ForeignKey, String, func from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship + class Base(DeclarativeBase): pass + class Project(Base): __tablename__ = "projects" id: Mapped[int] = mapped_column(primary_key=True) @@ -15,6 +17,7 @@ class Project(Base): archived: Mapped[bool] = mapped_column(default=False) sessions: Mapped[list["Session"]] = relationship(back_populates="project") + class Harness(Base): __tablename__ = "harnesses" id: Mapped[int] = mapped_column(primary_key=True) @@ -25,6 +28,7 @@ class Harness(Base): profile_json: Mapped[str | None] = mapped_column(default=None) sessions: Mapped[list["Session"]] = relationship(back_populates="harness") + class Session(Base): __tablename__ = "sessions" id: Mapped[int] = mapped_column(primary_key=True) @@ -40,6 +44,7 @@ class Session(Base): project: Mapped["Project"] = relationship(back_populates="sessions") harness: Mapped["Harness"] = relationship(back_populates="sessions") + class McpServer(Base): __tablename__ = "mcp_servers" id: Mapped[int] = mapped_column(primary_key=True) @@ -51,10 +56,14 @@ class McpServer(Base): env_json: Mapped[str | None] = mapped_column(default=None) headers_json: Mapped[str | None] = mapped_column(default=None) + class ProjectMcpServer(Base): __tablename__ = "project_mcp_servers" project_id: Mapped[int] = mapped_column(ForeignKey("projects.id"), primary_key=True) - mcp_server_id: Mapped[int] = mapped_column(ForeignKey("mcp_servers.id"), primary_key=True) + mcp_server_id: Mapped[int] = mapped_column( + ForeignKey("mcp_servers.id"), primary_key=True + ) + class ProjectSkill(Base): __tablename__ = "project_skills" diff --git a/src/hqt/harnesses/base.py b/src/hqt/harnesses/base.py index 8b7a86e..ba03a19 100644 --- a/src/hqt/harnesses/base.py +++ b/src/hqt/harnesses/base.py @@ -19,10 +19,14 @@ class HarnessConfigurator(ABC): def generate_session_id(self, hqt_session_id: int) -> str: ... @abstractmethod - def build_spawn_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: ... + def build_spawn_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: ... @abstractmethod - def build_resume_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: ... + def build_resume_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: ... def capture_session_id(self, project_path: Path, since: float) -> str | None: return None diff --git a/src/hqt/harnesses/configurators/claude.py b/src/hqt/harnesses/configurators/claude.py index 9de63e2..3adbded 100644 --- a/src/hqt/harnesses/configurators/claude.py +++ b/src/hqt/harnesses/configurators/claude.py @@ -12,24 +12,26 @@ class ClaudeConfigurator(HarnessConfigurator): # Markers for parse_status — conservative, case-insensitive substring checks. # None returned when none match; caller falls back to activity-based status. - _WORKING_MARKERS = ( - "esc to interrupt", - ) + _WORKING_MARKERS = ("esc to interrupt",) _WAITING_MARKERS = ( - "do you want", # permission / confirmation prompts ("Do you want to proceed?") - "❯ 1.", # ❯ 1. — numbered choice prompt in Claude's interactive UI + "do you want", # permission / confirmation prompts ("Do you want to proceed?") + "❯ 1.", # ❯ 1. — numbered choice prompt in Claude's interactive UI ) def generate_session_id(self, hqt_session_id: int) -> str: return str(uuid.uuid5(NAMESPACE, str(hqt_session_id))) - def build_spawn_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_spawn_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: cmd = ["claude", "--session-id", harness_session_id] if model: cmd += ["--model", model] return SpawnConfig(command=cmd, cwd=project_path) - def build_resume_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_resume_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: cmd = ["claude", "--resume", harness_session_id] if model: cmd += ["--model", model] diff --git a/src/hqt/harnesses/configurators/codex.py b/src/hqt/harnesses/configurators/codex.py index 9831915..dead135 100644 --- a/src/hqt/harnesses/configurators/codex.py +++ b/src/hqt/harnesses/configurators/codex.py @@ -1,4 +1,5 @@ import json +from datetime import datetime from pathlib import Path from hqt.harnesses.base import HarnessConfigurator, SpawnConfig @@ -10,20 +11,22 @@ class CodexConfigurator(HarnessConfigurator): captures_session_id = True # Markers for parse_status — conservative, case-insensitive substring checks. - _WORKING_MARKERS = ( - "esc to interrupt", - ) + _WORKING_MARKERS = ("esc to interrupt",) def generate_session_id(self, hqt_session_id: int) -> str: return str(hqt_session_id) - def build_spawn_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_spawn_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: cmd = ["codex"] if model: cmd += ["-m", model] return SpawnConfig(command=cmd, cwd=project_path) - def build_resume_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_resume_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: cmd = ["codex", "resume", harness_session_id] if model: cmd += ["-m", model] @@ -41,26 +44,34 @@ class CodexConfigurator(HarnessConfigurator): return "working" return None + def _meta_timestamp(self, payload: dict) -> float | None: + timestamp = payload.get("timestamp") + if not isinstance(timestamp, str): + return None + try: + return datetime.fromisoformat(timestamp.replace("Z", "+00:00")).timestamp() + except ValueError: + return None + def capture_session_id(self, project_path: Path, since: float) -> str | None: sessions_dir = Path.home() / ".codex" / "sessions" if not sessions_dir.exists(): return None - candidates: list[tuple[float, Path]] = [] + candidates: list[tuple[float, str]] = [] for rollout in sessions_dir.rglob("rollout-*.jsonl"): try: mtime = rollout.stat().st_mtime - except OSError: - continue - if mtime >= since: - candidates.append((mtime, rollout)) - # Sort newest-first - candidates.sort(key=lambda t: t[0], reverse=True) - for _, rollout in candidates: - try: with rollout.open() as f: meta = json.loads(f.readline()) - if meta.get("type") == "session_meta" and str(Path(meta["payload"]["cwd"])) == str(project_path): - return meta["payload"]["id"] - except (json.JSONDecodeError, KeyError, OSError): + payload = meta["payload"] + if meta.get("type") != "session_meta" or str( + Path(payload["cwd"]) + ) != str(project_path): + continue + started_at = self._meta_timestamp(payload) or mtime + if started_at >= since: + candidates.append((started_at, payload["id"])) + except (json.JSONDecodeError, KeyError, OSError, TypeError): continue - return None + candidates.sort(key=lambda t: t[0]) + return candidates[0][1] if candidates else None diff --git a/src/hqt/harnesses/configurators/generic.py b/src/hqt/harnesses/configurators/generic.py index 3e86bfa..505ea61 100644 --- a/src/hqt/harnesses/configurators/generic.py +++ b/src/hqt/harnesses/configurators/generic.py @@ -13,8 +13,12 @@ class GenericConfigurator(HarnessConfigurator): def generate_session_id(self, hqt_session_id: int) -> str: return str(hqt_session_id) - def build_spawn_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_spawn_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: return SpawnConfig(command=self._command, cwd=project_path) - def build_resume_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_resume_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: return SpawnConfig(command=self._command, cwd=project_path) diff --git a/src/hqt/harnesses/configurators/kiro.py b/src/hqt/harnesses/configurators/kiro.py index 8eaa8ad..c652aff 100644 --- a/src/hqt/harnesses/configurators/kiro.py +++ b/src/hqt/harnesses/configurators/kiro.py @@ -10,8 +10,12 @@ class KiroConfigurator(HarnessConfigurator): def generate_session_id(self, hqt_session_id: int) -> str: return str(hqt_session_id) - def build_spawn_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_spawn_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path) - def build_resume_config(self, project_path: Path, harness_session_id: str, model: str | None = None) -> SpawnConfig: + def build_resume_config( + self, project_path: Path, harness_session_id: str, model: str | None = None + ) -> SpawnConfig: return SpawnConfig(command=["kiro-cli", "chat"], cwd=project_path) diff --git a/src/hqt/mcp/service.py b/src/hqt/mcp/service.py index f1a5ade..48fc8e5 100644 --- a/src/hqt/mcp/service.py +++ b/src/hqt/mcp/service.py @@ -9,8 +9,21 @@ class McpService: def __init__(self, db: DBSession): self.db = db - def create(self, name: str, transport: str, command: str | None = None, args: list[str] | None = None, url: str | None = None) -> McpServer: - server = McpServer(name=name, transport=transport, command=command, args_json=json.dumps(args) if args else None, url=url) + def create( + self, + name: str, + transport: str, + command: str | None = None, + args: list[str] | None = None, + url: str | None = None, + ) -> McpServer: + server = McpServer( + name=name, + transport=transport, + command=command, + args_json=json.dumps(args) if args else None, + url=url, + ) self.db.add(server) self.db.commit() self.db.refresh(server) @@ -29,15 +42,24 @@ class McpService: self.db.commit() def bind_to_project(self, project_id: int, mcp_server_id: int) -> None: - self.db.add(ProjectMcpServer(project_id=project_id, mcp_server_id=mcp_server_id)) + self.db.add( + ProjectMcpServer(project_id=project_id, mcp_server_id=mcp_server_id) + ) self.db.commit() def unbind_from_project(self, project_id: int, mcp_server_id: int) -> None: - self.db.query(ProjectMcpServer).filter_by(project_id=project_id, mcp_server_id=mcp_server_id).delete() + self.db.query(ProjectMcpServer).filter_by( + project_id=project_id, mcp_server_id=mcp_server_id + ).delete() self.db.commit() def get_project_mcps(self, project_id: int) -> list[McpServer]: - ids = [r.mcp_server_id for r in self.db.query(ProjectMcpServer).filter_by(project_id=project_id).all()] + ids = [ + r.mcp_server_id + for r in self.db.query(ProjectMcpServer) + .filter_by(project_id=project_id) + .all() + ] if not ids: return [] return list(self.db.query(McpServer).filter(McpServer.id.in_(ids)).all()) diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 927e09f..880a6a6 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -1,13 +1,15 @@ import asyncio import logging import time +from collections.abc import Mapping from dataclasses import dataclass +from datetime import UTC, datetime from pathlib import Path from sqlalchemy.orm import Session as DBSession from hqt.db.models import Harness, Project, Session -from hqt.harnesses.base import HarnessConfigurator +from hqt.harnesses.base import HarnessConfigurator, SpawnConfig from hqt.status import window_label from hqt.tmux.manager import SpawnRequest, TmuxManager from hqt.tmux.runner import WindowInfo @@ -40,11 +42,22 @@ class CreateSessionResult: class SessionService: - def __init__(self, db: DBSession, tmux: TmuxManager, harnesses: dict[str, HarnessConfigurator]): + def __init__( + self, + db: DBSession, + tmux: TmuxManager, + harnesses: Mapping[str, HarnessConfigurator], + ): self.db = db self.tmux = tmux self.harnesses = harnesses + def _project_path(self, project_id: int) -> Path: + project = self.db.get(Project, project_id) + if project is None: + raise ValueError(f"Unknown project: {project_id}") + return Path(project.path) + async def _capture_session_id_with_retry( self, configurator: HarnessConfigurator, @@ -72,6 +85,30 @@ class SessionService: ) return captured_id + def _session_created_epoch(self, sess: Session) -> float: + created_at = sess.created_at + if isinstance(created_at, datetime): + if created_at.tzinfo is None: + created_at = created_at.replace(tzinfo=UTC) + return created_at.timestamp() + return 0.0 + + def _maybe_capture_missing_session_id(self, sess: Session) -> None: + configurator = self.harnesses.get(sess.harness.name) + if configurator is None or not configurator.captures_session_id: + return + + placeholder_id = configurator.generate_session_id(sess.id) + if sess.harness_session_id and sess.harness_session_id != placeholder_id: + return + + captured_id = configurator.capture_session_id( + self._project_path(sess.project_id), self._session_created_epoch(sess) + ) + if captured_id and captured_id != sess.harness_session_id: + sess.harness_session_id = captured_id + self.db.commit() + async def create_session( self, project_id: int, @@ -90,18 +127,38 @@ class SessionService: log.error("Harness not in DB: %s", harness_name) raise ValueError(f"Unknown harness: {harness_name}") configurator = self.harnesses[harness_name] - sess = Session(project_id=project_id, harness_id=harness_row.id, nickname=nickname, model=model, tmux_session_name="placeholder", archived=False) + sess = Session( + project_id=project_id, + harness_id=harness_row.id, + nickname=nickname, + model=model, + tmux_session_name="placeholder", + archived=False, + ) self.db.add(sess) self.db.flush() sess.tmux_session_name = f"hqt-{sess.id}" - sess.harness_session_id = configurator.generate_session_id(sess.id) - project = self.db.get(Project, project_id) - spawn_cfg = configurator.build_spawn_config(Path(project.path), sess.harness_session_id, model) - log.info("Spawning window %s: cmd=%s cwd=%s", sess.tmux_session_name, spawn_cfg.command, spawn_cfg.cwd) + project_path = self._project_path(project_id) + harness_session_id = configurator.generate_session_id(sess.id) + sess.harness_session_id = harness_session_id + spawn_cfg = configurator.build_spawn_config( + project_path, harness_session_id, model + ) + log.info( + "Spawning window %s: cmd=%s cwd=%s", + sess.tmux_session_name, + spawn_cfg.command, + spawn_cfg.cwd, + ) since = time.time() - spawn_result = await self.tmux.spawn(SpawnRequest( - window_name=sess.tmux_session_name, command=spawn_cfg.command, cwd=str(spawn_cfg.cwd), env=spawn_cfg.env, - )) + spawn_result = await self.tmux.spawn( + SpawnRequest( + window_name=sess.tmux_session_name, + command=spawn_cfg.command, + cwd=str(spawn_cfg.cwd), + env=spawn_cfg.env, + ) + ) if not spawn_result.ok: log.error( "Failed to spawn window %s: %s", @@ -112,7 +169,7 @@ class SessionService: # could inadvertently capture an unrelated session running in the same cwd. if spawn_result.ok and configurator.captures_session_id: captured_id = await self._capture_session_id_with_retry( - configurator, Path(project.path), since, sess.tmux_session_name + configurator, project_path, since, sess.tmux_session_name ) if captured_id: sess.harness_session_id = captured_id @@ -127,7 +184,10 @@ class SessionService: async def attach_session(self, session_id: int) -> bool: """Attach to a session. Handles all states: alive, dead pane, gone.""" sess = self.db.get(Session, session_id) + if sess is None: + return False window_name = sess.tmux_session_name + self._maybe_capture_missing_session_id(sess) if await self.tmux.is_alive(window_name): # Window exists, process running — just switch @@ -151,7 +211,9 @@ class SessionService: """ # Rung 1: resume (restore prior conversation) resume_cfg = self._get_resume_config(sess) - result = await self.tmux.respawn_verified(window_name, resume_cfg.command, str(resume_cfg.cwd), env=resume_cfg.env) + result = await self.tmux.respawn_verified( + window_name, resume_cfg.command, str(resume_cfg.cwd), env=resume_cfg.env + ) if result.ok: return True @@ -163,10 +225,17 @@ class SessionService: # Rung 2: fresh spawn (codex ignores the old session id; start fresh) configurator = self.harnesses[sess.harness.name] - project = self.db.get(Project, sess.project_id) - spawn_cfg = configurator.build_spawn_config(Path(project.path), sess.harness_session_id, sess.model) + project_path = self._project_path(sess.project_id) + harness_session_id = ( + sess.harness_session_id or configurator.generate_session_id(sess.id) + ) + spawn_cfg = configurator.build_spawn_config( + project_path, harness_session_id, sess.model + ) since = time.time() - result = await self.tmux.respawn_verified(window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env) + result = await self.tmux.respawn_verified( + window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env + ) if not result.ok: log.error( "Fallback spawn also failed for %s: %s", @@ -179,7 +248,7 @@ class SessionService: # so that future resumes target this fresh conversation. if configurator.captures_session_id: new_id = await self._capture_session_id_with_retry( - configurator, Path(project.path), since, window_name + configurator, project_path, since, window_name ) if new_id: sess.harness_session_id = new_id @@ -194,6 +263,8 @@ class SessionService: async def stop_session(self, session_id: int) -> None: sess = self.db.get(Session, session_id) + if sess is None: + return if await self.tmux.window_exists(sess.tmux_session_name): await self.tmux.kill(sess.tmux_session_name) @@ -209,17 +280,23 @@ class SessionService: poll (sync_window_labels), so no immediate tmux call is needed here. """ sess = self.db.get(Session, session_id) + if sess is None: + return sess.nickname = (nickname or "").strip() or None self.db.commit() async def delete_session(self, session_id: int) -> None: sess = self.db.get(Session, session_id) + if sess is None: + return if await self.tmux.window_exists(sess.tmux_session_name): await self.tmux.kill(sess.tmux_session_name) self.db.delete(sess) self.db.commit() - async def _status_for(self, sess: Session, wi: WindowInfo | None, now: float) -> str: + async def _status_for( + self, sess: Session, wi: WindowInfo | None, now: float + ) -> str: """Compute a session's status string from its window info. Dead/missing window → "dead". Alive window → the harness's parsed status @@ -243,14 +320,21 @@ class SessionService: ) -> list[SessionInfo]: if now is None: now = time.time() - sessions = self.db.query(Session).filter_by(project_id=project_id, archived=False).all() + sessions = ( + self.db.query(Session) + .filter_by(project_id=project_id, archived=False) + .all() + ) names = [s.tmux_session_name for s in sessions] info_map = await self.tmux.poll_info(names) result: list[SessionInfo] = [] for s in sessions: + self._maybe_capture_missing_session_id(s) wi = info_map.get(s.tmux_session_name) status = await self._status_for(s, wi, now) - result.append(SessionInfo(session=s, alive=bool(wi and wi.alive), status=status)) + result.append( + SessionInfo(session=s, alive=bool(wi and wi.alive), status=status) + ) return result async def sync_window_labels(self, now: float | None = None) -> list[SessionInfo]: @@ -269,6 +353,7 @@ class SessionService: info_map = await self.tmux.poll_info(names) result: list[SessionInfo] = [] for s in sessions: + self._maybe_capture_missing_session_id(s) wi = info_map.get(s.tmux_session_name) alive = bool(wi and wi.alive) status = await self._status_for(s, wi, now) @@ -279,7 +364,11 @@ class SessionService: ) return result - def _get_resume_config(self, sess: Session): + def _get_resume_config(self, sess: Session) -> SpawnConfig: configurator = self.harnesses[sess.harness.name] - project = self.db.get(Project, sess.project_id) - return configurator.build_resume_config(Path(project.path), sess.harness_session_id, sess.model) + harness_session_id = ( + sess.harness_session_id or configurator.generate_session_id(sess.id) + ) + return configurator.build_resume_config( + self._project_path(sess.project_id), harness_session_id, sess.model + ) diff --git a/src/hqt/skills/registry.py b/src/hqt/skills/registry.py index e44054e..a2c94f6 100644 --- a/src/hqt/skills/registry.py +++ b/src/hqt/skills/registry.py @@ -25,9 +25,11 @@ class SkillRegistry: if len(parts) < 3: continue meta = yaml.safe_load(parts[1]) - skills.append(Skill( - name=meta["name"], - description=meta.get("description", ""), - content=parts[2].strip(), - )) + skills.append( + Skill( + name=meta["name"], + description=meta.get("description", ""), + content=parts[2].strip(), + ) + ) return skills diff --git a/src/hqt/tmux/manager.py b/src/hqt/tmux/manager.py index 58a9fdb..5d7dd3a 100644 --- a/src/hqt/tmux/manager.py +++ b/src/hqt/tmux/manager.py @@ -31,13 +31,17 @@ class TmuxManager: async def spawn(self, req: SpawnRequest) -> SpawnResult: """Spawn a new harness window. Returns SpawnResult with ok, window_id, and error.""" cmd = shlex.join(req.command) - window_id = await self.runner.new_window(req.window_name, req.cwd, cmd, env=req.env) + window_id = await self.runner.new_window( + req.window_name, req.cwd, cmd, env=req.env + ) if window_id is None: return SpawnResult(ok=False, window_id=None, error="new_window failed") alive, pane_text = await self.runner.verify_window_alive(req.window_name) if not alive: - error = pane_text.strip() or f"Window {req.window_name!r} pane died immediately" + error = ( + pane_text.strip() or f"Window {req.window_name!r} pane died immediately" + ) return SpawnResult(ok=False, window_id=window_id, error=error) return SpawnResult(ok=True, window_id=window_id, error="") @@ -63,11 +67,19 @@ class TmuxManager: # Window gone entirely — create fresh, which gives us a window_id window_id = await self.runner.new_window(window_name, cwd, cmd, env=env) if window_id is None: - return SpawnResult(ok=False, window_id=None, error=f"new_window failed for {window_name!r}") + return SpawnResult( + ok=False, + window_id=None, + error=f"new_window failed for {window_name!r}", + ) else: ok = await self.runner.respawn_pane(window_name, cmd, cwd, env=env) if not ok: - return SpawnResult(ok=False, window_id=None, error=f"respawn_pane failed for {window_name!r}") + return SpawnResult( + ok=False, + window_id=None, + error=f"respawn_pane failed for {window_name!r}", + ) alive, pane_text = await self.runner.verify_window_alive(window_name) if not alive: diff --git a/src/hqt/tmux/runner.py b/src/hqt/tmux/runner.py index c6bb693..4b79ea7 100644 --- a/src/hqt/tmux/runner.py +++ b/src/hqt/tmux/runner.py @@ -15,6 +15,7 @@ WINDOW_STATUS_FORMAT = " #I:#{?@hqt_label,#{@hqt_label},#W}#F " # between the colored pills rather than a visible glyph. WINDOW_STATUS_SEPARATOR = " " + def _home_window_format(window_name: str) -> str: """Build the home/TUI window's status cell: just its first glyph (the house, "⌂") padded — no index, no flag, no "HQT" text. @@ -26,20 +27,21 @@ def _home_window_format(window_name: str) -> str: glyph = window_name.strip()[:1] or window_name return f" {glyph} " + # Catppuccin Frappé palette — mirrors the TUI's FRAPPE_THEME (see tui/app.py) so # the tmux status bar and the TUI render the same colors. Hex strings only; the # status bar is themed by apply_theme below. _FRAPPE = { "crust": "#232634", - "mantle": "#292c3c", # status-bar background (TUI $background) + "mantle": "#292c3c", # status-bar background (TUI $background) "base": "#303446", "surface0": "#414559", "surface1": "#51576d", "surface2": "#626880", "overlay1": "#838ba7", # inactive window text "text": "#c6d0f5", - "blue": "#8caaee", # primary — session badge / clock accent - "peach": "#ef9f76", # accent — current window (matches TUI focus) + "blue": "#8caaee", # primary — session badge / clock accent + "peach": "#ef9f76", # accent — current window (matches TUI focus) } # SESSION-scoped status-bar options. Each becomes `set-option -t ...`. @@ -50,13 +52,17 @@ _SESSION_THEME_OPTIONS: list[tuple[str, str]] = [ ("status-style", f"bg={_FRAPPE['mantle']},fg={_FRAPPE['text']}"), ("status-justify", "left"), ("status-left-length", "24"), - ("status-left", - f"#[fg={_FRAPPE['crust']},bg={_FRAPPE['blue']},bold] #S " - f"#[fg={_FRAPPE['blue']},bg={_FRAPPE['mantle']},nobold] "), + ( + "status-left", + f"#[fg={_FRAPPE['crust']},bg={_FRAPPE['blue']},bold] #S " + f"#[fg={_FRAPPE['blue']},bg={_FRAPPE['mantle']},nobold] ", + ), ("status-right-length", "40"), - ("status-right", - f"#[fg={_FRAPPE['overlay1']},bg={_FRAPPE['mantle']}] %H:%M " - f"#[fg={_FRAPPE['crust']},bg={_FRAPPE['blue']},bold] %d %b "), + ( + "status-right", + f"#[fg={_FRAPPE['overlay1']},bg={_FRAPPE['mantle']}] %H:%M " + f"#[fg={_FRAPPE['crust']},bg={_FRAPPE['blue']},bold] %d %b ", + ), ("message-style", f"fg={_FRAPPE['text']},bg={_FRAPPE['surface0']}"), ("message-command-style", f"fg={_FRAPPE['text']},bg={_FRAPPE['surface0']}"), ] @@ -73,7 +79,10 @@ _WINDOW_THEME_OPTIONS: list[tuple[str, str]] = [ ("window-status-separator", WINDOW_STATUS_SEPARATOR), ("window-status-style", f"fg={_FRAPPE['overlay1']},bg={_FRAPPE['mantle']}"), ("window-status-format", WINDOW_STATUS_FORMAT), - ("window-status-current-style", f"fg={_FRAPPE['crust']},bg={_FRAPPE['peach']},bold"), + ( + "window-status-current-style", + f"fg={_FRAPPE['crust']},bg={_FRAPPE['peach']},bold", + ), ("window-status-current-format", WINDOW_STATUS_FORMAT), ("mode-style", f"fg={_FRAPPE['crust']},bg={_FRAPPE['peach']}"), ("pane-border-style", f"fg={_FRAPPE['surface0']}"), @@ -99,6 +108,7 @@ def _window_theme_args(target: str) -> list[str]: @dataclass(frozen=True) class WindowInfo: """Activity metadata for a single tmux window.""" + alive: bool last_activity: int @@ -113,12 +123,13 @@ class TmuxRunner: async def _exec(self, *args: str) -> tuple[int, str, str]: """Run tmux command, return (returncode, stdout, stderr).""" proc = await asyncio.create_subprocess_exec( - self.tmux_path, *args, + self.tmux_path, + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, stderr = await proc.communicate() - return proc.returncode, stdout.decode(), stderr.decode() + return proc.returncode or 0, stdout.decode(), stderr.decode() def _target(self, window_name: str) -> str: return f"{self.session_name}:={window_name}" @@ -161,12 +172,24 @@ class TmuxRunner: if home_window: target = self._target(home_window) home_fmt = _home_window_format(home_window) - args.extend([ - ";", "set-option", "-w", "-t", target, - "window-status-format", home_fmt, - ";", "set-option", "-w", "-t", target, - "window-status-current-format", home_fmt, - ]) + args.extend( + [ + ";", + "set-option", + "-w", + "-t", + target, + "window-status-format", + home_fmt, + ";", + "set-option", + "-w", + "-t", + target, + "window-status-current-format", + home_fmt, + ] + ) rc, _, err = await self._exec(*args) if rc != 0: log.warning("apply_theme failed for session %s: %s", self.session_name, err) @@ -184,7 +207,11 @@ class TmuxRunner: session somehow has no windows. """ rc, stdout, _ = await self._exec( - "list-windows", "-t", self.session_name, "-F", "#{window_index}", + "list-windows", + "-t", + self.session_name, + "-F", + "#{window_index}", ) if rc != 0: return 0 @@ -226,10 +253,15 @@ class TmuxRunner: idx = await self._next_window_index() rc, stdout, err = await self._exec( "new-window", - "-t", f"{self.session_name}:{idx}", - "-n", window_name, - "-c", cwd, - "-P", "-F", "#{window_id}", + "-t", + f"{self.session_name}:{idx}", + "-n", + window_name, + "-c", + cwd, + "-P", + "-F", + "#{window_id}", ) if rc != 0: log.error("new-window failed: %s", err) @@ -245,10 +277,25 @@ class TmuxRunner: # would render with tmux's default (no Frappé current-window highlight). target = self._target(window_name) rc, _, err = await self._exec( - "set-option", "-t", target, "remain-on-exit", "on", - ";", "set-option", "-t", target, "allow-rename", "off", - ";", "set-option", "-t", target, "automatic-rename", "off", - ";", *_window_theme_args(target), + "set-option", + "-t", + target, + "remain-on-exit", + "on", + ";", + "set-option", + "-t", + target, + "allow-rename", + "off", + ";", + "set-option", + "-t", + target, + "automatic-rename", + "off", + ";", + *_window_theme_args(target), ) if rc != 0: log.error("set-option (window options) failed for %s: %s", window_name, err) @@ -268,8 +315,13 @@ class TmuxRunner: async def has_window(self, window_name: str) -> bool: """Check if a window with this name exists.""" rc, stdout, _ = await self._exec( - "list-windows", "-t", self.session_name, "-F", "#{window_name}", - "-f", f"#{{==:#{{window_name}},{window_name}}}", + "list-windows", + "-t", + self.session_name, + "-F", + "#{window_name}", + "-f", + f"#{{==:#{{window_name}},{window_name}}}", ) if rc != 0: return False @@ -278,7 +330,11 @@ class TmuxRunner: async def is_pane_dead(self, window_name: str) -> bool: """Check if the pane in this window has exited (remain-on-exit keeps it visible).""" rc, stdout, _ = await self._exec( - "list-panes", "-t", self._target(window_name), "-F", "#{pane_dead}", + "list-panes", + "-t", + self._target(window_name), + "-F", + "#{pane_dead}", ) if rc != 0: return True # Window doesn't exist, treat as dead @@ -320,7 +376,12 @@ class TmuxRunner: env, if provided, is passed as -e KEY=VALUE flags. Requires tmux ≥3.0. """ rc, _, err = await self._exec( - "respawn-pane", "-t", self._target(window_name), "-k", "-c", cwd, + "respawn-pane", + "-t", + self._target(window_name), + "-k", + "-c", + cwd, *self._env_flags(env), command, ) @@ -347,7 +408,12 @@ class TmuxRunner: """ target = self._target(window_name) await self._exec( - "set-option", "-w", "-t", target, "@hqt_label", label, + "set-option", + "-w", + "-t", + target, + "@hqt_label", + label, ) async def select_window(self, window_name: str) -> bool: @@ -363,7 +429,11 @@ class TmuxRunner: async def list_windows(self) -> list[str]: """List all window names in the session.""" rc, stdout, _ = await self._exec( - "list-windows", "-t", self.session_name, "-F", "#{window_name}", + "list-windows", + "-t", + self.session_name, + "-F", + "#{window_name}", ) if rc != 0: return [] @@ -379,7 +449,11 @@ class TmuxRunner: Malformed lines (not exactly 3 columns) are skipped silently. """ rc, stdout, _ = await self._exec( - "list-panes", "-s", "-t", self.session_name, "-F", + "list-panes", + "-s", + "-t", + self.session_name, + "-F", "#{window_name}\t#{pane_dead}\t#{window_activity}", ) if rc != 0: @@ -418,6 +492,10 @@ class TmuxRunner: async def capture_pane(self, window_name: str) -> str: _, stdout, _ = await self._exec( - "capture-pane", "-t", self._target(window_name), "-p", "-J", + "capture-pane", + "-t", + self._target(window_name), + "-p", + "-J", ) return stdout diff --git a/src/hqt/tui/app.py b/src/hqt/tui/app.py index a6f2494..e2f2da5 100644 --- a/src/hqt/tui/app.py +++ b/src/hqt/tui/app.py @@ -7,8 +7,6 @@ from textual.theme import Theme from textual.widgets import Footer from hqt.config import get_settings - -log = logging.getLogger(__name__) 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.projects.service import ProjectService @@ -21,32 +19,34 @@ from hqt.tui.screens.rename_session import RenameSessionScreen from hqt.tui.widgets.project_list import ProjectList, ProjectSelected from hqt.tui.widgets.session_list import SessionList, SessionSelected +log = logging.getLogger(__name__) + # Catppuccin Frappé, mirroring the structure of Textual's built-in # catppuccin-mocha theme so no colors are auto-derived off-palette. FRAPPE_THEME = Theme( name="catppuccin-frappe", - primary="#8caaee", # Blue - secondary="#ca9ee6", # Mauve - accent="#ef9f76", # Peach - success="#a6d189", # Green - warning="#e5c890", # Yellow - error="#e78284", # Red + primary="#8caaee", # Blue + secondary="#ca9ee6", # Mauve + accent="#ef9f76", # Peach + success="#a6d189", # Green + warning="#e5c890", # Yellow + error="#e78284", # Red foreground="#c6d0f5", # Text background="#292c3c", # Mantle - surface="#414559", # Surface0 - panel="#51576d", # Surface1 + surface="#414559", # Surface0 + panel="#51576d", # Surface1 dark=True, variables={ - "input-cursor-foreground": "#232634", # Crust - "input-cursor-background": "#f2d5cf", # Rosewater + "input-cursor-foreground": "#232634", # Crust + "input-cursor-background": "#f2d5cf", # Rosewater "input-selection-background": "#949cbb 30%", # Overlay2 30% - "border": "#babbf1", # Lavender - "border-blurred": "#626880", # Surface2 - "footer-background": "#51576d", # Surface1 - "block-cursor-foreground": "#303446", # Base + "border": "#babbf1", # Lavender + "border-blurred": "#626880", # Surface2 + "footer-background": "#51576d", # Surface1 + "block-cursor-foreground": "#303446", # Base "block-cursor-text-style": "none", - "button-color-foreground": "#292c3c", # Mantle + "button-color-foreground": "#292c3c", # Mantle }, ) @@ -82,6 +82,14 @@ class HqtApp(App): self._session_service: SessionService | None = None self._project_service: ProjectService | None = None + def _projects(self) -> ProjectService: + assert self._project_service is not None + return self._project_service + + def _sessions(self) -> SessionService: + assert self._session_service is not None + return self._session_service + def compose(self) -> ComposeResult: # No Header: the project and session panels share the top row directly # (ProjectList docks left, SessionList fills the rest), with the keybind @@ -114,15 +122,17 @@ class HqtApp(App): log.info("TUI mounted successfully") def _load_projects(self) -> None: - projects = self._project_service.list_all() + projects = self._projects().list_all() + async def _do() -> None: await self.query_one(ProjectList).refresh_projects(projects) + self.run_worker(_do()) async def _poll_sessions(self) -> None: # Sync every session's tmux window label (session-wide), then update the # UI for the selected project from the same computed status — no recompute. - infos = await self._session_service.sync_window_labels() + infos = await self._sessions().sync_window_labels() if self._selected_project_id is not None: subset = [ i for i in infos if i.session.project_id == self._selected_project_id @@ -132,12 +142,14 @@ class HqtApp(App): async def _refresh_sessions(self, restore_selection: bool = False) -> None: if self._selected_project_id is None: return - infos = await self._session_service.list_sessions(self._selected_project_id) + infos = await self._sessions().list_sessions(self._selected_project_id) sl = self.query_one(SessionList) if restore_selection: # Project switch: restore this project's remembered session, or let the # widget default to the first session when there's no memory yet. - remembered = self._selected_session_by_project.get(self._selected_project_id) + remembered = self._selected_session_by_project.get( + self._selected_project_id + ) await sl.refresh_sessions(infos, select_session_id=remembered) else: # Poll / post-action refresh: keep the current selection. @@ -149,7 +161,9 @@ class HqtApp(App): def on_session_selected(self, message: SessionSelected) -> None: if self._selected_project_id is not None: - self._selected_session_by_project[self._selected_project_id] = message.session_id + self._selected_session_by_project[self._selected_project_id] = ( + message.session_id + ) def action_toggle_focus(self) -> None: self.screen.focus_next() @@ -158,7 +172,7 @@ class HqtApp(App): def on_dismiss(result: tuple[str, str] | None) -> None: if result: name, path = result - self._project_service.create(name, path) + self._projects().create(name, path) self._load_projects() self.push_screen(ProjectFormScreen(), on_dismiss) @@ -168,7 +182,7 @@ class HqtApp(App): if project_id is None: self.notify("Select a project first", severity="warning") return - project = self._project_service.get(project_id) + project = self._projects().get(project_id) if project is None: self.notify("Project not found", severity="error") return @@ -177,7 +191,7 @@ class HqtApp(App): if result: name, path = result try: - self._project_service.update(project_id, name, path) + self._projects().update(project_id, name, path) except ValueError as err: self.notify(str(err), severity="error") return @@ -196,6 +210,7 @@ class HqtApp(App): if self._selected_project_id is None: self.notify("Select a project first", severity="warning") return + project_id = self._selected_project_id harness_names = list(discover_harnesses().keys()) if not harness_names: self.notify("No harnesses found", severity="error") @@ -204,13 +219,18 @@ class HqtApp(App): def on_dismiss(result: tuple[str, str, str | None] | None) -> None: if result: harness_name, nickname, model = result + async def _do() -> None: - create_result = await self._session_service.create_session( - self._selected_project_id, harness_name, nickname, model + create_result = await self._sessions().create_session( + project_id, harness_name, nickname, model ) if not create_result.spawn_ok: first_line = next( - (ln for ln in create_result.spawn_error.splitlines() if ln.strip()), + ( + ln + for ln in create_result.spawn_error.splitlines() + if ln.strip() + ), create_result.spawn_error, )[:120] self.notify( @@ -218,6 +238,7 @@ class HqtApp(App): severity="error", ) await self._refresh_sessions() + self.run_worker(_do()) self.push_screen(NewSessionScreen(harness_names), on_dismiss) @@ -236,18 +257,22 @@ class HqtApp(App): def action_attach_session(self) -> None: sid = self.query_one(SessionList).get_selected_session_id() if sid: + async def _do() -> None: - ok = await self._session_service.attach_session(sid) + ok = await self._sessions().attach_session(sid) if not ok: self.notify("Failed to attach session", severity="error") + self.run_worker(_do()) def action_delete_session(self) -> None: sid = self.query_one(SessionList).get_selected_session_id() if sid: + async def _do() -> None: - await self._session_service.delete_session(sid) + await self._sessions().delete_session(sid) await self._refresh_sessions() + self.run_worker(_do()) def action_rename_session(self) -> None: @@ -255,14 +280,14 @@ class HqtApp(App): if sid is None: self.notify("Select a session first", severity="warning") return - sess = self._session_service.get_session(sid) + sess = self._sessions().get_session(sid) if sess is None: self.notify("Session not found", severity="error") return def on_dismiss(result: str | None) -> None: if result is not None: - self._session_service.rename_session(sid, result) + self._sessions().rename_session(sid, result) self.run_worker(self._refresh_sessions()) self.push_screen( @@ -273,7 +298,9 @@ class HqtApp(App): def action_stop_session(self) -> None: sid = self.query_one(SessionList).get_selected_session_id() if sid: + async def _do() -> None: - await self._session_service.stop_session(sid) + await self._sessions().stop_session(sid) await self._refresh_sessions() + self.run_worker(_do()) diff --git a/src/hqt/tui/widgets/project_list.py b/src/hqt/tui/widgets/project_list.py index a4d9a87..a23942b 100644 --- a/src/hqt/tui/widgets/project_list.py +++ b/src/hqt/tui/widgets/project_list.py @@ -23,7 +23,6 @@ class ProjectList(Widget, can_focus=False): await lv.clear() for p in projects: item = ListItem(Label(p.name), id=f"proj-{p.id}") - item.data = p.id # type: ignore[attr-defined] lv.append(item) # Default to the first project so one is always selected on launch; keep # the prior position across reloads (e.g. after add/edit). Setting `index` @@ -35,11 +34,16 @@ class ProjectList(Widget, can_focus=False): lv.index = 0 def on_list_view_highlighted(self, event: ListView.Highlighted) -> None: - if event.item and hasattr(event.item, "data"): - self.post_message(ProjectSelected(event.item.data)) + project_id = self._project_id_from_item(event.item) + if project_id is not None: + self.post_message(ProjectSelected(project_id)) def get_selected_project_id(self) -> int | None: lv = self.query_one("#project-list", ListView) - if lv.highlighted_child and hasattr(lv.highlighted_child, "data"): - return lv.highlighted_child.data + return self._project_id_from_item(lv.highlighted_child) + + def _project_id_from_item(self, item: ListItem | None) -> int | None: + item_id = item.id if item else None + if item_id and item_id.startswith("proj-"): + return int(item_id.removeprefix("proj-")) return None diff --git a/src/hqt/tui/widgets/session_list.py b/src/hqt/tui/widgets/session_list.py index b9193ed..05f1789 100644 --- a/src/hqt/tui/widgets/session_list.py +++ b/src/hqt/tui/widgets/session_list.py @@ -18,12 +18,13 @@ class SessionSelected(Message): self.session_id = session_id super().__init__() + _STATUS_COLORS: dict[str, str] = { "working": "#a6d189", # Green "waiting": "#e5c890", # Yellow - "active": "#81c8be", # Teal - "idle": "#81c8be", # Teal - "dead": "#737994", # Overlay0 + "active": "#81c8be", # Teal + "idle": "#81c8be", # Teal + "dead": "#737994", # Overlay0 } @@ -40,7 +41,9 @@ class SessionList(Widget, can_focus=False): yield Label("Sessions", id="session-header") yield VimListView(id="session-list") - async def refresh_sessions(self, session_infos: list, select_session_id=_UNSET) -> None: + async def refresh_sessions( + self, session_infos: list, select_session_id=_UNSET + ) -> None: lv = self.query_one("#session-list", ListView) # Build new items and check if anything changed new_items = [] @@ -84,15 +87,19 @@ class SessionList(Widget, can_focus=False): await lv.clear() for item_id, text, sid in new_items: item = ListItem(Label(text), id=item_id) - item.data = sid # type: ignore[attr-defined] lv.append(item) def on_list_view_highlighted(self, event: ListView.Highlighted) -> None: - if event.item and hasattr(event.item, "data"): - self.post_message(SessionSelected(event.item.data)) + session_id = self._session_id_from_item(event.item) + if session_id is not None: + self.post_message(SessionSelected(session_id)) def get_selected_session_id(self) -> int | None: lv = self.query_one("#session-list", ListView) - if lv.highlighted_child and hasattr(lv.highlighted_child, "data"): - return lv.highlighted_child.data + return self._session_id_from_item(lv.highlighted_child) + + def _session_id_from_item(self, item: ListItem | None) -> int | None: + item_id = item.id if item else None + if item_id and item_id.startswith("sess-"): + return int(item_id.removeprefix("sess-")) return None diff --git a/tests/test_db.py b/tests/test_db.py index 6d55ef0..4cdaf1d 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1,6 +1,5 @@ from pathlib import Path from sqlalchemy import inspect -from sqlalchemy.orm import Session as DbSession from hqt.config import Settings from hqt.db.engine import ensure_db, get_engine, get_session_factory diff --git a/tests/test_harnesses.py b/tests/test_harnesses.py index 93bc66b..74dddb8 100644 --- a/tests/test_harnesses.py +++ b/tests/test_harnesses.py @@ -1,6 +1,7 @@ import json import os import time +from datetime import UTC, datetime from pathlib import Path from unittest.mock import patch @@ -73,8 +74,16 @@ def test_codex_resume_uses_uuid(): def test_codex_capture_session_id(tmp_path): sessions_dir = tmp_path / ".codex" / "sessions" / "2026" / "06" / "09" sessions_dir.mkdir(parents=True) - rollout = sessions_dir / "rollout-20260609-abcdef12-3456-7890-abcd-ef1234567890.jsonl" - meta = {"type": "session_meta", "payload": {"id": "abcdef12-3456-7890-abcd-ef1234567890", "cwd": "/projects/foo"}} + rollout = ( + sessions_dir / "rollout-20260609-abcdef12-3456-7890-abcd-ef1234567890.jsonl" + ) + meta = { + "type": "session_meta", + "payload": { + "id": "abcdef12-3456-7890-abcd-ef1234567890", + "cwd": "/projects/foo", + }, + } rollout.write_text(json.dumps(meta) + "\n") since = time.time() - 60 # far enough back that freshly-written file qualifies @@ -100,7 +109,9 @@ def test_codex_capture_session_id_wrong_cwd(tmp_path): @patch("hqt.harnesses.registry.shutil.which") def test_registry_discovers_harnesses(mock_which): - mock_which.side_effect = lambda x: "/usr/bin/" + x if x in ("claude", "codex") else None + mock_which.side_effect = lambda x: ( + "/usr/bin/" + x if x in ("claude", "codex") else None + ) result = discover_harnesses() assert "claude" in result assert "codex" in result @@ -120,6 +131,7 @@ def test_codex_captures_session_id_flag(): def test_base_captures_session_id_flag_false(): """HarnessConfigurator base class default must be False.""" from hqt.harnesses.base import HarnessConfigurator + assert HarnessConfigurator.captures_session_id is False @@ -128,8 +140,12 @@ def test_claude_captures_session_id_flag_false(): assert ClaudeConfigurator.captures_session_id is False -def _write_rollout(path: Path, session_id: str, cwd: str) -> None: +def _write_rollout( + path: Path, session_id: str, cwd: str, timestamp: str | None = None +) -> None: meta = {"type": "session_meta", "payload": {"id": session_id, "cwd": cwd}} + if timestamp: + meta["payload"]["timestamp"] = timestamp path.write_text(json.dumps(meta) + "\n") @@ -168,8 +184,8 @@ def test_codex_capture_accepts_file_after_since(tmp_path): assert result == "new-id" -def test_codex_capture_newest_post_since_cwd_match_wins(tmp_path): - """Among multiple post-since rollouts, newest matching cwd wins over older matching.""" +def test_codex_capture_first_post_since_cwd_match_wins(tmp_path): + """Among multiple post-since rollouts, the one started nearest since wins.""" sessions_dir = tmp_path / ".codex" / "sessions" / "dir" sessions_dir.mkdir(parents=True) @@ -186,7 +202,39 @@ def test_codex_capture_newest_post_since_cwd_match_wins(tmp_path): c = CodexConfigurator() with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path): result = c.capture_session_id(Path("/projects/foo"), since) - assert result == "newer-id" + assert result == "older-id" + + +def test_codex_capture_prefers_started_nearest_since_over_newest_file(tmp_path): + """Late capture should match the Codex session spawned at since, not a later same-cwd session.""" + sessions_dir = tmp_path / ".codex" / "sessions" / "dir" + sessions_dir.mkdir(parents=True) + + target = sessions_dir / "rollout-target.jsonl" + unrelated_later = sessions_dir / "rollout-later.jsonl" + + since = datetime(2026, 6, 10, 21, 9, 15, tzinfo=UTC).timestamp() + _write_rollout( + target, + "target-id", + "/projects/foo", + timestamp="2026-06-10T21:09:16.000Z", + ) + _write_rollout( + unrelated_later, + "later-id", + "/projects/foo", + timestamp="2026-06-10T21:45:45.000Z", + ) + + now = time.time() + os.utime(target, (now - 60, now - 60)) + os.utime(unrelated_later, (now, now)) + + c = CodexConfigurator() + with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path): + result = c.capture_session_id(Path("/projects/foo"), since) + assert result == "target-id" def test_codex_capture_skips_non_matching_cwd_picks_older_matching(tmp_path): @@ -273,6 +321,7 @@ CODEX_IDLE_SCREEN = """\ def test_base_parse_status_returns_none(): """Base HarnessConfigurator.parse_status always returns None.""" from hqt.harnesses.configurators.kiro import KiroConfigurator + k = KiroConfigurator() assert k.parse_status("anything") is None assert k.parse_status("") is None diff --git a/tests/test_integration.py b/tests/test_integration.py index 4dc4481..2091a54 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,5 +1,4 @@ import pytest -from pathlib import Path from unittest.mock import AsyncMock, MagicMock from sqlalchemy import create_engine @@ -47,14 +46,19 @@ async def test_full_lifecycle(setup): assert project.id == 1 # Create session - create_result = await sess_svc.create_session(project_id=project.id, harness_name="claude-code", nickname="e2e") + create_result = await sess_svc.create_session( + project_id=project.id, harness_name="claude-code", nickname="e2e" + ) session = create_result.session assert session.id is not None tmux.spawn.assert_called_once() # List sessions - alive from hqt.tmux.runner import WindowInfo - tmux.poll_info.return_value = {session.tmux_session_name: WindowInfo(alive=True, last_activity=0)} + + tmux.poll_info.return_value = { + session.tmux_session_name: WindowInfo(alive=True, last_activity=0) + } infos = await sess_svc.list_sessions(project_id=project.id) assert len(infos) == 1 assert infos[0].alive is True @@ -64,7 +68,9 @@ async def test_full_lifecycle(setup): tmux.kill.assert_called_once_with(session.tmux_session_name) # List sessions - dead - tmux.poll_info.return_value = {session.tmux_session_name: WindowInfo(alive=False, last_activity=0)} + tmux.poll_info.return_value = { + session.tmux_session_name: WindowInfo(alive=False, last_activity=0) + } infos = await sess_svc.list_sessions(project_id=project.id) assert len(infos) == 1 assert infos[0].alive is False diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 5479eb9..d5679bb 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -33,7 +33,9 @@ def tmux(): m.capture_pane = AsyncMock(return_value="") m.attach = AsyncMock() m.window_exists = AsyncMock(return_value=True) - m.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error="")) + m.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=True, window_id=None, error="") + ) return m @@ -42,7 +44,9 @@ def harnesses(): h = MagicMock() h.captures_session_id = False h.generate_session_id.return_value = "sess-1" - h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj")) + h.build_spawn_config.return_value = MagicMock( + command=["claude"], env={}, cwd=Path("/tmp/myproj") + ) h.parse_status = MagicMock(return_value=None) return {"claude-code": h} @@ -54,7 +58,9 @@ def service(db, tmux, harnesses): @pytest.mark.asyncio async def test_create_session(service, db, tmux, harnesses): - result = await service.create_session(project_id=1, harness_name="claude-code", nickname="test") + result = await service.create_session( + project_id=1, harness_name="claude-code", nickname="test" + ) sess = result.session assert sess.id is not None assert sess.tmux_session_name == f"hqt-{sess.id}" @@ -66,6 +72,7 @@ async def test_create_session(service, db, tmux, harnesses): @pytest.mark.asyncio async def test_list_sessions(service, db, tmux): from hqt.tmux.runner import WindowInfo + await service.create_session(project_id=1, harness_name="claude-code") tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1000)} result = await service.list_sessions(project_id=1) @@ -95,7 +102,10 @@ async def test_stop_session_no_kill_when_window_missing(service, db, tmux): async def test_create_session_logs_failed_spawn(service, db, tmux, caplog): """create_session logs error including pane text when spawn fails, but still commits.""" import logging - tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="bash: command not found") + + tmux.spawn.return_value = SpawnResult( + ok=False, window_id=None, error="bash: command not found" + ) with caplog.at_level(logging.ERROR): result = await service.create_session(project_id=1, harness_name="claude-code") assert result.session.id is not None # row was committed @@ -108,16 +118,21 @@ async def test_create_session_logs_failed_spawn(service, db, tmux, caplog): async def test_attach_session_passes_env(service, db, tmux, harnesses): """attach_session passes cfg.env to tmux.respawn_verified when the window needs respawning.""" from unittest.mock import AsyncMock + # Create a session first await service.create_session(project_id=1, harness_name="claude-code") # Set up harness to return a config with env - env_cfg = MagicMock(command=["claude"], env={"SESSION_VAR": "abc"}, cwd=Path("/tmp/myproj")) + env_cfg = MagicMock( + command=["claude"], env={"SESSION_VAR": "abc"}, cwd=Path("/tmp/myproj") + ) harnesses["claude-code"].build_resume_config = MagicMock(return_value=env_cfg) # Simulate window is not alive → respawn path tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error="")) + tmux.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=True, window_id=None, error="") + ) tmux.attach = AsyncMock(return_value=True) await service.attach_session(session_id=1) @@ -125,8 +140,9 @@ async def test_attach_session_passes_env(service, db, tmux, harnesses): tmux.respawn_verified.assert_called() first_call = tmux.respawn_verified.call_args_list[0] # env must be forwarded (passed as keyword or positional) - assert first_call.kwargs.get("env") == {"SESSION_VAR": "abc"} or \ - (len(first_call.args) >= 4 and first_call.args[3] == {"SESSION_VAR": "abc"}) + assert first_call.kwargs.get("env") == {"SESSION_VAR": "abc"} or ( + len(first_call.args) >= 4 and first_call.args[3] == {"SESSION_VAR": "abc"} + ) # --------------------------------------------------------------------------- @@ -135,18 +151,28 @@ async def test_attach_session_passes_env(service, db, tmux, harnesses): @pytest.mark.asyncio -async def test_attach_session_dead_resume_ok_attaches_once(service, db, tmux, harnesses): +async def test_attach_session_dead_resume_ok_attaches_once( + service, db, tmux, harnesses +): """attach_session: resume ok → attach called once, no fallback spawn attempted.""" await service.create_session(project_id=1, harness_name="claude-code") - resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")) + resume_cfg = MagicMock( + command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj") + ) harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg) harnesses["claude-code"].build_spawn_config = MagicMock( - return_value=MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")) + return_value=MagicMock( + command=["claude", "--session-id", "sess-1"], + env={}, + cwd=Path("/tmp/myproj"), + ) ) tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error="")) + tmux.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=True, window_id=None, error="") + ) tmux.attach = AsyncMock(return_value=True) result = await service.attach_session(session_id=1) @@ -160,20 +186,30 @@ async def test_attach_session_dead_resume_ok_attaches_once(service, db, tmux, ha @pytest.mark.asyncio -async def test_attach_session_dead_resume_fails_fallback_spawn(service, db, tmux, harnesses): +async def test_attach_session_dead_resume_fails_fallback_spawn( + service, db, tmux, harnesses +): """attach_session: resume dies → fallback spawn with session-id, attach called once.""" await service.create_session(project_id=1, harness_name="claude-code") - resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")) - spawn_cfg = MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")) + resume_cfg = MagicMock( + command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj") + ) + spawn_cfg = MagicMock( + command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj") + ) harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg) harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg) tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(side_effect=[ - SpawnResult(ok=False, window_id=None, error="no transcript"), # resume fails - SpawnResult(ok=True, window_id="@2", error=""), # fresh spawn ok - ]) + tmux.respawn_verified = AsyncMock( + side_effect=[ + SpawnResult( + ok=False, window_id=None, error="no transcript" + ), # resume fails + SpawnResult(ok=True, window_id="@2", error=""), # fresh spawn ok + ] + ) tmux.attach = AsyncMock(return_value=True) result = await service.attach_session(session_id=1) @@ -189,17 +225,25 @@ async def test_attach_session_dead_resume_fails_fallback_spawn(service, db, tmux @pytest.mark.asyncio -async def test_attach_session_both_rungs_fail_returns_false(service, db, tmux, harnesses): +async def test_attach_session_both_rungs_fail_returns_false( + service, db, tmux, harnesses +): """attach_session: both resume and spawn fail → False, no attach.""" await service.create_session(project_id=1, harness_name="claude-code") - resume_cfg = MagicMock(command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj")) - spawn_cfg = MagicMock(command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj")) + resume_cfg = MagicMock( + command=["claude", "--resume", "sess-1"], env={}, cwd=Path("/tmp/myproj") + ) + spawn_cfg = MagicMock( + command=["claude", "--session-id", "sess-1"], env={}, cwd=Path("/tmp/myproj") + ) harnesses["claude-code"].build_resume_config = MagicMock(return_value=resume_cfg) harnesses["claude-code"].build_spawn_config = MagicMock(return_value=spawn_cfg) tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(return_value=SpawnResult(ok=False, window_id=None, error="fatal")) + tmux.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=False, window_id=None, error="fatal") + ) tmux.attach = AsyncMock(return_value=True) result = await service.attach_session(session_id=1) @@ -219,7 +263,9 @@ def harnesses_no_capture(): h = MagicMock() h.captures_session_id = False h.generate_session_id.return_value = "sess-nc" - h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj")) + h.build_spawn_config.return_value = MagicMock( + command=["claude"], env={}, cwd=Path("/tmp/myproj") + ) return {"claude-code": h} @@ -229,12 +275,16 @@ def harnesses_capture(): h = MagicMock() h.captures_session_id = True h.generate_session_id.return_value = "placeholder-id" - h.build_spawn_config.return_value = MagicMock(command=["codex"], env={}, cwd=Path("/tmp/myproj")) + h.build_spawn_config.return_value = MagicMock( + command=["codex"], env={}, cwd=Path("/tmp/myproj") + ) return {"claude-code": h} @pytest.mark.asyncio -async def test_create_session_no_capture_flag_never_calls_capture(db, tmux, harnesses_no_capture): +async def test_create_session_no_capture_flag_never_calls_capture( + db, tmux, harnesses_no_capture +): """When captures_session_id=False, capture_session_id is never called.""" service = SessionService(db=db, tmux=tmux, harnesses=harnesses_no_capture) await service.create_session(project_id=1, harness_name="claude-code") @@ -242,11 +292,17 @@ async def test_create_session_no_capture_flag_never_calls_capture(db, tmux, harn @pytest.mark.asyncio -async def test_create_session_capture_retries_until_success(db, tmux, harnesses_capture): +async def test_create_session_capture_retries_until_success( + db, tmux, harnesses_capture +): """captures_session_id=True: returns None twice then an id → retries and stores captured id.""" import hqt.sessions.service as svc_mod - harnesses_capture["claude-code"].capture_session_id.side_effect = [None, None, "captured-id-xyz"] + harnesses_capture["claude-code"].capture_session_id.side_effect = [ + None, + None, + "captured-id-xyz", + ] sleep_calls = [] @@ -264,7 +320,9 @@ async def test_create_session_capture_retries_until_success(db, tmux, harnesses_ @pytest.mark.asyncio -async def test_create_session_capture_all_none_keeps_placeholder(db, tmux, harnesses_capture, caplog): +async def test_create_session_capture_all_none_keeps_placeholder( + db, tmux, harnesses_capture, caplog +): """All capture attempts return None → placeholder id kept, no crash, warning logged.""" import logging import hqt.sessions.service as svc_mod @@ -277,18 +335,27 @@ async def test_create_session_capture_all_none_keeps_placeholder(db, tmux, harne service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture) with caplog.at_level(logging.WARNING): with patch.object(svc_mod, "_capture_sleep", fake_sleep): - result = await service.create_session(project_id=1, harness_name="claude-code") + result = await service.create_session( + project_id=1, harness_name="claude-code" + ) assert result.session.harness_session_id == "placeholder-id" # Should have attempted up to 6 times (CAPTURE_MAX_ATTEMPTS) - assert harnesses_capture["claude-code"].capture_session_id.call_count == svc_mod.CAPTURE_MAX_ATTEMPTS + assert ( + harnesses_capture["claude-code"].capture_session_id.call_count + == svc_mod.CAPTURE_MAX_ATTEMPTS + ) # Exhaustion must emit a warning - warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] + warning_messages = [ + r.message for r in caplog.records if r.levelno == logging.WARNING + ] assert any("exhausted" in m for m in warning_messages) @pytest.mark.asyncio -async def test_create_session_capture_first_attempt_succeeds(db, tmux, harnesses_capture): +async def test_create_session_capture_first_attempt_succeeds( + db, tmux, harnesses_capture +): """captures_session_id=True: first attempt succeeds → no sleeps needed.""" import hqt.sessions.service as svc_mod @@ -316,6 +383,7 @@ async def test_create_session_capture_first_attempt_succeeds(db, tmux, harnesses async def test_list_sessions_dead_window(db, tmux, harnesses): """list_sessions: dead window → status='dead', alive=False.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) @@ -344,6 +412,7 @@ async def test_list_sessions_missing_window_is_dead(db, tmux, harnesses): async def test_list_sessions_alive_parse_working(db, tmux, harnesses): """Alive window + harness parse returns 'working' → status='working'.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) @@ -364,12 +433,15 @@ async def test_list_sessions_alive_parse_working(db, tmux, harnesses): async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesses): """Alive window + parse_status None + recent activity → status='active'.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) now = 2000.0 - tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1998)} # 2s ago + tmux.poll_info.return_value = { + "hqt-1": WindowInfo(alive=True, last_activity=1998) + } # 2s ago tmux.capture_pane = AsyncMock(return_value="some text") harnesses["claude-code"].parse_status = MagicMock(return_value=None) @@ -383,12 +455,15 @@ async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesse async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses): """Alive window + parse_status None + stale activity → status='idle'.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) now = 2000.0 - tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1990)} # 10s ago + tmux.poll_info.return_value = { + "hqt-1": WindowInfo(alive=True, last_activity=1990) + } # 10s ago tmux.capture_pane = AsyncMock(return_value="some text") harnesses["claude-code"].parse_status = MagicMock(return_value=None) @@ -402,6 +477,7 @@ async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harnesses): """capture_pane returns empty text → activity-based fallback, no crash.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) @@ -423,6 +499,7 @@ async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harn async def test_list_sessions_alive_parse_waiting(db, tmux, harnesses): """Alive window + parse_status 'waiting' → status='waiting'.""" from hqt.tmux.runner import WindowInfo + await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session( project_id=1, harness_name="claude-code" ) @@ -449,7 +526,9 @@ async def test_sync_window_labels_sets_status_symbol(db, tmux, harnesses): from hqt.tmux.runner import WindowInfo service = SessionService(db=db, tmux=tmux, harnesses=harnesses) - await service.create_session(project_id=1, harness_name="claude-code", nickname="mywork") + await service.create_session( + project_id=1, harness_name="claude-code", nickname="mywork" + ) now = 2000.0 tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)} @@ -471,7 +550,9 @@ async def test_sync_window_labels_dead_window_uses_dead_glyph(db, tmux, harnesse from hqt.tmux.runner import WindowInfo service = SessionService(db=db, tmux=tmux, harnesses=harnesses) - await service.create_session(project_id=1, harness_name="claude-code") # no nickname + await service.create_session( + project_id=1, harness_name="claude-code" + ) # no nickname tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)} tmux.set_window_label = AsyncMock() @@ -533,16 +614,22 @@ async def test_list_sessions_real_configurator_waiting(db): # spec=TmuxManager: only methods that exist on TmuxManager are accessible. # capture_pane MUST be on TmuxManager (via the delegate) for the spec to allow it. tmux_spec = MagicMock(spec=TmuxManager) - tmux_spec.spawn = AsyncMock(return_value=SpawnResult(ok=True, window_id="@1", error="")) + tmux_spec.spawn = AsyncMock( + return_value=SpawnResult(ok=True, window_id="@1", error="") + ) tmux_spec.kill = AsyncMock() tmux_spec.is_alive = AsyncMock(return_value=True) tmux_spec.poll_info = AsyncMock(return_value={}) # capture_pane returns realistic "waiting" pane text; the spec must allow this # attribute — fails with AttributeError before TmuxManager.capture_pane delegate exists. - tmux_spec.capture_pane = AsyncMock(return_value="Do you want to proceed?\n❯ 1. Yes\n") + tmux_spec.capture_pane = AsyncMock( + return_value="Do you want to proceed?\n❯ 1. Yes\n" + ) tmux_spec.attach = AsyncMock() tmux_spec.window_exists = AsyncMock(return_value=True) - tmux_spec.respawn_verified = AsyncMock(return_value=SpawnResult(ok=True, window_id=None, error="")) + tmux_spec.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=True, window_id=None, error="") + ) configurator = ClaudeConfigurator() harnesses = {"claude-code": configurator} @@ -553,7 +640,9 @@ async def test_list_sessions_real_configurator_waiting(db): now = 2000.0 window_name = "hqt-1" - tmux_spec.poll_info.return_value = {window_name: WindowInfo(alive=True, last_activity=1999)} + tmux_spec.poll_info.return_value = { + window_name: WindowInfo(alive=True, last_activity=1999) + } result = await service.list_sessions(project_id=1, now=now) @@ -615,17 +704,25 @@ def harnesses_capture_fallback(): h.name = "claude-code" h.captures_session_id = True h.generate_session_id.return_value = "placeholder-id" - h.build_spawn_config.return_value = MagicMock(command=["codex"], env={}, cwd=Path("/tmp/myproj")) - h.build_resume_config.return_value = MagicMock(command=["codex", "resume", "placeholder-id"], env={}, cwd=Path("/tmp/myproj")) + h.build_spawn_config.return_value = MagicMock( + command=["codex"], env={}, cwd=Path("/tmp/myproj") + ) + h.build_resume_config.return_value = MagicMock( + command=["codex", "resume", "placeholder-id"], env={}, cwd=Path("/tmp/myproj") + ) return {"claude-code": h} @pytest.mark.asyncio -async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harnesses_capture_fallback): +async def test_respawn_fallback_rung2_updates_harness_session_id( + db, tmux, harnesses_capture_fallback +): """Rung-2 success + captures_session_id=True + capture returns new id → sess.harness_session_id updated and committed.""" import hqt.sessions.service as svc_mod - harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "new-codex-id" + harnesses_capture_fallback[ + "claude-code" + ].capture_session_id.return_value = "new-codex-id" async def fake_sleep(t): pass @@ -638,13 +735,19 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harne # Simulate rung-1 resume fails, rung-2 fresh spawn succeeds harnesses_capture_fallback["claude-code"].capture_session_id.reset_mock() - harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "new-codex-id-after-rung2" + harnesses_capture_fallback[ + "claude-code" + ].capture_session_id.return_value = "new-codex-id-after-rung2" tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(side_effect=[ - SpawnResult(ok=False, window_id=None, error="no transcript"), # rung-1 resume fails - SpawnResult(ok=True, window_id="@2", error=""), # rung-2 fresh spawn ok - ]) + tmux.respawn_verified = AsyncMock( + side_effect=[ + SpawnResult( + ok=False, window_id=None, error="no transcript" + ), # rung-1 resume fails + SpawnResult(ok=True, window_id="@2", error=""), # rung-2 fresh spawn ok + ] + ) tmux.attach = AsyncMock(return_value=True) with patch.object(svc_mod, "_capture_sleep", fake_sleep): @@ -653,17 +756,61 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(db, tmux, harne assert ok is True # harness_session_id must now be the newly captured id from hqt.db.models import Session as DBSession + updated = db.get(DBSession, sess_id) assert updated.harness_session_id == "new-codex-id-after-rung2" @pytest.mark.asyncio -async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, harnesses_capture_fallback, caplog): +async def test_attach_late_capture_updates_id_before_resume( + db, tmux, harnesses_capture_fallback +): + """If create-time Codex capture missed, attach captures the real id before resuming.""" + import hqt.sessions.service as svc_mod + + harness = harnesses_capture_fallback["claude-code"] + harness.capture_session_id.side_effect = [None] * svc_mod.CAPTURE_MAX_ATTEMPTS + [ + "late-codex-id" + ] + + async def fake_sleep(t): + pass + + service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture_fallback) + with patch.object(svc_mod, "_capture_sleep", fake_sleep): + result = await service.create_session(project_id=1, harness_name="claude-code") + + assert result.session.harness_session_id == "placeholder-id" + + tmux.is_alive = AsyncMock(return_value=False) + tmux.respawn_verified = AsyncMock( + return_value=SpawnResult(ok=True, window_id="@2", error="") + ) + tmux.attach = AsyncMock(return_value=True) + + ok = await service.attach_session(session_id=result.session.id) + + assert ok is True + harness.build_resume_config.assert_called_with( + Path("/tmp/myproj"), "late-codex-id", result.session.model + ) + from hqt.db.models import Session as DBSession + + updated = db.get(DBSession, result.session.id) + assert updated.harness_session_id == "late-codex-id" + + +@pytest.mark.asyncio +async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id( + db, tmux, harnesses_capture_fallback, caplog +): """Rung-2 success + capture exhausted → old id kept + warning logged.""" import logging import hqt.sessions.service as svc_mod - harnesses_capture_fallback["claude-code"].capture_session_id.return_value = "placeholder-id" + harnesses_capture_fallback[ + "claude-code" + ].capture_session_id.return_value = "placeholder-id" async def fake_sleep(t): pass @@ -680,10 +827,12 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, h harnesses_capture_fallback["claude-code"].capture_session_id.return_value = None tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(side_effect=[ - SpawnResult(ok=False, window_id=None, error="no transcript"), - SpawnResult(ok=True, window_id="@2", error=""), - ]) + tmux.respawn_verified = AsyncMock( + side_effect=[ + SpawnResult(ok=False, window_id=None, error="no transcript"), + SpawnResult(ok=True, window_id="@2", error=""), + ] + ) tmux.attach = AsyncMock(return_value=True) with caplog.at_level(logging.WARNING): @@ -692,11 +841,14 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(db, tmux, h assert ok is True from hqt.db.models import Session as DBSession + updated = db.get(DBSession, sess_id) # Old id preserved assert updated.harness_session_id == old_id # Warning emitted - warning_messages = [r.message for r in caplog.records if r.levelno == logging.WARNING] + warning_messages = [ + r.message for r in caplog.records if r.levelno == logging.WARNING + ] assert any("exhausted" in m for m in warning_messages) @@ -707,8 +859,12 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux): h.name = "claude-code" h.captures_session_id = False h.generate_session_id.return_value = "sess-nc" - h.build_spawn_config.return_value = MagicMock(command=["claude"], env={}, cwd=Path("/tmp/myproj")) - h.build_resume_config.return_value = MagicMock(command=["claude", "--resume", "sess-nc"], env={}, cwd=Path("/tmp/myproj")) + h.build_spawn_config.return_value = MagicMock( + command=["claude"], env={}, cwd=Path("/tmp/myproj") + ) + h.build_resume_config.return_value = MagicMock( + command=["claude", "--resume", "sess-nc"], env={}, cwd=Path("/tmp/myproj") + ) harnesses_nc = {"claude-code": h} service = SessionService(db=db, tmux=tmux, harnesses=harnesses_nc) @@ -718,10 +874,12 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux): h.capture_session_id.reset_mock() tmux.is_alive = AsyncMock(return_value=False) - tmux.respawn_verified = AsyncMock(side_effect=[ - SpawnResult(ok=False, window_id=None, error="no transcript"), - SpawnResult(ok=True, window_id="@2", error=""), - ]) + tmux.respawn_verified = AsyncMock( + side_effect=[ + SpawnResult(ok=False, window_id=None, error="no transcript"), + SpawnResult(ok=True, window_id="@2", error=""), + ] + ) tmux.attach = AsyncMock(return_value=True) await service.attach_session(session_id=sess_id) @@ -739,7 +897,9 @@ async def test_create_session_no_capture_when_spawn_fails(db, tmux, harnesses_ca """create_session: spawn fails → capture_session_id must NOT be called.""" import hqt.sessions.service as svc_mod - tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="spawn failed") + tmux.spawn.return_value = SpawnResult( + ok=False, window_id=None, error="spawn failed" + ) async def fake_sleep(t): pass @@ -763,7 +923,9 @@ async def test_create_session_returns_result_dataclass(service, db, tmux, harnes """create_session returns a CreateSessionResult with spawn_ok=True on success.""" from hqt.sessions.service import CreateSessionResult - result = await service.create_session(project_id=1, harness_name="claude-code", nickname="test") + result = await service.create_session( + project_id=1, harness_name="claude-code", nickname="test" + ) assert isinstance(result, CreateSessionResult) assert result.spawn_ok is True assert result.spawn_error == "" @@ -775,7 +937,9 @@ async def test_create_session_result_propagates_spawn_failure(db, tmux, harnesse """create_session result reflects spawn failure.""" from hqt.sessions.service import CreateSessionResult - tmux.spawn.return_value = SpawnResult(ok=False, window_id=None, error="command not found: claude") + tmux.spawn.return_value = SpawnResult( + ok=False, window_id=None, error="command not found: claude" + ) service = SessionService(db=db, tmux=tmux, harnesses=harnesses) result = await service.create_session(project_id=1, harness_name="claude-code") diff --git a/tests/test_skills.py b/tests/test_skills.py index b220315..5ad0e82 100644 --- a/tests/test_skills.py +++ b/tests/test_skills.py @@ -6,7 +6,9 @@ from hqt.skills.registry import SkillRegistry def test_discover_skills(tmp_path: Path): skill_dir = tmp_path / "my-skill" skill_dir.mkdir() - (skill_dir / "SKILL.md").write_text("---\nname: my-skill\ndescription: A test skill\n---\nSkill content here") + (skill_dir / "SKILL.md").write_text( + "---\nname: my-skill\ndescription: A test skill\n---\nSkill content here" + ) registry = SkillRegistry(tmp_path) skills = registry.discover() diff --git a/tests/test_tmux.py b/tests/test_tmux.py index 02f9fa7..2b48878 100644 --- a/tests/test_tmux.py +++ b/tests/test_tmux.py @@ -15,10 +15,14 @@ def runner(): @pytest.mark.asyncio async def test_new_window(runner): runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows -F '#{window_index}' (next-index query) + (0, "0\n", ""), # list-windows -F '#{window_index}' (next-index query) (0, "@1\n", ""), # new-window -P -F - (0, "", ""), # set-option (combined: remain-on-exit + allow-rename + automatic-rename) - (0, "", ""), # respawn-pane + ( + 0, + "", + "", + ), # set-option (combined: remain-on-exit + allow-rename + automatic-rename) + (0, "", ""), # respawn-pane ] result = await runner.new_window("hqt-1", "/tmp", "kiro-cli chat") assert result == "@1" @@ -26,9 +30,23 @@ async def test_new_window(runner): # options, then the per-window Frappé theme options (see new_window). set_call = runner._exec.call_args_list[2].args assert set_call[:17] == ( - "set-option", "-t", "hqt-main:=hqt-1", "remain-on-exit", "on", - ";", "set-option", "-t", "hqt-main:=hqt-1", "allow-rename", "off", - ";", "set-option", "-t", "hqt-main:=hqt-1", "automatic-rename", "off", + "set-option", + "-t", + "hqt-main:=hqt-1", + "remain-on-exit", + "on", + ";", + "set-option", + "-t", + "hqt-main:=hqt-1", + "allow-rename", + "off", + ";", + "set-option", + "-t", + "hqt-main:=hqt-1", + "automatic-rename", + "off", ) assert "window-status-current-style" in set_call # respawn-pane runs the actual command @@ -51,9 +69,9 @@ async def test_new_window_targets_explicit_free_index(runner): and targeting `session:idx` sidesteps the resolution entirely.""" runner._exec.side_effect = [ (0, "0\n1\n2\n", ""), # list-windows -F '#{window_index}' (existing: 0,1,2) - (0, "@9\n", ""), # new-window -P -F -> window_id - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "@9\n", ""), # new-window -P -F -> window_id + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] window_id = await runner.new_window("hqt-9", "/tmp", "claude") assert window_id == "@9" @@ -61,7 +79,9 @@ async def test_new_window_targets_explicit_free_index(runner): new_win_args = runner._exec.call_args_list[1].args assert "new-window" in new_win_args # Explicit free index = max(0,1,2)+1 = 3, targeted as ":3". - assert "hqt-main:3" in new_win_args, f"expected explicit free index target, got {new_win_args}" + assert "hqt-main:3" in new_win_args, ( + f"expected explicit free index target, got {new_win_args}" + ) # Must NOT use the bare session name as the new-window target. ti = new_win_args.index("-t") assert new_win_args[ti + 1] == "hqt-main:3" @@ -102,7 +122,13 @@ async def test_select_window(runner): async def test_respawn_pane(runner): await runner.respawn_pane("hqt-1", "kiro-cli chat", "/tmp") runner._exec.assert_called_with( - "respawn-pane", "-t", "hqt-main:=hqt-1", "-k", "-c", "/tmp", "kiro-cli chat", + "respawn-pane", + "-t", + "hqt-main:=hqt-1", + "-k", + "-c", + "/tmp", + "kiro-cli chat", ) @@ -116,10 +142,10 @@ async def test_list_windows(runner): @pytest.mark.asyncio async def test_manager_spawn(runner): runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@3\n", ""), # new-window -P -F - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "0\n", ""), # list-windows (next-index query) + (0, "@3\n", ""), # new-window -P -F + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] runner.verify_window_alive = AsyncMock(return_value=(True, "")) mgr = TmuxManager(runner) @@ -150,7 +176,7 @@ async def test_manager_is_alive_true(runner): # is_pane_dead needs list-panes to return "0" runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window - (0, "0\n", ""), # is_pane_dead + (0, "0\n", ""), # is_pane_dead ] mgr = TmuxManager(runner) assert await mgr.is_alive("hqt-1") is True @@ -161,7 +187,7 @@ async def test_manager_is_alive_false_dead_pane(runner): """Window exists but pane dead.""" runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window - (0, "1\n", ""), # is_pane_dead + (0, "1\n", ""), # is_pane_dead ] mgr = TmuxManager(runner) assert await mgr.is_alive("hqt-1") is False @@ -199,10 +225,10 @@ async def test_has_window_false_single_exec(runner): async def test_new_window_race_free_sequence(runner): """new_window issues: new-window (no command), set-option, respawn-pane — in order.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@5\n", ""), # new-window -P -F -> window_id - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "0\n", ""), # list-windows (next-index query) + (0, "@5\n", ""), # new-window -P -F -> window_id + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] window_id = await runner.new_window("hqt-1", "/tmp", "kiro-cli chat") assert window_id == "@5" @@ -226,9 +252,23 @@ async def test_new_window_race_free_sequence(runner): set_args = calls[2].args assert set_args[:17] == ( - "set-option", "-t", "hqt-main:=hqt-1", "remain-on-exit", "on", - ";", "set-option", "-t", "hqt-main:=hqt-1", "allow-rename", "off", - ";", "set-option", "-t", "hqt-main:=hqt-1", "automatic-rename", "off", + "set-option", + "-t", + "hqt-main:=hqt-1", + "remain-on-exit", + "on", + ";", + "set-option", + "-t", + "hqt-main:=hqt-1", + "allow-rename", + "off", + ";", + "set-option", + "-t", + "hqt-main:=hqt-1", + "automatic-rename", + "off", ) # Window-scoped theme options painted on this window, current-style = peach. assert "window-status-current-style" in set_args @@ -249,12 +289,14 @@ async def test_new_window_race_free_sequence(runner): async def test_new_window_with_env(runner): """new_window passes -e KEY=VALUE flags on respawn-pane (step 3), NOT on new-window (step 1).""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) + (0, "0\n", ""), # list-windows (next-index query) (0, "@7\n", ""), (0, "", ""), (0, "", ""), ] - window_id = await runner.new_window("hqt-2", "/tmp", "claude", env={"FOO": "bar", "X": "1"}) + window_id = await runner.new_window( + "hqt-2", "/tmp", "claude", env={"FOO": "bar", "X": "1"} + ) assert window_id == "@7" calls_list = runner._exec.call_args_list @@ -278,10 +320,10 @@ async def test_new_window_with_env(runner): async def test_new_window_set_option_failure_aborts(runner): """If set-option fails, new_window kills the orphaned window and returns None.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@3\n", ""), # new-window succeeds + (0, "0\n", ""), # list-windows (next-index query) + (0, "@3\n", ""), # new-window succeeds (1, "", "some tmux error"), # set-option fails - (0, "", ""), # kill-window cleanup + (0, "", ""), # kill-window cleanup ] result = await runner.new_window("hqt-3", "/tmp", "claude") assert result is None @@ -296,11 +338,11 @@ async def test_new_window_set_option_failure_aborts(runner): async def test_new_window_respawn_failure_kills_window(runner): """If respawn-pane fails, new_window kills the orphaned window and returns None.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@4\n", ""), # new-window succeeds - (0, "", ""), # set-option succeeds - (1, "", "respawn error"), # respawn-pane fails - (0, "", ""), # kill-window cleanup + (0, "0\n", ""), # list-windows (next-index query) + (0, "@4\n", ""), # new-window succeeds + (0, "", ""), # set-option succeeds + (1, "", "respawn error"), # respawn-pane fails + (0, "", ""), # kill-window cleanup ] result = await runner.new_window("hqt-4", "/tmp", "claude") assert result is None @@ -314,7 +356,7 @@ async def test_new_window_respawn_failure_kills_window(runner): async def test_new_window_returns_window_id(runner): """new_window returns the window_id string on full success.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) + (0, "0\n", ""), # list-windows (next-index query) (0, "@42\n", ""), (0, "", ""), (0, "", ""), @@ -327,7 +369,7 @@ async def test_new_window_returns_window_id(runner): async def test_new_window_returns_none_on_new_window_failure(runner): """new_window returns None when the initial new-window command fails.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) + (0, "0\n", ""), # list-windows (next-index query) (1, "", "session not found"), # new-window fails ] result = await runner.new_window("hqt-x", "/tmp", "sleep 1") @@ -354,8 +396,8 @@ async def test_verify_window_alive_alive(runner): async def test_verify_window_alive_dead_with_text(runner): """Pane dies within timeout → (False, captured text).""" runner._exec.side_effect = [ - (0, "1\n", ""), # is_pane_dead → dead - (0, "command not found\n", ""), # capture_pane + (0, "1\n", ""), # is_pane_dead → dead + (0, "command not found\n", ""), # capture_pane ] ok, text = await runner.verify_window_alive("hqt-1", timeout=0.5, interval=0.05) assert ok is False @@ -366,8 +408,8 @@ async def test_verify_window_alive_dead_with_text(runner): async def test_verify_window_alive_dead_empty_text(runner): """Pane dies with empty capture → (False, '') still False.""" runner._exec.side_effect = [ - (0, "1\n", ""), # is_pane_dead - (0, "", ""), # capture_pane empty + (0, "1\n", ""), # is_pane_dead + (0, "", ""), # capture_pane empty ] ok, text = await runner.verify_window_alive("hqt-1", timeout=0.5, interval=0.05) assert ok is False @@ -379,9 +421,9 @@ async def test_verify_window_alive_dies_during_last_interval(runner): """Pane alive throughout polling loop but dies during the final sleep → (False, text).""" # All in-loop polls return alive (0), final post-loop check returns dead (1) runner._exec.side_effect = [ - (0, "0\n", ""), # is_pane_dead (in-loop) → alive - (0, "1\n", ""), # is_pane_dead (post-loop final check) → dead - (0, "died at last\n", ""), # capture_pane + (0, "0\n", ""), # is_pane_dead (in-loop) → alive + (0, "1\n", ""), # is_pane_dead (post-loop final check) → dead + (0, "died at last\n", ""), # capture_pane ] ok, text = await runner.verify_window_alive("hqt-1", timeout=0.05, interval=0.1) assert ok is False @@ -397,10 +439,10 @@ async def test_verify_window_alive_dies_during_last_interval(runner): async def test_spawn_result_ok(runner): """spawn returns SpawnResult(ok=True, window_id=...) on success.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@10\n", ""), # new-window - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "0\n", ""), # list-windows (next-index query) + (0, "@10\n", ""), # new-window + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] runner.verify_window_alive = AsyncMock(return_value=(True, "")) mgr = TmuxManager(runner) @@ -416,10 +458,10 @@ async def test_spawn_result_ok(runner): async def test_spawn_result_failure_verification(runner): """spawn returns SpawnResult(ok=False, error=...) when pane dies immediately.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@11\n", ""), # new-window - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "0\n", ""), # list-windows (next-index query) + (0, "@11\n", ""), # new-window + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] runner.verify_window_alive = AsyncMock(return_value=(False, "no such binary\n")) mgr = TmuxManager(runner) @@ -433,14 +475,16 @@ async def test_spawn_result_failure_verification(runner): async def test_spawn_shlex_join(runner): """spawn uses shlex.join so args with spaces are properly quoted.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) + (0, "0\n", ""), # list-windows (next-index query) (0, "@99\n", ""), (0, "", ""), (0, "", ""), ] runner.verify_window_alive = AsyncMock(return_value=(True, "")) mgr = TmuxManager(runner) - req = SpawnRequest(window_name="hqt-1", command=["my cmd", "arg with space"], cwd="/tmp") + req = SpawnRequest( + window_name="hqt-1", command=["my cmd", "arg with space"], cwd="/tmp" + ) await mgr.spawn(req) # The respawn-pane call (now the 4th _exec call) should have the shlex-joined string @@ -449,6 +493,7 @@ async def test_spawn_shlex_join(runner): joined_cmd = respawn_call[-1] # shlex.join of ["my cmd", "arg with space"] → "'my cmd' 'arg with space'" import shlex + expected = shlex.join(["my cmd", "arg with space"]) assert joined_cmd == expected @@ -457,14 +502,16 @@ async def test_spawn_shlex_join(runner): async def test_spawn_passes_env(runner): """spawn passes env dict through to respawn-pane (step 3), NOT to new-window (step 1).""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) + (0, "0\n", ""), # list-windows (next-index query) (0, "@20\n", ""), (0, "", ""), (0, "", ""), ] runner.verify_window_alive = AsyncMock(return_value=(True, "")) mgr = TmuxManager(runner) - req = SpawnRequest(window_name="hqt-env", command=["bash"], cwd="/tmp", env={"MY_VAR": "hello"}) + req = SpawnRequest( + window_name="hqt-env", command=["bash"], cwd="/tmp", env={"MY_VAR": "hello"} + ) result = await mgr.spawn(req) assert result.ok is True @@ -480,11 +527,12 @@ async def test_spawn_passes_env(runner): assert "MY_VAR=hello" in respawn_args - @pytest.mark.asyncio async def test_respawn_pane_with_env(runner): """respawn_pane passes -e KEY=VALUE flags when env is provided.""" - await runner.respawn_pane("hqt-1", "claude", "/tmp", env={"MYKEY": "myval", "A": "b"}) + await runner.respawn_pane( + "hqt-1", "claude", "/tmp", env={"MYKEY": "myval", "A": "b"} + ) call_args = runner._exec.call_args_list[0].args assert "respawn-pane" in call_args assert "-e" in call_args @@ -501,15 +549,14 @@ async def test_respawn_pane_no_env(runner): assert "-e" not in call_args - @pytest.mark.asyncio async def test_spawn_verify_dead_empty_text_returns_fallback_error(runner): """When verify_window_alive returns (False, ''), spawn returns the fallback error message.""" runner._exec.side_effect = [ - (0, "0\n", ""), # list-windows (next-index query) - (0, "@30\n", ""), # new-window - (0, "", ""), # set-option - (0, "", ""), # respawn-pane + (0, "0\n", ""), # list-windows (next-index query) + (0, "@30\n", ""), # new-window + (0, "", ""), # set-option + (0, "", ""), # respawn-pane ] # Empty pane text — verify_window_alive signals immediate death with no output runner.verify_window_alive = AsyncMock(return_value=(False, "")) @@ -531,7 +578,7 @@ async def test_respawn_verified_success_path(runner): runner.verify_window_alive = AsyncMock(return_value=(True, "")) runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window → True - (0, "", ""), # respawn-pane + (0, "", ""), # respawn-pane ] mgr = TmuxManager(runner) result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp") @@ -545,7 +592,7 @@ async def test_respawn_verified_pane_dies_after_respawn(runner): runner.verify_window_alive = AsyncMock(return_value=(False, "no such transcript\n")) runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window → True - (0, "", ""), # respawn-pane + (0, "", ""), # respawn-pane ] mgr = TmuxManager(runner) result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp") @@ -559,7 +606,7 @@ async def test_respawn_verified_pane_dies_empty_text_fallback_message(runner): runner.verify_window_alive = AsyncMock(return_value=(False, "")) runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window → True - (0, "", ""), # respawn-pane + (0, "", ""), # respawn-pane ] mgr = TmuxManager(runner) result = await mgr.respawn_verified("hqt-1", ["claude", "--resume", "abc"], "/tmp") @@ -572,14 +619,16 @@ async def test_respawn_verified_window_gone_branch(runner): """respawn_verified: window is gone → new_window + verify → SpawnResult(ok=True, window_id=...).""" runner.verify_window_alive = AsyncMock(return_value=(True, "")) runner._exec.side_effect = [ - (0, "other\n", ""), # has_window → False - (0, "0\n", ""), # new_window: list-windows (next-index query) - (0, "@9\n", ""), # new-window step 1 - (0, "", ""), # set-option step 2 - (0, "", ""), # respawn-pane step 3 + (0, "other\n", ""), # has_window → False + (0, "0\n", ""), # new_window: list-windows (next-index query) + (0, "@9\n", ""), # new-window step 1 + (0, "", ""), # set-option step 2 + (0, "", ""), # respawn-pane step 3 ] mgr = TmuxManager(runner) - result = await mgr.respawn_verified("hqt-gone", ["claude", "--resume", "abc"], "/tmp") + result = await mgr.respawn_verified( + "hqt-gone", ["claude", "--resume", "abc"], "/tmp" + ) assert result.ok is True assert result.window_id == "@9" @@ -588,12 +637,14 @@ async def test_respawn_verified_window_gone_branch(runner): async def test_respawn_verified_window_gone_new_window_fails(runner): """respawn_verified: window gone, new_window fails → SpawnResult(ok=False).""" runner._exec.side_effect = [ - (0, "other\n", ""), # has_window → False - (0, "0\n", ""), # new_window: list-windows (next-index query) - (1, "", "session not found"), # new-window fails + (0, "other\n", ""), # has_window → False + (0, "0\n", ""), # new_window: list-windows (next-index query) + (1, "", "session not found"), # new-window fails ] mgr = TmuxManager(runner) - result = await mgr.respawn_verified("hqt-gone", ["claude", "--resume", "abc"], "/tmp") + result = await mgr.respawn_verified( + "hqt-gone", ["claude", "--resume", "abc"], "/tmp" + ) assert result.ok is False assert result.error != "" @@ -604,7 +655,7 @@ async def test_respawn_verified_forwards_env(runner): runner.verify_window_alive = AsyncMock(return_value=(True, "")) runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window → True - (0, "", ""), # respawn-pane + (0, "", ""), # respawn-pane ] mgr = TmuxManager(runner) result = await mgr.respawn_verified( @@ -621,10 +672,11 @@ async def test_respawn_verified_forwards_env(runner): async def test_respawn_verified_uses_shlex_join(runner): """respawn_verified uses shlex.join so args with spaces are properly quoted.""" import shlex + runner.verify_window_alive = AsyncMock(return_value=(True, "")) runner._exec.side_effect = [ (0, "hqt-1\n", ""), # has_window → True - (0, "", ""), # respawn-pane + (0, "", ""), # respawn-pane ] mgr = TmuxManager(runner) await mgr.respawn_verified("hqt-1", ["cmd with space", "arg"], "/tmp") @@ -641,13 +693,15 @@ async def test_respawn_verified_window_gone_forwards_env(runner): runner.verify_window_alive = AsyncMock(return_value=(True, "")) runner._exec.side_effect = [ (0, "other-window\n", ""), # has_window → False - (0, "0\n", ""), # new_window: list-windows (next-index query) - (0, "@5\n", ""), # new-window step 1 - (0, "", ""), # set-option step 2 - (0, "", ""), # respawn-pane step 3 + (0, "0\n", ""), # new_window: list-windows (next-index query) + (0, "@5\n", ""), # new-window step 1 + (0, "", ""), # set-option step 2 + (0, "", ""), # respawn-pane step 3 ] mgr = TmuxManager(runner) - result = await mgr.respawn_verified("hqt-gone", ["bash"], "/tmp", env={"GONE_VAR": "x"}) + result = await mgr.respawn_verified( + "hqt-gone", ["bash"], "/tmp", env={"GONE_VAR": "x"} + ) assert result.ok is True # The respawn-pane (step 3 of new_window) should carry the env @@ -667,7 +721,12 @@ async def test_send_text_uses_literal_flag(runner): """send_text must pass -l and '--' before the text.""" await runner.send_text("hqt-1", "Enter; echo hi C-c") runner._exec.assert_called_once_with( - "send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "Enter; echo hi C-c", + "send-keys", + "-t", + "hqt-main:=hqt-1", + "-l", + "--", + "Enter; echo hi C-c", ) @@ -676,7 +735,10 @@ async def test_send_enter_no_literal_flag(runner): """send_enter intentionally sends the Enter key name — must NOT have -l.""" await runner.send_enter("hqt-1") runner._exec.assert_called_once_with( - "send-keys", "-t", "hqt-main:=hqt-1", "Enter", + "send-keys", + "-t", + "hqt-main:=hqt-1", + "Enter", ) @@ -690,11 +752,16 @@ async def test_list_windows_info_alive_and_activity(runner): """list_windows_info parses 3-column output; alive=True when pane_dead=0.""" runner._exec.return_value = (0, "hqt-1\t0\t1000000\nhqt-2\t1\t999999\n", "") from hqt.tmux.runner import WindowInfo + result = await runner.list_windows_info() assert result["hqt-1"] == WindowInfo(alive=True, last_activity=1000000) assert result["hqt-2"] == WindowInfo(alive=False, last_activity=999999) runner._exec.assert_called_once_with( - "list-panes", "-s", "-t", "hqt-main", "-F", + "list-panes", + "-s", + "-t", + "hqt-main", + "-F", "#{window_name}\t#{pane_dead}\t#{window_activity}", ) @@ -708,6 +775,7 @@ async def test_list_windows_info_multi_pane_alive_max_activity(runner): "", ) from hqt.tmux.runner import WindowInfo + result = await runner.list_windows_info() assert result["hqt-1"] == WindowInfo(alive=True, last_activity=2000) assert result["hqt-2"] == WindowInfo(alive=False, last_activity=600) @@ -747,7 +815,12 @@ async def test_send_text_includes_double_dash(runner): """send_text must pass '--' before the text to prevent flag parsing.""" await runner.send_text("hqt-1", "-flag-looking text") runner._exec.assert_called_once_with( - "send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "-flag-looking text", + "send-keys", + "-t", + "hqt-main:=hqt-1", + "-l", + "--", + "-flag-looking text", ) @@ -756,7 +829,12 @@ async def test_send_text_double_dash_with_normal_text(runner): """send_text includes '--' even for normal text (belt-and-suspenders).""" await runner.send_text("hqt-1", "hello world") runner._exec.assert_called_once_with( - "send-keys", "-t", "hqt-main:=hqt-1", "-l", "--", "hello world", + "send-keys", + "-t", + "hqt-main:=hqt-1", + "-l", + "--", + "hello world", ) @@ -765,6 +843,7 @@ async def test_poll_info_single_exec(runner): """poll_info calls _exec exactly once and returns WindowInfo per name.""" runner._exec.return_value = (0, "hqt-1\t0\t1234\nhqt-2\t1\t5678\n", "") from hqt.tmux.runner import WindowInfo + mgr = TmuxManager(runner) result = await mgr.poll_info(["hqt-1", "hqt-2", "hqt-missing"]) assert runner._exec.call_count == 1 @@ -881,7 +960,8 @@ async def test_apply_theme_paints_window_options_on_every_window(runner): for window in ("⌂ HQT", "hqt-1", "hqt-2"): target = f"hqt-main:={window}" idxs = [ - i for i, a in enumerate(args) + i + for i, a in enumerate(args) if a == "window-status-current-style" and args[i - 1] == target ] assert idxs, f"no current-style set for {window}" diff --git a/tests/test_tui.py b/tests/test_tui.py index ab54f69..d60ff2f 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -17,7 +17,7 @@ def mock_services(tmp_path, monkeypatch): @pytest.mark.asyncio async def test_app_mounts(): app = HqtApp() - async with app.run_test() as pilot: + async with app.run_test(): assert app.title == "HQ Terminal" @@ -27,16 +27,17 @@ async def test_no_header_bar(): from textual.widgets import Header app = HqtApp() - async with app.run_test() as pilot: + async with app.run_test(): assert len(app.query(Header)) == 0 @pytest.mark.asyncio async def test_app_has_project_and_session_panels(): app = HqtApp() - async with app.run_test() as pilot: + async with app.run_test(): from hqt.tui.widgets.project_list import ProjectList from hqt.tui.widgets.session_list import SessionList + assert app.query_one(ProjectList) is not None assert app.query_one(SessionList) is not None @@ -45,7 +46,7 @@ async def test_app_has_project_and_session_panels(): async def test_session_refresh_no_duplicate_ids(): """Regression test: rapid session refreshes must not crash with DuplicateIds.""" app = HqtApp() - async with app.run_test(size=(120, 40)) as pilot: + async with app.run_test(size=(120, 40)): from hqt.db.models import Project, Harness, Session proj = Project(name="test", path="/tmp/test") @@ -116,9 +117,12 @@ async def test_session_selection_preserved_across_refresh(): # --------------------------------------------------------------------------- -def _make_session_info(session_id: int, name: str, harness_name: str, status: str, alive: bool): +def _make_session_info( + session_id: int, name: str, harness_name: str, status: str, alive: bool +): """Build a minimal SessionInfo-like object for widget testing.""" from hqt.sessions.service import SessionInfo + harness = MagicMock() harness.name = harness_name session = MagicMock() @@ -150,13 +154,23 @@ async def test_session_list_label_includes_status(): label_texts = [str(child.query_one(Label).render()) for child in lv.children] # Every label must contain the status word - assert any("working" in t for t in label_texts), f"Expected 'working' in labels: {label_texts}" - assert any("idle" in t for t in label_texts), f"Expected 'idle' in labels: {label_texts}" - assert any("dead" in t for t in label_texts), f"Expected 'dead' in labels: {label_texts}" + assert any("working" in t for t in label_texts), ( + f"Expected 'working' in labels: {label_texts}" + ) + assert any("idle" in t for t in label_texts), ( + f"Expected 'idle' in labels: {label_texts}" + ) + assert any("dead" in t for t in label_texts), ( + f"Expected 'dead' in labels: {label_texts}" + ) # Symbols: working → ◐, idle → ●, dead → ○ - assert any("◐" in t for t in label_texts), f"Expected '◐' in labels: {label_texts}" - assert any("○" in t for t in label_texts), f"Expected '○' in labels: {label_texts}" + assert any("◐" in t for t in label_texts), ( + f"Expected '◐' in labels: {label_texts}" + ) + assert any("○" in t for t in label_texts), ( + f"Expected '○' in labels: {label_texts}" + ) @pytest.mark.asyncio @@ -270,7 +284,10 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): await asyncio.sleep(0.05) call_order.append("create") from hqt.sessions.service import CreateSessionResult - return CreateSessionResult(session=MagicMock(), spawn_ok=True, spawn_error="") + + return CreateSessionResult( + session=MagicMock(), spawn_ok=True, spawn_error="" + ) async def fake_list(project_id): call_order.append("list") @@ -284,6 +301,7 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): # Patch discover_harnesses so the early-return guard doesn't fire. from unittest.mock import patch + with patch("hqt.tui.app.discover_harnesses", return_value={"claude": object()}): app.action_new_session() await pilot.pause() @@ -296,7 +314,9 @@ async def test_new_session_on_dismiss_sequential_create_then_refresh(): await app.workers.wait_for_complete() await pilot.pause() - assert call_order == ["create", "list"], f"Expected create before list, got: {call_order}" + assert call_order == ["create", "list"], ( + f"Expected create before list, got: {call_order}" + ) # --------------------------------------------------------------------------- @@ -360,11 +380,17 @@ async def test_new_session_on_dismiss_notifies_on_spawn_failure(): await pilot.pause() # Should have notified with severity="error" - error_notifs = [(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"] - assert len(error_notifs) >= 1, f"Expected error notification, got: {notify_calls}" + error_notifs = [ + (m, kw) for m, kw in notify_calls if kw.get("severity") == "error" + ] + assert len(error_notifs) >= 1, ( + f"Expected error notification, got: {notify_calls}" + ) # Message should contain the first line of the error msg_text = error_notifs[0][0] - assert "command not found" in msg_text or "claude" in msg_text, f"Unexpected message: {msg_text}" + assert "command not found" in msg_text or "claude" in msg_text, ( + f"Unexpected message: {msg_text}" + ) @pytest.mark.asyncio @@ -394,7 +420,9 @@ async def test_new_session_on_dismiss_no_notify_on_success(): app._db_session.add(fake_sess) app._db_session.flush() - ok_result = CreateSessionResult(session=fake_sess, spawn_ok=True, spawn_error="") + ok_result = CreateSessionResult( + session=fake_sess, spawn_ok=True, spawn_error="" + ) mock_service = MagicMock() mock_service.create_session = AsyncMock(return_value=ok_result) @@ -417,8 +445,12 @@ async def test_new_session_on_dismiss_no_notify_on_success(): await app.workers.wait_for_complete() await pilot.pause() - error_notifs = [(m, kw) for m, kw in notify_calls if kw.get("severity") == "error"] - assert len(error_notifs) == 0, f"Expected no error notification, got: {error_notifs}" + error_notifs = [ + (m, kw) for m, kw in notify_calls if kw.get("severity") == "error" + ] + assert len(error_notifs) == 0, ( + f"Expected no error notification, got: {error_notifs}" + ) # --------------------------------------------------------------------------- @@ -469,7 +501,7 @@ async def test_new_session_enter_selects_highlighted_harness(): await pilot.press("enter") # open the dropdown await pilot.pause() - await pilot.press("down") # move to the second harness + await pilot.press("down") # move to the second harness await pilot.pause() await pilot.press("enter") # confirm the highlighted harness await pilot.pause() @@ -609,9 +641,9 @@ async def test_edit_project_no_selection_warns(): await app.run_action("edit_project") await pilot.pause() - assert any( - kw.get("severity") == "warning" for _, kw in notify_calls - ), f"Expected warning notification, got: {notify_calls}" + assert any(kw.get("severity") == "warning" for _, kw in notify_calls), ( + f"Expected warning notification, got: {notify_calls}" + ) @pytest.mark.asyncio @@ -694,14 +726,14 @@ async def test_edit_project_duplicate_path_notifies_error(): @pytest.mark.asyncio async def test_frappe_theme_fully_applied(): app = HqtApp() - async with app.run_test() as pilot: + async with app.run_test(): assert app.current_theme.name == "catppuccin-frappe" variables = app.get_css_variables() # Spot-check: explicit values, not Textual auto-derivations - assert variables["primary"] == "#8CAAEE" # Blue - assert variables["background"] == "#292C3C" # Mantle - assert variables["surface"] == "#414559" # Surface0 - assert variables["border"] == "#babbf1" # Lavender + assert variables["primary"] == "#8CAAEE" # Blue + assert variables["background"] == "#292C3C" # Mantle + assert variables["surface"] == "#414559" # Surface0 + assert variables["border"] == "#babbf1" # Lavender assert variables["footer-background"] == "#51576d" # Surface1 assert variables["input-cursor-background"] == "#f2d5cf" # Rosewater @@ -809,10 +841,14 @@ async def test_dialog_button_focus_is_visibly_highlighted(): # $accent (Frappé Peach #ef9f76) fill — peach is red-dominant, so the # focused button is unmistakably distinct from the blue/ghost defaults. # (Textual adds a small focus background-tint, so allow ±2 per channel.) - assert abs(bg.r - 0xEF) <= 2 and abs(bg.g - 0x9F) <= 2 and abs(bg.b - 0x76) <= 2, ( - f"{btn_id} focus background {bg.hex}, expected ~Peach #EF9F76" + assert ( + abs(bg.r - 0xEF) <= 2 + and abs(bg.g - 0x9F) <= 2 + and abs(bg.b - 0x76) <= 2 + ), f"{btn_id} focus background {bg.hex}, expected ~Peach #EF9F76" + assert bg.r > bg.b, ( + f"{btn_id} focus highlight should be peach, not blue ({bg.hex})" ) - assert bg.r > bg.b, f"{btn_id} focus highlight should be peach, not blue ({bg.hex})" assert btn.styles.color.hex == "#292C3C", ( f"{btn_id} focus text color {btn.styles.color.hex}, expected dark" ) @@ -837,7 +873,9 @@ async def test_first_project_selected_on_load(): await pilot.pause() pl = app.query_one(ProjectList) - first_id = pl.query_one("#project-list").children[0].data + first_item_id = pl.query_one("#project-list").children[0].id + assert first_item_id is not None + first_id = int(first_item_id.removeprefix("proj-")) assert pl.get_selected_project_id() == first_id assert app._selected_project_id == first_id @@ -857,9 +895,24 @@ async def test_first_session_selected_and_remembered_per_project(): pb = Project(name="B", path="/tmp/proj-B") app._db_session.add_all([pa, pb]) app._db_session.flush() - a1 = Session(project_id=pa.id, harness_id=h.id, tmux_session_name="hqt-a1", archived=False) - a2 = Session(project_id=pa.id, harness_id=h.id, tmux_session_name="hqt-a2", archived=False) - b1 = Session(project_id=pb.id, harness_id=h.id, tmux_session_name="hqt-b1", archived=False) + a1 = Session( + project_id=pa.id, + harness_id=h.id, + tmux_session_name="hqt-a1", + archived=False, + ) + a2 = Session( + project_id=pa.id, + harness_id=h.id, + tmux_session_name="hqt-a2", + archived=False, + ) + b1 = Session( + project_id=pb.id, + harness_id=h.id, + tmux_session_name="hqt-b1", + archived=False, + ) app._db_session.add_all([a1, a2, b1]) app._db_session.commit() @@ -901,17 +954,21 @@ def test_format_session_text_colors_and_escapes(): text = format_session_text("mywork", "claude", "working", "◐") assert text.startswith("[#a6d189]◐[/]") # Green symbol - assert "\\[claude]" in text # escaped, so brackets render + assert "\\[claude]" in text # escaped, so brackets render assert "working" in text def test_format_session_text_status_colors(): from hqt.tui.widgets.session_list import format_session_text - assert format_session_text("x", "h", "waiting", "◉").startswith("[#e5c890]") # Yellow - assert format_session_text("x", "h", "idle", "●").startswith("[#81c8be]") # Teal - assert format_session_text("x", "h", "active", "●").startswith("[#81c8be]") # Teal - assert format_session_text("x", "h", "dead", "○").startswith("[#737994]") # Overlay0 + assert format_session_text("x", "h", "waiting", "◉").startswith( + "[#e5c890]" + ) # Yellow + assert format_session_text("x", "h", "idle", "●").startswith("[#81c8be]") # Teal + assert format_session_text("x", "h", "active", "●").startswith("[#81c8be]") # Teal + assert format_session_text("x", "h", "dead", "○").startswith( + "[#737994]" + ) # Overlay0 @pytest.mark.asyncio @@ -1005,9 +1062,9 @@ async def test_rename_session_no_selection_warns(): await app.run_action("rename_session") await pilot.pause() - assert any( - kw.get("severity") == "warning" for _, kw in notify_calls - ), f"Expected warning notification, got: {notify_calls}" + assert any(kw.get("severity") == "warning" for _, kw in notify_calls), ( + f"Expected warning notification, got: {notify_calls}" + ) @pytest.mark.asyncio @@ -1128,5 +1185,3 @@ async def test_session_list_jk_navigation(): await pilot.press("k") await pilot.pause() assert sl.get_selected_session_id() == sessions[0].id - -