From f10264f695d1f435c3d4b2ad2e08e9fd3c264c0e Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Thu, 11 Jun 2026 08:40:16 -0400 Subject: [PATCH] feat: dispatch due scheduled prompts --- src/hqt/sessions/service.py | 147 ++++++++++++++++++++++++++++++++---- tests/test_sessions.py | 109 +++++++++++++++++++++++++- 2 files changed, 241 insertions(+), 15 deletions(-) diff --git a/src/hqt/sessions/service.py b/src/hqt/sessions/service.py index 83ee26c..c61fb64 100644 --- a/src/hqt/sessions/service.py +++ b/src/hqt/sessions/service.py @@ -54,6 +54,14 @@ class CreateSessionResult: spawn_error: str = "" +@dataclass +class DispatchResult: + session_id: int + window_name: str + ok: bool + error: str = "" + + class SessionService: """Session lifecycle operations. @@ -375,25 +383,42 @@ class SessionService: self._maybe_capture_missing_session_id(db, sess) - if await self.tmux.is_alive(window_name): - # Window exists, process running — just switch - return await self.tmux.attach(window_name) - - # Window exists but pane is dead, OR window is completely gone - log.info( - "Session %s not alive, respawning with fallback ladder", window_name - ) - ok = await self._respawn_with_fallback(db, sess, window_name) + 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( + self, db: DBSession, sess: Session, window_name: str + ) -> bool: + """Ensure a session's tmux pane is alive without selecting the window.""" + ok, _error = await self._ensure_session_alive_result(db, sess, window_name) + return ok + + async def _ensure_session_alive_result( + self, db: DBSession, sess: Session, window_name: str + ) -> tuple[bool, str]: + """Ensure a session is alive, returning the respawn error on failure.""" + if await self.tmux.is_alive(window_name): + return True, "" + log.info( + "Session %s not alive, respawning with fallback ladder", + window_name, + ) + return await self._respawn_with_fallback_result(db, sess, window_name) + async def _respawn_with_fallback( self, db: DBSession, sess: Session, window_name: str ) -> bool: + ok, _error = await self._respawn_with_fallback_result(db, sess, window_name) + return ok + + async def _respawn_with_fallback_result( + self, db: DBSession, sess: Session, window_name: str + ) -> tuple[bool, str]: """Try resume config first; if it dies, fall back to fresh spawn config. - Returns True if either rung succeeded, False if both failed. + Returns (True, "") if either rung succeeded, otherwise (False, error). On rung-2 success, if the harness captures_session_id, a capture-retry loop updates sess.harness_session_id so later resumes target the fresh @@ -405,7 +430,7 @@ class SessionService: window_name, resume_cfg.command, str(resume_cfg.cwd), env=resume_cfg.env ) if result.ok: - return True + return True, "" log.warning( "Resume failed for %s (%s), falling back to fresh spawn", @@ -436,12 +461,13 @@ class SessionService: window_name, command, str(spawn_cfg.cwd), env=spawn_cfg.env ) if not result.ok: + error = result.error or "respawn failed" log.error( "Fallback spawn also failed for %s: %s", window_name, result.error or "(no output)", ) - return False + return False, error # Rung-2 succeeded — capture the new session id if the harness # supports it so that future resumes target this fresh conversation. @@ -458,7 +484,7 @@ class SessionService: window_name, sess.harness_session_id, ) - return True + return True, "" async def stop_session(self, session_id: int) -> None: with self.factory() as db: @@ -516,6 +542,95 @@ class SessionService: db.expunge(scheduled) return scheduled + def _mark_prompt_failed( + self, db: DBSession, row: ScheduledPrompt, error: str + ) -> DispatchResult: + row.status = PROMPT_FAILED + row.error = error + db.commit() + session = db.get(Session, row.session_id) + window_name = session.tmux_session_name if session is not None else "" + return DispatchResult( + session_id=row.session_id, + window_name=window_name, + ok=False, + error=error, + ) + + async def dispatch_due_prompts( + self, now: datetime | None = None + ) -> list[DispatchResult]: + """Dispatch due pending prompts once; failed prompts are not retried.""" + if now is None: + now = datetime.now() + results: list[DispatchResult] = [] + with self.factory() as db: + rows = ( + db.query(ScheduledPrompt) + .filter( + ScheduledPrompt.status == PROMPT_PENDING, + ScheduledPrompt.due_at <= now, + ) + .all() + ) + for row in rows: + row.status = PROMPT_SENDING + row.error = None + db.commit() + + sess = db.get( + Session, + row.session_id, + options=[selectinload(Session.harness)], + ) + if sess is None: + results.append( + self._mark_prompt_failed(db, row, "Session not found") + ) + continue + + window_name = sess.tmux_session_name + try: + alive, error = await self._ensure_session_alive_result( + db, sess, window_name + ) + except ServiceError as exc: + results.append(self._mark_prompt_failed(db, row, str(exc))) + continue + if not alive: + results.append( + self._mark_prompt_failed( + db, + row, + error or "Failed to resume session before prompt", + ) + ) + continue + + if not await self.tmux.send_text(window_name, row.prompt): + results.append( + self._mark_prompt_failed(db, row, "send text failed") + ) + continue + if not await self.tmux.send_enter(window_name): + results.append( + self._mark_prompt_failed(db, row, "send enter failed") + ) + continue + + row.status = PROMPT_SENT + row.sent_at = now + row.error = None + db.commit() + results.append( + DispatchResult( + session_id=sess.id, + window_name=window_name, + ok=True, + ) + ) + return results + def cancel_scheduled_prompt(self, session_id: int) -> None: """Delete any scheduled prompt for a session.""" with self.factory() as db: @@ -720,6 +835,10 @@ class SessionService: sandboxed=policy is not None, ) command = self._wrap_command( - configurator, cfg.command, cfg.cwd, policy, self._worktree_git_binds(db, sess) + configurator, + cfg.command, + cfg.cwd, + policy, + self._worktree_git_binds(db, sess), ) return SpawnConfig(command=command, env=cfg.env, cwd=cfg.cwd) diff --git a/tests/test_sessions.py b/tests/test_sessions.py index cf3544f..90eeaa5 100644 --- a/tests/test_sessions.py +++ b/tests/test_sessions.py @@ -1,5 +1,5 @@ import pytest -from datetime import datetime +from datetime import datetime, timedelta from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch @@ -188,6 +188,113 @@ async def test_list_sessions_includes_pending_scheduled_prompt(service, db, tmux assert result[0].scheduled_prompt.status == "pending" +@pytest.mark.asyncio +async def test_dispatch_due_prompt_alive_window_sends_and_marks_sent(service, db, tmux): + sess = _seed_session(db) + due = datetime.now() - timedelta(seconds=1) + scheduled = service.schedule_prompt(sess.id, "continue", due) + tmux.is_alive = AsyncMock(return_value=True) + + results = await service.dispatch_due_prompts(now=datetime.now()) + + assert len(results) == 1 + assert results[0].ok is True + assert results[0].session_id == sess.id + assert results[0].window_name == "hqt-rename" + 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 == "sent" + assert row.sent_at is not None + + +@pytest.mark.asyncio +async def test_dispatch_due_prompt_dead_window_resumes_without_attach( + service, db, tmux, harnesses +): + sess = _seed_session(db) + 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=None, error="") + ) + tmux.attach = AsyncMock(return_value=True) + harnesses["claude-code"].build_resume_config = MagicMock( + return_value=MagicMock( + command=["claude", "--resume"], env={}, cwd=Path("/tmp/myproj") + ) + ) + + results = await service.dispatch_due_prompts(now=datetime.now()) + + assert len(results) == 1 + assert results[0].ok is True + tmux.respawn_verified.assert_awaited() + tmux.attach.assert_not_called() + tmux.send_text.assert_awaited_once_with("hqt-rename", "continue") + tmux.send_enter.assert_awaited_once_with("hqt-rename") + + +@pytest.mark.asyncio +async def test_dispatch_due_prompt_resume_failure_marks_failed( + service, db, tmux, harnesses +): + sess = _seed_session(db) + scheduled = 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=False, window_id=None, error="resume failed") + ) + 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"], env={}, cwd=Path("/tmp/myproj")) + ) + + results = await service.dispatch_due_prompts(now=datetime.now()) + + assert len(results) == 1 + assert results[0].ok is False + assert "resume failed" 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 failed" in row.error + + +@pytest.mark.asyncio +async def test_dispatch_due_prompt_send_failure_does_not_retry_next_poll( + 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(return_value=False) + + 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 second == [] + assert tmux.send_text.await_count == 1 + tmux.send_enter.assert_not_called() + db.expire_all() + row = db.get(ScheduledPrompt, scheduled.id) + assert row.status == "failed" + assert "send text failed" in row.error + + @pytest.mark.asyncio async def test_stop_session(service, tmux): await service.create_session(project_id=1, harness_name="claude-code")