diff --git a/docs/superpowers/plans/2026-06-11-delayed-prompt-injection.md b/docs/superpowers/plans/2026-06-11-delayed-prompt-injection.md new file mode 100644 index 0000000..0aa5041 --- /dev/null +++ b/docs/superpowers/plans/2026-06-11-delayed-prompt-injection.md @@ -0,0 +1,1656 @@ +# Delayed Prompt Injection Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Build TUI-only delayed prompt scheduling for one active prompt per session, with persisted state, compact row visualization, auto-resume before injection, and poll-time dispatch. + +**Architecture:** Add a `scheduled_prompts` table and expose scheduling through `SessionService`. The existing TUI poll loop becomes the scheduler runtime: it dispatches due prompts, shows notifications, then refreshes session rows. Tmux text injection stays in the tmux layer and returns success/failure so service dispatch can mark prompts `sent` or `failed`. + +**Tech Stack:** Python 3.12, SQLAlchemy 2, Textual 2, pytest/pytest-asyncio, tmux command wrappers. + +--- + +## File Structure + +- Modify `src/hqt/db/models.py`: add `ScheduledPrompt` ORM model and relationship from `Session`. +- Modify `src/hqt/db/migrations.py`: add v3 migration creating `scheduled_prompts`. +- Modify `tests/test_db.py`: verify fresh schema, v2-to-v3 migration, and unique `session_id`. +- Modify `src/hqt/tmux/runner.py`: make `send_text()` and `send_enter()` return `bool`. +- Modify `src/hqt/tmux/manager.py`: add `send_text()` and `send_enter()` delegates returning `bool`. +- Modify `tests/test_tmux.py`: cover send success/failure at runner and manager layers. +- Modify `src/hqt/sessions/service.py`: add scheduling dataclasses/methods, pending prompt hydration in `SessionInfo`, no-attach resume helper, and due dispatch. +- Modify `tests/test_sessions.py`: cover scheduling, row metadata, dispatch success/failure, auto-resume, and no repeated retry. +- Create `src/hqt/tui/screens/schedule_prompt.py`: modal for prompt text + relative delay. +- Modify `src/hqt/tui/styles.tcss`: include `SchedulePromptScreen` in modal alignment/styling selectors. +- Modify `src/hqt/tui/widgets/session_list.py`: format pending prompt indicators. +- Modify `src/hqt/tui/app.py`: add `p`/`P` bindings, schedule/cancel actions, and poll-time dispatch notifications. +- Modify `tests/test_tui.py`: cover modal validation, keybindings/actions, row indicator, and poll dispatch notifications. + +--- + +### Task 1: Persist ScheduledPrompt + +**Files:** +- Modify: `src/hqt/db/models.py` +- Modify: `src/hqt/db/migrations.py` +- Test: `tests/test_db.py` + +- [ ] **Step 1: Write failing DB schema tests** + +Append these tests to `tests/test_db.py`: + +```python +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_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() +``` + +Update the existing tests in `tests/test_db.py` that assert version `2` so they expect `migrations.LATEST_VERSION` or `3`: + +```python +assert _user_version(engine) == migrations.LATEST_VERSION +``` + +- [ ] **Step 2: Run DB tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_db.py -q +``` + +Expected: FAIL because `ScheduledPrompt` and the v3 migration do not exist. + +- [ ] **Step 3: Add the model** + +In `src/hqt/db/models.py`, update imports and add the relationship: + +```python +from sqlalchemy import ForeignKey, String, Text, func +``` + +Inside `class Session`, add: + +```python + scheduled_prompt: Mapped["ScheduledPrompt | None"] = relationship( + back_populates="session" + ) +``` + +After `class Session`, add: + +```python +class ScheduledPrompt(Base): + __tablename__ = "scheduled_prompts" + id: Mapped[int] = mapped_column(primary_key=True) + session_id: Mapped[int] = mapped_column( + ForeignKey("sessions.id"), + unique=True, + ) + prompt: Mapped[str] = mapped_column(Text) + due_at: Mapped[datetime] + status: Mapped[str] = mapped_column(String, default="pending") + error: Mapped[str | None] = mapped_column(Text, default=None) + created_at: Mapped[datetime] = mapped_column(insert_default=func.now()) + sent_at: Mapped[datetime | None] = mapped_column(default=None) + session: Mapped["Session"] = relationship(back_populates="scheduled_prompt") +``` + +- [ ] **Step 4: Add the migration** + +In `src/hqt/db/migrations.py`, add: + +```python +def _migrate_v3(conn: Connection) -> None: + conn.exec_driver_sql( + """ + CREATE TABLE scheduled_prompts ( + id INTEGER NOT NULL PRIMARY KEY, + session_id INTEGER NOT NULL UNIQUE, + prompt TEXT NOT NULL, + due_at DATETIME NOT NULL, + status VARCHAR NOT NULL, + error TEXT, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + sent_at DATETIME, + FOREIGN KEY(session_id) REFERENCES sessions (id) + ) + """ + ) +``` + +Update `MIGRATIONS`: + +```python +MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [ + (2, _migrate_v2), + (3, _migrate_v3), +] +``` + +- [ ] **Step 5: Run DB tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_db.py -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/hqt/db/models.py src/hqt/db/migrations.py tests/test_db.py +git commit -m "feat: persist scheduled prompts" +``` + +--- + +### Task 2: Return Injection Success From Tmux Layer + +**Files:** +- Modify: `src/hqt/tmux/runner.py` +- Modify: `src/hqt/tmux/manager.py` +- Test: `tests/test_tmux.py` + +- [ ] **Step 1: Write failing tmux send tests** + +Append these tests near the existing send-key tests in `tests/test_tmux.py`: + +```python +@pytest.mark.asyncio +async def test_send_text_returns_true_on_success(runner): + runner._exec.return_value = (0, "", "") + + assert await runner.send_text("hqt-1", "hello") is True + + +@pytest.mark.asyncio +async def test_send_text_returns_false_on_failure(runner): + runner._exec.return_value = (1, "", "no such target") + + assert await runner.send_text("hqt-1", "hello") is False + + +@pytest.mark.asyncio +async def test_send_enter_returns_true_on_success(runner): + runner._exec.return_value = (0, "", "") + + assert await runner.send_enter("hqt-1") is True + + +@pytest.mark.asyncio +async def test_send_enter_returns_false_on_failure(runner): + runner._exec.return_value = (1, "", "no such target") + + assert await runner.send_enter("hqt-1") is False + + +@pytest.mark.asyncio +async def test_manager_send_delegates(runner): + mgr = TmuxManager(runner) + runner.send_text = AsyncMock(return_value=True) + runner.send_enter = AsyncMock(return_value=True) + + assert await mgr.send_text("hqt-1", "continue") is True + assert await mgr.send_enter("hqt-1") is True + runner.send_text.assert_awaited_once_with("hqt-1", "continue") + runner.send_enter.assert_awaited_once_with("hqt-1") +``` + +- [ ] **Step 2: Run tmux tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_tmux.py -q +``` + +Expected: FAIL because `send_text()` / `send_enter()` return `None`, and `TmuxManager` has no send delegates. + +- [ ] **Step 3: Update runner send methods** + +In `src/hqt/tmux/runner.py`, replace the send methods with: + +```python + async def send_text(self, window_name: str, text: str) -> bool: + rc, _, err = await self._exec( + "send-keys", + "-t", + self._target(window_name), + "-l", + "--", + text, + ) + if rc != 0: + log.error("send_text failed for %s: %s", window_name, err) + return rc == 0 + + async def send_enter(self, window_name: str) -> bool: + rc, _, err = await self._exec( + "send-keys", + "-t", + self._target(window_name), + "Enter", + ) + if rc != 0: + log.error("send_enter failed for %s: %s", window_name, err) + return rc == 0 +``` + +- [ ] **Step 4: Add manager delegates** + +In `src/hqt/tmux/manager.py`, add methods near `capture_pane()`: + +```python + async def send_text(self, window_name: str, text: str) -> bool: + """Send literal text to a harness window.""" + return await self.runner.send_text(window_name, text) + + async def send_enter(self, window_name: str) -> bool: + """Send Enter to a harness window.""" + return await self.runner.send_enter(window_name) +``` + +- [ ] **Step 5: Run tmux tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_tmux.py -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/hqt/tmux/runner.py src/hqt/tmux/manager.py tests/test_tmux.py +git commit -m "feat: report tmux prompt injection failures" +``` + +--- + +### Task 3: Add Scheduling Service Methods And Row Metadata + +**Files:** +- Modify: `src/hqt/sessions/service.py` +- Test: `tests/test_sessions.py` + +- [ ] **Step 1: Extend the tmux fixture for new manager methods** + +In the `tmux()` fixture in `tests/test_sessions.py`, add: + +```python + m.send_text = AsyncMock(return_value=True) + m.send_enter = AsyncMock(return_value=True) +``` + +- [ ] **Step 2: Write failing service scheduling tests** + +Append this section after the rename/get-session tests in `tests/test_sessions.py`: + +```python +# --------------------------------------------------------------------------- +# delayed prompt scheduling +# --------------------------------------------------------------------------- + + +def _due_at(): + from datetime import datetime, timedelta + + return datetime.now() + timedelta(minutes=30) + + +def test_schedule_prompt_creates_row(service, db): + from hqt.db.models import ScheduledPrompt + + sess = _seed_session(db) + scheduled = service.schedule_prompt(sess.id, "continue", _due_at()) + + db.expire_all() + row = db.get(ScheduledPrompt, scheduled.id) + assert row.session_id == sess.id + assert row.prompt == "continue" + assert row.status == "pending" + assert row.error is None + + +def test_schedule_prompt_replaces_existing_row(service, db): + from hqt.db.models import ScheduledPrompt + + sess = _seed_session(db) + first = service.schedule_prompt(sess.id, "first", _due_at()) + second = service.schedule_prompt(sess.id, "second", _due_at()) + + db.expire_all() + rows = db.query(ScheduledPrompt).filter_by(session_id=sess.id).all() + assert len(rows) == 1 + assert rows[0].id == first.id + assert second.id == first.id + assert rows[0].prompt == "second" + assert rows[0].status == "pending" + + +def test_schedule_prompt_empty_raises(service, db): + sess = _seed_session(db) + + with pytest.raises(ServiceError, match="Prompt cannot be empty"): + service.schedule_prompt(sess.id, " ", _due_at()) + + +def test_schedule_prompt_missing_session_raises(service): + with pytest.raises(ServiceError, match="Session not found"): + service.schedule_prompt(99999, "continue", _due_at()) + + +def test_cancel_scheduled_prompt_deletes_row(service, db): + from hqt.db.models import ScheduledPrompt + + sess = _seed_session(db) + scheduled = service.schedule_prompt(sess.id, "continue", _due_at()) + + service.cancel_scheduled_prompt(sess.id) + + db.expire_all() + assert db.get(ScheduledPrompt, scheduled.id) is None + + +def test_get_scheduled_prompt_pending_only(service, db): + from hqt.db.models import ScheduledPrompt + + sess = _seed_session(db) + scheduled = service.schedule_prompt(sess.id, "continue", _due_at()) + db.expire_all() + row = db.get(ScheduledPrompt, scheduled.id) + 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) + service.schedule_prompt(sess.id, "continue", _due_at()) + tmux.poll_info.return_value = {"hqt-rename": WindowInfo(alive=True, last_activity=1000)} + + infos = await service.list_sessions(project_id=1, now=1000.0) + + assert infos[0].scheduled_prompt is not None + assert infos[0].scheduled_prompt.prompt == "continue" +``` + +- [ ] **Step 3: Run service tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_sessions.py -q +``` + +Expected: FAIL because service scheduling methods and `SessionInfo.scheduled_prompt` do not exist. + +- [ ] **Step 4: Add imports, constants, and dataclasses** + +In `src/hqt/sessions/service.py`, update imports: + +```python +from dataclasses import dataclass +from datetime import UTC, datetime +``` + +Update model imports: + +```python +from hqt.db.models import Harness, Project, ScheduledPrompt, Session +``` + +Add status constants near the existing module constants: + +```python +PROMPT_PENDING = "pending" +PROMPT_SENDING = "sending" +PROMPT_SENT = "sent" +PROMPT_FAILED = "failed" +``` + +Update `SessionInfo`: + +```python +@dataclass +class SessionInfo: + session: Session + alive: bool + status: str = "dead" + scheduled_prompt: ScheduledPrompt | None = None +``` + +- [ ] **Step 5: Add scheduling methods** + +Inside `class SessionService`, add these methods after `rename_session()`: + +```python + def schedule_prompt( + self, session_id: int, prompt: str, due_at: datetime + ) -> ScheduledPrompt: + """Create or replace the selected session's scheduled prompt.""" + prompt = prompt.strip() + if not prompt: + raise ServiceError("Prompt cannot be empty") + with self.factory() as db: + sess = db.get(Session, session_id) + if sess is None: + raise ServiceError("Session not found") + row = db.query(ScheduledPrompt).filter_by(session_id=session_id).first() + if row is None: + row = ScheduledPrompt(session_id=session_id, prompt=prompt, due_at=due_at) + db.add(row) + else: + row.prompt = prompt + row.due_at = due_at + row.status = PROMPT_PENDING + row.error = None + row.sent_at = None + db.commit() + return row + + def cancel_scheduled_prompt(self, session_id: int) -> None: + """Delete any scheduled prompt row for a session.""" + with self.factory() as db: + row = db.query(ScheduledPrompt).filter_by(session_id=session_id).first() + if row is None: + return + db.delete(row) + db.commit() + + def get_scheduled_prompt(self, session_id: int) -> ScheduledPrompt | None: + """Return the pending scheduled prompt for a session, if any.""" + with self.factory() as db: + return ( + db.query(ScheduledPrompt) + .filter_by(session_id=session_id, status=PROMPT_PENDING) + .first() + ) +``` + +- [ ] **Step 6: Hydrate pending prompt metadata for session rows** + +Add a helper inside `SessionService`: + +```python + def _pending_prompt_map( + self, db: DBSession, session_ids: list[int] + ) -> dict[int, ScheduledPrompt]: + if not session_ids: + return {} + rows = ( + db.query(ScheduledPrompt) + .filter( + ScheduledPrompt.session_id.in_(session_ids), + ScheduledPrompt.status == PROMPT_PENDING, + ) + .all() + ) + return {row.session_id: row for row in rows} +``` + +In both `list_sessions()` and `sync_window_labels()`, after building `names`, add: + +```python + pending_prompts = self._pending_prompt_map(db, [s.id for s in sessions]) +``` + +When appending `SessionInfo`, pass: + +```python + scheduled_prompt=pending_prompts.get(s.id), +``` + +For the one-line append in `sync_window_labels()`, replace it with: + +```python + result.append( + SessionInfo( + session=s, + alive=alive, + status=status, + scheduled_prompt=pending_prompts.get(s.id), + ) + ) +``` + +- [ ] **Step 7: Run service tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_sessions.py -q +``` + +Expected: PASS. + +- [ ] **Step 8: Commit** + +```bash +git add src/hqt/sessions/service.py tests/test_sessions.py +git commit -m "feat: schedule prompts in session service" +``` + +--- + +### Task 4: Dispatch Due Prompts + +**Files:** +- Modify: `src/hqt/sessions/service.py` +- Test: `tests/test_sessions.py` + +- [ ] **Step 1: Write failing dispatch tests** + +Append these tests to the delayed prompt scheduling section in `tests/test_sessions.py`: + +```python +@pytest.mark.asyncio +async def test_dispatch_due_prompt_alive_window_sends_and_marks_sent( + service, db, tmux +): + from datetime import datetime, timedelta + from hqt.db.models import ScheduledPrompt + + 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 + 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 +): + from datetime import datetime, timedelta + + 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 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") + + +@pytest.mark.asyncio +async def test_dispatch_due_prompt_resume_failure_marks_failed( + service, db, tmux, harnesses +): + from datetime import datetime, timedelta + from hqt.db.models import ScheduledPrompt + + 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 results[0].ok is False + assert "resume failed" in results[0].error + tmux.send_text.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 +): + from datetime import datetime, timedelta + from hqt.db.models import ScheduledPrompt + + 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 first[0].ok is False + assert second == [] + assert tmux.send_text.await_count == 1 + db.expire_all() + row = db.get(ScheduledPrompt, scheduled.id) + assert row.status == "failed" + assert "send text failed" in row.error +``` + +- [ ] **Step 2: Run dispatch tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_sessions.py -q +``` + +Expected: FAIL because `DispatchResult`, `_ensure_session_alive()`, and `dispatch_due_prompts()` do not exist. + +- [ ] **Step 3: Add dispatch result dataclass** + +In `src/hqt/sessions/service.py`, add after `CreateSessionResult`: + +```python +@dataclass +class DispatchResult: + session_id: int + window_name: str + ok: bool + error: str = "" +``` + +- [ ] **Step 4: Factor no-attach resume helper** + +In `attach_session()`, replace the alive/respawn block: + +```python + 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) + if not ok: + return False + return await self.tmux.attach(window_name) +``` + +with: + +```python + ok = await self._ensure_session_alive(db, sess, window_name) + if not ok: + return False + return await self.tmux.attach(window_name) +``` + +Add the helper before `_respawn_with_fallback()`: + +```python + 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.""" + 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(db, sess, window_name) +``` + +- [ ] **Step 5: Add prompt failure helper and dispatcher** + +Add these methods to `SessionService` after `get_scheduled_prompt()`: + +```python + 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 = await self._ensure_session_alive(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, "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 +``` + +- [ ] **Step 6: Run dispatch and attach regression tests** + +Run: + +```bash +uv run pytest tests/test_sessions.py::test_attach_session_dead_resume_ok_attaches_once tests/test_sessions.py::test_dispatch_due_prompt_alive_window_sends_and_marks_sent tests/test_sessions.py::test_dispatch_due_prompt_dead_window_resumes_without_attach tests/test_sessions.py::test_dispatch_due_prompt_resume_failure_marks_failed tests/test_sessions.py::test_dispatch_due_prompt_send_failure_does_not_retry_next_poll -q +``` + +Expected: PASS. + +- [ ] **Step 7: Run all session tests** + +Run: + +```bash +uv run pytest tests/test_sessions.py -q +``` + +Expected: PASS. + +- [ ] **Step 8: Commit** + +```bash +git add src/hqt/sessions/service.py tests/test_sessions.py +git commit -m "feat: dispatch due scheduled prompts" +``` + +--- + +### Task 5: Build Schedule Prompt Modal + +**Files:** +- Create: `src/hqt/tui/screens/schedule_prompt.py` +- Modify: `src/hqt/tui/styles.tcss` +- Test: `tests/test_tui.py` + +- [ ] **Step 1: Write failing modal tests** + +Append this section in `tests/test_tui.py` near the other screen/modal tests: + +```python +# --------------------------------------------------------------------------- +# SchedulePromptScreen +# --------------------------------------------------------------------------- + + +def test_parse_delay_accepts_seconds_minutes_hours(): + from datetime import timedelta + from hqt.tui.screens.schedule_prompt import parse_delay + + assert parse_delay("90s") == timedelta(seconds=90) + assert parse_delay("30m") == timedelta(minutes=30) + assert parse_delay("2h") == timedelta(hours=2) + assert parse_delay("1h30m") == timedelta(hours=1, minutes=30) + + +def test_parse_delay_rejects_invalid_values(): + from hqt.tui.screens.schedule_prompt import parse_delay + + for value in ["", "0m", "abc", "1d", "1h 30m"]: + with pytest.raises(ValueError): + parse_delay(value) + + +@pytest.mark.asyncio +async def test_schedule_prompt_screen_submits_prompt_and_delay(): + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + from hqt.tui.screens.schedule_prompt import SchedulePromptScreen + from textual.widgets import Input + + results = [] + app.push_screen(SchedulePromptScreen(), results.append) + await pilot.pause() + + app.screen.query_one("#prompt-input", Input).value = "continue" + app.screen.query_one("#delay-input", Input).value = "30m" + app.screen.query_one("#prompt-input", Input).focus() + await pilot.press("enter") + await pilot.pause() + + assert len(results) == 1 + assert results[0][0] == "continue" + assert results[0][1].total_seconds() == 1800 + + +@pytest.mark.asyncio +async def test_schedule_prompt_screen_invalid_delay_stays_open(): + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + from hqt.tui.screens.schedule_prompt import SchedulePromptScreen + from textual.widgets import Input, Label + + results = [] + app.push_screen(SchedulePromptScreen(), results.append) + await pilot.pause() + + app.screen.query_one("#prompt-input", Input).value = "continue" + app.screen.query_one("#delay-input", Input).value = "bogus" + await pilot.press("enter") + await pilot.pause() + + assert results == [] + error_text = str(app.screen.query_one("#schedule-error", Label).render()) + assert "Use a delay like" in error_text +``` + +- [ ] **Step 2: Run modal tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_parse_delay_accepts_seconds_minutes_hours tests/test_tui.py::test_parse_delay_rejects_invalid_values tests/test_tui.py::test_schedule_prompt_screen_submits_prompt_and_delay tests/test_tui.py::test_schedule_prompt_screen_invalid_delay_stays_open -q +``` + +Expected: FAIL because `schedule_prompt.py` does not exist. + +- [ ] **Step 3: Create the modal** + +Create `src/hqt/tui/screens/schedule_prompt.py`: + +```python +import re +from datetime import timedelta + +from textual.app import ComposeResult +from textual.containers import Horizontal, Vertical +from textual.screen import ModalScreen +from textual.widgets import Button, Input, Label + +_DELAY_RE = re.compile(r"(?P\d+)(?P[smh])") + + +def parse_delay(value: str) -> timedelta: + text = value.strip().lower() + if not text: + raise ValueError("empty delay") + pos = 0 + seconds = 0 + for match in _DELAY_RE.finditer(text): + if match.start() != pos: + raise ValueError("invalid delay") + amount = int(match.group("value")) + unit = match.group("unit") + if unit == "s": + seconds += amount + elif unit == "m": + seconds += amount * 60 + elif unit == "h": + seconds += amount * 3600 + pos = match.end() + if pos != len(text) or seconds <= 0: + raise ValueError("invalid delay") + return timedelta(seconds=seconds) + + +class SchedulePromptScreen(ModalScreen[tuple[str, timedelta] | None]): + """Collect prompt text and a relative delay.""" + + def compose(self) -> ComposeResult: + with Vertical(id="schedule-prompt-dialog"): + yield Label("Prompt Later") + yield Label("Prompt:") + yield Input(placeholder="message to send", id="prompt-input") + yield Label("Delay:") + yield Input(placeholder="30m, 2h, 1h30m, 90s", id="delay-input") + yield Label("", id="schedule-error") + with Horizontal(classes="dialog-actions"): + yield Button("OK", variant="primary", id="ok-btn") + yield Button("Cancel", id="cancel-btn") + + def on_button_pressed(self, event: Button.Pressed) -> None: + if event.button.id == "ok-btn": + self._submit() + else: + self.dismiss(None) + + def on_input_submitted(self, event: Input.Submitted) -> None: + self._submit() + + def _submit(self) -> None: + prompt_input = self.query_one("#prompt-input", Input) + delay_input = self.query_one("#delay-input", Input) + error = self.query_one("#schedule-error", Label) + prompt = prompt_input.value.strip() + if not prompt: + error.update("Prompt cannot be empty") + prompt_input.focus() + return + try: + delay = parse_delay(delay_input.value) + except ValueError: + error.update("Use a delay like 30m, 2h, 1h30m, or 90s") + delay_input.focus() + return + self.dismiss((prompt, delay)) +``` + +- [ ] **Step 4: Style the modal** + +In `src/hqt/tui/styles.tcss`, update: + +```css +ProjectFormScreen, NewSessionScreen { +``` + +to: + +```css +ProjectFormScreen, NewSessionScreen, SchedulePromptScreen { +``` + +Update: + +```css +#project-form-dialog, #new-session-dialog { +``` + +to: + +```css +#project-form-dialog, #new-session-dialog, #schedule-prompt-dialog { +``` + +Add: + +```css +#schedule-error { + color: $error; + height: 1; +} +``` + +- [ ] **Step 5: Run modal tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_parse_delay_accepts_seconds_minutes_hours tests/test_tui.py::test_parse_delay_rejects_invalid_values tests/test_tui.py::test_schedule_prompt_screen_submits_prompt_and_delay tests/test_tui.py::test_schedule_prompt_screen_invalid_delay_stays_open -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/hqt/tui/screens/schedule_prompt.py src/hqt/tui/styles.tcss tests/test_tui.py +git commit -m "feat: add delayed prompt modal" +``` + +--- + +### Task 6: Render Pending Prompt Indicators + +**Files:** +- Modify: `src/hqt/tui/widgets/session_list.py` +- Test: `tests/test_tui.py` + +- [ ] **Step 1: Write failing row-format tests** + +Append near the existing `SessionList` tests in `tests/test_tui.py`: + +```python +def _make_scheduled_prompt(due_at): + prompt = MagicMock() + prompt.due_at = due_at + prompt.status = "pending" + return prompt + + +def test_format_session_text_shows_scheduled_prompt_countdown(): + from datetime import datetime, timedelta + from hqt.tui.widgets.session_list import format_session_text + + now = datetime(2026, 6, 11, 14, 0, 0) + scheduled = _make_scheduled_prompt(now + timedelta(minutes=12, seconds=1)) + + text = format_session_text( + "mywork", + "claude", + "idle", + "●", + scheduled_prompt=scheduled, + now=now, + ) + + assert "@ 13m" in text + + +def test_format_session_text_shows_scheduled_prompt_clock_time(): + from datetime import datetime, timedelta + from hqt.tui.widgets.session_list import format_session_text + + now = datetime(2026, 6, 11, 14, 0, 0) + scheduled = _make_scheduled_prompt(now + timedelta(hours=2, minutes=30)) + + text = format_session_text( + "mywork", + "claude", + "idle", + "●", + scheduled_prompt=scheduled, + now=now, + ) + + assert "@ 16:30" in text +``` + +- [ ] **Step 2: Run row-format tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_format_session_text_shows_scheduled_prompt_countdown tests/test_tui.py::test_format_session_text_shows_scheduled_prompt_clock_time -q +``` + +Expected: FAIL because `format_session_text()` has no scheduled prompt support. + +- [ ] **Step 3: Update formatter signature and helper** + +In `src/hqt/tui/widgets/session_list.py`, add imports: + +```python +from datetime import datetime +from math import ceil +``` + +Add helper above `format_session_text()`: + +```python +def format_scheduled_prompt(due_at: datetime, now: datetime | None = None) -> str: + if now is None: + now = datetime.now() + seconds = max(0, (due_at - now).total_seconds()) + if seconds < 3600: + return f"@ {max(1, ceil(seconds / 60))}m" + return f"@ {due_at.strftime('%H:%M')}" +``` + +Change `format_session_text()` signature: + +```python +def format_session_text( + nickname: str, + harness_name: str, + status_text: str, + symbol: str, + worktree_branch: str | None = None, + scheduled_prompt=None, + now: datetime | None = None, +) -> str: +``` + +Before the return, add: + +```python + schedule = "" + if scheduled_prompt is not None: + schedule = f" {format_scheduled_prompt(scheduled_prompt.due_at, now=now)}" +``` + +Replace the return with: + +```python + return f"[{color}]{symbol}[/] {label} {status_text}{schedule}" +``` + +- [ ] **Step 4: Pass scheduled prompt metadata from refresh** + +In `SessionList.refresh_sessions()`, update the formatter call: + +```python + text = format_session_text( + s.nickname or s.tmux_session_name, + s.harness.name, + status_text, + symbol, + getattr(s, "worktree_branch", None), + scheduled_prompt=getattr(info, "scheduled_prompt", None), + ) +``` + +- [ ] **Step 5: Run TUI row tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_format_session_text_shows_scheduled_prompt_countdown tests/test_tui.py::test_format_session_text_shows_scheduled_prompt_clock_time tests/test_tui.py::test_session_list_label_includes_status -q +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add src/hqt/tui/widgets/session_list.py tests/test_tui.py +git commit -m "feat: show scheduled prompts in session rows" +``` + +--- + +### Task 7: Wire TUI Scheduling, Canceling, And Poll Dispatch + +**Files:** +- Modify: `src/hqt/tui/app.py` +- Test: `tests/test_tui.py` + +- [ ] **Step 1: Write failing TUI action tests** + +Append these tests to `tests/test_tui.py`: + +```python +@pytest.mark.asyncio +async def test_prompt_later_binding_exists(): + bindings = {b.key: b for b in HqtApp.BINDINGS} + assert "p" in bindings + assert bindings["p"].action == "schedule_prompt" + assert "P" in bindings + assert bindings["P"].action == "cancel_scheduled_prompt" + + +@pytest.mark.asyncio +async def test_schedule_prompt_action_calls_service_and_refreshes(): + from datetime import datetime + from unittest.mock import patch + from hqt.db.models import Harness, Project, Session + from hqt.tui.widgets.session_list import SessionList + + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + seed_db = app._db_factory() + proj = Project(name="prompt-proj", path="/tmp/prompt-proj") + seed_db.add(proj) + seed_db.flush() + harness = seed_db.query(Harness).first() + sess = Session( + project_id=proj.id, + harness_id=harness.id, + tmux_session_name="hqt-prompt-test", + archived=False, + ) + seed_db.add(sess) + seed_db.commit() + proj_id = proj.id + sess_id = sess.id + seed_db.close() + + app._selected_project_id = proj_id + await app._refresh_sessions() + await pilot.pause() + app.query_one(SessionList).query_one("#session-list").focus() + + mock_service = MagicMock() + mock_service.schedule_prompt = MagicMock() + mock_service.list_sessions = AsyncMock(return_value=[]) + mock_service.sync_window_labels = AsyncMock(return_value=[]) + mock_service.dispatch_due_prompts = AsyncMock(return_value=[]) + app._session_service = mock_service + + notify_calls = [] + + def capture_notify(message, **kwargs): + notify_calls.append((message, kwargs)) + + with patch.object(app, "notify", side_effect=capture_notify): + app.action_schedule_prompt() + await pilot.pause() + app.screen.dismiss(("continue", __import__("datetime").timedelta(minutes=30))) + await pilot.pause() + await app.workers.wait_for_complete() + + assert mock_service.schedule_prompt.call_count == 1 + args = mock_service.schedule_prompt.call_args.args + assert args[0] == sess_id + assert args[1] == "continue" + assert isinstance(args[2], datetime) + mock_service.list_sessions.assert_awaited() + assert any("Prompt scheduled" in call[0] for call in notify_calls) + + +@pytest.mark.asyncio +async def test_cancel_scheduled_prompt_action_calls_service_and_refreshes(): + from hqt.db.models import Harness, Project, Session + from hqt.tui.widgets.session_list import SessionList + + app = HqtApp() + async with app.run_test(size=(120, 40)) as pilot: + seed_db = app._db_factory() + proj = Project(name="cancel-proj", path="/tmp/cancel-proj") + seed_db.add(proj) + seed_db.flush() + harness = seed_db.query(Harness).first() + sess = Session( + project_id=proj.id, + harness_id=harness.id, + tmux_session_name="hqt-cancel-test", + archived=False, + ) + seed_db.add(sess) + seed_db.commit() + proj_id = proj.id + sess_id = sess.id + seed_db.close() + + app._selected_project_id = proj_id + await app._refresh_sessions() + await pilot.pause() + app.query_one(SessionList).query_one("#session-list").focus() + + mock_service = MagicMock() + mock_service.cancel_scheduled_prompt = MagicMock() + mock_service.list_sessions = AsyncMock(return_value=[]) + mock_service.sync_window_labels = AsyncMock(return_value=[]) + mock_service.dispatch_due_prompts = AsyncMock(return_value=[]) + app._session_service = mock_service + + await app.run_action("cancel_scheduled_prompt") + await pilot.pause() + await app.workers.wait_for_complete() + + mock_service.cancel_scheduled_prompt.assert_called_once_with(sess_id) + mock_service.list_sessions.assert_awaited() + + +@pytest.mark.asyncio +async def test_poll_sessions_dispatches_due_prompts_and_notifies(): + app = HqtApp() + async with app.run_test(size=(120, 40)): + result = MagicMock() + result.ok = True + result.window_name = "hqt-1" + result.error = "" + + mock_service = MagicMock() + mock_service.dispatch_due_prompts = AsyncMock(return_value=[result]) + mock_service.sync_window_labels = AsyncMock(return_value=[]) + app._session_service = mock_service + + notify_calls = [] + + def capture_notify(message, **kwargs): + notify_calls.append((message, kwargs)) + + from unittest.mock import patch + + with patch.object(app, "notify", side_effect=capture_notify): + await app._poll_sessions() + + mock_service.dispatch_due_prompts.assert_awaited_once() + assert any("Prompt sent" in call[0] for call in notify_calls) +``` + +- [ ] **Step 2: Run TUI action tests and verify they fail** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_prompt_later_binding_exists tests/test_tui.py::test_schedule_prompt_action_calls_service_and_refreshes tests/test_tui.py::test_cancel_scheduled_prompt_action_calls_service_and_refreshes tests/test_tui.py::test_poll_sessions_dispatches_due_prompts_and_notifies -q +``` + +Expected: FAIL because app bindings/actions/poll dispatch do not exist. + +- [ ] **Step 3: Import datetime and modal** + +In `src/hqt/tui/app.py`, add: + +```python +from datetime import datetime +``` + +Add the screen import: + +```python +from hqt.tui.screens.schedule_prompt import SchedulePromptScreen +``` + +- [ ] **Step 4: Add bindings** + +In `HqtApp.BINDINGS`, add after rename or stop: + +```python + Binding("p", "schedule_prompt", "Prompt Later"), + Binding("P", "cancel_scheduled_prompt", "Cancel Prompt"), +``` + +- [ ] **Step 5: Dispatch due prompts in the poller** + +At the start of `_poll_sessions()`, before `sync_window_labels()`, add: + +```python + dispatch_results = await self._sessions().dispatch_due_prompts() + for result in dispatch_results: + if result.ok: + self.notify(f"Prompt sent to {result.window_name}") + else: + target = result.window_name or f"session {result.session_id}" + self.notify( + f"Prompt failed for {target}: {result.error}", + severity="error", + ) +``` + +The final method keeps the existing `ServiceError` handling: + +```python + async def _poll_sessions(self) -> None: + try: + dispatch_results = await self._sessions().dispatch_due_prompts() + for result in dispatch_results: + if result.ok: + self.notify(f"Prompt sent to {result.window_name}") + else: + target = result.window_name or f"session {result.session_id}" + self.notify( + f"Prompt failed for {target}: {result.error}", + severity="error", + ) + infos = await self._sessions().sync_window_labels() + except ServiceError as err: + log.warning("Session poll failed: %s", err) + return + if self._selected_project_id is not None: + subset = [ + i for i in infos if i.session.project_id == self._selected_project_id + ] + await self.query_one(SessionList).refresh_sessions(subset) +``` + +- [ ] **Step 6: Add schedule and cancel actions** + +At the end of `HqtApp`, add: + +```python + def action_schedule_prompt(self) -> None: + sid = self.query_one(SessionList).get_selected_session_id() + if sid is None: + self.notify("Select a session first", severity="warning") + return + + def on_dismiss(result) -> None: + if result is None: + return + prompt, delay = result + due_at = datetime.now() + delay + try: + self._sessions().schedule_prompt(sid, prompt, due_at) + except ServiceError as err: + self.notify(str(err), severity="error") + return + self.notify(f"Prompt scheduled for session {sid} at {due_at:%H:%M}") + self._run_service_worker(self._refresh_sessions()) + + self.push_screen(SchedulePromptScreen(), on_dismiss) + + def action_cancel_scheduled_prompt(self) -> None: + sid = self.query_one(SessionList).get_selected_session_id() + if sid is None: + self.notify("Select a session first", severity="warning") + return + try: + self._sessions().cancel_scheduled_prompt(sid) + except ServiceError as err: + self.notify(str(err), severity="error") + return + self.notify(f"Scheduled prompt canceled for session {sid}") + self._run_service_worker(self._refresh_sessions()) +``` + +- [ ] **Step 7: Run TUI action tests and verify they pass** + +Run: + +```bash +uv run pytest tests/test_tui.py::test_prompt_later_binding_exists tests/test_tui.py::test_schedule_prompt_action_calls_service_and_refreshes tests/test_tui.py::test_cancel_scheduled_prompt_action_calls_service_and_refreshes tests/test_tui.py::test_poll_sessions_dispatches_due_prompts_and_notifies -q +``` + +Expected: PASS. + +- [ ] **Step 8: Run all TUI tests** + +Run: + +```bash +uv run pytest tests/test_tui.py -q +``` + +Expected: PASS. + +- [ ] **Step 9: Commit** + +```bash +git add src/hqt/tui/app.py tests/test_tui.py +git commit -m "feat: wire delayed prompt TUI actions" +``` + +--- + +### Task 8: Final Integration Verification + +**Files:** +- No planned source changes. Modify only files touched in earlier tasks when verification exposes a specific failure. + +- [ ] **Step 1: Run focused test suite** + +Run: + +```bash +uv run pytest tests/test_db.py tests/test_tmux.py tests/test_sessions.py tests/test_tui.py -q +``` + +Expected: PASS. + +- [ ] **Step 2: Run full test suite** + +Run: + +```bash +uv run pytest -q +``` + +Expected: PASS. + +- [ ] **Step 3: Run required project checks** + +Run: + +```bash +uv run ruff format src tests +uv run ruff check src tests +uv run ty check +``` + +Expected: + +- `ruff format`: completes with either "files left unchanged" or formatted files. +- `ruff check`: "All checks passed!" +- `ty check`: no errors. + +- [ ] **Step 4: Inspect final diff** + +Run: + +```bash +git status --short +git diff --stat +``` + +Expected: only intentional source/test files are modified. No unrelated files. + +- [ ] **Step 5: Commit verification fixes when verification changed files** + +If Step 3 formatted files or Step 1/2 required small fixes, commit them: + +```bash +git add src tests +git commit -m "test: verify delayed prompt injection" +``` + +If no files changed after verification, skip this commit. + +--- + +## Self-Review + +- Spec coverage: the plan covers persisted state, one prompt per session, TUI-only runtime, `p` schedule and `P` cancel bindings, row visualization, poll dispatch notifications, auto-resume without attach/select, literal tmux injection, failure marking, and no automatic retries. +- Scope control: CLI scheduling, background dispatch while hqt is closed, absolute timestamps, recurring prompts, multiple queued prompts, prompt history, and harness-specific templates are not implemented. +- Type consistency: `ScheduledPrompt`, `SessionInfo.scheduled_prompt`, `DispatchResult`, `schedule_prompt()`, `cancel_scheduled_prompt()`, `get_scheduled_prompt()`, `dispatch_due_prompts()`, `SchedulePromptScreen`, and `parse_delay()` are introduced before later tasks reference them.