Latest Codex app exposed a bug where redrawing the screen could cause speech to slip through even when it's turned off. Hopefully fixed now.

This commit is contained in:
Storm Dragon
2026-08-21 18:36:04 -04:00
parent 45019091ab
commit 9e6e806947
10 changed files with 272 additions and 15 deletions
@@ -408,6 +408,20 @@ class command:
text, text,
re.IGNORECASE, re.IGNORECASE,
) )
if not interruptible_activity_match:
interruptible_activity_match = re.search(
(
r'^\s*(?:thinking|working|processing|analyzing|searching|'
r'reading|writing|planning|running|executing|updating|'
r'building|installing|compiling|downloading|reviewing|'
r'generating|responding|applying|fixing|editing|creating|'
r'preparing|checking|opening|loading|fetching|retrieving|'
r'scanning|indexing|summarizing)\b.*'
r'\([^)]*(?:esc|ctrl\+c) to interrupt[^)]*\)\s*$'
),
text,
re.IGNORECASE,
)
if interruptible_activity_match: if interruptible_activity_match:
if current_time - self.env["commandBuffer"]["lastProgressTime"] >= 1.0: if current_time - self.env["commandBuffer"]["lastProgressTime"] >= 1.0:
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
+31 -7
View File
@@ -249,13 +249,20 @@ class FenrirManager:
): ):
self.environment["runtime"]["InputManager"].clear_last_deep_input() self.environment["runtime"]["InputManager"].clear_last_deep_input()
if not self._input_target_is_active():
self.environment["runtime"]["InputManager"].clear_last_deep_input()
return
if ( if (
self.environment["runtime"][ self._has_direct_user_input()
"CursorManager" and (
].is_cursor_vertical_move() self.environment["runtime"][
or self.environment["runtime"][ "CursorManager"
"CursorManager" ].is_cursor_vertical_move()
].is_cursor_horizontal_move() or self.environment["runtime"][
"CursorManager"
].is_cursor_horizontal_move()
)
): ):
self.environment["runtime"][ self.environment["runtime"][
"CommandManager" "CommandManager"
@@ -264,7 +271,22 @@ class FenrirManager:
self.environment["runtime"]["CommandManager"].execute_default_trigger( self.environment["runtime"]["CommandManager"].execute_default_trigger(
"onScreenUpdate" "onScreenUpdate"
) )
self.environment["runtime"]["InputManager"].clear_last_deep_input()
def _input_target_is_active(self):
input_driver = self.environment["runtime"].get("InputDriver")
if input_driver is None:
return True
return input_driver.is_active()
def _has_direct_user_input(self):
deepest_input = self.environment["runtime"][
"InputManager"
].get_last_deepest_input()
if not deepest_input:
return False
return not any(
key in deepest_input for key in ("KEY_FENRIR", "KEY_SCRIPT")
)
def handle_plug_input_device(self, event): def handle_plug_input_device(self, event):
try: try:
@@ -288,6 +310,8 @@ class FenrirManager:
) )
def handle_heart_beat(self, event): def handle_heart_beat(self, event):
if not self._input_target_is_active():
return
self.environment["runtime"]["CommandManager"].execute_default_trigger( self.environment["runtime"]["CommandManager"].execute_default_trigger(
"onHeartBeat", force=True "onHeartBeat", force=True
) )
@@ -26,6 +26,10 @@ class InputDriver:
time.sleep(0.1) time.sleep(0.1)
return None return None
def is_active(self):
"""Return whether this input driver's target currently has focus."""
return True
def clear_event_buffer(self): def clear_event_buffer(self):
if not self._initialized: if not self._initialized:
return return
+1 -1
View File
@@ -4,5 +4,5 @@
# Fenrir TTY screen reader # Fenrir TTY screen reader
# By Chrys, Storm Dragon, and contributors. # By Chrys, Storm Dragon, and contributors.
version = "2026.08.17" version = "2026.08.21"
code_name = "testing" code_name = "testing"
@@ -267,6 +267,9 @@ class driver(inputDriver):
return None return None
return int(prop.value[0]) return int(prop.value[0])
def is_active(self):
return self.active
def format_window_id(self, window_id): def format_window_id(self, window_id):
if window_id is None: if window_id is None:
return "None" return "None"
@@ -331,6 +334,9 @@ class driver(inputDriver):
if event_type == X.FocusOut: if event_type == X.FocusOut:
self.active = False self.active = False
self.reset_input_state() self.reset_input_state()
output_manager = self.env["runtime"].get("OutputManager")
if output_manager:
output_manager.interrupt_output_async()
return return
if event_type not in [X.KeyPress, X.KeyRelease]: if event_type not in [X.KeyPress, X.KeyRelease]:
return return
@@ -462,11 +462,10 @@ class driver(screenDriver):
) )
def record_stdin_keypress(self, msg_bytes): def record_stdin_keypress(self, msg_bytes):
if msg_bytes != b"\t": key_name = self.stdin_key_name(msg_bytes)
return
try: try:
self.env["runtime"]["InputManager"].record_unmanaged_keypress( self.env["runtime"]["InputManager"].record_unmanaged_keypress(
"KEY_TAB" key_name
) )
except Exception as e: except Exception as e:
self.env["runtime"]["DebugManager"].write_debug_out( self.env["runtime"]["DebugManager"].write_debug_out(
@@ -474,6 +473,38 @@ class driver(screenDriver):
debug.DebugLevel.ERROR, debug.DebugLevel.ERROR,
) )
def stdin_key_name(self, msg_bytes):
key_map = {
b"\t": "KEY_TAB",
b"\x08": "KEY_BACKSPACE",
b"\x7f": "KEY_BACKSPACE",
b"\x1b[A": "KEY_UP",
b"\x1bOA": "KEY_UP",
b"\x1b[B": "KEY_DOWN",
b"\x1bOB": "KEY_DOWN",
b"\x1b[C": "KEY_RIGHT",
b"\x1bOC": "KEY_RIGHT",
b"\x1b[D": "KEY_LEFT",
b"\x1bOD": "KEY_LEFT",
b"\x1b[H": "KEY_HOME",
b"\x1bOH": "KEY_HOME",
b"\x1b[F": "KEY_END",
b"\x1bOF": "KEY_END",
b"\x1b[5~": "KEY_PAGEUP",
b"\x1b[6~": "KEY_PAGEDOWN",
b"\x1b[3~": "KEY_DELETE",
b"\r": "KEY_ENTER",
b"\n": "KEY_ENTER",
b" ": "KEY_SPACE",
}
if msg_bytes in key_map:
return key_map[msg_bytes]
if len(msg_bytes) == 1:
char = chr(msg_bytes[0])
if char.isascii() and char.isalnum():
return "KEY_" + char.upper()
return "KEY_TERMINAL_INPUT"
def synthesize_backspace_shortcut(self, msg_bytes, event_queue): def synthesize_backspace_shortcut(self, msg_bytes, event_queue):
if not self.is_backspace_shortcut_sequence(msg_bytes): if not self.is_backspace_shortcut_sequence(msg_bytes):
return False return False
@@ -0,0 +1,117 @@
import time
from unittest.mock import Mock, call
import pytest
from fenrirscreenreader.core.fenrirManager import FenrirManager
def _build_manager(input_active=True, deepest_input=None):
manager = FenrirManager.__new__(FenrirManager)
input_manager = Mock(
get_last_input_time=Mock(return_value=time.time()),
get_last_deepest_input=Mock(return_value=deepest_input or []),
)
command_manager = Mock(execute_default_trigger=Mock())
screen_manager = Mock(handle_screen_update=Mock())
cursor_manager = Mock(
is_cursor_vertical_move=Mock(return_value=True),
is_cursor_horizontal_move=Mock(return_value=False),
)
manager.environment = {
"runtime": {
"InputDriver": Mock(is_active=Mock(return_value=input_active)),
"InputManager": input_manager,
"ScreenManager": screen_manager,
"CursorManager": cursor_manager,
"CommandManager": command_manager,
}
}
return manager, screen_manager, input_manager, command_manager
@pytest.mark.unit
def test_screen_update_while_x11_target_is_unfocused_updates_state_silently():
manager, screen_manager, input_manager, command_manager = _build_manager(
input_active=False,
deepest_input=["KEY_DOWN"],
)
event = {"data": {"text": "background repaint"}}
manager.handle_screen_update(event)
screen_manager.handle_screen_update.assert_called_once_with(event["data"])
command_manager.execute_default_trigger.assert_not_called()
input_manager.clear_last_deep_input.assert_called_once_with()
@pytest.mark.unit
def test_programmatic_cursor_move_does_not_run_cursor_change_triggers():
manager, _screen_manager, _input_manager, command_manager = _build_manager(
deepest_input=[]
)
manager.handle_screen_update({"data": {"text": "spinner repaint"}})
command_manager.execute_default_trigger.assert_called_once_with(
"onScreenUpdate"
)
@pytest.mark.unit
def test_navigation_cursor_move_runs_cursor_change_triggers():
manager, _screen_manager, _input_manager, command_manager = _build_manager(
deepest_input=["KEY_DOWN"]
)
manager.handle_screen_update({"data": {"text": "cursor moved"}})
assert command_manager.execute_default_trigger.call_args_list == [
call("onCursorChange"),
call("onScreenUpdate"),
]
@pytest.mark.unit
@pytest.mark.parametrize("key_name", ["KEY_RIGHT", "KEY_BACKSPACE"])
def test_recent_cursor_input_survives_update_before_cursor_moves(key_name):
manager, _screen_manager, input_manager, command_manager = _build_manager(
deepest_input=[key_name]
)
manager.environment["runtime"][
"CursorManager"
].is_cursor_vertical_move.return_value = False
manager.environment["runtime"][
"CursorManager"
].is_cursor_horizontal_move.return_value = False
manager.handle_screen_update({"data": {"text": "intermediate repaint"}})
command_manager.execute_default_trigger.assert_called_once_with(
"onScreenUpdate"
)
input_manager.clear_last_deep_input.assert_not_called()
@pytest.mark.unit
def test_fenrir_review_command_does_not_enable_screen_cursor_speech():
manager, _screen_manager, _input_manager, command_manager = _build_manager(
deepest_input=["KEY_FENRIR", "KEY_KP8"]
)
manager.handle_screen_update({"data": {"text": "background repaint"}})
command_manager.execute_default_trigger.assert_called_once_with(
"onScreenUpdate"
)
@pytest.mark.unit
def test_heartbeat_triggers_are_silent_while_x11_target_is_unfocused():
manager, _screen_manager, _input_manager, command_manager = _build_manager(
input_active=False
)
manager.handle_heart_beat({"data": None})
command_manager.execute_default_trigger.assert_not_called()
+31
View File
@@ -142,6 +142,37 @@ def test_progress_detector_beeps_for_interruptible_status_without_ellipsis():
command.play_activity_beep.assert_called_once_with() command.play_activity_beep.assert_called_once_with()
@pytest.mark.unit
def test_progress_detector_beeps_for_plain_working_elapsed_status():
progress_module = _load_progress_module()
command = progress_module.command()
sample = "Working (21s • esc to interrupt)"
command.env = {
"commandBuffer": {
"progress_monitoring": True,
"lastProgressValue": -1,
"lastProgressTime": 0,
},
"runtime": {
"DebugManager": Mock(write_debug_out=Mock()),
"ScreenManager": Mock(is_screen_change=Mock(return_value=False)),
"CursorManager": Mock(is_cursor_vertical_move=Mock(return_value=False)),
},
"screen": {
"new_delta": sample,
"new_delta_is_typing": False,
"new_content_text": sample,
"old_cursor": {"x": 0, "y": 0},
"new_cursor": {"x": 0, "y": 0},
},
}
command.play_activity_beep = Mock()
command.run()
command.play_activity_beep.assert_called_once_with()
@pytest.mark.unit @pytest.mark.unit
def test_progress_detector_allows_long_key_value_status_delta(): def test_progress_detector_allows_long_key_value_status_delta():
progress_module = _load_progress_module() progress_module = _load_progress_module()
+19 -4
View File
@@ -160,7 +160,22 @@ def test_pty_raw_tab_records_recent_tab_keypress():
@pytest.mark.unit @pytest.mark.unit
def test_pty_plain_stdin_does_not_record_tab_keypress(): @pytest.mark.parametrize(
("sequence", "key_name"),
[
(b"a", "KEY_A"),
(b"\x7f", "KEY_BACKSPACE"),
(b"\x1b[D", "KEY_LEFT"),
(b"\x1bOD", "KEY_LEFT"),
(b"\x1b[C", "KEY_RIGHT"),
(b"\x1bOC", "KEY_RIGHT"),
(b"\x1b[A", "KEY_UP"),
(b"\x1bOA", "KEY_UP"),
(b"\x1b[B", "KEY_DOWN"),
(b"\x1bOB", "KEY_DOWN"),
],
)
def test_pty_forwarded_stdin_records_recent_terminal_input(sequence, key_name):
pty_driver = PtyDriver() pty_driver = PtyDriver()
settings_manager = Mock() settings_manager = Mock()
settings_manager.get_setting_as_bool.return_value = False settings_manager.get_setting_as_bool.return_value = False
@@ -175,10 +190,10 @@ def test_pty_plain_stdin_does_not_record_tab_keypress():
} }
pty_driver.inject_text_to_screen = Mock() pty_driver.inject_text_to_screen = Mock()
pty_driver.handle_stdin_input(b"a", Mock()) pty_driver.handle_stdin_input(sequence, Mock())
input_manager.record_unmanaged_keypress.assert_not_called() input_manager.record_unmanaged_keypress.assert_called_once_with(key_name)
pty_driver.inject_text_to_screen.assert_called_once_with(b"a") pty_driver.inject_text_to_screen.assert_called_once_with(sequence)
@pytest.mark.unit @pytest.mark.unit
+15
View File
@@ -531,12 +531,27 @@ def test_x11_normal_focus_out_resets_input_state():
x11 = X11Driver() x11 = X11Driver()
x11.active = True x11.active = True
x11.reset_input_state = Mock() x11.reset_input_state = Mock()
output_manager = Mock()
x11.env = {"runtime": {"OutputManager": output_manager}}
event = Mock(type=X.FocusOut, mode=X.NotifyNormal) event = Mock(type=X.FocusOut, mode=X.NotifyNormal)
x11.handle_x_event(event, Mock()) x11.handle_x_event(event, Mock())
assert x11.active is False assert x11.active is False
x11.reset_input_state.assert_called_once_with() x11.reset_input_state.assert_called_once_with()
output_manager.interrupt_output_async.assert_called_once_with()
@pytest.mark.unit
def test_x11_reports_target_focus_state():
x11 = X11Driver()
x11.active = False
assert x11.is_active() is False
x11.active = True
assert x11.is_active() is True
@pytest.mark.unit @pytest.mark.unit