Parcourir la source

Added support for command type-in

Efren Yevale Varela il y a 1 mois
Parent
commit
2ef6488014
2 fichiers modifiés avec 283 ajouts et 22 suppressions
  1. 27 15
      src/stt_daemon/tui/app.py
  2. 256 7
      tests/test_tui_app.py

+ 27 - 15
src/stt_daemon/tui/app.py

@@ -28,13 +28,13 @@ class AssistantTUI(App):
     """Live view of transcriptions, commands, and chat."""
 
     ALLOW_IN_MAXIMIZED_VIEW = ".__never__"
+    AUTO_FOCUS = None
 
     CSS = """
     #status { height: 1; }
     #transcript { height: 25%; border: solid $accent; }
     #command { height: 3; border: solid $accent; }
     #chat { height: 1fr; border: solid $accent; }
-    #edit { display: none; }
     #main { height: 1fr; }
     #main #triage { width: 1fr; height: 100%; border: solid $accent; }
     #main #chat { width: 2fr; height: 100%; }
@@ -74,18 +74,17 @@ class AssistantTUI(App):
     def compose(self) -> ComposeResult:
         yield Static(id="status")
         yield RichLog(id="transcript", wrap=True, markup=False)
-        yield Static(id="command")
+        yield Input(id="command", placeholder="type a command, Enter to run")
         if self.triage_enabled:
             with MainPane(id="main"):
                 yield RichLog(id="triage", wrap=True, markup=False)
                 yield RichLog(id="chat", wrap=True, markup=False)
         else:
             yield RichLog(id="chat", wrap=True, markup=False)
-        yield Input(placeholder="edit the command, then press Enter", id="edit")
         yield Footer()
 
     def _set_titles(self) -> None:
-        self.query_one("#command", Static).border_title = "command buffer"
+        self.query_one("#command", Input).border_title = "command buffer"
         if self.triage_enabled:
             self.query_one("#triage", RichLog).border_title = "triage"
         self._update_titles(self.state)
@@ -148,10 +147,15 @@ class AssistantTUI(App):
         self.query_one("#status", Static).update(status)
         self._update_titles(state)
 
-        edited = " (edited)" if state.command_edited else ""
-        self.query_one("#command", Static).update(
-            f"command{edited}: {state.command_text or '<empty>'}"
-        )
+        cmd_input = self.query_one("#command", Input)
+        if not cmd_input.has_focus:
+            cmd_input.value = state.command_text
+        title = "command buffer"
+        if state.mode == "command_armed":
+            title += " · armed"
+        if state.command_edited:
+            title += " (edited)"
+        cmd_input.border_title = title
 
         transcript_log = self.query_one("#transcript", RichLog)
         lines = state.transcript_lines()
@@ -202,10 +206,10 @@ class AssistantTUI(App):
             self._send("clear")
 
     def action_edit(self) -> None:
-        edit = self.query_one("#edit", Input)
-        edit.value = self.state.command_text
-        edit.display = True
-        edit.focus()
+        cmd = self.query_one("#command", Input)
+        cmd.value = self.state.command_text
+        cmd.focus()
+        cmd.cursor_position = len(cmd.value)
 
     def _active_pane(self):
         focused = self.screen.focused
@@ -282,9 +286,17 @@ class AssistantTUI(App):
                 pass
 
     def on_input_submitted(self, event: Input.Submitted) -> None:
-        self._send("command_edit", text=event.value)
-        edit = self.query_one("#edit", Input)
-        edit.display = False
+        if event.input.id != "command":
+            return
+        text = event.value.strip()
+        if not text:
+            return
+        if self.state.mode != "command_armed":
+            self._send("command_start")
+        self._send("command_edit", text=text)
+        self._send("command_execute")
+        event.input.value = ""
+        self.set_focus(None)
 
     def on_unmount(self) -> None:
         if self._timer is not None:

+ 256 - 7
tests/test_tui_app.py

@@ -40,7 +40,7 @@ def server():
     shutil.rmtree(directory, ignore_errors=True)
 
 
-def test_ctrl_r_sends_execute_and_e_opens_editor(server):
+def test_ctrl_r_sends_execute_and_e_focuses_command(server):
     pytest.importorskip("textual")
     from textual.widgets import Input
 
@@ -65,16 +65,265 @@ def test_ctrl_r_sends_execute_and_e_opens_editor(server):
 
             await pilot.press("e")
             await pilot.pause()
-            edit = app.query_one("#edit", Input)
-            observed["edit_display"] = edit.display
-            observed["edit_focused"] = edit.has_focus
+            cmd = app.query_one("#command", Input)
+            observed["cmd_focused"] = cmd.has_focus
 
     asyncio.run(scenario())
     assert ("command_execute", {}) in sent
     assert ("command_start", {}) in sent
     assert ("command_cancel", {}) in sent
-    assert observed["edit_display"] is True
-    assert observed["edit_focused"] is True
+    assert observed["cmd_focused"] is True
+
+
+def test_type_command_enter_from_idle_sends_start_edit_execute(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    sent: list[tuple] = []
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.client.send = lambda type_, **fields: sent.append((type_, fields))
+            assert app.state.mode == "idle"
+            app.query_one("#command", Input).focus()
+            await pilot.pause()
+            for ch in "take notes":
+                await pilot.press(ch)
+            await pilot.pause()
+            await pilot.press("enter")
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    assert [t for t, _ in sent] == ["command_start", "command_edit", "command_execute"], sent
+    assert sent[1][1] == {"text": "take notes"}, sent
+
+
+def test_type_command_enter_when_armed_skips_start(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    sent: list[tuple] = []
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.client.send = lambda type_, **fields: sent.append((type_, fields))
+            app._timer.stop()  # no background drain: server status must not clobber state
+            app._drain()
+            app.state.ingest({"type": "mode", "mode": "command_armed"})
+            assert app.state.mode == "command_armed"
+            app.query_one("#command", Input).focus()
+            await pilot.pause()
+            for ch in "open calendar":
+                await pilot.press(ch)
+            await pilot.pause()
+            await pilot.press("enter")
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    assert [t for t, _ in sent] == ["command_edit", "command_execute"], sent
+    assert sent[0][1] == {"text": "open calendar"}, sent
+
+
+def test_empty_command_enter_sends_nothing(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    sent: list[tuple] = []
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.client.send = lambda type_, **fields: sent.append((type_, fields))
+            app.query_one("#command", Input).focus()
+            await pilot.pause()
+            await pilot.press("enter")
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    assert sent == []
+
+
+def test_e_focuses_command_input(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    observed: dict = {}
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._timer.stop()  # no background drain: server status must not clobber state
+            app._drain()
+            app.state.ingest({"type": "command_text", "text": "voice text", "edited": False})
+            app._render()
+            await pilot.pause()
+            await pilot.press("e")
+            await pilot.pause()
+            cmd = app.query_one("#command", Input)
+            observed["focused"] = cmd.has_focus
+            observed["value"] = cmd.value
+
+    asyncio.run(scenario())
+    assert observed["focused"] is True
+    assert observed["value"] == "voice text"
+
+
+def test_voice_text_updates_unfocused_but_not_focused(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    observed: dict = {}
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._timer.stop()  # no background drain: server status must not clobber state
+            app._drain()
+            cmd = app.query_one("#command", Input)
+            assert not cmd.has_focus
+            app.state.ingest({"type": "command_text", "text": "voice built", "edited": False})
+            app._render()
+            await pilot.pause()
+            observed["unfocused_value"] = app.query_one("#command", Input).value
+
+            app.query_one("#command", Input).focus()
+            await pilot.pause()
+            for ch in "user typing":
+                await pilot.press(ch)
+            await pilot.pause()
+            app.state.ingest({"type": "command_text", "text": "voice late", "edited": False})
+            app._render()
+            await pilot.pause()
+            observed["focused_value"] = app.query_one("#command", Input).value
+
+    asyncio.run(scenario())
+    assert observed["unfocused_value"] == "voice built"
+    assert observed["focused_value"] == "user typing"
+
+
+def test_command_input_border_title(server):
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._render()
+            await pilot.pause()
+            return app.query_one("#command", Input).border_title
+
+    title = asyncio.run(scenario())
+    assert "command buffer" in title
+
+
+def test_no_autofocus_on_mount(server):
+    """Nothing is focused on mount: single-letter hotkeys must not be swallowed by #command."""
+    pytest.importorskip("textual")
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            await pilot.pause()
+            return app.focused
+
+    assert asyncio.run(scenario()) is None
+
+
+def test_submit_blurs_command_and_hotkey_still_fires(server):
+    """Type + Enter sends messages, clears focus; a hotkey (t) still fires afterwards."""
+    pytest.importorskip("textual")
+    from textual.widgets import Input
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path)
+    sent: list[tuple] = []
+    observed: dict = {}
+
+    async def scenario():
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app.client.send = lambda type_, **fields: sent.append((type_, fields))
+            app._timer.stop()  # no background drain: server status must not clobber state
+            app._drain()
+            app.query_one("#command", Input).focus()
+            await pilot.pause()
+            for ch in "take notes":
+                await pilot.press(ch)
+            await pilot.pause()
+            await pilot.press("enter")
+            await pilot.pause()
+            observed["focused_after_submit"] = app.focused
+            observed["value_after_submit"] = app.query_one("#command", Input).value
+            await pilot.press("t")
+            await pilot.pause()
+
+    asyncio.run(scenario())
+    assert [t for t, _ in sent] == [
+        "command_start",
+        "command_edit",
+        "command_execute",
+        "chat_toggle",
+    ], sent
+    assert observed["focused_after_submit"] is None
+    assert observed["value_after_submit"] == ""
+
+
+def test_focus_triage_then_m_maximizes_deterministically(server):
+    """Focus #triage, assert focus stuck, then m maximizes #main (no auto-focus race)."""
+    pytest.importorskip("textual")
+    from textual.widgets import RichLog
+
+    from stt_daemon.tui.app import AssistantTUI
+
+    path, _service = server
+    app = AssistantTUI(path, triage_enabled=True)
+    observed: dict = {}
+
+    async def scenario():
+        async with app.run_test(size=(80, 40)) as pilot:
+            await pilot.pause()
+            app.query_one("#triage", RichLog).focus()
+            await pilot.pause()
+            assert app.focused is not None and app.focused.id == "triage", app.focused
+            await pilot.press("m")
+            await pilot.pause()
+            observed["maximized_is_main"] = app.screen.maximized is app.query_one("#main")
+
+    asyncio.run(scenario())
+    assert observed["maximized_is_main"] is True
 
 
 def test_c_key_confirms_before_clearing(server):
@@ -379,7 +628,6 @@ def test_triage_disabled_composes_no_triage_widget(server):
             app.query_one("#status")
             app.query_one("#command")
             app.query_one("#chat")
-            app.query_one("#edit")
 
     asyncio.run(scenario())
 
@@ -642,6 +890,7 @@ def test_maximize_triage_hides_chat_and_fills_screen(server):
             await pilot.pause()
             app.query_one("#triage", RichLog).focus()
             await pilot.pause()
+            assert app.focused is not None and app.focused.id == "triage", app.focused
             await pilot.press("m")
             await pilot.pause()
             observed["maximized_is_main"] = app.screen.maximized is app.query_one("#main")