session.py 2.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788
  1. """tmux session management for driving agent CLI sessions."""
  2. from __future__ import annotations
  3. import subprocess
  4. import time
  5. class TmuxSession:
  6. def __init__(self, name: str, cols: int = 200, rows: int = 50) -> None:
  7. self.name = name
  8. self.cols = cols
  9. self.rows = rows
  10. def create(self) -> None:
  11. subprocess.run(
  12. [
  13. "tmux",
  14. "new-session",
  15. "-d",
  16. "-s",
  17. self.name,
  18. "-x",
  19. str(self.cols),
  20. "-y",
  21. str(self.rows),
  22. ],
  23. check=True,
  24. )
  25. def launch(self, command: list[str], cwd: str) -> None:
  26. cmd_str = " ".join(command)
  27. self.send_keys(f"cd {cwd} && {cmd_str}")
  28. def send_keys(self, text: str) -> None:
  29. if text:
  30. buffer_name = f"{self.name}-input"
  31. subprocess.run(
  32. ["tmux", "set-buffer", "-b", buffer_name, text],
  33. check=True,
  34. )
  35. subprocess.run(
  36. ["tmux", "paste-buffer", "-d", "-b", buffer_name, "-t", self.name],
  37. check=True,
  38. )
  39. time.sleep(0.1)
  40. subprocess.run(
  41. ["tmux", "send-keys", "-t", self.name, "Enter"],
  42. check=True,
  43. )
  44. def send_special_key(self, key: str) -> None:
  45. key_map = {
  46. "ctrl-c": "C-c",
  47. "ctrl-d": "C-d",
  48. "ctrl-z": "C-z",
  49. "enter": "Enter",
  50. "escape": "Escape",
  51. }
  52. tmux_key = key_map.get(key, key)
  53. subprocess.run(
  54. ["tmux", "send-keys", "-t", self.name, tmux_key],
  55. check=True,
  56. )
  57. def capture(self) -> str:
  58. result = subprocess.run(
  59. ["tmux", "capture-pane", "-t", self.name, "-p"],
  60. capture_output=True,
  61. text=True,
  62. check=True,
  63. )
  64. return result.stdout
  65. def is_process_alive(self) -> bool:
  66. result = subprocess.run(
  67. ["tmux", "list-panes", "-t", self.name, "-F", "#{pane_dead}"],
  68. capture_output=True,
  69. text=True,
  70. )
  71. return result.stdout.strip() == "0"
  72. def kill(self) -> None:
  73. subprocess.run(
  74. ["tmux", "kill-session", "-t", self.name],
  75. capture_output=True,
  76. )