diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py
index 880a6a6..4289905 100644
--- a/src/hqt/sessions/service.py
+++ b/src/hqt/sessions/service.py
@@ -7,8 +7,10 @@ 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
@@ -42,20 +44,34 @@ class CreateSessionResult:
 
 
 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,
-        db: DBSession,
+        factory: sessionmaker,
         tmux: TmuxManager,
         harnesses: Mapping[str, HarnessConfigurator],
     ):
-        self.db = db
+        self.factory = factory
         self.tmux = tmux
         self.harnesses = harnesses
 
-    def _project_path(self, project_id: int) -> Path:
-        project = self.db.get(Project, project_id)
+    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 ValueError(f"Unknown project: {project_id}")
+            raise ServiceError(f"Project {project_id} no longer exists")
         return Path(project.path)
 
     async def _capture_session_id_with_retry(
@@ -93,7 +109,7 @@ class SessionService:
             return created_at.timestamp()
         return 0.0
 
-    def _maybe_capture_missing_session_id(self, sess: Session) -> None:
+    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
@@ -102,12 +118,16 @@ class SessionService:
         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(
-            self._project_path(sess.project_id), self._session_created_epoch(sess)
+            Path(project.path), self._session_created_epoch(sess)
         )
         if captured_id and captured_id != sess.harness_session_id:
             sess.harness_session_id = captured_id
-            self.db.commit()
+            db.commit()
 
     async def create_session(
         self,
@@ -122,85 +142,94 @@ class SessionService:
         The capture-retry loop only runs when spawn succeeded.
         """
         log.info("Creating session: project=%s harness=%s", project_id, harness_name)
-        harness_row = self.db.query(Harness).filter_by(name=harness_name).first()
-        if not harness_row:
-            log.error("Harness not in DB: %s", harness_name)
-            raise ValueError(f"Unknown harness: {harness_name}")
-        configurator = self.harnesses[harness_name]
-        sess = Session(
-            project_id=project_id,
-            harness_id=harness_row.id,
-            nickname=nickname,
-            model=model,
-            tmux_session_name="placeholder",
-            archived=False,
-        )
-        self.db.add(sess)
-        self.db.flush()
-        sess.tmux_session_name = f"hqt-{sess.id}"
-        project_path = self._project_path(project_id)
-        harness_session_id = configurator.generate_session_id(sess.id)
-        sess.harness_session_id = harness_session_id
-        spawn_cfg = configurator.build_spawn_config(
-            project_path, harness_session_id, model
-        )
-        log.info(
-            "Spawning window %s: cmd=%s cwd=%s",
-            sess.tmux_session_name,
-            spawn_cfg.command,
-            spawn_cfg.cwd,
-        )
-        since = time.time()
-        spawn_result = await self.tmux.spawn(
-            SpawnRequest(
-                window_name=sess.tmux_session_name,
-                command=spawn_cfg.command,
-                cwd=str(spawn_cfg.cwd),
-                env=spawn_cfg.env,
+        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,
             )
-        )
-        if not spawn_result.ok:
-            log.error(
-                "Failed to spawn window %s: %s",
+            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_result.error or "(no output captured)",
+                spawn_cfg.command,
+                spawn_cfg.cwd,
             )
-        # 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
+            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 "",
             )
-            if captured_id:
-                sess.harness_session_id = captured_id
-        self.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."""
-        sess = self.db.get(Session, session_id)
-        if sess is None:
-            return False
-        window_name = sess.tmux_session_name
-        self._maybe_capture_missing_session_id(sess)
+        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
+            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)
 
-        # 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(sess, window_name)
-        if not ok:
-            return False
-        return await self.tmux.attach(window_name)
-
-    async def _respawn_with_fallback(self, sess: Session, window_name: str) -> bool:
+    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.
@@ -210,7 +239,7 @@ class SessionService:
         conversation rather than the stale/nonexistent one.
         """
         # Rung 1: resume (restore prior conversation)
-        resume_cfg = self._get_resume_config(sess)
+        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
         )
@@ -224,8 +253,8 @@ class SessionService:
         )
 
         # Rung 2: fresh spawn (codex ignores the old session id; start fresh)
-        configurator = self.harnesses[sess.harness.name]
-        project_path = self._project_path(sess.project_id)
+        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)
         )
@@ -244,15 +273,15 @@ class SessionService:
             )
             return False
 
-        # Rung-2 succeeded — capture the new session id if the harness supports it
-        # so that future resumes target this fresh conversation.
+        # 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
-                self.db.commit()
+                db.commit()
             else:
                 log.warning(
                     "capture_session_id exhausted after rung-2 for %s; keeping old id %s",
@@ -262,15 +291,18 @@ class SessionService:
         return True
 
     async def stop_session(self, session_id: int) -> None:
-        sess = self.db.get(Session, session_id)
-        if sess is None:
-            return
-        if await self.tmux.window_exists(sess.tmux_session_name):
-            await self.tmux.kill(sess.tmux_session_name)
+        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."""
-        return self.db.get(Session, session_id)
+        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.
@@ -279,20 +311,22 @@ class SessionService:
         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.
         """
-        sess = self.db.get(Session, session_id)
-        if sess is None:
-            return
-        sess.nickname = (nickname or "").strip() or None
-        self.db.commit()
+        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:
-        sess = self.db.get(Session, session_id)
-        if sess is None:
-            return
-        if await self.tmux.window_exists(sess.tmux_session_name):
-            await self.tmux.kill(sess.tmux_session_name)
-        self.db.delete(sess)
-        self.db.commit()
+        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
@@ -320,22 +354,24 @@ class SessionService:
     ) -> list[SessionInfo]:
         if now is None:
             now = time.time()
-        sessions = (
-            self.db.query(Session)
-            .filter_by(project_id=project_id, archived=False)
-            .all()
-        )
-        names = [s.tmux_session_name for s in sessions]
-        info_map = await self.tmux.poll_info(names)
-        result: list[SessionInfo] = []
-        for s in sessions:
-            self._maybe_capture_missing_session_id(s)
-            wi = info_map.get(s.tmux_session_name)
-            status = await self._status_for(s, wi, now)
-            result.append(
-                SessionInfo(session=s, alive=bool(wi and wi.alive), status=status)
+        with self.factory() as db:
+            sessions = (
+                db.query(Session)
+                .options(selectinload(Session.harness))
+                .filter_by(project_id=project_id, archived=False)
+                .all()
             )
-        return result
+            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.
@@ -348,27 +384,33 @@ class SessionService:
         """
         if now is None:
             now = time.time()
-        sessions = self.db.query(Session).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(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)
+        with self.factory() as db:
+            sessions = (
+                db.query(Session)
+                .options(selectinload(Session.harness))
+                .filter_by(archived=False)
+                .all()
             )
-        return result
+            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, sess: Session) -> SpawnConfig:
-        configurator = self.harnesses[sess.harness.name]
+    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(sess.project_id), harness_session_id, sess.model
+            self._project_path(db, sess.project_id), harness_session_id, sess.model
         )
