Merge branch 'main' into worktree-sessions

# Conflicts:
#	src/hqt/sessions/service.py
#	tests/test_sessions.py

tests/test_harnesses.py

@@ -1,7 +1,6 @@
 import json
 import os
 import time
-from datetime import UTC, datetime
 from pathlib import Path
 from unittest.mock import patch
 
@@ -184,57 +183,25 @@ def test_codex_capture_accepts_file_after_since(tmp_path):
     assert result == "new-id"
 
 
-def test_codex_capture_first_post_since_cwd_match_wins(tmp_path):
-    """Among multiple post-since rollouts, the one started nearest since wins."""
+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_older = sessions_dir / "rollout-older.jsonl"
-    rollout_newer = sessions_dir / "rollout-newer.jsonl"
-    _write_rollout(rollout_older, "older-id", "/projects/foo")
-    _write_rollout(rollout_newer, "newer-id", "/projects/foo")
+    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_older, (now - 30, now - 30))
-    os.utime(rollout_newer, (now - 10, now - 10))
+    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 == "older-id"
-
-
-def test_codex_capture_prefers_started_nearest_since_over_newest_file(tmp_path):
-    """Late capture should match the Codex session spawned at since, not a later same-cwd session."""
-    sessions_dir = tmp_path / ".codex" / "sessions" / "dir"
-    sessions_dir.mkdir(parents=True)
-
-    target = sessions_dir / "rollout-target.jsonl"
-    unrelated_later = sessions_dir / "rollout-later.jsonl"
-
-    since = datetime(2026, 6, 10, 21, 9, 15, tzinfo=UTC).timestamp()
-    _write_rollout(
-        target,
-        "target-id",
-        "/projects/foo",
-        timestamp="2026-06-10T21:09:16.000Z",
-    )
-    _write_rollout(
-        unrelated_later,
-        "later-id",
-        "/projects/foo",
-        timestamp="2026-06-10T21:45:45.000Z",
-    )
-
-    now = time.time()
-    os.utime(target, (now - 60, now - 60))
-    os.utime(unrelated_later, (now, now))
-
-    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 == "target-id"
+    assert result is None
 
 
 def test_codex_capture_skips_non_matching_cwd_picks_older_matching(tmp_path):

tests/test_integration.py

@@ -43,11 +43,11 @@ def setup():
 
 
 @pytest.mark.asyncio
-async def test_full_lifecycle(setup):
+async def test_full_lifecycle(setup, tmp_path):
     proj_svc, sess_svc, tmux = setup
 
     # Create project
-    project = proj_svc.create(name="test-proj", path="/tmp/test-proj")
+    project = proj_svc.create(name="test-proj", path=str(tmp_path))
     assert project.id == 1
 
     # Create session

tests/test_services.py

@@ -24,48 +24,95 @@ def factory():
 
 
 class TestProjectService:
-    def test_create_and_get(self, factory):
+    def test_create_and_get(self, factory, tmp_path):
         svc = ProjectService(factory)
-        p = svc.create("test", "/tmp/test")
+        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):
+    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", "/a")
-        p2 = svc.create("b", "/b")
+        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):
+    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", "/old/path")
-        updated = svc.update(p.id, "new", "/new/path")
+        p = svc.create("old", str(old))
+        updated = svc.update(p.id, "new", str(new))
         assert updated.name == "new"
-        assert updated.path == "/new/path"
-        assert svc.get(p.id).path == "/new/path"
+        assert updated.path == str(new.resolve())
+        assert svc.get(p.id).path == str(new.resolve())
 
-    def test_update_duplicate_path_raises(self, factory):
+    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", "/a")
-        p2 = svc.create("b", "/b")
+        svc.create("a", str(a))
+        p2 = svc.create("b", str(b))
         with pytest.raises(ServiceError, match="already uses path"):
-            svc.update(p2.id, "b", "/a")
+            svc.update(p2.id, "b", str(a))
         # DB session must remain usable after rollback, values unchanged
-        assert svc.get(p2.id).path == "/b"
+        assert svc.get(p2.id).path == str(b.resolve())
         assert svc.get(p2.id).name == "b"
 
-    def test_update_unknown_id_raises(self, factory):
+    def test_update_unknown_id_raises(self, factory, tmp_path):
         svc = ProjectService(factory)
         with pytest.raises(ServiceError, match="not found"):
-            svc.update(9999, "x", "/x")
+            svc.update(9999, "x", str(tmp_path))
 
