agent_base/tools/tool_runtime.py

import argparse
import atexit
import itertools
import os
import pty
import re
import select
import shutil
import signal
import struct
import subprocess
import termios
import threading
import time
from pathlib import Path
from typing import Optional, Union
import sys

from agent_base.utils import PROJECT_ROOT, load_dotenv
from agent_base.tools.tooling import (
    ToolBase,
    command_safety_issue,
    sanitized_subprocess_env,
    validate_tool_path,
    workspace_root,
)


DEFAULT_BUFFER_LIMIT = 200000
DEFAULT_OUTPUT_CHARS = 16384
DEFAULT_YIELD_MS = 200
REPEAT_COLLAPSE_THRESHOLD = 3


def _default_shell() -> str:
    return shutil.which("bash") or "/bin/bash"


def _resolve_cwd(path_value: Optional[str], *, base_root: Optional[Path] = None) -> Path:
    if not path_value:
        return (base_root or workspace_root()).resolve()
    return validate_tool_path(path_value, "Working directory", base_root=base_root)


def _set_terminal_size(fd: int, rows: int, cols: int) -> None:
    winsize = struct.pack("HHHH", rows, cols, 0, 0)
    try:
        import fcntl

        fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
    except (ImportError, OSError):
        return


def _disable_echo(fd: int) -> None:
    try:
        attrs = termios.tcgetattr(fd)
        attrs[3] &= ~termios.ECHO
        termios.tcsetattr(fd, termios.TCSANOW, attrs)
    except termios.error:
        return


def _collapse_repeated_lines(text: str, *, threshold: int = REPEAT_COLLAPSE_THRESHOLD) -> str:
    if not text:
        return text
    lines = text.splitlines(keepends=True)
    if not lines:
        return text
    collapsed: list[str] = []
    current = lines[0]
    count = 1
    for line in lines[1:]:
        if line == current:
            count += 1
            continue
        if count >= threshold:
            collapsed.append(current)
            collapsed.append(f"[previous line repeated {count - 1} additional times]\n")
        else:
            collapsed.extend([current] * count)
        current = line
        count = 1
    if count >= threshold:
        collapsed.append(current)
        collapsed.append(f"[previous line repeated {count - 1} additional times]\n")
    else:
        collapsed.extend([current] * count)
    return "".join(collapsed)


def _bounded_output(text: str, *, max_output_chars: int = DEFAULT_OUTPUT_CHARS) -> str:
    if not text:
        return text
    compressed = _collapse_repeated_lines(text)
    if len(compressed) <= max_output_chars:
        return compressed
    omitted = len(compressed) - max_output_chars
    suffix = f"\n[output truncated: omitted {omitted} chars]\n"
    keep = max(0, max_output_chars - len(suffix))
    return compressed[:keep] + suffix


def _looks_binary_output(data: bytes) -> bool:
    if not data:
        return False
    sample = data[:4096]
    if b"\x00" in sample:
        return True
    allowed_controls = {9, 10, 12, 13}
    control_count = sum(1 for byte in sample if byte < 32 and byte not in allowed_controls)
    return control_count / max(len(sample), 1) > 0.05


def _truncate_output_bytes(data: bytes, *, max_output_chars: int) -> tuple[bytes, bool, int]:
    max_output_bytes = max(1, int(max_output_chars))
    if len(data) <= max_output_bytes:
        return data, False, 0
    omitted = len(data) - max_output_bytes
    suffix = f"\n[output truncated before decoding: omitted {omitted} bytes]\n".encode("utf-8")
    keep = max(0, max_output_bytes - len(suffix))
    return data[:keep] + suffix, True, omitted


