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 <dev@disqualifier.me>
This commit is contained in:
disqualifier 2026-06-25 18:42:34 -04:00
parent d43004e4b9
commit 84c74603e5

View File

@ -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, the production handler (TmuxOutputHandler) injects via ``tmux send-keys`` the ONLY
never OS-level keyboard input. it works regardless of which window is focused and mechanism by which claudedo affects claude code. PTY injection, never OS-level
never touches Windows input or a game/anticheat's view (it is text into a linux keyboard input: it works regardless of which window is focused and never touches
pseudo-terminal). do not replace this with OS keystroke injection. 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 from __future__ import annotations
import logging import logging
import subprocess import subprocess
from abc import ABC, abstractmethod
from . import keys, target from . import keys, target
@ -17,10 +26,62 @@ log = logging.getLogger(__name__)
class InjectError(Exception): 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: 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] cmd = ["tmux", "send-keys", "-t", session]
if literal: if literal:
cmd.append("-l") cmd.append("-l")
@ -30,55 +91,68 @@ def _send_keys(session: str, args: list[str], literal: bool) -> None:
err = result.stderr.decode("utf-8", "replace").strip() err = result.stderr.decode("utf-8", "replace").strip()
raise InjectError(f"tmux send-keys failed: {err}") raise InjectError(f"tmux send-keys failed: {err}")
def send_named(self, session: str, key_tokens: list[str]) -> None:
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): if not target.session_exists(session):
log.warning("refusing to inject — session %r does not exist", session) log.warning("refusing to inject — session %r does not exist", session)
return return
for token in key_tokens: for token in key_tokens:
_send_keys(session, [token], literal=False) self._send_keys(session, [token], literal=False)
log.info("injected keys %s -> %s", key_tokens, session) log.info("injected keys %s -> %s", key_tokens, session)
def send_literal(self, session: str, text: str) -> None:
def send_literal(session: str, text: str) -> None:
"""insert literal text into the input box without submitting (``type``)."""
if not text: if not text:
return return
if not target.session_exists(session): if not target.session_exists(session):
log.warning("refusing to inject — session %r does not exist", session) log.warning("refusing to inject — session %r does not exist", session)
return return
_send_keys(session, [text], literal=True) self._send_keys(session, [text], literal=True)
log.info("injected literal text (%d chars) -> %s", len(text), session) log.info("injected literal text (%d chars) -> %s", len(text), session)
def perform(session: str, action) -> bool: class StdoutOutputHandler(OutputHandler):
"""resolve a grammar.Action to keystrokes and inject them. returns acted?. """test handler — prints what would be injected instead of touching tmux.
``switch`` and ``mode`` are handled by the daemon (they change daemon state, not no session existence check (there is no real session); lets grammar + keymap be
the claude session), so they are ignored here. exercised end-to-end without a live claude session. records the last emission on
``self.last`` for assertions.
""" """
name = action.name
if name == "yes": def __init__(self, stream=None) -> None:
send_named(session, keys.YES) import sys
elif name == "no":
send_named(session, keys.NO) self.stream = stream if stream is not None else sys.stdout
elif name == "approve": self.last: tuple[str, object] | None = None
send_named(session, keys.APPROVE)
elif name == "deny": def send_named(self, session: str, key_tokens: list[str]) -> None:
send_named(session, keys.DENY) self.last = ("named", list(key_tokens))
elif name == "submit": print(f"[stdout] keys {key_tokens} -> {session}", file=self.stream)
send_named(session, keys.SUBMIT)
elif name == "cancel": def send_literal(self, session: str, text: str) -> None:
send_named(session, keys.CANCEL) if not text:
elif name == "select": return
seq = keys.SELECT_BY_INDEX.get(int(action.arg)) self.last = ("literal", text)
if seq is None: print(f"[stdout] literal {text!r} -> {session}", file=self.stream)
log.warning("no keymap for select index %r", action.arg)
return False
send_named(session, seq) _default_handler: OutputHandler = TmuxOutputHandler()
elif name == "type":
send_literal(session, str(action.arg))
else: def set_default_handler(handler: OutputHandler) -> None:
return False """swap the module-level handler the daemon drives (tmux in prod, stdout in tests)"""
return True global _default_handler
_default_handler = handler
def send_named(session: str, key_tokens: list[str]) -> None:
"""module-level shim delegating to the default handler"""
_default_handler.send_named(session, key_tokens)
def send_literal(session: str, text: str) -> None:
"""module-level shim delegating to the default handler"""
_default_handler.send_literal(session, text)
def perform(session: str, action) -> bool:
"""module-level shim delegating to the default handler"""
return _default_handler.perform(session, action)