From 84c74603e529f248a22de920c03165e3c3f9f7f2 Mon Sep 17 00:00:00 2001 From: disqualifier Date: Thu, 25 Jun 2026 18:42:34 -0400 Subject: [PATCH] feat: output-handler seam with tmux and stdout handlers extract an OutputHandler abstract base; TmuxOutputHandler is production (send-keys, PTY-only), StdoutOutputHandler prints what would be injected so grammar+keymap run end-to-end without a live claude session (the deterministic test path). module-level shims default to tmux so the daemon is unchanged. Signed-off-by: disqualifier --- src/claudedo/inject.py | 192 ++++++++++++++++++++++++++++------------- 1 file changed, 133 insertions(+), 59 deletions(-) diff --git a/src/claudedo/inject.py b/src/claudedo/inject.py index ccdabbb..64b1a46 100644 --- a/src/claudedo/inject.py +++ b/src/claudedo/inject.py @@ -1,15 +1,24 @@ -"""inject keystrokes into a tmux session via ``tmux send-keys``. +"""output handlers: resolve a grammar.Action to keystrokes and emit them. -this is the ONLY mechanism by which claudedo affects claude code — PTY injection, -never OS-level keyboard input. it works regardless of which window is focused and -never touches Windows input or a game/anticheat's view (it is text into a linux -pseudo-terminal). do not replace this with OS keystroke injection. +the production handler (TmuxOutputHandler) injects via ``tmux send-keys`` — the ONLY +mechanism by which claudedo affects claude code. PTY injection, never OS-level +keyboard input: it works regardless of which window is focused and never touches +Windows input or a game/anticheat's view (it is text into a linux pseudo-terminal). +do not replace this with OS keystroke injection. this is also why claudedo is a +standalone daemon and not an MCP server — MCP tools can only return content to claude, +not inject into its input stream. + +StdoutOutputHandler prints what WOULD be injected instead of touching tmux, so the +grammar + keymap can be exercised end-to-end without a live claude session — the +deterministic test path. both implement the same OutputHandler seam and are +interchangeable. """ from __future__ import annotations import logging import subprocess +from abc import ABC, abstractmethod from . import keys, target @@ -17,68 +26,133 @@ log = logging.getLogger(__name__) class InjectError(Exception): - """raised when a tmux send-keys call fails.""" + """raised when a tmux send-keys call fails""" -def _send_keys(session: str, args: list[str], literal: bool) -> None: - cmd = ["tmux", "send-keys", "-t", session] - if literal: - cmd.append("-l") - cmd.extend(args) - result = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) - if result.returncode != 0: - err = result.stderr.decode("utf-8", "replace").strip() - raise InjectError(f"tmux send-keys failed: {err}") +class OutputHandler(ABC): + """abstract sink for resolved keystrokes. + + concretes implement send_named (a sequence of named tmux keys) and send_literal + (literal text, no submit). perform() maps a grammar.Action onto these and is shared + by all handlers. + """ + + @abstractmethod + def send_named(self, session: str, key_tokens: list[str]) -> None: + """emit a sequence of named keys (e.g. ['1'] or ['Down', 'Enter'])""" + + @abstractmethod + def send_literal(self, session: str, text: str) -> None: + """emit literal text into the input box without submitting (``type``)""" + + def perform(self, session: str, action) -> bool: + """resolve a grammar.Action to keystrokes and emit them. returns acted?. + + ``switch`` and ``mode`` are handled by the daemon (they change daemon state, + not the claude session), so they are ignored here. + """ + name = action.name + if name == "yes": + self.send_named(session, keys.YES) + elif name == "no": + self.send_named(session, keys.NO) + elif name == "approve": + self.send_named(session, keys.APPROVE) + elif name == "deny": + self.send_named(session, keys.DENY) + elif name == "submit": + self.send_named(session, keys.SUBMIT) + elif name == "cancel": + self.send_named(session, keys.CANCEL) + elif name == "select": + seq = keys.SELECT_BY_INDEX.get(int(action.arg)) + if seq is None: + log.warning("no keymap for select index %r", action.arg) + return False + self.send_named(session, seq) + elif name == "type": + self.send_literal(session, str(action.arg)) + else: + return False + return True + + +class TmuxOutputHandler(OutputHandler): + """production handler — injects keystrokes into a tmux session via send-keys""" + + @staticmethod + def _send_keys(session: str, args: list[str], literal: bool) -> None: + cmd = ["tmux", "send-keys", "-t", session] + if literal: + cmd.append("-l") + cmd.extend(args) + result = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + if result.returncode != 0: + err = result.stderr.decode("utf-8", "replace").strip() + raise InjectError(f"tmux send-keys failed: {err}") + + def send_named(self, session: str, key_tokens: list[str]) -> None: + if not target.session_exists(session): + log.warning("refusing to inject — session %r does not exist", session) + return + for token in key_tokens: + self._send_keys(session, [token], literal=False) + log.info("injected keys %s -> %s", key_tokens, session) + + def send_literal(self, session: str, text: str) -> None: + if not text: + return + if not target.session_exists(session): + log.warning("refusing to inject — session %r does not exist", session) + return + self._send_keys(session, [text], literal=True) + log.info("injected literal text (%d chars) -> %s", len(text), session) + + +class StdoutOutputHandler(OutputHandler): + """test handler — prints what would be injected instead of touching tmux. + + no session existence check (there is no real session); lets grammar + keymap be + exercised end-to-end without a live claude session. records the last emission on + ``self.last`` for assertions. + """ + + def __init__(self, stream=None) -> None: + import sys + + self.stream = stream if stream is not None else sys.stdout + self.last: tuple[str, object] | None = None + + def send_named(self, session: str, key_tokens: list[str]) -> None: + self.last = ("named", list(key_tokens)) + print(f"[stdout] keys {key_tokens} -> {session}", file=self.stream) + + def send_literal(self, session: str, text: str) -> None: + if not text: + return + self.last = ("literal", text) + print(f"[stdout] literal {text!r} -> {session}", file=self.stream) + + +_default_handler: OutputHandler = TmuxOutputHandler() + + +def set_default_handler(handler: OutputHandler) -> None: + """swap the module-level handler the daemon drives (tmux in prod, stdout in tests)""" + global _default_handler + _default_handler = handler def send_named(session: str, key_tokens: list[str]) -> None: - """send a sequence of named tmux keys (e.g. ['1'] or ['Down', 'Enter']).""" - if not target.session_exists(session): - log.warning("refusing to inject — session %r does not exist", session) - return - for token in key_tokens: - _send_keys(session, [token], literal=False) - log.info("injected keys %s -> %s", key_tokens, session) + """module-level shim delegating to the default handler""" + _default_handler.send_named(session, key_tokens) def send_literal(session: str, text: str) -> None: - """insert literal text into the input box without submitting (``type``).""" - if not text: - return - if not target.session_exists(session): - log.warning("refusing to inject — session %r does not exist", session) - return - _send_keys(session, [text], literal=True) - log.info("injected literal text (%d chars) -> %s", len(text), session) + """module-level shim delegating to the default handler""" + _default_handler.send_literal(session, text) def perform(session: str, action) -> bool: - """resolve a grammar.Action to keystrokes and inject them. returns acted?. - - ``switch`` and ``mode`` are handled by the daemon (they change daemon state, not - the claude session), so they are ignored here. - """ - name = action.name - if name == "yes": - send_named(session, keys.YES) - elif name == "no": - send_named(session, keys.NO) - elif name == "approve": - send_named(session, keys.APPROVE) - elif name == "deny": - send_named(session, keys.DENY) - elif name == "submit": - send_named(session, keys.SUBMIT) - elif name == "cancel": - send_named(session, keys.CANCEL) - elif name == "select": - seq = keys.SELECT_BY_INDEX.get(int(action.arg)) - if seq is None: - log.warning("no keymap for select index %r", action.arg) - return False - send_named(session, seq) - elif name == "type": - send_literal(session, str(action.arg)) - else: - return False - return True + """module-level shim delegating to the default handler""" + return _default_handler.perform(session, action)