1657 lines
49 KiB
Markdown
1657 lines
49 KiB
Markdown
# 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<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:
|
|
|
|
```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.
|