docs: implementation plan for P2 fixes

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/superpowers/plans/2026-06-10-p2-fixes.md

@@ -0,0 +1,822 @@
+# P2 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 the three P2 findings in `TODO.md` — validate project paths at create/update, harden Codex session-id capture against the multi-session race, and add isolated real-tmux smoke tests.
+
+**Architecture:** Three independent fixes. (1) `ProjectService` gains a `_normalize_path` helper that rejects non-directories. (2) `CodexConfigurator.capture_session_id` refuses to guess when more than one rollout matches, and `SessionService` serializes the spawn→capture window with an `asyncio.Lock` for capturing harnesses. (3) A new `tests/test_tmux_smoke.py` drives the real `TmuxRunner` against a throwaway tmux server isolated via `TMUX_TMPDIR`, auto-skipping when tmux is absent.
+
+**Tech Stack:** Python 3.12, SQLAlchemy 2.0, Textual, pytest + pytest-asyncio, `uv`, tmux.
+
+**Design spec:** `docs/superpowers/specs/2026-06-10-p2-fixes-design.md`
+
+**Conventions for every task:**
+- Run quality gates after code/test changes: `uv run ruff format src tests && uv run ruff check src tests && uv run ty check`.
+- Run tests with `uv run pytest`.
+- Commit messages end with the trailer:
+  ```
+  Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
+  ```
+- **NEVER `git add -A` or `git add .`** — `src/hqt/tmux/runner.py` carries unrelated uncommitted WIP that must stay out of every commit. Always `git add` the exact files listed in the task. No task in this plan modifies `runner.py`.
+- Work on `main`, committing directly to it.
+
+---
+
+## File Structure
+
+| File | Responsibility | Tasks |
+|------|----------------|-------|
+| `src/hqt/projects/service.py` | Project CRUD; add path normalization/validation | 1 |
+| `tests/test_services.py` | ProjectService tests; add path-validation cases | 1 |
+| `src/hqt/harnesses/configurators/codex.py` | Codex capture; refuse ambiguous matches | 2 |
+| `tests/test_harnesses.py` | Codex capture tests; replace earliest-wins cases with ambiguity case | 2 |
+| `src/hqt/sessions/service.py` | Session lifecycle; serialize spawn→capture with a lock | 3 |
+| `tests/test_sessions.py` | SessionService tests; assert lock behavior | 3 |
+| `tests/test_tmux_smoke.py` (new) | Real-tmux smoke tests on an isolated server | 4 |
+| `pyproject.toml` | Register the `tmux` pytest marker | 4 |
+| `TODO.md` | Remove the three resolved P2 lines | 5 |
+
+---
+
+## Task 1: Project path validation
+
+**Files:**
+- Modify: `src/hqt/projects/service.py`
+- Test: `tests/test_services.py`
+
+- [ ] **Step 1: Write the failing tests**
+
+Add these tests to the `TestProjectService` class in `tests/test_services.py`. They need a real directory, so they use pytest's `tmp_path` fixture.
+
+```python
+    def test_create_rejects_nonexistent_path(self, factory, tmp_path):
+        svc = ProjectService(factory)
+        missing = tmp_path / "does-not-exist"
+        with pytest.raises(ServiceError, match="does not exist or is not a directory"):
+            svc.create("p", str(missing))
+
+    def test_create_rejects_file_path(self, factory, tmp_path):
+        svc = ProjectService(factory)
+        a_file = tmp_path / "afile"
+        a_file.write_text("x")
+        with pytest.raises(ServiceError, match="does not exist or is not a directory"):
+            svc.create("p", str(a_file))
+
+    def test_create_stores_resolved_absolute_path(self, factory, tmp_path):
+        svc = ProjectService(factory)
+        p = svc.create("p", str(tmp_path))
+        assert p.path == str(tmp_path.resolve())
+
+    def test_create_expands_user_home(self, factory, tmp_path, monkeypatch):
+        # Point ~ at tmp_path; "~" must expand to the existing tmp_path dir.
+        monkeypatch.setenv("HOME", str(tmp_path))
+        svc = ProjectService(factory)
+        p = svc.create("p", "~")
+        assert p.path == str(tmp_path.resolve())
+
+    def test_update_rejects_nonexistent_path_and_leaves_row(self, factory, tmp_path):
+        svc = ProjectService(factory)
+        p = svc.create("p", str(tmp_path))
+        missing = tmp_path / "nope"
+        with pytest.raises(ServiceError, match="does not exist or is not a directory"):
+            svc.update(p.id, "p", str(missing))
+        assert svc.get(p.id).path == str(tmp_path.resolve())
+```
+
+Several existing tests pass made-up paths like `/tmp/test`, `/a`, `/old/path` that are not guaranteed to exist. Update those literals to real directories so they keep passing. Replace the existing `TestProjectService` methods `test_create_and_get`, `test_list_excludes_archived`, `test_update_name_and_path`, `test_update_duplicate_path_raises`, and `test_create_duplicate_path_raises_service_error` to seed paths from `tmp_path`:
+
+```python
+    def test_create_and_get(self, factory, tmp_path):
+        svc = ProjectService(factory)
+        p = svc.create("test", str(tmp_path))
+        assert p.id is not None
+        assert svc.get(p.id).name == "test"
+
+    def test_list_excludes_archived(self, factory, tmp_path):
+        a = tmp_path / "a"
+        a.mkdir()
+        b = tmp_path / "b"
+        b.mkdir()
+        svc = ProjectService(factory)
+        svc.create("a", str(a))
+        p2 = svc.create("b", str(b))
+        svc.archive(p2.id)
+        assert len(svc.list_all()) == 1
+        assert len(svc.list_all(include_archived=True)) == 2
+
+    def test_update_name_and_path(self, factory, tmp_path):
+        old = tmp_path / "old"
+        old.mkdir()
+        new = tmp_path / "new"
+        new.mkdir()
+        svc = ProjectService(factory)
+        p = svc.create("old", str(old))
+        updated = svc.update(p.id, "new", str(new))
+        assert updated.name == "new"
+        assert updated.path == str(new.resolve())
+        assert svc.get(p.id).path == str(new.resolve())
+
+    def test_update_duplicate_path_raises(self, factory, tmp_path):
+        a = tmp_path / "a"
+        a.mkdir()
+        b = tmp_path / "b"
+        b.mkdir()
+        svc = ProjectService(factory)
+        svc.create("a", str(a))
+        p2 = svc.create("b", str(b))
+        with pytest.raises(ServiceError, match="already uses path"):
+            svc.update(p2.id, "b", str(a))
+        # DB session must remain usable after rollback, values unchanged
+        assert svc.get(p2.id).path == str(b.resolve())
+        assert svc.get(p2.id).name == "b"
+
+    def test_create_duplicate_path_raises_service_error(self, factory, tmp_path):
+        dup = tmp_path / "dup"
+        dup.mkdir()
+        svc = ProjectService(factory)
+        svc.create("a", str(dup))
+        with pytest.raises(ServiceError, match="already uses path"):
+            svc.create("b", str(dup))
+```
+
+Note: `TestMcpService.test_bind_unbind` creates a project at `/proj`; change that literal to `str(tmp_path)` and add `tmp_path` to its signature:
+
+```python
+    def test_bind_unbind(self, factory, tmp_path):
+        psvc = ProjectService(factory)
+        msvc = McpService(factory)
+        project = psvc.create("proj", str(tmp_path))
+        server = msvc.create("s1", "stdio", command="cmd")
+        msvc.bind_to_project(project.id, server.id)
+        assert len(msvc.get_project_mcps(project.id)) == 1
+        msvc.unbind_from_project(project.id, server.id)
+        assert len(msvc.get_project_mcps(project.id)) == 0
+```
+
+- [ ] **Step 2: Run the new tests to verify they fail**
+
+Run: `uv run pytest tests/test_services.py -k "rejects or resolved or expands_user" -v`
+Expected: FAIL — `ServiceError` not raised / path not resolved (validation not implemented yet).
+
+- [ ] **Step 3: Implement `_normalize_path` and call it from `create`/`update`**
+
+In `src/hqt/projects/service.py`, add the `Path` import and the helper, and call it at the top of `create` and `update`. Full file:
+
+```python
+from pathlib import Path
+
+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):
+        # factory must be created with expire_on_commit=False so rows returned
+        # from these methods stay readable after their session closes.
+        self.factory = factory
+
+    def _normalize_path(self, raw: str) -> str:
+        """Expand ~, require an existing directory, return the resolved absolute path.
+
+        Raises ServiceError if the path does not exist or is not a directory, so
+        the TUI surfaces it as a notification instead of letting the bad path
+        become a dead session later.
+        """
+        path = Path(raw).expanduser()
+        if not path.is_dir():
+            raise ServiceError(f"Path does not exist or is not a directory: {raw}")
+        return str(path.resolve())
+
+    def create(self, name: str, path: str) -> Project:
+        path = self._normalize_path(path)
+        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:
+        path = self._normalize_path(path)
+        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()
+```
+
+Note ordering: `_normalize_path` runs **before** opening the DB session, so a bad path is rejected without touching the DB. `update` validates the path before checking the project exists — that's fine; both raise `ServiceError`.
+
+- [ ] **Step 4: Run the full service test file**
+
+Run: `uv run pytest tests/test_services.py -v`
+Expected: PASS (new validation tests + all updated existing tests).
+
+- [ ] **Step 5: Run quality gates**
+
+Run: `uv run ruff format src tests && uv run ruff check src tests && uv run ty check`
+Expected: all green.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/hqt/projects/service.py tests/test_services.py
+git commit -m "feat: validate project paths at create/update
+
+Reject non-existent or non-directory paths with ServiceError (already
+surfaced as a TUI notification), and store the resolved absolute path.
+Closes the P2 finding where bad paths became dead sessions later.
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 2: Codex capture refuses ambiguous matches
+
+**Files:**
+- Modify: `src/hqt/harnesses/configurators/codex.py`
+- Test: `tests/test_harnesses.py`
+
+**Context:** `capture_session_id` scans `~/.codex/sessions/rollout-*.jsonl` for rollouts whose `cwd` matches the project and whose start time is `>= since`. Today it sorts the matches and returns the earliest. When two Codex sessions ran in the same directory in the capture window, that guess can be the wrong conversation. The fix: if more than one rollout matches, refuse to guess (return `None` + warn). The single-match normal case is unchanged.
+
+- [ ] **Step 1: Replace the earliest-wins tests with an ambiguity test**
+
+In `tests/test_harnesses.py`:
+
+1. **Delete** `test_codex_capture_first_post_since_cwd_match_wins` (it asserts the now-removed earliest-wins behavior with two matching rollouts).
+2. **Delete** `test_codex_capture_prefers_started_nearest_since_over_newest_file` (also asserts earliest-wins among two matching rollouts).
+3. **Add** this test (place it where the deleted ones were):
+
+```python
+def test_codex_capture_ambiguous_returns_none(tmp_path):
+    """Two rollouts match cwd + since window -> refuse to guess, return None."""
+    sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
+    sessions_dir.mkdir(parents=True)
+
+    rollout_a = sessions_dir / "rollout-a.jsonl"
+    rollout_b = sessions_dir / "rollout-b.jsonl"
+    _write_rollout(rollout_a, "id-a", "/projects/foo")
+    _write_rollout(rollout_b, "id-b", "/projects/foo")
+
+    now = time.time()
+    since = now - 120
+    os.utime(rollout_a, (now - 30, now - 30))
+    os.utime(rollout_b, (now - 10, now - 10))
+
+    c = CodexConfigurator()
+    with patch("hqt.harnesses.configurators.codex.Path.home", return_value=tmp_path):
+        result = c.capture_session_id(Path("/projects/foo"), since)
+    assert result is None
+```
+
+The single-match regression cases (`test_codex_capture_session_id`,
+`test_codex_capture_accepts_file_after_since`,
+`test_codex_capture_skips_non_matching_cwd_picks_older_matching` — which has only
+one cwd-matching rollout) stay as-is and must keep passing.
+
+- [ ] **Step 2: Run the capture tests to verify the new one fails**
+
+Run: `uv run pytest tests/test_harnesses.py -k "capture" -v`
+Expected: `test_codex_capture_ambiguous_returns_none` FAILS (current code returns `"id-a"`, not `None`). The deleted tests are gone.
+
+- [ ] **Step 3: Implement the ambiguity guard**
+
+In `src/hqt/harnesses/configurators/codex.py`, add logging and change `capture_session_id`. Replace the imports block at the top:
+
+```python
+import json
+import logging
+from datetime import datetime
+from pathlib import Path
+
+from hqt.harnesses.base import HarnessConfigurator, SpawnConfig
+
+log = logging.getLogger(__name__)
+```
+
+Replace the tail of `capture_session_id` (the `candidates.sort(...)` / return lines) with the ambiguity guard. The full method:
+
+```python
+    def capture_session_id(self, project_path: Path, since: float) -> str | None:
+        sessions_dir = Path.home() / ".codex" / "sessions"
+        if not sessions_dir.exists():
+            return None
+        candidates: list[tuple[float, str]] = []
+        for rollout in sessions_dir.rglob("rollout-*.jsonl"):
+            try:
+                mtime = rollout.stat().st_mtime
+                with rollout.open() as f:
+                    meta = json.loads(f.readline())
+                payload = meta["payload"]
+                if meta.get("type") != "session_meta" or str(
+                    Path(payload["cwd"])
+                ) != str(project_path):
+                    continue
+                started_at = self._meta_timestamp(payload) or mtime
+                if started_at >= since:
+                    candidates.append((started_at, payload["id"]))
+            except (json.JSONDecodeError, KeyError, OSError, TypeError):
+                continue
+        if len(candidates) > 1:
+            # Ambiguous: more than one Codex rollout matches this cwd + time
+            # window (e.g. a second Codex started in the same project). Guessing
+            # risks storing the wrong conversation id, so keep the placeholder.
+            log.warning(
+                "capture_session_id: %d rollouts match cwd=%s since=%s; refusing to guess",
+                len(candidates),
+                project_path,
+                since,
+            )
+            return None
+        return candidates[0][1] if candidates else None
+```
+
+(The `candidates.sort(...)` line is removed — with at most one returned candidate there is nothing to sort.)
+
+- [ ] **Step 4: Run the capture tests**
+
+Run: `uv run pytest tests/test_harnesses.py -k "capture" -v`
+Expected: PASS, including `test_codex_capture_ambiguous_returns_none`.
+
+- [ ] **Step 5: Run quality gates**
+
+Run: `uv run ruff format src tests && uv run ruff check src tests && uv run ty check`
+Expected: all green.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add src/hqt/harnesses/configurators/codex.py tests/test_harnesses.py
+git commit -m "fix: refuse ambiguous Codex session-id capture
+
+When more than one rollout matches the project cwd within the capture
+window, return None and warn instead of guessing the earliest. The worst
+case is now a kept placeholder id, never a wrong one. Part of the P2
+capture-race fix.
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 3: Serialize the spawn→capture window in SessionService
+
+**Files:**
+- Modify: `src/hqt/sessions/service.py`
+- Test: `tests/test_sessions.py`
+
+**Context:** The Task 2 guard makes wrong captures impossible, but two hqt-launched Codex sessions starting in the same project would now each see an ambiguous window and capture nothing. An `asyncio.Lock` held across the spawn→capture region (only for harnesses that `captures_session_id`) keeps hqt-launched starts from overlapping, so each capture sees exactly its own rollout.
+
+- [ ] **Step 1: Write the failing tests**
+
+Add to `tests/test_sessions.py`. The first asserts the lock is held while a capturing harness spawns+captures; the second asserts it is NOT held for a non-capturing harness (so we only pay the serialization cost when needed).
+
+```python
+@pytest.mark.asyncio
+async def test_create_session_holds_capture_lock_for_capturing_harness(
+    factory, db, tmux
+):
+    observed = {}
+
+    def _capture(project_path, since):
+        # capture runs inside the spawn->capture critical section
+        observed["locked_during_capture"] = service._capture_lock.locked()
+        return "real-codex-id"
+
+    h = MagicMock()
+    h.captures_session_id = True
+    h.generate_session_id.return_value = "placeholder"
+    h.build_spawn_config.return_value = MagicMock(
+        command=["codex"], env={}, cwd=Path("/tmp/myproj")
+    )
+    h.capture_session_id.side_effect = _capture
+    service = SessionService(factory=factory, tmux=tmux, harnesses={"claude-code": h})
+
+    result = await service.create_session(project_id=1, harness_name="claude-code")
+
+    assert observed["locked_during_capture"] is True
+    assert result.session.harness_session_id == "real-codex-id"
+    # Lock is released after create_session returns.
+    assert service._capture_lock.locked() is False
+
+
+@pytest.mark.asyncio
+async def test_create_session_no_lock_for_non_capturing_harness(service, db, tmux):
+    observed = {}
+
+    async def _spawn(req):
+        observed["locked_during_spawn"] = service._capture_lock.locked()
+        return SpawnResult(ok=True, window_id="@1", error="")
+
+    tmux.spawn.side_effect = _spawn
+
+    await service.create_session(project_id=1, harness_name="claude-code")
+
+    # The `service` fixture's harness has captures_session_id = False, so the
+    # spawn must NOT be serialized under the capture lock.
+    assert observed["locked_during_spawn"] is False
+```
+
+- [ ] **Step 2: Run the new tests to verify they fail**
+
+Run: `uv run pytest tests/test_sessions.py -k "capture_lock or no_lock" -v`
+Expected: FAIL with `AttributeError: 'SessionService' object has no attribute '_capture_lock'`.
+
+- [ ] **Step 3: Add the lock and the conditional guard helper**
+
+In `src/hqt/sessions/service.py`:
+
+Add `import contextlib` near the top imports (alongside `import asyncio`):
+
+```python
+import asyncio
+import contextlib
+import logging
+import time
+```
+
+Add the lock in `__init__` and a helper context manager. Update `__init__`:
+
+```python
+    def __init__(
+        self,
+        factory: sessionmaker,
+        tmux: TmuxManager,
+        harnesses: Mapping[str, HarnessConfigurator],
+    ):
+        self.factory = factory
+        self.tmux = tmux
+        self.harnesses = harnesses
+        # Serializes the spawn->capture window for harnesses that capture a
+        # session id (codex), so two concurrent hqt starts in the same project
+        # never overlap and confuse capture_session_id.
+        self._capture_lock = asyncio.Lock()
+
+    @contextlib.asynccontextmanager
+    async def _maybe_capture_lock(self, needed: bool):
+        """Hold the capture lock only when the harness captures a session id."""
+        if needed:
+            async with self._capture_lock:
+                yield
+        else:
+            yield
+```
+
+- [ ] **Step 4: Wrap the spawn→capture region in `create_session`**
+
+In `create_session`, wrap from `since = time.time()` through the capture block in the guard. The relevant region becomes (note the added `async with` and the indentation of the lines it now encloses):
+
+```python
+            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,
+            )
+            async with self._maybe_capture_lock(configurator.captures_session_id):
+                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()
+```
+
+`spawn_result` is referenced after the `async with` block (in the `return CreateSessionResult(...)` that follows) — it remains in scope because `async with` does not introduce a new scope. Leave the rest of the method (the `log.info` and `return CreateSessionResult(...)`) unchanged.
+
+- [ ] **Step 5: Wrap the rung-2 fresh-spawn region in `_respawn_with_fallback`**
+
+In `_respawn_with_fallback`, wrap the rung-2 region (from `since = time.time()` through the capture block) in the same guard. The rung-2 portion becomes:
+
+```python
+        # 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
+        )
+        async with self._maybe_capture_lock(configurator.captures_session_id):
+            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
+```
+
+The `return False` inside the `async with` correctly releases the lock on the way out. Leave rung 1 (the resume attempt above this region) unchanged — it does not spawn a new conversation, so it needs no lock.
+
+- [ ] **Step 6: Run the session tests**
+
+Run: `uv run pytest tests/test_sessions.py -v`
+Expected: PASS (new lock tests + all existing session tests).
+
+- [ ] **Step 7: Run quality gates**
+
+Run: `uv run ruff format src tests && uv run ruff check src tests && uv run ty check`
+Expected: all green.
+
+- [ ] **Step 8: Commit**
+
+```bash
+git add src/hqt/sessions/service.py tests/test_sessions.py
+git commit -m "fix: serialize Codex spawn->capture with a lock
+
+Hold an asyncio.Lock across the spawn->capture window for harnesses that
+capture a session id, so two concurrent hqt starts in the same project
+never produce overlapping capture windows. Completes the P2 capture-race
+fix alongside the ambiguity guard.
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 4: Real-tmux smoke tests
+
+**Files:**
+- Create: `tests/test_tmux_smoke.py`
+- Modify: `pyproject.toml`
+
+**Context:** tmux behavior is otherwise tested with mocked `_exec`. These tests drive the real `TmuxRunner` against a throwaway tmux server isolated via `TMUX_TMPDIR`, so they never touch the user's live sessions, and skip when tmux is absent.
+
+- [ ] **Step 1: Register the `tmux` pytest marker**
+
+In `pyproject.toml`, add a `[tool.pytest.ini_options]` block (none exists yet — add it after the `[dependency-groups]` block, before `[tool.hqt.quality]`):
+
+```toml
+[tool.pytest.ini_options]
+markers = [
+    "tmux: real-tmux smoke tests; require a tmux binary and run on an isolated TMUX_TMPDIR server",
+]
+```
+
+Note: the existing async tests already run with explicit `@pytest.mark.asyncio` decorators under pytest-asyncio's default strict mode, so do **not** add an `asyncio_mode` setting — only the `markers` entry is needed here.
+
+- [ ] **Step 2: Write the smoke tests**
+
+Create `tests/test_tmux_smoke.py`:
+
+```python
+"""Real-tmux smoke tests.
+
+These drive the actual TmuxRunner against a throwaway tmux server whose socket
+lives in a temp TMUX_TMPDIR, so they never touch the user's live tmux sessions.
+They skip when no tmux binary is available.
+"""
+
+import asyncio
+import os
+import shutil
+import subprocess
+
+import pytest
+
+from hqt.tmux.runner import TmuxRunner
+
+pytestmark = [
+    pytest.mark.tmux,
+    pytest.mark.skipif(shutil.which("tmux") is None, reason="tmux not installed"),
+]
+
+SESSION = "hqt-test"
+
+
+def _tmux(env, *args):
+    """Run a raw tmux command against the isolated server, return stdout."""
+    return subprocess.run(
+        ["tmux", *args],
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+@pytest.fixture
+def tmux_env(tmp_path, monkeypatch):
+    # Point tmux's server socket at the temp dir: a fully isolated server.
+    monkeypatch.setenv("TMUX_TMPDIR", str(tmp_path))
+    env = {**os.environ, "TMUX_TMPDIR": str(tmp_path)}
+    _tmux(env, "new-session", "-d", "-s", SESSION, "-x", "200", "-y", "50")
+    try:
+        yield env
+    finally:
+        _tmux(env, "kill-server")
+
+
+@pytest.fixture
+def runner(tmux_env):
+    return TmuxRunner(tmux_path="tmux", session_name=SESSION)
+
+
+async def _wait_pane_dead(runner, window, timeout=2.0):
+    deadline = asyncio.get_running_loop().time() + timeout
+    while asyncio.get_running_loop().time() < deadline:
+        if await runner.is_pane_dead(window):
+            return True
+        await asyncio.sleep(0.05)
+    return await runner.is_pane_dead(window)
+
+
+@pytest.mark.asyncio
+async def test_new_window_appears_in_list(runner, tmp_path):
+    window_id = await runner.new_window("hqt-1", str(tmp_path), "sleep 60")
+    assert window_id is not None
+    assert "hqt-1" in await runner.list_windows()
+
+
+@pytest.mark.asyncio
+async def test_set_window_label_round_trips(runner, tmp_path, tmux_env):
+    await runner.new_window("hqt-1", str(tmp_path), "sleep 60")
+    await runner.set_window_label("hqt-1", "•hqt-1")
+    shown = _tmux(
+        tmux_env,
+        "show-options",
+        "-w",
+        "-t",
+        f"{SESSION}:=hqt-1",
+        "@hqt_label",
+    ).stdout
+    assert "•hqt-1" in shown
+
+
+@pytest.mark.asyncio
+async def test_respawn_revives_dead_pane(runner, tmp_path):
+    # Window whose command exits immediately -> pane dies (remain-on-exit keeps it).
+    await runner.new_window("hqt-1", str(tmp_path), "true")
+    assert await _wait_pane_dead(runner, "hqt-1") is True
+    assert await runner.respawn_pane("hqt-1", "sleep 60", str(tmp_path)) is True
+    alive, _ = await runner.verify_window_alive("hqt-1", timeout=1.0)
+    assert alive is True
+
+
+@pytest.mark.asyncio
+async def test_apply_theme_sets_session_options(runner, tmux_env):
+    await runner.apply_theme()
+    shown = _tmux(tmux_env, "show-options", "-t", SESSION, "status-justify").stdout
+    assert "left" in shown
+```
+
+- [ ] **Step 3: Run the smoke tests**
+
+Run: `uv run pytest tests/test_tmux_smoke.py -v`
+Expected: PASS where tmux is installed (4 tests). If tmux is absent: all SKIPPED with reason "tmux not installed".
+
+- [ ] **Step 4: Verify the marker is registered (no warnings)**
+
+Run: `uv run pytest tests/test_tmux_smoke.py -v -W error::pytest.PytestUnknownMarkWarning`
+Expected: PASS/SKIP with no `PytestUnknownMarkWarning` (proves the marker is registered).
+
+- [ ] **Step 5: Run the full suite + quality gates**
+
+Run: `uv run pytest && uv run ruff format src tests && uv run ruff check src tests && uv run ty check`
+Expected: full suite green (smoke tests pass or skip), gates green.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add tests/test_tmux_smoke.py pyproject.toml
+git commit -m "test: add isolated real-tmux smoke tests
+
+Drive the real TmuxRunner against a throwaway tmux server isolated via
+TMUX_TMPDIR (never touches live sessions), covering new windows, label
+round-trip, respawn, and theming. Auto-skip when tmux is absent; register
+the 'tmux' marker. Closes the P2 real-tmux testing gap.
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+## Task 5: Remove resolved P2 findings from TODO.md
+
+**Files:**
+- Modify: `TODO.md`
+
+**Context:** All three P2 findings are now resolved. `TODO.md` currently contains exactly those three lines and nothing else.
+
+- [ ] **Step 1: Empty out the resolved findings**
+
+`TODO.md` currently reads:
+
+```
+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.
+```
+
+Replace the entire file contents with a single placeholder line so the file is not left dangling:
+
+```
+No open findings.
+```
+
+- [ ] **Step 2: Verify**
+
+Run: `cat TODO.md`
+Expected: `No open findings.`
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add TODO.md
+git commit -m "chore: clear resolved P2 findings from TODO
+
+Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>"
+```
+
+---
+
+## Final verification (after all tasks)
+
+- [ ] Run the full suite and gates once more:
+  ```
+  uv run pytest && uv run ruff format src tests && uv run ruff check src tests && uv run ty check
+  ```
+  Expected: all tests pass (or smoke tests skip without tmux), all gates green.
+- [ ] Confirm `src/hqt/tmux/runner.py` is still only modified in the working tree (its pre-existing WIP) and was never committed:
+  ```
+  git log --oneline -6
+  git status --short
+  ```
+  Expected: the six new commits are present; `git status --short` shows only ` M src/hqt/tmux/runner.py`.