-    def test_create_duplicate_path_raises_service_error(self, factory):
+    def test_create_duplicate_path_raises_service_error(self, factory, tmp_path):
+        dup = tmp_path / "dup"
+        dup.mkdir()
         svc = ProjectService(factory)
-        svc.create("a", "/dup")
+        svc.create("a", str(dup))
         with pytest.raises(ServiceError, match="already uses path"):
-            svc.create("b", "/dup")
+            svc.create("b", str(dup))
+
+    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())
 
 
 class TestMcpService:
@@ -78,10 +125,10 @@ class TestMcpService:
         svc.delete("test-mcp")
         assert svc.get("test-mcp") is None
 
-    def test_bind_unbind(self, factory):
+    def test_bind_unbind(self, factory, tmp_path):
         psvc = ProjectService(factory)
         msvc = McpService(factory)
-        project = psvc.create("proj", "/proj")
+        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

tests/test_sessions.py

@@ -1468,3 +1468,53 @@ async def test_repo_path_fallback_slash_branch_when_project_gone(
 
     # The fallback must recover the real repo, not <repo>/.worktrees.
     assert fake_worktree.state_calls == [(repo, wt_path, branch)]
+
+
+# ---------------------------------------------------------------------------
+# Task 3 (P2): serialize the spawn->capture window with a lock
+# ---------------------------------------------------------------------------
+
+
+@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

tests/test_tmux_smoke.py

@@ -0,0 +1,115 @@
+"""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 the result."""
+    return subprocess.run(
+        ["tmux", *args],
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+@pytest.fixture
+def tmux_env(tmp_path, monkeypatch):
+    # Isolate tmux completely from the user's live server:
+    #  - TMUX_TMPDIR points the default socket at a throwaway temp dir.
+    #  - TMUX / TMUX_PANE are unset. When TMUX is set (i.e. the test runner is
+    #    itself inside a tmux session), tmux talks to THAT server and ignores
+    #    TMUX_TMPDIR, so new-session/kill-server would hit the user's live
+    #    server and evict them. delenv removes them from os.environ so both the
+    #    raw `_tmux` calls AND the TmuxRunner subprocesses (which inherit
+    #    os.environ) stay on the isolated server.
+    monkeypatch.setenv("TMUX_TMPDIR", str(tmp_path))
+    monkeypatch.delenv("TMUX", raising=False)
+    monkeypatch.delenv("TMUX_PANE", raising=False)
+    env = {**os.environ, "TMUX_TMPDIR": str(tmp_path)}
+    env.pop("TMUX", None)
+    env.pop("TMUX_PANE", None)
+    _tmux(env, "new-session", "-d", "-s", SESSION, "-x", "200", "-y", "50")
+    # Tripwire: a freshly isolated server has exactly one session. If isolation
+    # silently failed we'd see the user's sessions here — fail loudly BEFORE any
+    # test does something destructive. (kill-server in teardown still targets
+    # only the isolated socket, because env has TMUX removed + TMUX_TMPDIR set.)
+    listed = _tmux(env, "list-sessions", "-F", "#{session_name}").stdout.split()
+    assert listed == [SESSION], f"tmux isolation failed; saw sessions: {listed}"
+    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

tests/test_tui.py

@@ -626,12 +626,12 @@ async def test_project_form_add_mode_defaults_name_to_basename():
 
 
 @pytest.mark.asyncio
-async def test_project_list_get_selected_project_id():
+async def test_project_list_get_selected_project_id(tmp_path):
     app = HqtApp()
     async with app.run_test(size=(120, 40)) as pilot:
         from hqt.tui.widgets.project_list import ProjectList
 
-        p = app._project_service.create("selproj", "/tmp/selproj")
+        p = app._project_service.create("selproj", str(tmp_path))
         app._load_projects()
         await app.workers.wait_for_complete()
         await pilot.pause()
@@ -676,14 +676,22 @@ async def test_edit_project_no_selection_warns():
 
 
 @pytest.mark.asyncio
-async def test_edit_project_opens_prefilled_form_and_updates():
+async def test_edit_project_opens_prefilled_form_and_updates(tmp_path):
     app = HqtApp()
     async with app.run_test(size=(120, 40)) as pilot:
+        from pathlib import Path
         from textual.widgets import Input
         from hqt.tui.screens.project_form import ProjectFormScreen
         from hqt.tui.widgets.project_list import ProjectList
 
-        p = app._project_service.create("oldname", "/tmp/oldpath")
+        old_dir = tmp_path / "oldpath"
+        old_dir.mkdir()
+        new_dir = tmp_path / "newpath"
+        new_dir.mkdir()
+        old_path = str(old_dir)
+        new_path = str(new_dir)
+
+        p = app._project_service.create("oldname", old_path)
         app._load_projects()
         await app.workers.wait_for_complete()
         await pilot.pause()
@@ -699,25 +707,35 @@ async def test_edit_project_opens_prefilled_form_and_updates():
 
         assert isinstance(app.screen, ProjectFormScreen)
         assert app.screen.query_one("#name-input", Input).value == "oldname"
-        assert app.screen.query_one("#path-input", Input).value == "/tmp/oldpath"
+        assert app.screen.query_one("#path-input", Input).value == str(
+            Path(old_path).resolve()
+        )
 
-        app.screen.dismiss(("newname", "/tmp/newpath"))
+        app.screen.dismiss(("newname", new_path))
         await pilot.pause()
 
         refreshed = app._project_service.get(p.id)
         assert refreshed.name == "newname"
-        assert refreshed.path == "/tmp/newpath"
+        assert refreshed.path == str(Path(new_path).resolve())
 
 
 @pytest.mark.asyncio
-async def test_edit_project_duplicate_path_notifies_error():
+async def test_edit_project_duplicate_path_notifies_error(tmp_path):
     app = HqtApp()
     async with app.run_test(size=(120, 40)) as pilot:
+        from pathlib import Path
         from unittest.mock import patch
         from hqt.tui.widgets.project_list import ProjectList
 
-        app._project_service.create("first", "/tmp/first")
-        p2 = app._project_service.create("second", "/tmp/second")
+        first_dir = tmp_path / "first"
+        first_dir.mkdir()
+        second_dir = tmp_path / "second"
+        second_dir.mkdir()
+        first_path = str(first_dir)
+        second_path = str(second_dir)
+
+        app._project_service.create("first", first_path)
+        p2 = app._project_service.create("second", second_path)
         app._load_projects()
         await app.workers.wait_for_complete()
         await pilot.pause()
@@ -736,7 +754,8 @@ async def test_edit_project_duplicate_path_notifies_error():
         ):
             await app.run_action("edit_project")
             await pilot.pause()
-            app.screen.dismiss(("second", "/tmp/first"))
+            # Move "second" onto "first"'s path to trigger the duplicate error.
+            app.screen.dismiss(("second", first_path))
             await pilot.pause()
 
         assert any(
@@ -744,7 +763,7 @@ async def test_edit_project_duplicate_path_notifies_error():
             for msg, kw in notify_calls
         ), f"Expected duplicate-path error notification, got: {notify_calls}"
         # DB unchanged
-        assert app._project_service.get(p2.id).path == "/tmp/second"
+        assert app._project_service.get(p2.id).path == str(Path(second_path).resolve())
 
 
 # ---------------------------------------------------------------------------
@@ -889,14 +908,18 @@ async def test_dialog_button_focus_is_visibly_highlighted():
 
 
 @pytest.mark.asyncio
-async def test_first_project_selected_on_load():
+async def test_first_project_selected_on_load(tmp_path):
     """If a project exists, the first is selected on load with no key press."""
     app = HqtApp()
     async with app.run_test(size=(120, 40)) as pilot:
         from hqt.tui.widgets.project_list import ProjectList
 
-        app._project_service.create("p1", "/tmp/p1-auto")
-        app._project_service.create("p2", "/tmp/p2-auto")
+        d1 = tmp_path / "p1-auto"
+        d1.mkdir()
+        d2 = tmp_path / "p2-auto"
+        d2.mkdir()
+        app._project_service.create("p1", str(d1))
+        app._project_service.create("p2", str(d2))
         app._load_projects()
         await app.workers.wait_for_complete()
         await pilot.pause()
@@ -1159,14 +1182,18 @@ async def test_rename_session_opens_prefilled_and_updates():
 
 
 @pytest.mark.asyncio
-async def test_project_list_jk_navigation():
+async def test_project_list_jk_navigation(tmp_path):
     """j/k move the highlight down/up in the project list like the arrow keys."""
     app = HqtApp()
     async with app.run_test(size=(120, 40)) as pilot:
         from hqt.tui.widgets.project_list import ProjectList
 
-        p1 = app._project_service.create("first", "/tmp/jk-first")
-        p2 = app._project_service.create("second", "/tmp/jk-second")
+        d1 = tmp_path / "jk-first"
+        d1.mkdir()
+        d2 = tmp_path / "jk-second"
+        d2.mkdir()
+        p1 = app._project_service.create("first", str(d1))
+        p2 = app._project_service.create("second", str(d2))
         app._load_projects()
         await app.workers.wait_for_complete()
         await pilot.pause()