feat: schedule prompts in session service

tests/test_sessions.py

@@ -1,4 +1,5 @@
 import pytest
+from datetime import datetime
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -6,7 +7,7 @@ from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
 
-from hqt.db.models import Base, Harness, Project
+from hqt.db.models import Base, Harness, Project, ScheduledPrompt, Session
 from hqt.errors import ServiceError
 from hqt.sessions.service import SessionInfo, SessionService
 from hqt.tmux.manager import SpawnResult, TmuxManager
@@ -43,6 +44,8 @@ def tmux():
     m.poll_info = AsyncMock(return_value={})
     m.capture_pane = AsyncMock(return_value="")
     m.attach = AsyncMock()
+    m.send_text = AsyncMock(return_value=True)
+    m.send_enter = AsyncMock(return_value=True)
     m.window_exists = AsyncMock(return_value=True)
     m.respawn_verified = AsyncMock(
         return_value=SpawnResult(ok=True, window_id=None, error="")
@@ -93,6 +96,97 @@ async def test_list_sessions(service, db, tmux):
     assert result[0].alive is True
 
 
+def test_schedule_prompt_creates_pending_row(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+
+    prompt = service.schedule_prompt(sess.id, "run the tests", due_at)
+
+    assert prompt.session_id == sess.id
+    assert prompt.prompt == "run the tests"
+    assert prompt.due_at == due_at
+    assert prompt.status == "pending"
+    assert prompt.error is None
+    rows = db.query(ScheduledPrompt).all()
+    assert len(rows) == 1
+
+
+def test_schedule_prompt_replaces_existing_row_for_session(service, db):
+    sess = _seed_session(db)
+    first_due = datetime(2026, 6, 11, 12)
+    second_due = datetime(2026, 6, 11, 13)
+    first = service.schedule_prompt(sess.id, "old prompt", first_due)
+
+    second = service.schedule_prompt(sess.id, "new prompt", second_due)
+
+    rows = db.query(ScheduledPrompt).all()
+    assert len(rows) == 1
+    assert second.id == first.id
+    assert rows[0].id == first.id
+    assert rows[0].prompt == "new prompt"
+    assert rows[0].due_at == second_due
+    assert rows[0].status == "pending"
+    assert rows[0].error is None
+
+
+def test_schedule_prompt_rejects_empty_prompt(service):
+    due_at = datetime(2026, 6, 11, 12)
+
+    with pytest.raises(ServiceError, match="Prompt cannot be empty"):
+        service.schedule_prompt(1, "   ", due_at)
+
+
+def test_schedule_prompt_rejects_missing_session(service):
+    due_at = datetime(2026, 6, 11, 12)
+
+    with pytest.raises(ServiceError, match="Session not found"):
+        service.schedule_prompt(999, "hello", due_at)
+
+
+def test_cancel_scheduled_prompt_deletes_row(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+
+    service.cancel_scheduled_prompt(sess.id)
+
+    assert db.query(ScheduledPrompt).all() == []
+
+
+def test_get_scheduled_prompt_returns_pending_only(service, db):
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+
+    prompt = service.get_scheduled_prompt(sess.id)
+    assert prompt is not None
+    assert prompt.status == "pending"
+
+    row = db.query(ScheduledPrompt).filter_by(session_id=sess.id).one()
+    row.status = "failed"
+    db.commit()
+
+    assert service.get_scheduled_prompt(sess.id) is None
+
+
+@pytest.mark.asyncio
+async def test_list_sessions_includes_pending_scheduled_prompt(service, db, tmux):
+    from hqt.tmux.runner import WindowInfo
+
+    sess = _seed_session(db)
+    due_at = datetime(2026, 6, 11, 12)
+    service.schedule_prompt(sess.id, "run the tests", due_at)
+    tmux.poll_info.return_value = {
+        sess.tmux_session_name: WindowInfo(alive=True, last_activity=1000)
+    }
+
+    result = await service.list_sessions(project_id=1)
+
+    assert result[0].scheduled_prompt is not None
+    assert result[0].scheduled_prompt.prompt == "run the tests"
+    assert result[0].scheduled_prompt.status == "pending"
+
+
 @pytest.mark.asyncio
 async def test_stop_session(service, tmux):
     await service.create_session(project_id=1, harness_name="claude-code")
@@ -985,16 +1079,15 @@ async def test_create_session_result_propagates_spawn_failure(
 # ---------------------------------------------------------------------------
 
 
-def _seed_session(db, nickname=None):
-    """Create a session for project 1 / harness 'claude-code' (seeded by the db fixture)."""
-    from hqt.db.models import Harness, Session
-
+def _seed_session(db, project_id: int = 1, nickname=None) -> Session:
+    """Create a session for the seeded 'claude-code' harness."""
     harness = db.query(Harness).filter_by(name="claude-code").first()
     sess = Session(
-        project_id=1,
+        project_id=project_id,
         harness_id=harness.id,
         nickname=nickname,
         tmux_session_name="hqt-rename",
+        harness_session_id="sess-seeded",
         archived=False,
     )
     db.add(sess)