hqt/docs/superpowers/plans/2026-06-10-p1-fixes.md

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):

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 <path> -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):

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

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):

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

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:

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:

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

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

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:

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:

    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

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

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).

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:

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):

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:

    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:

@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:

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.

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:

        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:

from hqt.errors import ServiceError

In __init__, replace self._db_session = None with:

        self._db_factory = None

In on_mount (lines 101-122), replace the DB wiring:

        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:

        self._project_service = ProjectService(self._db_factory)
        self._session_service = SessionService(self._db_factory, tmux, harnesses)

Add a worker helper after _sessions():

    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:

        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):

    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

    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

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

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:

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

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

git add TODO.md
git commit -m "chore: close out P1 findings"