Files
superpowers/evals/drill/session.py
Jesse Vincent 3c046f579e Lift drill into evals/ at 013fcb8b7dbefd6d3fa4653493e5d2ec8e7f985b
rsync of obra/drill@013fcb8b7d into superpowers/evals/, excluding
.git/, .venv/, results/, .env/, __pycache__/, *.egg-info/,
.private-journal/.

The drill repo is unaffected by this commit; archival is a separate
manual step after this PR merges.

Source SHA recorded at evals/.drill-source-sha for divergence
detection.
2026-05-06 12:15:46 -07:00

89 lines
2.3 KiB
Python

"""tmux session management for driving agent CLI sessions."""
from __future__ import annotations
import subprocess
import time
class TmuxSession:
def __init__(self, name: str, cols: int = 200, rows: int = 50) -> None:
self.name = name
self.cols = cols
self.rows = rows
def create(self) -> None:
subprocess.run(
[
"tmux",
"new-session",
"-d",
"-s",
self.name,
"-x",
str(self.cols),
"-y",
str(self.rows),
],
check=True,
)
def launch(self, command: list[str], cwd: str) -> None:
cmd_str = " ".join(command)
self.send_keys(f"cd {cwd} && {cmd_str}")
def send_keys(self, text: str) -> None:
if text:
buffer_name = f"{self.name}-input"
subprocess.run(
["tmux", "set-buffer", "-b", buffer_name, text],
check=True,
)
subprocess.run(
["tmux", "paste-buffer", "-d", "-b", buffer_name, "-t", self.name],
check=True,
)
time.sleep(0.1)
subprocess.run(
["tmux", "send-keys", "-t", self.name, "Enter"],
check=True,
)
def send_special_key(self, key: str) -> None:
key_map = {
"ctrl-c": "C-c",
"ctrl-d": "C-d",
"ctrl-z": "C-z",
"enter": "Enter",
"escape": "Escape",
}
tmux_key = key_map.get(key, key)
subprocess.run(
["tmux", "send-keys", "-t", self.name, tmux_key],
check=True,
)
def capture(self) -> str:
result = subprocess.run(
["tmux", "capture-pane", "-t", self.name, "-p"],
capture_output=True,
text=True,
check=True,
)
return result.stdout
def is_process_alive(self) -> bool:
result = subprocess.run(
["tmux", "list-panes", "-t", self.name, "-F", "#{pane_dead}"],
capture_output=True,
text=True,
)
return result.stdout.strip() == "0"
def kill(self) -> None:
subprocess.run(
["tmux", "kill-session", "-t", self.name],
capture_output=True,
)