fix: preserve resume setup for scheduled dispatch
This commit is contained in:
@@ -364,6 +364,14 @@ class SessionService:
|
|||||||
raise ServiceError("Session not found")
|
raise ServiceError("Session not found")
|
||||||
window_name = sess.tmux_session_name
|
window_name = sess.tmux_session_name
|
||||||
|
|
||||||
|
await self._prepare_session_for_resume(db, sess)
|
||||||
|
ok = await self._ensure_session_alive(db, sess, window_name)
|
||||||
|
if not ok:
|
||||||
|
return False
|
||||||
|
return await self.tmux.attach(window_name)
|
||||||
|
|
||||||
|
async def _prepare_session_for_resume(self, db: DBSession, sess: Session) -> None:
|
||||||
|
"""Run pre-resume repair/capture work without selecting the window."""
|
||||||
# Rung 0: a worktree session whose directory was removed from disk
|
# Rung 0: a worktree session whose directory was removed from disk
|
||||||
# but whose branch still exists can be recreated before anything
|
# but whose branch still exists can be recreated before anything
|
||||||
# else runs. If recreation fails, the branch is gone too.
|
# else runs. If recreation fails, the branch is gone too.
|
||||||
@@ -383,11 +391,6 @@ class SessionService:
|
|||||||
|
|
||||||
self._maybe_capture_missing_session_id(db, sess)
|
self._maybe_capture_missing_session_id(db, sess)
|
||||||
|
|
||||||
ok = await self._ensure_session_alive(db, sess, window_name)
|
|
||||||
if not ok:
|
|
||||||
return False
|
|
||||||
return await self.tmux.attach(window_name)
|
|
||||||
|
|
||||||
async def _ensure_session_alive(
|
async def _ensure_session_alive(
|
||||||
self, db: DBSession, sess: Session, window_name: str
|
self, db: DBSession, sess: Session, window_name: str
|
||||||
) -> bool:
|
) -> bool:
|
||||||
@@ -557,6 +560,9 @@ class SessionService:
|
|||||||
error=error,
|
error=error,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def _dispatch_error_message(self, exc: Exception) -> str:
|
||||||
|
return str(exc) or exc.__class__.__name__
|
||||||
|
|
||||||
async def dispatch_due_prompts(
|
async def dispatch_due_prompts(
|
||||||
self, now: datetime | None = None
|
self, now: datetime | None = None
|
||||||
) -> list[DispatchResult]:
|
) -> list[DispatchResult]:
|
||||||
@@ -591,11 +597,16 @@ class SessionService:
|
|||||||
|
|
||||||
window_name = sess.tmux_session_name
|
window_name = sess.tmux_session_name
|
||||||
try:
|
try:
|
||||||
|
await self._prepare_session_for_resume(db, sess)
|
||||||
alive, error = await self._ensure_session_alive_result(
|
alive, error = await self._ensure_session_alive_result(
|
||||||
db, sess, window_name
|
db, sess, window_name
|
||||||
)
|
)
|
||||||
except ServiceError as exc:
|
except Exception as exc:
|
||||||
results.append(self._mark_prompt_failed(db, row, str(exc)))
|
results.append(
|
||||||
|
self._mark_prompt_failed(
|
||||||
|
db, row, self._dispatch_error_message(exc)
|
||||||
|
)
|
||||||
|
)
|
||||||
continue
|
continue
|
||||||
if not alive:
|
if not alive:
|
||||||
results.append(
|
results.append(
|
||||||
@@ -607,12 +618,30 @@ class SessionService:
|
|||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
|
|
||||||
if not await self.tmux.send_text(window_name, row.prompt):
|
try:
|
||||||
|
send_text_ok = await self.tmux.send_text(window_name, row.prompt)
|
||||||
|
except Exception as exc:
|
||||||
|
results.append(
|
||||||
|
self._mark_prompt_failed(
|
||||||
|
db, row, self._dispatch_error_message(exc)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
if not send_text_ok:
|
||||||
results.append(
|
results.append(
|
||||||
self._mark_prompt_failed(db, row, "send text failed")
|
self._mark_prompt_failed(db, row, "send text failed")
|
||||||
)
|
)
|
||||||
continue
|
continue
|
||||||
if not await self.tmux.send_enter(window_name):
|
try:
|
||||||
|
send_enter_ok = await self.tmux.send_enter(window_name)
|
||||||
|
except Exception as exc:
|
||||||
|
results.append(
|
||||||
|
self._mark_prompt_failed(
|
||||||
|
db, row, self._dispatch_error_message(exc)
|
||||||
|
)
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
if not send_enter_ok:
|
||||||
results.append(
|
results.append(
|
||||||
self._mark_prompt_failed(db, row, "send enter failed")
|
self._mark_prompt_failed(db, row, "send enter failed")
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -295,6 +295,169 @@ async def test_dispatch_due_prompt_send_failure_does_not_retry_next_poll(
|
|||||||
assert "send text failed" in row.error
|
assert "send text failed" in row.error
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_missing_worktree_recreates_without_attach(
|
||||||
|
factory, db, tmux, harnesses, fake_worktree
|
||||||
|
):
|
||||||
|
missing = "/tmp/does-not-exist/.worktrees/feature-x"
|
||||||
|
sess = _seed_worktree_session(db, path=missing)
|
||||||
|
service = SessionService(factory=factory, tmux=tmux, harnesses=harnesses)
|
||||||
|
service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
|
||||||
|
harnesses["claude-code"].build_resume_config = MagicMock(
|
||||||
|
return_value=MagicMock(command=["claude"], env={}, cwd=Path(missing))
|
||||||
|
)
|
||||||
|
tmux.is_alive = AsyncMock(return_value=False)
|
||||||
|
tmux.respawn_verified = AsyncMock(
|
||||||
|
return_value=SpawnResult(ok=True, window_id=None, error="")
|
||||||
|
)
|
||||||
|
tmux.attach = AsyncMock(return_value=True)
|
||||||
|
|
||||||
|
results = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert results[0].ok is True
|
||||||
|
assert fake_worktree.add_calls == [(Path("/tmp/myproj"), "feature-x")]
|
||||||
|
tmux.attach.assert_not_called()
|
||||||
|
tmux.send_text.assert_awaited_once_with("hqt-wt", "continue")
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_late_capture_updates_id_before_resume(
|
||||||
|
factory, db, tmux, harnesses_capture_fallback
|
||||||
|
):
|
||||||
|
harness = harnesses_capture_fallback["claude-code"]
|
||||||
|
harness.capture_session_id.return_value = "late-codex-id"
|
||||||
|
sess = _seed_session(db)
|
||||||
|
sess.harness_session_id = "placeholder-id"
|
||||||
|
db.commit()
|
||||||
|
service = SessionService(
|
||||||
|
factory=factory, tmux=tmux, harnesses=harnesses_capture_fallback
|
||||||
|
)
|
||||||
|
service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
|
||||||
|
tmux.is_alive = AsyncMock(return_value=False)
|
||||||
|
tmux.respawn_verified = AsyncMock(
|
||||||
|
return_value=SpawnResult(ok=True, window_id="@2", error="")
|
||||||
|
)
|
||||||
|
|
||||||
|
results = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert results[0].ok is True
|
||||||
|
harness.build_resume_config.assert_called_with(
|
||||||
|
Path("/tmp/myproj"), "late-codex-id", sess.model, sandboxed=False
|
||||||
|
)
|
||||||
|
db.expire_all()
|
||||||
|
assert db.get(Session, sess.id).harness_session_id == "late-codex-id"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_send_exception_marks_failed_and_does_not_retry(
|
||||||
|
service, db, tmux
|
||||||
|
):
|
||||||
|
sess = _seed_session(db)
|
||||||
|
scheduled = service.schedule_prompt(
|
||||||
|
sess.id, "continue", datetime.now() - timedelta(seconds=1)
|
||||||
|
)
|
||||||
|
tmux.is_alive = AsyncMock(return_value=True)
|
||||||
|
tmux.send_text = AsyncMock(side_effect=RuntimeError("send exploded"))
|
||||||
|
|
||||||
|
first = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
second = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert len(first) == 1
|
||||||
|
assert first[0].ok is False
|
||||||
|
assert "send exploded" in first[0].error
|
||||||
|
assert second == []
|
||||||
|
tmux.send_text.assert_awaited_once()
|
||||||
|
tmux.send_enter.assert_not_called()
|
||||||
|
db.expire_all()
|
||||||
|
row = db.get(ScheduledPrompt, scheduled.id)
|
||||||
|
assert row.status == "failed"
|
||||||
|
assert "send exploded" in row.error
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_resume_exception_marks_failed_without_send(
|
||||||
|
service, db, tmux, harnesses
|
||||||
|
):
|
||||||
|
sess = _seed_session(db)
|
||||||
|
scheduled = service.schedule_prompt(
|
||||||
|
sess.id, "continue", datetime.now() - timedelta(seconds=1)
|
||||||
|
)
|
||||||
|
harnesses["claude-code"].build_resume_config = MagicMock(
|
||||||
|
return_value=MagicMock(
|
||||||
|
command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
|
||||||
|
)
|
||||||
|
)
|
||||||
|
tmux.is_alive = AsyncMock(return_value=False)
|
||||||
|
tmux.respawn_verified = AsyncMock(side_effect=RuntimeError("resume exploded"))
|
||||||
|
|
||||||
|
results = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert len(results) == 1
|
||||||
|
assert results[0].ok is False
|
||||||
|
assert "resume exploded" in results[0].error
|
||||||
|
tmux.send_text.assert_not_called()
|
||||||
|
tmux.send_enter.assert_not_called()
|
||||||
|
db.expire_all()
|
||||||
|
row = db.get(ScheduledPrompt, scheduled.id)
|
||||||
|
assert row.status == "failed"
|
||||||
|
assert "resume exploded" in row.error
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_send_enter_failure_marks_failed(service, db, tmux):
|
||||||
|
sess = _seed_session(db)
|
||||||
|
scheduled = service.schedule_prompt(
|
||||||
|
sess.id, "continue", datetime.now() - timedelta(seconds=1)
|
||||||
|
)
|
||||||
|
tmux.is_alive = AsyncMock(return_value=True)
|
||||||
|
tmux.send_enter = AsyncMock(return_value=False)
|
||||||
|
|
||||||
|
results = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert len(results) == 1
|
||||||
|
assert results[0].ok is False
|
||||||
|
assert "send enter failed" in results[0].error
|
||||||
|
tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
|
||||||
|
tmux.send_enter.assert_awaited_once_with("hqt-rename")
|
||||||
|
db.expire_all()
|
||||||
|
row = db.get(ScheduledPrompt, scheduled.id)
|
||||||
|
assert row.status == "failed"
|
||||||
|
assert "send enter failed" in row.error
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_dispatch_due_prompt_resume_fails_then_fallback_spawn_succeeds(
|
||||||
|
service, db, tmux, harnesses
|
||||||
|
):
|
||||||
|
sess = _seed_session(db)
|
||||||
|
service.schedule_prompt(sess.id, "continue", datetime.now() - timedelta(seconds=1))
|
||||||
|
harnesses["claude-code"].build_resume_config = MagicMock(
|
||||||
|
return_value=MagicMock(
|
||||||
|
command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj")
|
||||||
|
)
|
||||||
|
)
|
||||||
|
harnesses["claude-code"].build_spawn_config = MagicMock(
|
||||||
|
return_value=MagicMock(
|
||||||
|
command=["claude", "--session-id", "sess-seeded"],
|
||||||
|
env={},
|
||||||
|
cwd=Path("/tmp/myproj"),
|
||||||
|
)
|
||||||
|
)
|
||||||
|
tmux.is_alive = AsyncMock(return_value=False)
|
||||||
|
tmux.respawn_verified = AsyncMock(
|
||||||
|
side_effect=[
|
||||||
|
SpawnResult(ok=False, window_id=None, error="resume failed"),
|
||||||
|
SpawnResult(ok=True, window_id="@2", error=""),
|
||||||
|
]
|
||||||
|
)
|
||||||
|
|
||||||
|
results = await service.dispatch_due_prompts(now=datetime.now())
|
||||||
|
|
||||||
|
assert results[0].ok is True
|
||||||
|
assert tmux.respawn_verified.await_count == 2
|
||||||
|
tmux.send_text.assert_awaited_once_with("hqt-rename", "continue")
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_stop_session(service, tmux):
|
async def test_stop_session(service, tmux):
|
||||||
await service.create_session(project_id=1, harness_name="claude-code")
|
await service.create_session(project_id=1, harness_name="claude-code")
|
||||||
|
|||||||
Reference in New Issue
Block a user