felixm/hqt

Files

T

felixm a5f2f69faa docs: plan delayed prompt injection

2026-06-10 22:46:55 -04:00

49 KiB

Raw Blame History

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:

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:

assert _user_version(engine) == migrations.LATEST_VERSION

Step 2: Run DB tests and verify they fail

Run:

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:

from sqlalchemy import ForeignKey, String, Text, func

Inside class Session, add:

    scheduled_prompt: Mapped["ScheduledPrompt | None"] = relationship(
        back_populates="session"
    )

After class Session, add:

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:

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:

MIGRATIONS: list[tuple[int, Callable[[Connection], None]]] = [
    (2, _migrate_v2),
    (3, _migrate_v3),
]

Step 5: Run DB tests and verify they pass

Run:

uv run pytest tests/test_db.py -q

Expected: PASS.

Step 6: Commit

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:

@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:

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:

    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():

    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:

uv run pytest tests/test_tmux.py -q

Expected: PASS.

Step 6: Commit

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:

    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:

# ---------------------------------------------------------------------------
# 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:

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:

from dataclasses import dataclass
from datetime import UTC, datetime

Update model imports:

from hqt.db.models import Harness, Project, ScheduledPrompt, Session

Add status constants near the existing module constants:

PROMPT_PENDING = "pending"
PROMPT_SENDING = "sending"
PROMPT_SENT = "sent"
PROMPT_FAILED = "failed"

Update SessionInfo:

@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():

    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:

    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:

            pending_prompts = self._pending_prompt_map(db, [s.id for s in sessions])

When appending SessionInfo, pass:

                    scheduled_prompt=pending_prompts.get(s.id),

For the one-line append in sync_window_labels(), replace it with:

                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:

uv run pytest tests/test_sessions.py -q

Expected: PASS.

Step 8: Commit

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:

@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:

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:

@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:

            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:

            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():

    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():

    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:

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:

uv run pytest tests/test_sessions.py -q

Expected: PASS.

Step 8: Commit

git add src/hqt/sessions/service.py tests/test_sessions.py
git commit -m "feat: dispatch due scheduled prompts"

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:

# ---------------------------------------------------------------------------
# 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:

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:

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<value>\d+)(?P<unit>[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:

ProjectFormScreen, NewSessionScreen {

to:

ProjectFormScreen, NewSessionScreen, SchedulePromptScreen {

Update:

#project-form-dialog, #new-session-dialog {

to:

#project-form-dialog, #new-session-dialog, #schedule-prompt-dialog {

Add:

#schedule-error {
    color: $error;
    height: 1;
}

Step 5: Run modal tests and verify they pass

Run:

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

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:

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:

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:

from datetime import datetime
from math import ceil

Add helper above format_session_text():

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:

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:

    schedule = ""
    if scheduled_prompt is not None:
        schedule = f"  {format_scheduled_prompt(scheduled_prompt.due_at, now=now)}"

Replace the return with:

    return f"[{color}]{symbol}[/] {label} {status_text}{schedule}"

Step 4: Pass scheduled prompt metadata from refresh

In SessionList.refresh_sessions(), update the formatter call:

            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:

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

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:

@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:

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:

from datetime import datetime

Add the screen import:

from hqt.tui.screens.schedule_prompt import SchedulePromptScreen

Step 4: Add bindings

In HqtApp.BINDINGS, add after rename or stop:

        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:

            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:

    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:

    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:

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:

uv run pytest tests/test_tui.py -q

Expected: PASS.

Step 9: Commit

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:

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:

uv run pytest -q

Expected: PASS.

Step 3: Run required project checks

Run:

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:

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:

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.

49 KiB Raw Blame History

Delayed Prompt Injection Implementation Plan

File Structure

Task 1: Persist ScheduledPrompt

Task 2: Return Injection Success From Tmux Layer

Task 3: Add Scheduling Service Methods And Row Metadata

Task 4: Dispatch Due Prompts

Task 5: Build Schedule Prompt Modal

Task 6: Render Pending Prompt Indicators

Task 7: Wire TUI Scheduling, Canceling, And Poll Dispatch

Task 8: Final Integration Verification

Self-Review

49 KiB

Raw Blame History