def _safe_process_output(data: Union[str, bytes, None], *, max_output_chars: int = DEFAULT_OUTPUT_CHARS) -> str:
    if data is None or data == "":
        return ""
    if isinstance(data, str):
        return _bounded_output(data, max_output_chars=max_output_chars)

    if not data:
        return ""
    if _looks_binary_output(data):
        note = (
            f"[binary output omitted: {len(data)} bytes. "
            "Use file or a format-aware reader such as openpyxl/scipy.io.loadmat for binary files.]"
        )
        if len(data) > max_output_chars:
            note += f"\n[output truncated before decoding: omitted {len(data) - max_output_chars} bytes]"
        return note

    bounded, _, _ = _truncate_output_bytes(data, max_output_chars=max_output_chars)
    try:
        decoded = bounded.decode("utf-8")
        prefix = ""
    except UnicodeDecodeError:
        decoded = bounded.decode("utf-8", errors="replace")
        prefix = "[non-UTF-8 bytes decoded with replacement characters]\n"
    return _bounded_output(prefix + decoded, max_output_chars=max_output_chars)


class Bash(ToolBase):
    name = "Bash"
    description = (
        "Run a local bash command and return stdout and stderr. This is the primary local execution tool for "
        "shell commands, path operations, ripgrep, git, temporary python3 heredoc scripts, parsing, validation, "
        "and local result transformation."
    )
    parameters = {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "The shell command to execute.",
            },
            "timeout": {
                "type": "integer",
                "description": "Timeout in seconds. Default is 30.",
            },
            "workdir": {
                "type": "string",
                "description": "Optional working directory for the command. Defaults to the current workspace root.",
            },
            "max_output_chars": {
                "type": "integer",
                "description": f"Maximum combined stdout/stderr characters returned after repeated-line compression. Default is {DEFAULT_OUTPUT_CHARS}.",
            },
        },
        "required": ["command"],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[Bash] {exc}"
        base_root = kwargs.get("workspace_root")
        runtime_deadline = kwargs.get("runtime_deadline")

        command = str(params["command"])
        workdir = params.get("workdir")
        try:
            timeout = int(params.get("timeout", 30))
            max_output_chars = int(params.get("max_output_chars", DEFAULT_OUTPUT_CHARS))
        except (TypeError, ValueError):
            return "[Bash] timeout and max_output_chars must be integers."

        issue = command_safety_issue(str(command))
        if issue:
            return f"[Bash] Blocked by safety policy: {issue}"

        try:
            cwd = _resolve_cwd(workdir, base_root=base_root)
        except ValueError as exc:
            return f"[Bash] Invalid or blocked working directory: {exc}"
        if not cwd.exists():
            return f"[Bash] Working directory does not exist: {cwd}"
        if not cwd.is_dir():
            return f"[Bash] Working directory is not a directory: {cwd}"
        if timeout <= 0:
            return "[Bash] timeout must be > 0."
        if max_output_chars <= 0:
            return "[Bash] max_output_chars must be > 0."

        effective_timeout: float = float(timeout)
        if runtime_deadline is not None:
            remaining = float(runtime_deadline) - time.time()
            if remaining <= 0:
                return "[Bash] Agent runtime limit reached before command execution."
            effective_timeout = min(effective_timeout, max(remaining, 0.001))

        try:
            proc = subprocess.run(
                command,
                shell=True,
                capture_output=True,
                text=False,
                timeout=effective_timeout,
                cwd=str(cwd),
                env=sanitized_subprocess_env(base_root=base_root),
                executable=shutil.which("bash") or "/bin/bash",
            )
        except subprocess.TimeoutExpired:
            return "[Bash] TimeoutError: Execution timed out."
        except (OSError, subprocess.SubprocessError) as exc:
            return f"[Bash] Error executing command: {exc}"

        parts = [f"exit_code: {proc.returncode}"]
        stdout = _safe_process_output(proc.stdout, max_output_chars=max_output_chars)
        stderr = _safe_process_output(proc.stderr, max_output_chars=max_output_chars)
        if stdout:
            parts.append(f"stdout:\n{stdout}")
        if stderr:
            parts.append(f"stderr:\n{stderr}")
        return "\n".join(parts)

