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>
This commit is contained in:
+63
-47
@@ -1,4 +1,5 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
|
import contextlib
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from collections.abc import Mapping
|
from collections.abc import Mapping
|
||||||
@@ -61,6 +62,19 @@ class SessionService:
|
|||||||
self.factory = factory
|
self.factory = factory
|
||||||
self.tmux = tmux
|
self.tmux = tmux
|
||||||
self.harnesses = harnesses
|
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:
|
def _configurator(self, harness_name: str) -> HarnessConfigurator:
|
||||||
configurator = self.harnesses.get(harness_name)
|
configurator = self.harnesses.get(harness_name)
|
||||||
@@ -171,30 +185,31 @@ class SessionService:
|
|||||||
spawn_cfg.command,
|
spawn_cfg.command,
|
||||||
spawn_cfg.cwd,
|
spawn_cfg.cwd,
|
||||||
)
|
)
|
||||||
since = time.time()
|
async with self._maybe_capture_lock(configurator.captures_session_id):
|
||||||
spawn_result = await self.tmux.spawn(
|
since = time.time()
|
||||||
SpawnRequest(
|
spawn_result = await self.tmux.spawn(
|
||||||
window_name=sess.tmux_session_name,
|
SpawnRequest(
|
||||||
command=spawn_cfg.command,
|
window_name=sess.tmux_session_name,
|
||||||
cwd=str(spawn_cfg.cwd),
|
command=spawn_cfg.command,
|
||||||
env=spawn_cfg.env,
|
cwd=str(spawn_cfg.cwd),
|
||||||
|
env=spawn_cfg.env,
|
||||||
|
)
|
||||||
)
|
)
|
||||||
)
|
if not spawn_result.ok:
|
||||||
if not spawn_result.ok:
|
log.error(
|
||||||
log.error(
|
"Failed to spawn window %s: %s",
|
||||||
"Failed to spawn window %s: %s",
|
sess.tmux_session_name,
|
||||||
sess.tmux_session_name,
|
spawn_result.error or "(no output captured)",
|
||||||
spawn_result.error or "(no output captured)",
|
)
|
||||||
)
|
# Only attempt capture when the spawn actually succeeded; a failed
|
||||||
# Only attempt capture when the spawn actually succeeded; a failed
|
# spawn could inadvertently capture an unrelated session running in
|
||||||
# spawn could inadvertently capture an unrelated session running in
|
# the same cwd.
|
||||||
# the same cwd.
|
if spawn_result.ok and configurator.captures_session_id:
|
||||||
if spawn_result.ok and configurator.captures_session_id:
|
captured_id = await self._capture_session_id_with_retry(
|
||||||
captured_id = await self._capture_session_id_with_retry(
|
configurator, project_path, since, sess.tmux_session_name
|
||||||
configurator, project_path, since, sess.tmux_session_name
|
)
|
||||||
)
|
if captured_id:
|
||||||
if captured_id:
|
sess.harness_session_id = captured_id
|
||||||
sess.harness_session_id = captured_id
|
|
||||||
db.commit()
|
db.commit()
|
||||||
log.info(
|
log.info(
|
||||||
"Session created: id=%s window=%s", sess.id, sess.tmux_session_name
|
"Session created: id=%s window=%s", sess.id, sess.tmux_session_name
|
||||||
@@ -261,33 +276,34 @@ class SessionService:
|
|||||||
spawn_cfg = configurator.build_spawn_config(
|
spawn_cfg = configurator.build_spawn_config(
|
||||||
project_path, harness_session_id, sess.model
|
project_path, harness_session_id, sess.model
|
||||||
)
|
)
|
||||||
since = time.time()
|
async with self._maybe_capture_lock(configurator.captures_session_id):
|
||||||
result = await self.tmux.respawn_verified(
|
since = time.time()
|
||||||
window_name, spawn_cfg.command, str(spawn_cfg.cwd), env=spawn_cfg.env
|
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
|
if not result.ok:
|
||||||
|
log.error(
|
||||||
# Rung-2 succeeded — capture the new session id if the harness supports
|
"Fallback spawn also failed for %s: %s",
|
||||||
# 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,
|
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
|
return True
|
||||||
|
|
||||||
async def stop_session(self, session_id: int) -> None:
|
async def stop_session(self, session_id: int) -> None:
|
||||||
|
|||||||
@@ -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"
|
harnesses_capture["claude-code"].capture_session_id.return_value = "would-capture"
|
||||||
infos = await service.list_sessions(1)
|
infos = await service.list_sessions(1)
|
||||||
assert len(infos) == 1 # no ServiceError escaped
|
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
|
||||||
|
|||||||
Reference in New Issue
Block a user