feat: dispatch due scheduled prompts

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)

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")