class TerminalSession:
    def __init__(self, cwd: Path, shell: str, rows: int, cols: int, *, base_root: Optional[Path] = None):
        self.cwd = cwd
        self.shell = shell
        self.rows = rows
        self.cols = cols
        self._buffer_limit = DEFAULT_BUFFER_LIMIT
        self._pending_output = ""
        self._dropped_output_chars = 0
        self._lock = threading.Lock()

        master_fd, slave_fd = pty.openpty()
        _set_terminal_size(slave_fd, rows, cols)
        _disable_echo(slave_fd)

        env = sanitized_subprocess_env(base_root=base_root)
        env.setdefault("TERM", "xterm-256color")
        env.setdefault("PS1", "")
        env.setdefault("PROMPT_COMMAND", "")

        self._proc = subprocess.Popen(
            [shell, "--noprofile", "--norc"],
            stdin=slave_fd,
            stdout=slave_fd,
            stderr=slave_fd,
            cwd=str(cwd),
            env=env,
            text=False,
            close_fds=True,
            start_new_session=True,
        )
        os.close(slave_fd)
        self._master_fd = master_fd
        self._reader = threading.Thread(target=self._reader_loop, daemon=True)
        self._reader.start()

    @property
    def pid(self) -> int:
        return self._proc.pid

    @property
    def alive(self) -> bool:
        return self._proc.poll() is None

    @property
    def returncode(self) -> Optional[int]:
        return self._proc.poll()

    def _reader_loop(self) -> None:
        while True:
            try:
                ready, _, _ = select.select([self._master_fd], [], [], 0.1)
            except (OSError, ValueError):
                break

            if not ready:
                if self._proc.poll() is not None:
                    break
                continue

            try:
                data = os.read(self._master_fd, 4096)
            except OSError:
                break

            if not data:
                if self._proc.poll() is not None:
                    break
                continue

            decoded = data.decode("utf-8", errors="replace")
            with self._lock:
                self._pending_output += decoded
                overflow = len(self._pending_output) - self._buffer_limit
                if overflow > 0:
                    self._pending_output = self._pending_output[overflow:]
                    self._dropped_output_chars += overflow

        try:
            os.close(self._master_fd)
        except OSError:
            pass

    def write(self, data: str) -> None:
        if not self.alive:
            raise RuntimeError("session is not running")
        os.write(self._master_fd, data.encode("utf-8", errors="replace"))

    def read(self, yield_time_ms: int = DEFAULT_YIELD_MS, max_output_chars: int = DEFAULT_OUTPUT_CHARS) -> dict:
        if yield_time_ms > 0:
            time.sleep(yield_time_ms / 1000.0)

        with self._lock:
            output = self._pending_output[:max_output_chars]
            self._pending_output = self._pending_output[max_output_chars:]
            remaining_output_chars = len(self._pending_output)
            dropped_output_chars = self._dropped_output_chars
            self._dropped_output_chars = 0

        return {
            "alive": self.alive,
            "returncode": self.returncode,
            "output": output,
            "remaining_output_chars": remaining_output_chars,
            "dropped_output_chars": dropped_output_chars,
            "truncated": remaining_output_chars > 0,
        }

    def interrupt(self, *, max_output_chars: int = DEFAULT_OUTPUT_CHARS) -> dict:
        if not self.alive:
            raise RuntimeError("session is not running")
        os.write(self._master_fd, b"\x03")
        return self.read(yield_time_ms=DEFAULT_YIELD_MS, max_output_chars=max_output_chars)

    def terminate(self, force: bool = False) -> Optional[int]:
        if self.alive:
            try:
                os.killpg(os.getpgid(self.pid), signal.SIGKILL if force else signal.SIGTERM)
            except ProcessLookupError:
                pass
            except OSError:
                self._proc.kill() if force else self._proc.terminate()
            try:
                self._proc.wait(timeout=2 if not force else 1)
            except subprocess.TimeoutExpired:
                if not force:
                    return self.terminate(force=True)
        return self.returncode


