felixm/hqt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: wire delayed prompt TUI actions

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Felix Martin
2026-06-11 09:06:36 -04:00
parent dfea5a5630
commit 2e58b191fa
2 changed files with 197 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/hqt/tui/app.py
+48
View File
@@ -1,5 +1,6 @@
import logging
from collections.abc import Coroutine
from datetime import datetime
from pathlib import Path
from typing import Any
@@ -24,6 +25,7 @@ from hqt.tui.screens.confirm_delete import ConfirmDeleteScreen
from hqt.tui.screens.project_form import ProjectFormScreen
from hqt.tui.screens.new_session import NewSessionScreen
from hqt.tui.screens.rename_session import RenameSessionScreen
from hqt.tui.screens.schedule_prompt import SchedulePromptScreen
from hqt.tui.widgets.project_list import ProjectList, ProjectSelected
from hqt.tui.widgets.session_list import SessionList, SessionSelected
@@ -71,6 +73,8 @@ class HqtApp(App):
Binding("enter", "attach_session", "Attach", priority=True),
Binding("r", "rename_session", "Rename"),
Binding("s", "stop_session", "Stop"),
Binding("p", "schedule_prompt", "Prompt Later"),
Binding("P", "cancel_scheduled_prompt", "Cancel Prompt"),
]
CSS_PATH = "styles.tcss"
@@ -159,6 +163,16 @@ class HqtApp(App):
# Sync every session's tmux window label (session-wide), then update the
# UI for the selected project from the same computed status — no recompute.
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)
@@ -411,3 +425,37 @@ class HqtApp(App):
await self._refresh_sessions()
self._run_service_worker(_do())
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())
tests/test_tui.py
+149
View File
@@ -1718,3 +1718,152 @@ async def test_sandbox_switch_enabled_when_available():
)
await pilot.pause()
assert app.screen.query_one("#sandbox-switch", Switch).disabled is False
# ---------------------------------------------------------------------------
# delayed prompt: bindings, actions, poll dispatch
# ---------------------------------------------------------------------------
@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()
await pilot.press("down")
await pilot.pause()
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()
await pilot.press("down")
await pilot.pause()
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)