feat: persist scheduled prompts

tests/test_db.py

@@ -95,6 +95,7 @@ def test_unversioned_existing_db_stamped_latest(tmp_path):
     engine = get_engine(settings)
     Base.metadata.create_all(engine)
     with engine.begin() as conn:
+        conn.exec_driver_sql("DROP TABLE scheduled_prompts")
         conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path")
         conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch")
     migrate(engine)
@@ -113,7 +114,7 @@ def test_db_from_newer_hqt_raises(tmp_path):
 
 def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch):
     settings = _tmp_settings(tmp_path)
-    ensure_db(settings)  # stamps migrations.LATEST_VERSION (currently 2)
+    ensure_db(settings)  # stamps migrations.LATEST_VERSION
     applied: list[int] = []
     # Use versions above current LATEST_VERSION so none have been applied yet.
     monkeypatch.setattr(
@@ -121,17 +122,17 @@ def test_pending_migrations_applied_in_order_once(tmp_path, monkeypatch):
         "MIGRATIONS",
         [
             *migrations.MIGRATIONS,
-            (3, lambda conn: applied.append(3)),
             (4, lambda conn: applied.append(4)),
+            (5, lambda conn: applied.append(5)),
         ],
     )
-    monkeypatch.setattr(migrations, "LATEST_VERSION", 4)
+    monkeypatch.setattr(migrations, "LATEST_VERSION", 5)
     engine = get_engine(settings)
     migrate(engine)
-    assert applied == [3, 4]
-    assert _user_version(engine) == 4
+    assert applied == [4, 5]
+    assert _user_version(engine) == 5
     migrate(engine)  # re-run is a no-op
-    assert applied == [3, 4]
+    assert applied == [4, 5]
 
 
 def _session_column_names(engine) -> list[str]:
@@ -140,6 +141,12 @@ def _session_column_names(engine) -> list[str]:
     return [row[1] for row in rows]
 
 
+def _scheduled_prompt_column_names(engine) -> list[str]:
+    with engine.connect() as conn:
+        rows = conn.exec_driver_sql("PRAGMA table_info(scheduled_prompts)").fetchall()
+    return [row[1] for row in rows]
+
+
 def test_fresh_db_has_worktree_columns(tmp_path):
     settings = _tmp_settings(tmp_path)
     ensure_db(settings)
@@ -147,7 +154,7 @@ def test_fresh_db_has_worktree_columns(tmp_path):
     cols = _session_column_names(engine)
     assert "worktree_path" in cols
     assert "worktree_branch" in cols
-    assert _user_version(engine) == 2
+    assert _user_version(engine) == migrations.LATEST_VERSION
 
 
 def test_v1_db_upgraded_to_v2(tmp_path):
@@ -157,6 +164,7 @@ def test_v1_db_upgraded_to_v2(tmp_path):
     engine = get_engine(settings)
     Base.metadata.create_all(engine)
     with engine.begin() as conn:
+        conn.exec_driver_sql("DROP TABLE scheduled_prompts")
         conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_path")
         conn.exec_driver_sql("ALTER TABLE sessions DROP COLUMN worktree_branch")
         conn.exec_driver_sql("PRAGMA user_version = 1")
@@ -166,7 +174,75 @@ def test_v1_db_upgraded_to_v2(tmp_path):
     cols = _session_column_names(engine)
     assert "worktree_path" in cols
     assert "worktree_branch" in cols
-    assert _user_version(engine) == 2
+    assert _user_version(engine) == migrations.LATEST_VERSION
+
+
+def test_fresh_db_has_scheduled_prompts_table(tmp_path):
+    settings = _tmp_settings(tmp_path)
+    ensure_db(settings)
+    engine = get_engine(settings)
+
+    tables = inspect(engine).get_table_names()
+    assert "scheduled_prompts" in tables
+    cols = _scheduled_prompt_column_names(engine)
+    assert cols == [
+        "id",
+        "session_id",
+        "prompt",
+        "due_at",
+        "status",
+        "error",
+        "created_at",
+        "sent_at",
+    ]
+    assert _user_version(engine) == 3
+
+
+def test_v2_db_upgraded_to_v3_adds_scheduled_prompts(tmp_path):
+    settings = _tmp_settings(tmp_path)
+    settings.db_path.parent.mkdir(parents=True, exist_ok=True)
+    engine = get_engine(settings)
+    Base.metadata.create_all(engine)
+    with engine.begin() as conn:
+        conn.exec_driver_sql("DROP TABLE scheduled_prompts")
+        conn.exec_driver_sql("PRAGMA user_version = 2")
+
+    migrate(engine)
+
+    assert "scheduled_prompts" in inspect(engine).get_table_names()
+    assert _user_version(engine) == 3
+
+
+def test_scheduled_prompt_one_row_per_session(tmp_path):
+    from datetime import datetime, timedelta
+
+    from sqlalchemy.exc import IntegrityError
+
+    from hqt.db.models import ScheduledPrompt
+
+    settings = _tmp_settings(tmp_path)
+    ensure_db(settings)
+    factory = get_session_factory(get_engine(settings))
+
+    with factory() as session:
+        p = Project(name="sched-proj", path="/tmp/sched-proj")
+        h = Harness(name="claude-code-scheduled")
+        session.add_all([p, h])
+        session.flush()
+        s = Session(
+            project_id=p.id,
+            harness_id=h.id,
+            tmux_session_name="hqt-scheduled",
+        )
+        session.add(s)
+        session.flush()
+        due = datetime.now() + timedelta(minutes=30)
+        session.add(ScheduledPrompt(session_id=s.id, prompt="continue", due_at=due))
+        session.commit()
+
+        session.add(ScheduledPrompt(session_id=s.id, prompt="again", due_at=due))
+        with pytest.raises(IntegrityError):
+            session.commit()
 
 
 def test_session_worktree_fields_default_none(tmp_path):