diff --git a/TODO.md b/TODO.md
index 103f1d8..1c66402 100644
--- a/TODO.md
+++ b/TODO.md
@@ -1,6 +1,3 @@
-P1: One long-lived SQLAlchemy session is shared across Textual workers and the 3-second poller, creating concurrency and stale-state risk.
-P1: Missing session rows or stale harness rows can crash service methods instead of returning user-facing errors.
-P1: Database setup uses `Base.metadata.create_all` only, so existing user databases will not be migrated when schema changes.
 P2: Codex session-id capture is heuristic and can store the wrong rollout if another Codex session starts in the same project during the retry window.
 P2: tmux behavior is mostly tested with mocks; add isolated real-tmux smoke tests for theming, new windows, respawn, and labels.
 P2: Project paths are accepted without existence validation, so bad paths become dead sessions later instead of being rejected early.
diff --git a/docs/superpowers/plans/2026-06-10-p1-fixes.md b/docs/superpowers/plans/2026-06-10-p1-fixes.md
new file mode 100644
index 0000000..efcd905
--- /dev/null
+++ b/docs/superpowers/plans/2026-06-10-p1-fixes.md
@@ -0,0 +1,1248 @@
+# 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 <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):
+
+```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"
+```