From 2e58b191fa31b06c6875f7f68bbcfce46537395f Mon Sep 17 00:00:00 2001 From: Felix Martin Date: Thu, 11 Jun 2026 09:06:36 -0400 Subject: [PATCH] feat: wire delayed prompt TUI actions Co-Authored-By: Claude Opus 4.8 (1M context) --- src/hqt/tui/app.py | 48 +++++++++++++++ tests/test_tui.py | 149 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 197 insertions(+) diff --git a/src/hqt/tui/app.py b/src/hqt/tui/app.py index 2726734..5f00532 100644 --- a/src/hqt/tui/app.py +++ b/src/hqt/tui/app.py @@ -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()) diff --git a/tests/test_tui.py b/tests/test_tui.py index 964afc0..1bf7a65 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -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)