class TerminalSessionManager:
    def __init__(self):
        self._lock = threading.Lock()
        self._counter = itertools.count(1)
        self._sessions: dict[str, TerminalSession] = {}

    def start(self, cwd: Path, shell: str, rows: int, cols: int, *, base_root: Optional[Path] = None) -> tuple[str, TerminalSession]:
        session = TerminalSession(cwd=cwd, shell=shell, rows=rows, cols=cols, base_root=base_root)
        session_id = f"term_{next(self._counter)}"
        with self._lock:
            self._sessions[session_id] = session
        return session_id, session

    def get(self, session_id: str) -> Optional[TerminalSession]:
        with self._lock:
            return self._sessions.get(session_id)

    def pop(self, session_id: str) -> Optional[TerminalSession]:
        with self._lock:
            return self._sessions.pop(session_id, None)

    def cleanup(self) -> None:
        with self._lock:
            sessions = list(self._sessions.items())
            self._sessions.clear()
        for _, session in sessions:
            session.terminate(force=True)


SESSION_MANAGER = TerminalSessionManager()
atexit.register(SESSION_MANAGER.cleanup)


def _format_terminal_response(
    prefix: str,
    session_id: str,
    payload: dict,
    cwd: Optional[Path] = None,
    shell: Optional[str] = None,
    pid: Optional[int] = None,
) -> str:
    lines = [prefix, f"session_id: {session_id}"]
    if pid is not None:
        lines.append(f"pid: {pid}")
    if cwd is not None:
        lines.append(f"cwd: {cwd}")
    if shell is not None:
        lines.append(f"shell: {shell}")
    if "alive" in payload:
        lines.append(f"alive: {str(payload['alive']).lower()}")
    if "returncode" in payload:
        lines.append(f"returncode: {payload['returncode']}")
    if "truncated" in payload:
        lines.append(f"truncated: {str(payload['truncated']).lower()}")
    if "remaining_output_chars" in payload:
        lines.append(f"remaining_output_chars: {payload['remaining_output_chars']}")
    if "dropped_output_chars" in payload:
        lines.append(f"dropped_output_chars: {payload['dropped_output_chars']}")
    if "output" in payload:
        lines.append("output:")
        lines.append(payload["output"])
    return "\n".join(lines)


