Merge pull request #2 from stefanc-ai2/fix-sending-message

fix(wezterm): improve enter injection reliability

Author: stefanc-ai2

README.md

@@ -125,3 +125,16 @@ cq-mounted --json
 python -m compileall -q lib bin cq test
 python -m pytest test/ -v --tb=short
 ```
+
+---
+
+## WezTerm notes
+
+`ask` sends text to a pane and then injects **Enter** so the target TUI processes it. If you see text get pasted but not submitted, tune these env vars:
+
+- `CQ_WEZTERM_ENTER_METHOD=auto|key|text` (default: `auto`)
+  - `auto`: try `wezterm cli send-key Enter`, fall back to a raw CR byte
+  - `key`: force `send-key` only (no fallback)
+  - `text`: legacy mode (CR byte only)
+- `CQ_WEZTERM_ENTER_DELAY` (seconds): delay before injecting Enter
+- `CQ_WEZTERM_PASTE_DELAY` (seconds): delay between paste-mode send and Enter injection (multiline)

lib/terminal.py

@@ -6,6 +6,7 @@
 import shlex
 import shutil
 import subprocess
+import sys
 import time
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
@@ -641,24 +642,24 @@ def _send_enter(self, pane_id: str) -> None:
             time.sleep(enter_delay)
 
         env_method_raw = os.environ.get("CQ_WEZTERM_ENTER_METHOD")
-        # Default behavior is intentionally unchanged on non-Windows platforms:
-        # previously we used `send-text` with a CR byte; keep that unless the user overrides.
-        default_method = "auto" if os.name == "nt" else "text"
+        # Default to "auto": try `wezterm cli send-key` first (real key event), then fall back to a CR byte.
+        # Users can force the legacy behavior via `CQ_WEZTERM_ENTER_METHOD=text`.
+        default_method = "auto"
         method = (env_method_raw or default_method).strip().lower()
         if method not in {"auto", "key", "text"}:
             method = default_method
 
         # Retry mechanism for reliability (Windows native occasionally drops Enter)
         max_retries = 3
         for attempt in range(max_retries):
-            # Only enable "auto key" behavior by default on native Windows.
-            # Users can force key injection everywhere via CQ_WEZTERM_ENTER_METHOD=key.
-            if method == "key" or (method == "auto" and os.name == "nt"):
+            # Prefer a real key event. Fall back to a CR byte for older WezTerm CLIs or raw-mode TUIs.
+            if method in {"auto", "key"}:
                 if self._send_key_cli(pane_id, "Enter"):
                     return
 
             # Fallback: send CR byte; works for shells/readline, but not for all raw-mode TUIs.
-            if method in {"auto", "text", "key"}:
+            # NOTE: `key` mode is strict: if `send-key` fails, do not fall back.
+            if method in {"auto", "text"}:
                 result = _run(
                     [*self._cli_base_args(), "send-text", "--pane-id", pane_id, "--no-paste"],
                     input=b"\r",
@@ -668,7 +669,21 @@ def _send_enter(self, pane_id: str) -> None:
                     return
 
             if attempt < max_retries - 1:
-                time.sleep(0.05)
+                time.sleep(0.05 * (attempt + 1))
+
+        # If we got here, all attempts failed. The message text may be waiting in the input buffer.
+        debug = (os.environ.get("CQ_DEBUG") or "").strip().lower() in ("1", "true", "yes")
+        if debug or method == "key":
+            hint = (
+                "Press Enter manually in the target pane. "
+                "If this happens often, try `CQ_WEZTERM_ENTER_METHOD=auto` and/or increase "
+                "`CQ_WEZTERM_ENTER_DELAY` / `CQ_WEZTERM_PASTE_DELAY`."
+            )
+            level = "DEBUG" if debug else "WARN"
+            print(
+                f"[{level}] WezTerm Enter injection failed (method={method}, pane_id={pane_id}). {hint}",
+                file=sys.stderr,
+            )
 
     def send_text(self, pane_id: str, text: str) -> None:
         sanitized = text.replace("\r", "").strip()

test/test_wezterm_enter.py

@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import terminal
+
+
+def test_wezterm_send_enter_defaults_to_auto_tries_key_first(monkeypatch) -> None:
+    """Default should try send-key first (real key event), then return on success."""
+    monkeypatch.delenv("CQ_WEZTERM_ENTER_METHOD", raising=False)
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-key" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-key" in cmd for cmd in calls)
+    assert not any("send-text" in cmd for cmd in calls)
+
+
+def test_wezterm_send_enter_text_mode_does_not_use_send_key(monkeypatch) -> None:
+    """Legacy mode should only use send-text CR injection."""
+    monkeypatch.setenv("CQ_WEZTERM_ENTER_METHOD", "text")
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-text" in cmd:
+            assert kwargs.get("input") == b"\r"
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-text" in cmd for cmd in calls)
+    assert not any("send-key" in cmd for cmd in calls)
+
+
+def test_wezterm_send_enter_auto_falls_back_to_cr(monkeypatch) -> None:
+    """Auto should fall back to CR when send-key isn't supported."""
+    monkeypatch.delenv("CQ_WEZTERM_ENTER_METHOD", raising=False)
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-key" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="nope")
+        if "send-text" in cmd:
+            assert kwargs.get("input") == b"\r"
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-key" in cmd for cmd in calls)
+    assert any("send-text" in cmd for cmd in calls)
+
+
+def test_wezterm_send_enter_key_mode_is_strict(monkeypatch) -> None:
+    """Key mode should not fall back to CR injection if send-key fails."""
+    monkeypatch.setenv("CQ_WEZTERM_ENTER_METHOD", "key")
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-key" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="nope")
+        if "send-text" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-key" in cmd for cmd in calls)
+    assert not any("send-text" in cmd for cmd in calls)
+    send_key_calls = [cmd for cmd in calls if "send-key" in cmd]
+    assert len(send_key_calls) >= 3
+
+
+def test_wezterm_send_enter_invalid_method_falls_back_to_auto(monkeypatch) -> None:
+    """Invalid methods should behave like auto (try send-key then fall back)."""
+    monkeypatch.setenv("CQ_WEZTERM_ENTER_METHOD", "bogus")
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-key" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="nope")
+        if "send-text" in cmd:
+            assert kwargs.get("input") == b"\r"
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-key" in cmd for cmd in calls)
+    assert any("send-text" in cmd for cmd in calls)
+
+
+def test_wezterm_send_enter_key_mode_success(monkeypatch) -> None:
+    """Key mode should early-return when send-key works."""
+    monkeypatch.setenv("CQ_WEZTERM_ENTER_METHOD", "key")
+    monkeypatch.setattr(terminal, "_get_wezterm_bin", lambda: "/usr/bin/wezterm")
+    monkeypatch.setattr(terminal.time, "sleep", lambda _: None)
+
+    calls: list[list[str]] = []
+
+    def fake_run(*args, **kwargs):
+        cmd = args[0]
+        calls.append(cmd)
+        if "send-key" in cmd:
+            return terminal.subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
+        return terminal.subprocess.CompletedProcess(args=cmd, returncode=1, stdout="", stderr="unexpected")
+
+    monkeypatch.setattr(terminal, "_run", fake_run)
+
+    backend = terminal.WeztermBackend()
+    backend._send_enter("123")
+
+    assert any("send-key" in cmd for cmd in calls)
+    assert not any("send-text" in cmd for cmd in calls)