diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 4289905..c54a6f0 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -1,4 +1,5 @@ import asyncio +import contextlib import logging import time from collections.abc import Mapping @@ -61,6 +62,19 @@ class SessionService: 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 def _configurator(self, harness_name: str) -> HarnessConfigurator: configurator = self.harnesses.get(harness_name) @@ -171,30 +185,31 @@ class SessionService: 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, + 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 + 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 @@ -261,33 +276,34 @@ class SessionService: spawn_cfg = configurator.build_spawn_config( project_path, harness_session_id, sess.model ) - since = time.time() - result = await self.tmux.respawn_verified( - window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env - ) - if not result.ok: - log.error( - "Fallback spawn also failed for %s: %s", - window_name, - result.error or "(no output)", + 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 ) - 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", + if not result.ok: + log.error( + "Fallback spawn also failed for %s: %s", window_name, - sess.harness_session_id, + result.error or "(no output)", ) + return False + + # Rung-2 succeeded — capture the new session id if the harness + # supports it so that future resumes target this fresh conversation. + if configurator.captures_session_id: + new_id = await self._capture_session_id_with_retry( + configurator, project_path, since, window_name + ) + if new_id: + sess.harness_session_id = new_id + db.commit() + else: + log.warning( + "capture_session_id exhausted after rung-2 for %s; keeping old id %s", + window_name, + sess.harness_session_id, + ) return True async def stop_session(self, session_id: int) -> None: diff --git a/tests/test_sessions.py b/tests/test_sessions.py index 60eabc2..d5895e1 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -1106,3 +1106,53 @@ async def test_list_sessions_skips_capture_for_deleted_project( harnesses_capture["claude-code"].capture_session_id.return_value = "would-capture" infos = await service.list_sessions(1) assert len(infos) == 1 # no ServiceError escaped + + +# --------------------------------------------------------------------------- +# 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