diff --git a/tests/test_integration.py b/tests/test_integration.py
index 2091a54..fbbfa00 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -3,6 +3,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
 
 from hqt.db.models import Base, Harness
 from hqt.projects.service import ProjectService
@@ -13,9 +14,13 @@ from hqt.harnesses.configurators.claude import ClaudeConfigurator
 
 @pytest.fixture
 def setup():
-    engine = create_engine("sqlite:///:memory:")
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
     Base.metadata.create_all(engine)
-    factory = sessionmaker(bind=engine)
+    factory = sessionmaker(bind=engine, expire_on_commit=False)
     db = factory()
     db.add(Harness(name="claude-code", display_name="Claude Code"))
     db.commit()
@@ -32,8 +37,8 @@ def setup():
     configurator = ClaudeConfigurator()
     harnesses = {"claude-code": configurator}
 
-    proj_svc = ProjectService(db)
-    sess_svc = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    proj_svc = ProjectService(factory)
+    sess_svc = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     return proj_svc, sess_svc, tmux
 
 
diff --git a/tests/test_sessions.py b/tests/test_sessions.py
index d5679bb..60eabc2 100644
--- a/tests/test_sessions.py
+++ b/tests/test_sessions.py
@@ -4,17 +4,28 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
 
 from hqt.db.models import Base, Harness, Project
+from hqt.errors import ServiceError
 from hqt.sessions.service import SessionInfo, SessionService
 from hqt.tmux.manager import SpawnResult, TmuxManager
 
 
 @pytest.fixture
