# P1 Fixes Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** Resolve all three P1 findings in TODO.md: (1) replace the single shared long-lived SQLAlchemy session with short-lived per-operation sessions, (2) turn missing/stale DB-row crashes into user-facing error notifications, (3) add a schema-migration mechanism so existing user databases survive schema changes. **Architecture:** Services (`ProjectService`, `McpService`, `SessionService`) stop holding a live `Session` and instead hold a `sessionmaker` factory; every public method opens a scoped session (`with self.factory() as db:`). The factory is created with `expire_on_commit=False` and list queries eager-load the `harness` relationship with `selectinload`, so ORM rows returned to the TUI stay readable after their session closes. A new `ServiceError` exception carries user-facing messages for expected failures (missing project/session rows, uninstalled harnesses, duplicate paths); the TUI catches it and calls `self.notify(...)` instead of crashing the worker. Schema versioning uses SQLite's `PRAGMA user_version` with an ordered in-code migration list — no Alembic dependency. **Tech Stack:** Python 3.12+, SQLAlchemy 2.0 ORM, Textual, pytest + pytest-asyncio, SQLite. **Quality gates (run before EVERY commit, from repo root):** ```bash uv run ruff format src tests && uv run ruff check src tests && uv run ty check ``` **IMPORTANT — pre-existing dirty file:** `src/hqt/tmux/runner.py` has uncommitted changes unrelated to this plan. Never `git add -A` or `git add .`; always `git add` the specific files listed in each commit step so that file stays out of these commits. **Run tests with:** `uv run pytest -v` --- ## File Structure | File | Action | Responsibility | |---|---|---| | `src/hqt/db/migrations.py` | Create | `PRAGMA user_version` migration runner + ordered migration list | | `src/hqt/db/engine.py` | Modify | `ensure_db` delegates to `migrate()`; factory gets `expire_on_commit=False` | | `src/hqt/errors.py` | Create | `ServiceError` exception | | `src/hqt/projects/service.py` | Modify | Per-operation sessions; `ServiceError` for duplicates/missing rows | | `src/hqt/mcp/service.py` | Modify | Per-operation sessions | | `src/hqt/sessions/service.py` | Modify | Per-operation sessions; `ServiceError` for missing project/session/harness; eager-load `harness` | | `src/hqt/tui/app.py` | Modify | Hold factory instead of session; catch `ServiceError` → `notify` | | `src/hqt/cli.py` | Modify | `list_cmd` passes factory to `ProjectService` | | `tests/test_db.py` | Modify | Migration tests + detached-row test | | `tests/test_services.py` | Modify | Factory fixture; `ServiceError` assertions | | `tests/test_sessions.py` | Modify | Factory fixture; new error-path tests | | `tests/test_integration.py` | Modify | Factory wiring | | `tests/test_tui.py` | Modify | Seed via `app._db_factory()` instead of `app._db_session` | | `TODO.md` | Modify | Remove the three resolved P1 lines | --- ### Task 1: Schema migration framework (P1 #3) **Files:** - Create: `src/hqt/db/migrations.py` - Modify: `src/hqt/db/engine.py` - Test: `tests/test_db.py` Versioning model: `PRAGMA user_version` stores the schema version. Version 1 is the baseline (the schema `Base.metadata.create_all` produces today). Pre-existing user DBs have `user_version == 0` and the baseline schema, so version 0 with tables present is treated as version 1. Future schema changes append `(version, fn)` entries to `MIGRATIONS` and bump nothing else — `LATEST_VERSION` derives from the list. - [ ] **Step 1: Write the failing tests** Append to `tests/test_db.py` (note: it already imports `Path`, `inspect`, `Settings`, `ensure_db`, `get_engine`, `get_session_factory`, and models at top — add the new imports shown): ```python import pytest from hqt.db import migrations from hqt.db.migrations import migrate from hqt.db.models import Base def _user_version(engine) -> int: with engine.connect() as conn: return conn.exec_driver_sql("PRAGMA user_version").scalar() or 0 def test_fresh_db_stamped_latest_version(tmp_path): settings = _tmp_settings(tmp_path) ensure_db(settings) assert _user_version(get_engine(settings)) == migrations.LATEST_VERSION def test_unversioned_existing_db_stamped_baseline(tmp_path): # Simulates a user DB created before versioning existed: tables present, # user_version still 0. settings = _tmp_settings(tmp_path) settings.db_path.parent.mkdir(parents=True, exist_ok=True) engine = get_engine(settings) Base.metadata.create_all(engine) migrate(engine) assert _user_version(engine) == migrations.BASELINE_VERSION def test_db_from_newer_hqt_raises(tmp_path): settings = _tmp_settings(tmp_path) ensure_db(settings) engine = get_engine(settings) with engine.begin() as conn: conn.exec_driver_sql("PRAGMA user_version = 9999") with pytest.raises(RuntimeError, match="newer"): migrate(engine) def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch): settings = _tmp_settings(tmp_path) ensure_db(settings) # stamps BASELINE_VERSION applied: list[int] = [] monkeypatch.setattr( migrations, "MIGRATIONS", [(2, lambda conn: applied.append(2)), (3, lambda conn: applied.append(3))], ) monkeypatch.setattr(migrations, "LATEST_VERSION", 3) engine = get_engine(settings) migrate(engine) assert applied == [2, 3] assert _user_version(engine) == 3 migrate(engine) # re-run is a no-op assert applied == [2, 3] ``` - [ ] **Step 2: Run tests to verify they fail** Run: `uv run pytest tests/test_db.py -v` Expected: the four new tests FAIL with `ModuleNotFoundError: No module named 'hqt.db.migrations'`; pre-existing tests still pass. - [ ] **Step 3: Create `src/hqt/db/migrations.py`** ```python import logging from collections.abc import Callable from sqlalchemy import Connection, Engine, inspect from hqt.db.models import Base log = logging.getLogger(__name__) # Ordered schema migrations. Each entry upgrades the schema from the previous # version to `version`. Version 1 (BASELINE_VERSION) is the schema produced by # Base.metadata.create_all, so entries here start at version 2, e.g.: # (2, lambda conn: conn.exec_driver_sql( # "ALTER TABLE sessions ADD COLUMN foo TEXT")), MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [] BASELINE_VERSION = 1 LATEST_VERSION = MIGRATIONS[-1][0] if MIGRATIONS else BASELINE_VERSION def migrate(engine: Engine) -> None: """Create or upgrade the database schema to LATEST_VERSION. Fresh databases are created with create_all and stamped directly. Databases at user_version 0 that already have tables predate versioning and are treated as the baseline schema. """ with engine.begin() as conn: version = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0 if version > LATEST_VERSION: raise RuntimeError( f"Database schema version {version} is newer than this hqt " f"supports ({LATEST_VERSION}); upgrade hqt or use a fresh database." ) if not inspect(conn).get_table_names(): Base.metadata.create_all(conn) conn.exec_driver_sql(f"PRAGMA user_version = {LATEST_VERSION}") return if version == 0: version = BASELINE_VERSION for target, fn in MIGRATIONS: if target > version: log.info("Migrating database schema to version %d", target) fn(conn) version = target conn.exec_driver_sql(f"PRAGMA user_version = {version}") ``` - [ ] **Step 4: Point `ensure_db` at the migration runner** Replace `ensure_db` in `src/hqt/db/engine.py` (and drop the now-unused `Base` import): ```python from sqlalchemy import Engine, create_engine from sqlalchemy.orm import sessionmaker from hqt.config import Settings from hqt.db.migrations import migrate 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) migrate(get_engine(settings)) ``` - [ ] **Step 5: Run tests to verify they pass** Run: `uv run pytest tests/test_db.py -v` Expected: ALL tests PASS (including pre-existing `test_ensure_db_creates_tables`). - [ ] **Step 6: Quality gates + commit** ```bash uv run ruff format src tests && uv run ruff check src tests && uv run ty check git add src/hqt/db/migrations.py src/hqt/db/engine.py tests/test_db.py git commit -m "feat: add user_version-based schema migrations" ``` --- ### Task 2: `ServiceError` + detached-row-safe session factory **Files:** - Create: `src/hqt/errors.py` - Modify: `src/hqt/db/engine.py:13-14` - Test: `tests/test_db.py` - [ ] **Step 1: Write the failing test** Append to `tests/test_db.py`: ```python def test_rows_stay_readable_after_session_closes(tmp_path): # Services use per-operation sessions; rows they return must remain # readable after the originating session closes (expire_on_commit=False). settings = _tmp_settings(tmp_path) ensure_db(settings) factory = get_session_factory(get_engine(settings)) with factory() as session: p = Project(name="x", path="/x") session.add(p) session.commit() assert p.name == "x" # would raise DetachedInstanceError without the flag ``` - [ ] **Step 2: Run test to verify it fails** Run: `uv run pytest tests/test_db.py::test_rows_stay_readable_after_session_closes -v` Expected: FAIL with `DetachedInstanceError` (commit expired the instance, refresh after close is impossible). - [ ] **Step 3: Set `expire_on_commit=False` in the factory** In `src/hqt/db/engine.py`: ```python def get_session_factory(engine: Engine) -> sessionmaker: # expire_on_commit=False: services open a session per operation and return # ORM rows to the TUI after the session closes; loaded attributes must # stay readable on those detached instances. return sessionmaker(bind=engine, expire_on_commit=False) ``` - [ ] **Step 4: Create `src/hqt/errors.py`** ```python class ServiceError(Exception): """Expected, user-facing failure in a service operation. Raised for recoverable conditions: missing DB rows (project/session deleted underneath an action), harnesses no longer installed, duplicate project paths. The TUI catches this and shows a notification instead of letting the worker crash. """ ``` - [ ] **Step 5: Run tests to verify they pass** Run: `uv run pytest tests/test_db.py -v` Expected: ALL PASS. - [ ] **Step 6: Quality gates + commit** ```bash uv run ruff format src tests && uv run ruff check src tests && uv run ty check git add src/hqt/errors.py src/hqt/db/engine.py tests/test_db.py git commit -m "feat: add ServiceError and detached-safe session factory" ``` --- ### Task 3: `ProjectService` and `McpService` use per-operation sessions **Files:** - Modify: `src/hqt/projects/service.py` - Modify: `src/hqt/mcp/service.py` - Test: `tests/test_services.py` - [ ] **Step 1: Rewrite the test fixture and update assertions** In `tests/test_services.py`, replace the imports and the `db` fixture at the top of the file with: ```python import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.pool import StaticPool from hqt.db.models import Base from hqt.errors import ServiceError from hqt.mcp.service import McpService from hqt.projects.service import ProjectService @pytest.fixture def factory(): # StaticPool + check_same_thread=False: every session created by the # factory shares the single in-memory connection, so per-operation # sessions all see the same database. engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(engine) return sessionmaker(bind=engine, expire_on_commit=False) ``` Then mechanically update the test bodies: - Replace every `(self, db)` test parameter with `(self, factory)` and every `ProjectService(db)` / `McpService(db)` with `ProjectService(factory)` / `McpService(factory)`. - In `test_update_duplicate_path_raises` and `test_update_unknown_id_raises`, change `pytest.raises(ValueError, ...)` to `pytest.raises(ServiceError, ...)` (match strings stay the same). - Add one new test to `TestProjectService`: ```python def test_create_duplicate_path_raises_service_error(self, factory): svc = ProjectService(factory) svc.create("a", "/dup") with pytest.raises(ServiceError, match="already uses path"): svc.create("b", "/dup") ``` - [ ] **Step 2: Run tests to verify they fail** Run: `uv run pytest tests/test_services.py -v` Expected: FAIL — `ProjectService.__init__` receives a `sessionmaker`, methods break on `self.db.query` (`sessionmaker` has no `query`). - [ ] **Step 3: Rewrite `src/hqt/projects/service.py`** ```python from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import sessionmaker from hqt.db.models import Project from hqt.errors import ServiceError class ProjectService: def __init__(self, factory: sessionmaker): self.factory = factory def create(self, name: str, path: str) -> Project: with self.factory() as db: project = Project(name=name, path=path) db.add(project) try: db.commit() except IntegrityError as err: db.rollback() raise ServiceError( f"Another project already uses path {path}" ) from err return project def list_all(self, include_archived: bool = False) -> list[Project]: with self.factory() as db: q = db.query(Project) if not include_archived: q = q.filter_by(archived=False) return list(q.all()) def get(self, project_id: int) -> Project | None: with self.factory() as db: return db.get(Project, project_id) def update(self, project_id: int, name: str, path: str) -> Project: with self.factory() as db: project = db.get(Project, project_id) if project is None: raise ServiceError(f"Project {project_id} not found") project.name = name project.path = path try: db.commit() except IntegrityError as err: db.rollback() raise ServiceError( f"Another project already uses path {path}" ) from err return project def archive(self, project_id: int) -> None: with self.factory() as db: project = db.get(Project, project_id) if project: project.archived = True db.commit() ``` (`db.refresh(project)` calls are gone — `expire_on_commit=False` keeps attributes loaded.) - [ ] **Step 4: Rewrite `src/hqt/mcp/service.py`** ```python import json from sqlalchemy.orm import sessionmaker from hqt.db.models import McpServer, ProjectMcpServer class McpService: def __init__(self, factory: sessionmaker): self.factory = factory def create( self, name: str, transport: str, command: str | None = None, args: list[str] | None = None, url: str | None = None, ) -> McpServer: with self.factory() as db: server = McpServer( name=name, transport=transport, command=command, args_json=json.dumps(args) if args else None, url=url, ) db.add(server) db.commit() return server def list_all(self) -> list[McpServer]: with self.factory() as db: return list(db.query(McpServer).all()) def get(self, name: str) -> McpServer | None: with self.factory() as db: return db.query(McpServer).filter_by(name=name).first() def delete(self, name: str) -> None: with self.factory() as db: server = db.query(McpServer).filter_by(name=name).first() if server: db.delete(server) db.commit() def bind_to_project(self, project_id: int, mcp_server_id: int) -> None: with self.factory() as db: db.add( ProjectMcpServer(project_id=project_id, mcp_server_id=mcp_server_id) ) db.commit() def unbind_from_project(self, project_id: int, mcp_server_id: int) -> None: with self.factory() as db: db.query(ProjectMcpServer).filter_by( project_id=project_id, mcp_server_id=mcp_server_id ).delete() db.commit() def get_project_mcps(self, project_id: int) -> list[McpServer]: with self.factory() as db: ids = [ r.mcp_server_id for r in db.query(ProjectMcpServer) .filter_by(project_id=project_id) .all() ] if not ids: return [] return list(db.query(McpServer).filter(McpServer.id.in_(ids)).all()) ``` (Note `delete()` re-queries inside its own session rather than calling `self.get()`, because `get()` returns a row detached from a closed session.) - [ ] **Step 5: Run tests to verify they pass** Run: `uv run pytest tests/test_services.py -v` Expected: ALL PASS. (`tests/test_integration.py` and `src/hqt/tui/app.py` / `src/hqt/cli.py` are still broken at this point — they're fixed in Tasks 4 and 5; do not run the full suite yet.) - [ ] **Step 6: Quality gates + commit** `uv run ty check` may flag `cli.py`/`app.py` passing a `Session` where `sessionmaker` is now expected — that is expected breakage fixed in Task 5; if it blocks, note it and commit anyway (gates must be fully green by end of Task 5). ```bash uv run ruff format src tests && uv run ruff check src tests git add src/hqt/projects/service.py src/hqt/mcp/service.py tests/test_services.py git commit -m "refactor: per-operation DB sessions in Project/Mcp services" ``` --- ### Task 4: `SessionService` — per-operation sessions + user-facing errors (P1 #1, #2) **Files:** - Modify: `src/hqt/sessions/service.py` - Test: `tests/test_sessions.py`, `tests/test_integration.py` - [ ] **Step 1: Update the fixtures in `tests/test_sessions.py`** Replace the `db` fixture (lines 13-23) with a `factory` + `db` pair, and update the `service` fixture: ```python from sqlalchemy.pool import StaticPool from hqt.errors import ServiceError @pytest.fixture def factory(): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) Base.metadata.create_all(engine) return sessionmaker(bind=engine, expire_on_commit=False) @pytest.fixture def db(factory): # Seeding/assertion handle onto the same in-memory DB the service uses. session = factory() session.add(Harness(name="claude-code", display_name="Claude Code")) session.add(Project(name="myproj", path="/tmp/myproj")) session.commit() yield session session.close() @pytest.fixture def service(factory, db, tmux, harnesses): # depends on db so the seed rows exist before the service runs return SessionService(factory=factory, tmux=tmux, harnesses=harnesses) ``` Then mechanically update every inline construction (the `db` fixture must stay in those tests' parameter lists wherever it seeds/asserts): ```bash sed -i 's/SessionService(db=db,/SessionService(factory=factory,/' tests/test_sessions.py tests/test_integration.py ``` After the sed, every test that constructs `SessionService(factory=factory, ...)` inline must have `factory` in its parameter list (add it where missing — pytest will error loudly on any you miss). **Stale-read rule for assertions:** the `db` fixture session caches rows in its identity map and will NOT see service-side mutations on instances it already loaded, and objects returned by the service are detached snapshots. Wherever a test asserts on DB state *after* a service call mutated it (e.g. the capture tests asserting `harness_session_id`, `test_respawn_fallback_rung2_updates_harness_session_id`), insert `db.expire_all()` immediately before the assertion and re-read via `db.get(Session, sess.id)` / `db.query(...)`. Example: ```python db.expire_all() fresh = db.get(Session, result.session.id) assert fresh.harness_session_id == "captured-id" ``` - [ ] **Step 2: Add the new error-path and eager-load tests** Append to `tests/test_sessions.py`: ```python @pytest.mark.asyncio async def test_create_session_unknown_project_raises(service): with pytest.raises(ServiceError, match="no longer exists"): await service.create_session(9999, "claude-code") @pytest.mark.asyncio async def test_create_session_uninstalled_harness_raises(factory, db, tmux): # Harness row exists in the DB but the binary is gone from PATH # (self.harnesses has no configurator for it). service = SessionService(factory=factory, tmux=tmux, harnesses={}) with pytest.raises(ServiceError, match="not installed"): await service.create_session(1, "claude-code") @pytest.mark.asyncio async def test_attach_session_missing_row_raises(service): with pytest.raises(ServiceError, match="not found"): await service.attach_session(9999) @pytest.mark.asyncio async def test_list_sessions_rows_usable_after_return(service, db, tmux): await service.create_session(1, "claude-code") infos = await service.list_sessions(1) # The widget reads session.harness.name on detached rows; selectinload # must have eager-loaded the relationship. assert infos[0].session.harness.name == "claude-code" @pytest.mark.asyncio async def test_list_sessions_skips_capture_for_deleted_project( factory, db, tmux, harnesses_capture ): # A session whose project row was deleted must not crash the poller. service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture) await service.create_session(1, "claude-code") db.query(Project).delete() db.commit() infos = await service.list_sessions(1) assert len(infos) == 1 # no ServiceError escaped ``` (Check the `harnesses_capture` fixture near line 273 for its mock shape; if its `capture_session_id` mock asserts call args, the deleted-project test needs no call to happen at all — which is the behavior under test.) - [ ] **Step 3: Run tests to verify they fail** Run: `uv run pytest tests/test_sessions.py -v 2>&1 | tail -20` Expected: widespread FAIL/ERROR — `SessionService` has no `factory` kwarg yet. - [ ] **Step 4: Rewrite `src/hqt/sessions/service.py`** Full replacement file: ```python 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 sqlalchemy.orm import selectinload, sessionmaker from hqt.db.models import Harness, Project, Session from hqt.errors import ServiceError 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 log = logging.getLogger(__name__) CAPTURE_MAX_ATTEMPTS = 6 CAPTURE_RETRY_INTERVAL = 0.5 ACTIVITY_RECENT_SECS = 5 async def _capture_sleep(t: float) -> None: await asyncio.sleep(t) @dataclass class SessionInfo: session: Session alive: bool status: str = "dead" @dataclass class CreateSessionResult: """Result of create_session, carrying the DB row plus spawn outcome.""" session: Session spawn_ok: bool spawn_error: str = "" class SessionService: """Session lifecycle operations. Holds a sessionmaker, not a live session: every public method opens its own short-lived DB session, so concurrent Textual workers and the 3s poller never share ORM state. Rows returned to callers are detached; queries eager-load the harness relationship so widgets can read it. """ def __init__( self, factory: sessionmaker, tmux: TmuxManager, harnesses: Mapping[str, HarnessConfigurator], ): self.factory = factory self.tmux = tmux self.harnesses = harnesses def _configurator(self, harness_name: str) -> HarnessConfigurator: configurator = self.harnesses.get(harness_name) if configurator is None: raise ServiceError(f"Harness '{harness_name}' is not installed") return configurator def _project_path(self, db: DBSession, project_id: int) -> Path: project = db.get(Project, project_id) if project is None: raise ServiceError(f"Project {project_id} no longer exists") return Path(project.path) async def _capture_session_id_with_retry( self, configurator: HarnessConfigurator, project_path: Path, since: float, window_name: str, ) -> str | None: """Retry capture_session_id up to CAPTURE_MAX_ATTEMPTS times. Returns the captured id, or None if all attempts fail (a warning is logged on exhaustion). """ captured_id: str | None = None for attempt in range(CAPTURE_MAX_ATTEMPTS): if attempt > 0: await _capture_sleep(CAPTURE_RETRY_INTERVAL) captured_id = configurator.capture_session_id(project_path, since) if captured_id: break if not captured_id: log.warning( "capture_session_id exhausted %d attempts for window %s; keeping placeholder id", CAPTURE_MAX_ATTEMPTS, window_name, ) 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, db: DBSession, 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 # Polling path: a deleted project just means we skip capture, never crash. project = db.get(Project, sess.project_id) if project is None: return captured_id = configurator.capture_session_id( Path(project.path), self._session_created_epoch(sess) ) if captured_id and captured_id != sess.harness_session_id: sess.harness_session_id = captured_id db.commit() async def create_session( self, project_id: int, harness_name: str, nickname: str | None = None, model: str | None = None, ) -> "CreateSessionResult": """Create a new session row, spawn the harness window, and return a CreateSessionResult with the Session row and spawn outcome. The capture-retry loop only runs when spawn succeeded. """ log.info("Creating session: project=%s harness=%s", project_id, harness_name) configurator = self._configurator(harness_name) with self.factory() as db: harness_row = db.query(Harness).filter_by(name=harness_name).first() if not harness_row: log.error("Harness not in DB: %s", harness_name) raise ServiceError(f"Unknown harness: {harness_name}") project_path = self._project_path(db, project_id) sess = Session( project_id=project_id, harness_id=harness_row.id, nickname=nickname, model=model, tmux_session_name="placeholder", archived=False, ) db.add(sess) db.flush() sess.tmux_session_name = f"hqt-{sess.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, ) ) if not spawn_result.ok: log.error( "Failed to spawn window %s: %s", sess.tmux_session_name, spawn_result.error or "(no output captured)", ) # Only attempt capture when the spawn actually succeeded; a failed # spawn 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, project_path, since, sess.tmux_session_name ) if captured_id: sess.harness_session_id = captured_id db.commit() log.info( "Session created: id=%s window=%s", sess.id, sess.tmux_session_name ) return CreateSessionResult( session=sess, spawn_ok=spawn_result.ok, spawn_error=spawn_result.error or "", ) async def attach_session(self, session_id: int) -> bool: """Attach to a session. Handles all states: alive, dead pane, gone.""" with self.factory() as db: sess = db.get( Session, session_id, options=[selectinload(Session.harness)] ) if sess is None: raise ServiceError("Session not found") window_name = sess.tmux_session_name self._maybe_capture_missing_session_id(db, sess) if await self.tmux.is_alive(window_name): # Window exists, process running — just switch return await self.tmux.attach(window_name) # Window exists but pane is dead, OR window is completely gone log.info( "Session %s not alive, respawning with fallback ladder", window_name ) ok = await self._respawn_with_fallback(db, sess, window_name) if not ok: return False return await self.tmux.attach(window_name) async def _respawn_with_fallback( self, db: DBSession, sess: Session, window_name: str ) -> bool: """Try resume config first; if it dies, fall back to fresh spawn config. Returns True if either rung succeeded, False if both failed. On rung-2 success, if the harness captures_session_id, a capture-retry loop updates sess.harness_session_id so later resumes target the fresh conversation rather than the stale/nonexistent one. """ # Rung 1: resume (restore prior conversation) resume_cfg = self._get_resume_config(db, sess) result = await self.tmux.respawn_verified( window_name, resume_cfg.command, str(resume_cfg.cwd), env=resume_cfg.env ) if result.ok: return True log.warning( "Resume failed for %s (%s), falling back to fresh spawn", window_name, result.error or "(no output)", ) # Rung 2: fresh spawn (codex ignores the old session id; start fresh) configurator = self._configurator(sess.harness.name) project_path = self._project_path(db, 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 ) if not result.ok: log.error( "Fallback spawn also failed for %s: %s", window_name, result.error or "(no output)", ) return False # Rung-2 succeeded — capture the new session id if the harness supports # it so that future resumes target this fresh conversation. if configurator.captures_session_id: new_id = await self._capture_session_id_with_retry( configurator, project_path, since, window_name ) if new_id: sess.harness_session_id = new_id db.commit() else: log.warning( "capture_session_id exhausted after rung-2 for %s; keeping old id %s", window_name, sess.harness_session_id, ) return True async def stop_session(self, session_id: int) -> None: with self.factory() as db: sess = db.get(Session, session_id) if sess is None: return window_name = sess.tmux_session_name if await self.tmux.window_exists(window_name): await self.tmux.kill(window_name) def get_session(self, session_id: int) -> Session | None: """Return the session row, or None if it does not exist.""" with self.factory() as db: return db.get( Session, session_id, options=[selectinload(Session.harness)] ) def rename_session(self, session_id: int, nickname: str | None) -> None: """Update a session's display nickname. An empty/blank nickname clears it to None, so the label falls back to the tmux window name. The tmux window label is refreshed by the next poll (sync_window_labels), so no immediate tmux call is needed here. """ with self.factory() as db: sess = db.get(Session, session_id) if sess is None: return sess.nickname = (nickname or "").strip() or None db.commit() async def delete_session(self, session_id: int) -> None: with self.factory() as db: sess = 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) db.delete(sess) db.commit() 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 (working/waiting) when recognizable, else an activity-based active/idle. capture_pane is only called for alive windows. """ if wi is None or not wi.alive: return "dead" configurator = self.harnesses.get(sess.harness.name) pane_text = await self.tmux.capture_pane(sess.tmux_session_name) parsed: str | None = None if pane_text and configurator is not None: parsed = configurator.parse_status(pane_text) if parsed is not None: return parsed age = now - wi.last_activity return "active" if age <= ACTIVITY_RECENT_SECS else "idle" async def list_sessions( self, project_id: int, now: float | None = None ) -> list[SessionInfo]: if now is None: now = time.time() with self.factory() as db: sessions = ( db.query(Session) .options(selectinload(Session.harness)) .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(db, 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) ) return result async def sync_window_labels(self, now: float | None = None) -> list[SessionInfo]: """Refresh every session's tmux window label to reflect its status. Runs session-wide (all projects), so the tmux status bar stays accurate regardless of which project is selected in the TUI. Returns SessionInfo for all sessions so the caller can reuse the computed status for the UI without recomputing. Labels are set by NAME; windows that no longer exist are harmless no-ops. """ if now is None: now = time.time() with self.factory() as db: sessions = ( db.query(Session) .options(selectinload(Session.harness)) .filter_by(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(db, s) wi = info_map.get(s.tmux_session_name) alive = bool(wi and wi.alive) status = await self._status_for(s, wi, now) result.append(SessionInfo(session=s, alive=alive, status=status)) name = s.nickname or s.tmux_session_name await self.tmux.set_window_label( s.tmux_session_name, window_label(status, name, alive=alive) ) return result def _get_resume_config(self, db: DBSession, sess: Session) -> SpawnConfig: configurator = self._configurator(sess.harness.name) harness_session_id = ( sess.harness_session_id or configurator.generate_session_id(sess.id) ) return configurator.build_resume_config( self._project_path(db, sess.project_id), harness_session_id, sess.model ) ``` - [ ] **Step 5: Update `tests/test_integration.py` wiring** In the `setup` fixture: keep the local `factory` variable, change `ProjectService(db)` → `ProjectService(factory)` (the `SessionService(db=db,` line was already handled by the Step-1 sed). The seeding `db = factory()` handle stays as-is. - [ ] **Step 6: Run tests, fix stale-read assertions per the rule in Step 1** Run: `uv run pytest tests/test_sessions.py tests/test_integration.py tests/test_services.py -v 2>&1 | tail -30` Any remaining failures should be exactly the stale-identity-map pattern (assertion sees a pre-mutation value). For each, apply the `db.expire_all()` + re-read pattern from Step 1. Re-run until: ALL PASS. - [ ] **Step 7: Quality gates + commit** `ty` may still flag `src/hqt/tui/app.py` and `src/hqt/cli.py` (fixed next task) — everything else must be clean. ```bash uv run ruff format src tests && uv run ruff check src tests git add src/hqt/sessions/service.py tests/test_sessions.py tests/test_integration.py git commit -m "refactor: per-operation DB sessions and ServiceError in SessionService" ``` --- ### Task 5: Wire the TUI and CLI — factory + `ServiceError` → notifications **Files:** - Modify: `src/hqt/tui/app.py` - Modify: `src/hqt/cli.py:134-138` - Test: `tests/test_tui.py` - [ ] **Step 1: Update `tests/test_tui.py` seeding** Tests seed via `app._db_session` (around lines 50-63, 81-95, 205-219, 270+). The app will now expose `_db_factory` instead. In each seeding block, open a session from the factory: ```python seed_db = app._db_factory() proj = Project(name="p", path="/tmp/p") seed_db.add(proj) seed_db.flush() harness = seed_db.query(Harness).first() ... seed_db.add(sess) seed_db.commit() seed_db.close() ``` i.e. mechanically: insert `seed_db = app._db_factory()` at the top of each block, replace `app._db_session` → `seed_db`, and add `seed_db.close()` after the final `commit()` in the block. (Keep any later assertions reading via a fresh `app._db_factory()` session the same way.) - [ ] **Step 2: Run TUI tests to verify they fail** Run: `uv run pytest tests/test_tui.py -v 2>&1 | tail -10` Expected: FAIL — `HqtApp` has no `_db_factory` yet. - [ ] **Step 3: Update `src/hqt/tui/app.py`** Add the import: ```python from hqt.errors import ServiceError ``` In `__init__`, replace `self._db_session = None` with: ```python self._db_factory = None ``` In `on_mount` (lines 101-122), replace the DB wiring: ```python settings = get_settings() ensure_db(settings) engine = get_engine(settings) self._db_factory = get_session_factory(engine) with self._db_factory() as db: ensure_harnesses_in_db(db) ``` and the service construction: ```python self._project_service = ProjectService(self._db_factory) self._session_service = SessionService(self._db_factory, tmux, harnesses) ``` Add a worker helper after `_sessions()`: ```python def _run_service_worker(self, coro) -> None: """run_worker, but ServiceError surfaces as a notification. Services raise ServiceError for expected failures (rows deleted underneath an action, harness uninstalled); an uncaught exception would kill the worker and crash the app. """ async def _wrapped() -> None: try: await coro except ServiceError as err: self.notify(str(err), severity="error") self.run_worker(_wrapped()) ``` Then route every service-touching worker through it — replace `self.run_worker(_do())` with `self._run_service_worker(_do())` in: `action_new_session.on_dismiss`, `action_attach_session`, `action_delete_session`, `action_stop_session`, and replace `self.run_worker(self._refresh_sessions(restore_selection=True))` in `on_project_selected` and `self.run_worker(self._refresh_sessions())` in `action_rename_session` with `self._run_service_worker(...)` of the same coroutine. Leave `_load_projects`'s widget-refresh worker and the theme worker on plain `run_worker`. Wrap the synchronous service calls: In `action_add_project.on_dismiss`: ```python def on_dismiss(result: tuple[str, str] | None) -> None: if result: name, path = result try: self._projects().create(name, path) except ServiceError as err: self.notify(str(err), severity="error") return self._load_projects() ``` In `action_edit_project.on_dismiss`, change `except ValueError` to `except ServiceError`. In `_poll_sessions`, guard the interval callback (a crash here kills polling for the rest of the app's life): ```python 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. try: infos = await self._sessions().sync_window_labels() except ServiceError as err: log.warning("Session poll failed: %s", err) return if self._selected_project_id is not None: subset = [ i for i in infos if i.session.project_id == self._selected_project_id ] await self.query_one(SessionList).refresh_sessions(subset) ``` - [ ] **Step 4: Update `src/hqt/cli.py` `list_cmd`** ```python settings = get_settings() ensure_db(settings) engine = get_engine(settings) factory = get_session_factory(engine) svc = ProjectService(factory) for p in svc.list_all(): click.echo(f" {p.name} ({p.path})") ``` (The `with Session() as db:` block is gone — the service scopes its own sessions.) - [ ] **Step 5: Run the FULL test suite** Run: `uv run pytest -v 2>&1 | tail -15` Expected: ALL PASS, no skips introduced. - [ ] **Step 6: Quality gates (must be fully green now) + commit** ```bash uv run ruff format src tests && uv run ruff check src tests && uv run ty check git add src/hqt/tui/app.py src/hqt/cli.py tests/test_tui.py git commit -m "feat: factory-based DB wiring and ServiceError notifications in TUI" ``` --- ### Task 6: Manual smoke test + close out TODO.md **Files:** - Modify: `TODO.md` (remove lines 1-3, the three P1 findings) - [ ] **Step 1: Smoke-test against a real pre-existing DB** ```bash cp ~/.local/share/hqt/hqt.db /tmp/hqt-backup.db 2>/dev/null || echo "no existing db (fresh install) — skip restore step" uv run hqt list ``` Expected: project list prints without traceback (this exercises `ensure_db` → `migrate()` against a real unversioned DB, stamping it to baseline). Then verify the stamp: ```bash sqlite3 ~/.local/share/hqt/hqt.db "PRAGMA user_version" ``` Expected output: `1` If anything broke, restore: `cp /tmp/hqt-backup.db ~/.local/share/hqt/hqt.db`. - [ ] **Step 2: Remove the three P1 lines from `TODO.md`** Delete lines 1-3 so the file starts with the first P2 finding. Leave all P2 lines untouched. - [ ] **Step 3: Final full verification** ```bash uv run pytest -q && uv run ruff format src tests && uv run ruff check src tests && uv run ty check ``` Expected: tests all pass, gates green. - [ ] **Step 4: Commit** ```bash git add TODO.md git commit -m "chore: close out P1 findings" ```