diff --git a/lib/main.py b/lib/main.py index f96bb7bb..4a4b2198 100644 --- a/lib/main.py +++ b/lib/main.py @@ -75,7 +75,7 @@ def _looks_like_wlroots_session() -> bool: from audio_capture import AudioCapture from whisper_manager import WhisperManager from session_environment import ensure_wayland_display -from text_injector import TextInjector +from text_injector import TextInjector, InjectionOutcome from global_shortcuts import GlobalShortcuts from audio_manager import AudioManager from audio_ducker import AudioDucker @@ -977,8 +977,14 @@ def process(): if self._continuous_cancelled: print("[CONTINUOUS] Cancelled — discarding transcription", flush=True) return - self._inject_text(text) - print(f"[CONTINUOUS] Pasted: {text[:80]}{'...' if len(text) > 80 else ''}", flush=True) + outcome = self._inject_text(text) + preview = f"{text[:80]}{'...' if len(text) > 80 else ''}" + if outcome == InjectionOutcome.INJECTED: + print(f"[CONTINUOUS] Pasted: {preview}", flush=True) + elif outcome == InjectionOutcome.CONSUMED: + print(f"[CONTINUOUS] Consumed by hook: {preview}", flush=True) + else: + print(f"[CONTINUOUS] Injection failed: {preview}", flush=True) else: print("[CONTINUOUS] No transcription from flushed audio", flush=True) except Exception as e: @@ -1425,8 +1431,8 @@ def _process_audio(self, audio_data): self.current_transcription = text # Inject text - self._inject_text(self.current_transcription) - success = True + outcome = self._inject_text(self.current_transcription) + success = outcome != InjectionOutcome.FAILED else: print("[WARN] No transcription generated") self.audio_manager.play_error_sound() @@ -1446,16 +1452,20 @@ def _inject_text(self, text): # Capture mode: route text to client instead of injecting into active app if self._recording_control_server.has_capture_subscriber(): self._recording_control_server.notify_capture(text, final=True) - return True + return InjectionOutcome.INJECTED try: - if not self.text_injector.inject_text(text): + outcome = self.text_injector.inject_text(text) + if outcome == InjectionOutcome.FAILED: print(f"[ERROR] Text injection failed ({len(text)} chars)", flush=True) - return False + return InjectionOutcome.FAILED - print(f"[INJECT] Text injected ({len(text)} chars)", flush=True) + if outcome == InjectionOutcome.CONSUMED: + print("[INJECT] Post-transcription hook consumed transcription", flush=True) + else: + print(f"[INJECT] Text injected ({len(text)} chars)", flush=True) - # Text injection succeeded - system is fully healthy + # Text injection succeeded (or was intentionally consumed) - system is fully healthy # Cancel any pending background recovery if self._background_recovery_needed.is_set(): print("[HEALTH] Successful recording detected - canceling background recovery", flush=True) @@ -1470,10 +1480,10 @@ def _inject_text(self, text): self.audio_capture.abort_recovery() except Exception: pass - return True + return outcome except Exception as e: print(f"[ERROR] Text injection failed: {e}", flush=True) - return False + return InjectionOutcome.FAILED def _is_zero_volume(self, audio_data) -> bool: """Check if audio data has zero or near-zero volume""" diff --git a/lib/src/longform_controller.py b/lib/src/longform_controller.py index c6feee3b..034d4d75 100644 --- a/lib/src/longform_controller.py +++ b/lib/src/longform_controller.py @@ -5,9 +5,11 @@ try: from .paths import LONGFORM_STATE_FILE from .segment_manager import SegmentManager + from .text_injector import InjectionOutcome except ImportError: from paths import LONGFORM_STATE_FILE from segment_manager import SegmentManager + from text_injector import InjectionOutcome class LongFormController: @@ -248,7 +250,7 @@ def submit(self, retry=False, audio_data=None): self._submission_failed(audio_data) return - if not self.inject_text(text): + if self.inject_text(text) == InjectionOutcome.FAILED: self._submission_failed(audio_data) return diff --git a/lib/src/text_injector.py b/lib/src/text_injector.py index a0b2bcca..d1940f9a 100644 --- a/lib/src/text_injector.py +++ b/lib/src/text_injector.py @@ -71,6 +71,12 @@ class _PostTranscriptionHookResult: text: str +class InjectionOutcome(Enum): + INJECTED = "injected" + CONSUMED = "consumed" + FAILED = "failed" + + DEFAULT_PASTE_KEYCODE = 47 # Linux evdev KEY_V on QWERTY NON_XKB_INPUT_METHOD_LAYOUT = '__non_xkb_input_method__' @@ -1112,7 +1118,7 @@ def _send_enter_if_auto_submit(self): # ------------------------ Public API ------------------------ - def inject_text(self, text: str) -> bool: + def inject_text(self, text: str) -> InjectionOutcome: """ Inject text into the currently focused application @@ -1120,18 +1126,18 @@ def inject_text(self, text: str) -> bool: text: Text to inject Returns: - True if successful, False otherwise + InjectionOutcome.INJECTED if pasted, CONSUMED if a post-transcription + hook consumed the transcription, FAILED otherwise """ if not text or text.strip() == "": print("No text to inject (empty or whitespace)") - return True + return InjectionOutcome.INJECTED # Preprocess; also trim trailing newlines (avoid unwanted Enter) processed_text = self._preprocess_text(text).rstrip("\r\n") hook_result = self._run_post_transcription_hook(processed_text) if hook_result.outcome == _PostTranscriptionHookOutcome.CONSUME: - print("Post-transcription hook consumed transcription") - return True + return InjectionOutcome.CONSUMED processed_text = hook_result.text + ' ' try: @@ -1143,11 +1149,12 @@ def inject_text(self, text: str) -> bool: print(f"⚠️ inject_mode='{inject_mode}' is deprecated: direct typing drops characters at speed. " f"Using clipboard+paste instead.") - return self._inject_via_clipboard_and_hotkey(processed_text) + injected = self._inject_via_clipboard_and_hotkey(processed_text) + return InjectionOutcome.INJECTED if injected else InjectionOutcome.FAILED except Exception as e: print(f"Primary injection method failed: {e}") - return False + return InjectionOutcome.FAILED # ------------------------ Helpers ------------------------ diff --git a/tests/test_longform_reliability.py b/tests/test_longform_reliability.py index a7246bac..2d71d91b 100644 --- a/tests/test_longform_reliability.py +++ b/tests/test_longform_reliability.py @@ -12,6 +12,7 @@ sys.path.insert(0, str(ROOT / "lib" / "src")) from longform_controller import LongFormController +from text_injector import InjectionOutcome class ImmediateTimer: @@ -68,7 +69,7 @@ def _controller(self, timer_factory=ImmediateTimer): whisper_manager=SimpleNamespace( transcribe_audio=mock.Mock(return_value="hello") ), - inject_text=mock.Mock(return_value=True), + inject_text=mock.Mock(return_value=InjectionOutcome.INJECTED), notify_capture=mock.Mock(), set_visualizer_state=mock.Mock(), show_mic_osd=mock.Mock(), @@ -138,7 +139,7 @@ def test_failed_final_write_can_submit_combined_audio(self): def test_injection_failure_retains_audio_until_successful_retry(self): controller, persisted = self._controller() - controller.inject_text.side_effect = [False, True] + controller.inject_text.side_effect = [InjectionOutcome.FAILED, InjectionOutcome.INJECTED] controller.submit() diff --git a/tests/test_main_startup_safety.py b/tests/test_main_startup_safety.py index 889d13c5..7e1f3ff5 100644 --- a/tests/test_main_startup_safety.py +++ b/tests/test_main_startup_safety.py @@ -178,11 +178,10 @@ def test_inject_text_checks_injector_result_before_success_log(self): success_log_line = None for node in ast.walk(inject_func): if ( - isinstance(node, ast.UnaryOp) - and isinstance(node.op, ast.Not) - and isinstance(node.operand, ast.Call) - and isinstance(node.operand.func, ast.Attribute) - and node.operand.func.attr == "inject_text" + isinstance(node, ast.Assign) + and isinstance(node.value, ast.Call) + and isinstance(node.value.func, ast.Attribute) + and node.value.func.attr == "inject_text" ): result_checked_line = node.lineno elif ( @@ -202,6 +201,58 @@ def test_inject_text_checks_injector_result_before_success_log(self): self.assertIsNotNone(success_log_line) self.assertLess(result_checked_line, success_log_line) + def test_process_audio_success_reflects_injection_outcome(self): + tree = ast.parse((ROOT / "lib" / "main.py").read_text(encoding="utf-8")) + + process_func = self._find_function(tree, "_process_audio") + self.assertIsNotNone(process_func) + + hardcoded_success_after_injection = False + success_compares_outcome = False + for node in ast.walk(process_func): + if not (isinstance(node, ast.Assign) and any( + isinstance(t, ast.Name) and t.id == "success" for t in node.targets + )): + continue + if isinstance(node.value, ast.Constant) and node.value.value is True: + hardcoded_success_after_injection = True + if ( + isinstance(node.value, ast.Compare) + and isinstance(node.value.left, ast.Name) + and node.value.left.id == "outcome" + ): + success_compares_outcome = True + + self.assertFalse( + hardcoded_success_after_injection, + "success must be derived from _inject_text's outcome, not hardcoded True", + ) + self.assertTrue( + success_compares_outcome, + "success should compare the injection outcome (e.g. against InjectionOutcome.FAILED)", + ) + + def test_continuous_flush_distinguishes_consumed_from_injected(self): + tree = ast.parse((ROOT / "lib" / "main.py").read_text(encoding="utf-8")) + + flush_func = self._find_function(tree, "_continuous_flush_audio") + self.assertIsNotNone(flush_func) + + references_consumed = any( + isinstance(node, ast.Attribute) and node.attr == "CONSUMED" + for node in ast.walk(flush_func) + ) + references_injected = any( + isinstance(node, ast.Attribute) and node.attr == "INJECTED" + for node in ast.walk(flush_func) + ) + + self.assertTrue( + references_consumed and references_injected, + "continuous-mode flush should branch on InjectionOutcome instead of " + "unconditionally logging every transcription as pasted", + ) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_text_injector_injection.py b/tests/test_text_injector_injection.py index 6ef92e43..8bc18bb5 100644 --- a/tests/test_text_injector_injection.py +++ b/tests/test_text_injector_injection.py @@ -14,6 +14,7 @@ from text_injector import ( POST_TRANSCRIPTION_HOOK_CONSUMED_EXIT_CODE, + InjectionOutcome, TextInjector, _LazyPyperclip, _PostTranscriptionHookOutcome, @@ -1051,10 +1052,30 @@ def test_consume_result_skips_injection(self): mock.patch.object(injector, "_run_post_transcription_hook", return_value=consumed), mock.patch.object(injector, "_inject_via_clipboard_and_hotkey") as inject, ): - self.assertTrue(injector.inject_text("open terminal")) + self.assertEqual(injector.inject_text("open terminal"), InjectionOutcome.CONSUMED) inject.assert_not_called() + def test_inject_text_returns_injected_on_successful_paste(self): + injector = self._injector() + injector.config_manager = ConfigStub({}) + + with ( + mock.patch.object(injector, "_preprocess_text", return_value="hello"), + mock.patch.object(injector, "_inject_via_clipboard_and_hotkey", return_value=True), + ): + self.assertEqual(injector.inject_text("hello"), InjectionOutcome.INJECTED) + + def test_inject_text_returns_failed_on_failed_paste(self): + injector = self._injector() + injector.config_manager = ConfigStub({}) + + with ( + mock.patch.object(injector, "_preprocess_text", return_value="hello"), + mock.patch.object(injector, "_inject_via_clipboard_and_hotkey", return_value=False), + ): + self.assertEqual(injector.inject_text("hello"), InjectionOutcome.FAILED) + class LazyPyperclipTests(unittest.TestCase): def test_selects_xclip_then_xsel_on_first_use(self):