class TerminalStart(ToolBase):
    name = "TerminalStart"
    description = "Start a persistent local terminal session backed by a PTY shell."
    parameters = {
        "type": "object",
        "properties": {
            "cwd": {
                "type": "string",
                "description": "Optional working directory for the terminal session. Default is the current workspace root.",
            },
            "shell": {
                "type": "string",
                "description": "Optional shell executable path. Default is bash.",
            },
            "rows": {
                "type": "integer",
                "description": "Terminal row count. Default is 30.",
            },
            "cols": {
                "type": "integer",
                "description": "Terminal column count. Default is 120.",
            },
        },
        "required": [],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[TerminalStart] {exc}"
        base_root = kwargs.get("workspace_root")
        try:
            cwd = _resolve_cwd(params.get("cwd"), base_root=base_root)
            shell = params.get("shell") or _default_shell()
            rows = int(params.get("rows", 30))
            cols = int(params.get("cols", 120))
        except ValueError as exc:
            return f"[TerminalStart] {exc}"
        except (TypeError, OverflowError):
            return "[TerminalStart] rows and cols must be integers."

        if not cwd.exists():
            return f"[TerminalStart] Working directory does not exist: {cwd}"
        if not cwd.is_dir():
            return f"[TerminalStart] Working directory is not a directory: {cwd}"
        if not Path(shell).exists() and shutil.which(shell) is None:
            return f"[TerminalStart] Shell not found: {shell}"
        if rows <= 0 or cols <= 0:
            return "[TerminalStart] rows and cols must both be > 0."

        try:
            session_id, session = SESSION_MANAGER.start(cwd=cwd, shell=shell, rows=rows, cols=cols, base_root=base_root)
        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
            return f"[TerminalStart] Failed to start terminal session: {exc}"

        return _format_terminal_response(
            "[TerminalStart] Started terminal session.",
            session_id=session_id,
            payload={"alive": session.alive, "returncode": session.returncode},
            cwd=cwd,
            shell=shell,
            pid=session.pid,
        )


class TerminalWrite(ToolBase):
    name = "TerminalWrite"
    description = "Write input into an existing terminal session and read back newly produced output."
    parameters = {
        "type": "object",
        "properties": {
            "session_id": {
                "type": "string",
                "description": "The terminal session ID returned by TerminalStart.",
            },
            "input": {
                "type": "string",
                "description": "The text to send to the terminal session.",
            },
            "append_newline": {
                "type": "boolean",
                "description": "Whether to append a newline after the provided input. Default is true.",
            },
            "yield_time_ms": {
                "type": "integer",
                "description": "Milliseconds to wait before reading output. Default is 200.",
            },
            "max_output_chars": {
                "type": "integer",
                "description": f"Maximum number of output characters to return. Default is {DEFAULT_OUTPUT_CHARS}.",
            },
        },
        "required": ["session_id", "input"],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[TerminalWrite] {exc}"

        session_id = str(params["session_id"])
        input_text = str(params["input"])
        append_newline = bool(params.get("append_newline", True))
        try:
            yield_time_ms = int(params.get("yield_time_ms", DEFAULT_YIELD_MS))
            max_output_chars = int(params.get("max_output_chars", DEFAULT_OUTPUT_CHARS))
        except (TypeError, ValueError):
            return "[TerminalWrite] yield_time_ms and max_output_chars must be integers."

        issue = command_safety_issue(input_text)
        if issue:
            return f"[TerminalWrite] Blocked by safety policy: {issue}"

        session = SESSION_MANAGER.get(session_id)
        if session is None:
            return f"[TerminalWrite] Session not found: {session_id}"
        if max_output_chars <= 0:
            return "[TerminalWrite] max_output_chars must be > 0."
        if yield_time_ms < 0:
            return "[TerminalWrite] yield_time_ms must be >= 0."

        payload_input = input_text + ("\n" if append_newline else "")
        try:
            session.write(payload_input)
            payload = session.read(yield_time_ms=yield_time_ms, max_output_chars=max_output_chars)
        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
            return f"[TerminalWrite] Failed to write to session {session_id}: {exc}"

        return _format_terminal_response("[TerminalWrite] Session updated.", session_id=session_id, payload=payload)


class TerminalRead(ToolBase):
    name = "TerminalRead"
    description = "Read unread output from an existing terminal session."
    parameters = {
        "type": "object",
        "properties": {
            "session_id": {
                "type": "string",
                "description": "The terminal session ID returned by TerminalStart.",
            },
            "yield_time_ms": {
                "type": "integer",
                "description": "Milliseconds to wait before reading output. Default is 200.",
            },
            "max_output_chars": {
                "type": "integer",
                "description": f"Maximum number of output characters to return. Default is {DEFAULT_OUTPUT_CHARS}.",
            },
        },
        "required": ["session_id"],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[TerminalRead] {exc}"

        session_id = str(params["session_id"])
        try:
            yield_time_ms = int(params.get("yield_time_ms", DEFAULT_YIELD_MS))
            max_output_chars = int(params.get("max_output_chars", DEFAULT_OUTPUT_CHARS))
        except (TypeError, ValueError):
            return "[TerminalRead] yield_time_ms and max_output_chars must be integers."

        session = SESSION_MANAGER.get(session_id)
        if session is None:
            return f"[TerminalRead] Session not found: {session_id}"
        if max_output_chars <= 0:
            return "[TerminalRead] max_output_chars must be > 0."
        if yield_time_ms < 0:
            return "[TerminalRead] yield_time_ms must be >= 0."

        try:
            payload = session.read(yield_time_ms=yield_time_ms, max_output_chars=max_output_chars)
        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
            return f"[TerminalRead] Failed to read session {session_id}: {exc}"

        return _format_terminal_response("[TerminalRead] Session output fetched.", session_id=session_id, payload=payload)


class TerminalKill(ToolBase):
    name = "TerminalKill"
    description = "Terminate an existing terminal session and release its resources."
    parameters = {
        "type": "object",
        "properties": {
            "session_id": {
                "type": "string",
                "description": "The terminal session ID returned by TerminalStart.",
            },
            "force": {
                "type": "boolean",
                "description": "Whether to force kill the terminal session immediately. Default is false.",
            },
        },
        "required": ["session_id"],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[TerminalKill] {exc}"

        session_id = str(params["session_id"])
        force = bool(params.get("force", False))

        session = SESSION_MANAGER.pop(session_id)
        if session is None:
            return f"[TerminalKill] Session not found: {session_id}"

        try:
            returncode = session.terminate(force=force)
        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
            return f"[TerminalKill] Failed to terminate session {session_id}: {exc}"

        return _format_terminal_response(
            "[TerminalKill] Terminal session terminated.",
            session_id=session_id,
            payload={"alive": False, "returncode": returncode},
        )


class TerminalInterrupt(ToolBase):
    name = "TerminalInterrupt"
    description = "Send Ctrl-C to the foreground process in an existing terminal session while keeping the session alive."
    parameters = {
        "type": "object",
        "properties": {
            "session_id": {
                "type": "string",
                "description": "The terminal session ID returned by TerminalStart.",
            },
            "max_output_chars": {
                "type": "integer",
                "description": f"Maximum number of output characters to return after the interrupt. Default is {DEFAULT_OUTPUT_CHARS}.",
            },
        },
        "required": ["session_id"],
    }

    def __init__(self, cfg: Optional[dict] = None):
        super().__init__(cfg)

    def call(self, params: Union[str, dict], **kwargs) -> str:
        try:
            params = self.parse_json_args(params)
        except ValueError as exc:
            return f"[TerminalInterrupt] {exc}"

        session_id = str(params["session_id"])
        try:
            max_output_chars = int(params.get("max_output_chars", DEFAULT_OUTPUT_CHARS))
        except (TypeError, ValueError):
            return "[TerminalInterrupt] max_output_chars must be an integer."

        session = SESSION_MANAGER.get(session_id)
        if session is None:
            return f"[TerminalInterrupt] Session not found: {session_id}"
        if max_output_chars <= 0:
            return "[TerminalInterrupt] max_output_chars must be > 0."

        try:
            payload = session.interrupt(max_output_chars=max_output_chars)
        except (OSError, RuntimeError, subprocess.SubprocessError) as exc:
            return f"[TerminalInterrupt] Failed to interrupt session {session_id}: {exc}"

        return _format_terminal_response(
            "[TerminalInterrupt] Sent Ctrl-C to terminal session.",
            session_id=session_id,
            payload=payload,
        )


def main(argv: Optional[list[str]] = None) -> int:
    parser = argparse.ArgumentParser(description="Run runtime and terminal tools directly.")
    subparsers = parser.add_subparsers(dest="tool", required=True)

    bash_parser = subparsers.add_parser("bash", help="Run the Bash tool.")
    bash_parser.add_argument("command")
    bash_parser.add_argument("--timeout", type=int, default=30)
    bash_parser.add_argument("--workdir")

    terminal_parser = subparsers.add_parser("terminal", help="Run a minimal terminal session demo.")
    terminal_parser.add_argument("input", help="Input to send after starting the session.")
    terminal_parser.add_argument("--cwd")
    terminal_parser.add_argument("--yield-time-ms", type=int, default=200)

    args = parser.parse_args(argv)
    load_dotenv(PROJECT_ROOT / ".env")
    workdir_root = Path(args.workdir).expanduser().resolve() if getattr(args, "workdir", None) else None

    if args.tool == "bash":
        result = Bash().call(
            {"command": args.command, "timeout": args.timeout, "workdir": args.workdir},
            workspace_root=workdir_root,
        )
        print(result)
        return 0

    terminal_root = Path(args.cwd).expanduser().resolve() if args.cwd else workspace_root()
    start_result = TerminalStart().call({"cwd": str(terminal_root)}, workspace_root=terminal_root)
    print(start_result)
    session_match = re.search(r"session_id: (term_\d+)", start_result)
    if not session_match:
        return 1
    session_id = session_match.group(1)
    write_result = TerminalWrite().call(
        {
            "session_id": session_id,
            "input": args.input,
            "yield_time_ms": args.yield_time_ms,
        },
        workspace_root=terminal_root,
    )
    print(write_result)
    print(TerminalKill().call({"session_id": session_id}, workspace_root=terminal_root))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))