-def db():
-    engine = create_engine("sqlite:///:memory:")
+def factory():
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
     Base.metadata.create_all(engine)
-    factory = sessionmaker(bind=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"))
@@ -52,8 +63,9 @@ def harnesses():
 
 
 @pytest.fixture
-def service(db, tmux, harnesses):
-    return SessionService(db=db, tmux=tmux, harnesses=harnesses)
+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)
 
 
 @pytest.mark.asyncio
@@ -283,17 +295,17 @@ def harnesses_capture():
 
 @pytest.mark.asyncio
 async def test_create_session_no_capture_flag_never_calls_capture(
-    db, tmux, harnesses_no_capture
+    factory, db, tmux, harnesses_no_capture
 ):
     """When captures_session_id=False, capture_session_id is never called."""
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_no_capture)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_no_capture)
     await service.create_session(project_id=1, harness_name="claude-code")
     harnesses_no_capture["claude-code"].capture_session_id.assert_not_called()
 
 
 @pytest.mark.asyncio
 async def test_create_session_capture_retries_until_success(
-    db, tmux, harnesses_capture
+    factory, db, tmux, harnesses_capture
 ):
     """captures_session_id=True: returns None twice then an id → retries and stores captured id."""
     import hqt.sessions.service as svc_mod
@@ -309,7 +321,7 @@ async def test_create_session_capture_retries_until_success(
     async def fake_sleep(t):
         sleep_calls.append(t)
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -321,7 +333,7 @@ async def test_create_session_capture_retries_until_success(
 
 @pytest.mark.asyncio
 async def test_create_session_capture_all_none_keeps_placeholder(
-    db, tmux, harnesses_capture, caplog
+    factory, db, tmux, harnesses_capture, caplog
 ):
     """All capture attempts return None → placeholder id kept, no crash, warning logged."""
     import logging
@@ -332,7 +344,7 @@ async def test_create_session_capture_all_none_keeps_placeholder(
     async def fake_sleep(t):
         pass
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
     with caplog.at_level(logging.WARNING):
         with patch.object(svc_mod, "_capture_sleep", fake_sleep):
             result = await service.create_session(
@@ -354,7 +366,7 @@ async def test_create_session_capture_all_none_keeps_placeholder(
 
 @pytest.mark.asyncio
 async def test_create_session_capture_first_attempt_succeeds(
-    db, tmux, harnesses_capture
+    factory, db, tmux, harnesses_capture
 ):
     """captures_session_id=True: first attempt succeeds → no sleeps needed."""
     import hqt.sessions.service as svc_mod
@@ -366,7 +378,7 @@ async def test_create_session_capture_first_attempt_succeeds(
     async def fake_sleep(t):
         sleep_calls.append(t)
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -380,15 +392,15 @@ async def test_create_session_capture_first_attempt_succeeds(
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_dead_window(db, tmux, harnesses):
+async def test_list_sessions_dead_window(factory, db, tmux, harnesses):
     """list_sessions: dead window → status='dead', alive=False."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
     tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1)
     assert len(result) == 1
     assert result[0].alive is False
@@ -396,33 +408,33 @@ async def test_list_sessions_dead_window(db, tmux, harnesses):
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_missing_window_is_dead(db, tmux, harnesses):
+async def test_list_sessions_missing_window_is_dead(factory, db, tmux, harnesses):
     """list_sessions: window not in poll_info result → status='dead', alive=False."""
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
     tmux.poll_info.return_value = {}  # no windows at all
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1)
     assert result[0].alive is False
     assert result[0].status == "dead"
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_alive_parse_working(db, tmux, harnesses):
+async def test_list_sessions_alive_parse_working(factory, db, tmux, harnesses):
     """Alive window + harness parse returns 'working' → status='working'."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
 
     now = 2000.0
     tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
     tmux.capture_pane = AsyncMock(return_value="Esc to interrupt\n")
     harnesses["claude-code"].parse_status = MagicMock(return_value="working")
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1, now=now)
 
     assert result[0].status == "working"
@@ -430,13 +442,15 @@ async def test_list_sessions_alive_parse_working(db, tmux, harnesses):
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesses):
+async def test_list_sessions_alive_parse_none_recent_activity(
+    factory, db, tmux, harnesses
+):
     """Alive window + parse_status None + recent activity → status='active'."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
 
     now = 2000.0
     tmux.poll_info.return_value = {
@@ -445,20 +459,22 @@ async def test_list_sessions_alive_parse_none_recent_activity(db, tmux, harnesse
     tmux.capture_pane = AsyncMock(return_value="some text")
     harnesses["claude-code"].parse_status = MagicMock(return_value=None)
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1, now=now)
 
     assert result[0].status == "active"
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses):
+async def test_list_sessions_alive_parse_none_stale_activity(
+    factory, db, tmux, harnesses
+):
     """Alive window + parse_status None + stale activity → status='idle'."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
 
     now = 2000.0
     tmux.poll_info.return_value = {
@@ -467,27 +483,29 @@ async def test_list_sessions_alive_parse_none_stale_activity(db, tmux, harnesses
     tmux.capture_pane = AsyncMock(return_value="some text")
     harnesses["claude-code"].parse_status = MagicMock(return_value=None)
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1, now=now)
 
     assert result[0].status == "idle"
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harnesses):
+async def test_list_sessions_capture_empty_falls_back_to_activity(
+    factory, db, tmux, harnesses
+):
     """capture_pane returns empty text → activity-based fallback, no crash."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
 
     now = 2000.0
     tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
     tmux.capture_pane = AsyncMock(return_value="")  # empty — treat as capture failure
     harnesses["claude-code"].parse_status = MagicMock(return_value=None)
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1, now=now)
 
     # Should still produce a valid status via activity
@@ -496,20 +514,20 @@ async def test_list_sessions_capture_empty_falls_back_to_activity(db, tmux, harn
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_alive_parse_waiting(db, tmux, harnesses):
+async def test_list_sessions_alive_parse_waiting(factory, db, tmux, harnesses):
     """Alive window + parse_status 'waiting' → status='waiting'."""
     from hqt.tmux.runner import WindowInfo
 
-    await SessionService(db=db, tmux=tmux, harnesses=harnesses).create_session(
-        project_id=1, harness_name="claude-code"
-    )
+    await SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses
+    ).create_session(project_id=1, harness_name="claude-code")
 
     now = 2000.0
     tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=True, last_activity=1999)}
     tmux.capture_pane = AsyncMock(return_value="Do you want to proceed?\n❯ 1. Yes\n")
     harnesses["claude-code"].parse_status = MagicMock(return_value="waiting")
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.list_sessions(project_id=1, now=now)
 
     assert result[0].status == "waiting"
@@ -521,11 +539,11 @@ async def test_list_sessions_alive_parse_waiting(db, tmux, harnesses):
 
 
 @pytest.mark.asyncio
-async def test_sync_window_labels_sets_status_symbol(db, tmux, harnesses):
+async def test_sync_window_labels_sets_status_symbol(factory, db, tmux, harnesses):
     """sync_window_labels pushes a '<symbol> <nickname>' label per session."""
     from hqt.tmux.runner import WindowInfo
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     await service.create_session(
         project_id=1, harness_name="claude-code", nickname="mywork"
     )
@@ -545,11 +563,13 @@ async def test_sync_window_labels_sets_status_symbol(db, tmux, harnesses):
 
 
 @pytest.mark.asyncio
-async def test_sync_window_labels_dead_window_uses_dead_glyph(db, tmux, harnesses):
+async def test_sync_window_labels_dead_window_uses_dead_glyph(
+    factory, db, tmux, harnesses
+):
     """A dead/missing window gets the ○ glyph and falls back to the window name."""
     from hqt.tmux.runner import WindowInfo
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     await service.create_session(
         project_id=1, harness_name="claude-code"
     )  # no nickname
@@ -565,7 +585,7 @@ async def test_sync_window_labels_dead_window_uses_dead_glyph(db, tmux, harnesse
 
 
 @pytest.mark.asyncio
-async def test_sync_window_labels_covers_all_projects(db, tmux, harnesses):
+async def test_sync_window_labels_covers_all_projects(factory, db, tmux, harnesses):
     """sync_window_labels is session-wide: it labels sessions across all projects."""
     from hqt.db.models import Project
     from hqt.tmux.runner import WindowInfo
@@ -573,7 +593,7 @@ async def test_sync_window_labels_covers_all_projects(db, tmux, harnesses):
     db.add(Project(name="proj2", path="/tmp/proj2"))
     db.commit()
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     await service.create_session(project_id=1, harness_name="claude-code", nickname="a")
     await service.create_session(project_id=2, harness_name="claude-code", nickname="b")
 
@@ -596,7 +616,7 @@ async def test_sync_window_labels_covers_all_projects(db, tmux, harnesses):
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_real_configurator_waiting(db):
+async def test_list_sessions_real_configurator_waiting(factory, db):
     """list_sessions with real ClaudeConfigurator + spec'd TmuxManager mock → status='waiting'.
 
     Uses the REAL ClaudeConfigurator (not a mock) to close the decorative-fixture
@@ -633,7 +653,7 @@ async def test_list_sessions_real_configurator_waiting(db):
 
     configurator = ClaudeConfigurator()
     harnesses = {"claude-code": configurator}
-    service = SessionService(db=db, tmux=tmux_spec, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux_spec, harnesses=harnesses)
 
     # Create a session so there is a row in the DB
     await service.create_session(project_id=1, harness_name="claude-code")
@@ -658,13 +678,13 @@ async def test_list_sessions_real_configurator_waiting(db):
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_dead_window_no_capture(db, tmux):
+async def test_list_sessions_dead_window_no_capture(factory, db, tmux):
     """list_sessions: dead window → capture_pane must NOT be called."""
     from hqt.tmux.runner import WindowInfo
     from hqt.harnesses.configurators.claude import ClaudeConfigurator
 
     harnesses = {"claude-code": ClaudeConfigurator()}
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     await service.create_session(project_id=1, harness_name="claude-code")
 
     tmux.poll_info.return_value = {"hqt-1": WindowInfo(alive=False, last_activity=0)}
@@ -676,12 +696,12 @@ async def test_list_sessions_dead_window_no_capture(db, tmux):
 
 
 @pytest.mark.asyncio
-async def test_list_sessions_missing_window_no_capture(db, tmux):
+async def test_list_sessions_missing_window_no_capture(factory, db, tmux):
     """list_sessions: window missing from poll_info → capture_pane must NOT be called."""
     from hqt.harnesses.configurators.claude import ClaudeConfigurator
 
     harnesses = {"claude-code": ClaudeConfigurator()}
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     await service.create_session(project_id=1, harness_name="claude-code")
 
     tmux.poll_info.return_value = {}  # window not present at all
@@ -715,7 +735,7 @@ def harnesses_capture_fallback():
 
 @pytest.mark.asyncio
 async def test_respawn_fallback_rung2_updates_harness_session_id(
-    db, tmux, harnesses_capture_fallback
+    factory, db, tmux, harnesses_capture_fallback
 ):
     """Rung-2 success + captures_session_id=True + capture returns new id → sess.harness_session_id updated and committed."""
     import hqt.sessions.service as svc_mod
@@ -727,7 +747,9 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(
     async def fake_sleep(t):
         pass
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture_fallback)
+    service = SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
+    )
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -763,7 +785,7 @@ async def test_respawn_fallback_rung2_updates_harness_session_id(
 
 @pytest.mark.asyncio
 async def test_attach_late_capture_updates_id_before_resume(
-    db, tmux, harnesses_capture_fallback
+    factory, db, tmux, harnesses_capture_fallback
 ):
     """If create-time Codex capture missed, attach captures the real id before resuming."""
     import hqt.sessions.service as svc_mod
@@ -776,7 +798,9 @@ async def test_attach_late_capture_updates_id_before_resume(
     async def fake_sleep(t):
         pass
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture_fallback)
+    service = SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
+    )
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -802,7 +826,7 @@ async def test_attach_late_capture_updates_id_before_resume(
 
 @pytest.mark.asyncio
 async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(
-    db, tmux, harnesses_capture_fallback, caplog
+    factory, db, tmux, harnesses_capture_fallback, caplog
 ):
     """Rung-2 success + capture exhausted → old id kept + warning logged."""
     import logging
@@ -815,7 +839,9 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(
     async def fake_sleep(t):
         pass
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture_fallback)
+    service = SessionService(
+        factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
+    )
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -853,7 +879,7 @@ async def test_respawn_fallback_rung2_capture_exhausted_keeps_old_id(
 
 
 @pytest.mark.asyncio
-async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux):
+async def test_respawn_fallback_rung2_no_capture_when_flag_false(factory, db, tmux):
     """Rung-2 success + captures_session_id=False → capture_session_id never called."""
     h = MagicMock()
     h.name = "claude-code"
@@ -867,7 +893,7 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux):
     )
     harnesses_nc = {"claude-code": h}
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_nc)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_nc)
     result = await service.create_session(project_id=1, harness_name="claude-code")
     sess_id = result.session.id
 
@@ -893,7 +919,9 @@ async def test_respawn_fallback_rung2_no_capture_when_flag_false(db, tmux):
 
 
 @pytest.mark.asyncio
-async def test_create_session_no_capture_when_spawn_fails(db, tmux, harnesses_capture):
+async def test_create_session_no_capture_when_spawn_fails(
+    factory, db, tmux, harnesses_capture
+):
     """create_session: spawn fails → capture_session_id must NOT be called."""
     import hqt.sessions.service as svc_mod
 
@@ -904,7 +932,7 @@ async def test_create_session_no_capture_when_spawn_fails(db, tmux, harnesses_ca
     async def fake_sleep(t):
         pass
 
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses_capture)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
     with patch.object(svc_mod, "_capture_sleep", fake_sleep):
         result = await service.create_session(project_id=1, harness_name="claude-code")
 
@@ -933,14 +961,16 @@ async def test_create_session_returns_result_dataclass(service, db, tmux, harnes
 
 
 @pytest.mark.asyncio
-async def test_create_session_result_propagates_spawn_failure(db, tmux, harnesses):
+async def test_create_session_result_propagates_spawn_failure(
+    factory, db, tmux, harnesses
+):
     """create_session result reflects spawn failure."""
     from hqt.sessions.service import CreateSessionResult
 
     tmux.spawn.return_value = SpawnResult(
         ok=False, window_id=None, error="command not found: claude"
     )
-    service = SessionService(db=db, tmux=tmux, harnesses=harnesses)
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
     result = await service.create_session(project_id=1, harness_name="claude-code")
 
     assert isinstance(result, CreateSessionResult)
@@ -977,6 +1007,7 @@ def test_rename_session_sets_nickname(service, db):
 
     sess = _seed_session(db, nickname="old")
     service.rename_session(sess.id, "new-name")
+    db.expire_all()
     assert db.get(Session, sess.id).nickname == "new-name"
 
 
@@ -985,6 +1016,7 @@ def test_rename_session_empty_clears_to_none(service, db):
 
     sess = _seed_session(db, nickname="old")
     service.rename_session(sess.id, "")
+    db.expire_all()
     assert db.get(Session, sess.id).nickname is None
 
 
@@ -993,6 +1025,7 @@ def test_rename_session_whitespace_clears_to_none(service, db):
 
     sess = _seed_session(db, nickname="old")
     service.rename_session(sess.id, "   ")
+    db.expire_all()
     assert db.get(Session, sess.id).nickname is None
 
 
@@ -1003,3 +1036,73 @@ def test_get_session_returns_row(service, db):
 
 def test_get_session_missing_returns_none(service):
     assert service.get_session(99999) is None
+
+
+# ---------------------------------------------------------------------------
+# Task 4: per-operation sessions, ServiceError, eager-load harness
+# ---------------------------------------------------------------------------
+
+
+@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.
+    # Capture returns None at create time so the placeholder id is kept; the
+    # poller then re-enters the capture path and must SKIP on the missing
+    # project rather than raise.
+    import hqt.sessions.service as svc_mod
+    from hqt.db.models import Session
+
+    harnesses_capture["claude-code"].capture_session_id.return_value = None
+
+    async def fake_sleep(t):
+        pass
+
+    service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses_capture)
+    with patch.object(svc_mod, "_capture_sleep", fake_sleep):
+        await service.create_session(1, "claude-code")
+
+    # The placeholder id must still be in place so the poller attempts capture.
+    db.expire_all()
+    fresh = db.query(Session).filter_by(project_id=1).first()
+    assert fresh.harness_session_id == "placeholder-id"
+
+    db.query(Project).delete()
+    db.commit()
+
+    # Capture would now find an id, but the project lookup must short-circuit
+    # before it is consulted; either way no ServiceError may escape.
+    harnesses_capture["claude-code"].capture_session_id.return_value = "would-capture"
+    infos = await service.list_sessions(1)
+    assert len(infos) == 1  # no ServiceError escaped