Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 31236b95a1 | |||
| a00e98d8d6 | |||
| 32f85256d3 | |||
| b350e57395 | |||
| 25f0d91c18 | |||
| db3c714027 | |||
| 3b0738c928 | |||
| ee14664958 |
@@ -2,41 +2,12 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
|
||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
|
||||||
from typing import Callable, cast
|
from typing import Callable, cast
|
||||||
|
|
||||||
from ...agent_provider import PromptMode, prompt_args
|
from ...agent_provider import PromptMode, prompt_args
|
||||||
from .. import Bottle, ExecResult
|
from .. import Bottle, ExecResult
|
||||||
from ..terminal import exec_shell_script
|
from ..terminal import exec_shell_script
|
||||||
from . import pty_forward as _pty_forward
|
|
||||||
|
|
||||||
|
|
||||||
_PTY_FORWARD_SCRIPT = _pty_forward.__file__
|
|
||||||
_TERMINAL_ENV_NAMES = (
|
|
||||||
"TERM",
|
|
||||||
"COLORTERM",
|
|
||||||
"TERM_PROGRAM",
|
|
||||||
"TERM_PROGRAM_VERSION",
|
|
||||||
"KITTY_WINDOW_ID",
|
|
||||||
"KITTY_PID",
|
|
||||||
"WEZTERM_PANE",
|
|
||||||
"WEZTERM_UNIX_SOCKET",
|
|
||||||
"GHOSTTY_BIN_DIR",
|
|
||||||
"GHOSTTY_RESOURCES_DIR",
|
|
||||||
"ITERM_SESSION_ID",
|
|
||||||
"VTE_VERSION",
|
|
||||||
"KONSOLE_VERSION",
|
|
||||||
"ALACRITTY_WINDOW_ID",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _terminal_env_names() -> tuple[str, ...]:
|
|
||||||
return tuple(
|
|
||||||
name for name in _TERMINAL_ENV_NAMES
|
|
||||||
if name == "TERM" or os.environ.get(name)
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class MacosContainerBottle(Bottle):
|
class MacosContainerBottle(Bottle):
|
||||||
@@ -73,24 +44,13 @@ class MacosContainerBottle(Bottle):
|
|||||||
argv=full_argv,
|
argv=full_argv,
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
container_exec = ["container", "exec"]
|
cmd = ["container", "exec"]
|
||||||
if tty:
|
if tty:
|
||||||
container_exec.extend(["--interactive", "--tty"])
|
cmd.extend(["--interactive", "--tty"])
|
||||||
# Forward terminal capability hints so TUIs can enable modified-key
|
|
||||||
# protocols. Use bare env names: values stay in the child env, not
|
|
||||||
# on argv, and pty_forward supplies a TERM fallback when needed.
|
|
||||||
for name in _terminal_env_names():
|
|
||||||
container_exec.extend(["--env", name])
|
|
||||||
if self.agent_workdir and self.agent_workdir != "/home/node":
|
if self.agent_workdir and self.agent_workdir != "/home/node":
|
||||||
container_exec.extend(["--workdir", self.agent_workdir])
|
cmd.extend(["--workdir", self.agent_workdir])
|
||||||
container_exec.extend([self.name, self.agent_command, *full_argv])
|
cmd.extend([self.name, self.agent_command, *full_argv])
|
||||||
if tty:
|
return cmd
|
||||||
# Wrap with the raw-mode forwarder: container exec does not put
|
|
||||||
# the host terminal into raw mode itself, so the line discipline
|
|
||||||
# buffers modifier-key sequences until CR. The wrapper sets raw
|
|
||||||
# mode before exec and restores it on exit.
|
|
||||||
return [sys.executable, _PTY_FORWARD_SCRIPT, "--", *container_exec]
|
|
||||||
return container_exec
|
|
||||||
|
|
||||||
def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
|
def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
|
||||||
agent_argv = self.agent_argv(argv, tty=tty)
|
agent_argv = self.agent_argv(argv, tty=tty)
|
||||||
|
|||||||
@@ -1,70 +0,0 @@
|
|||||||
"""Host-side raw-mode wrapper for `container exec --interactive --tty`.
|
|
||||||
|
|
||||||
Apple's `container exec --interactive --tty` does not set the host terminal to
|
|
||||||
raw mode before starting its I/O relay. Without raw mode the kernel line
|
|
||||||
discipline buffers modifier-key escape sequences (e.g. Shift+Enter in
|
|
||||||
modifyOtherKeys mode produces \\x1b[13;2~) until a carriage-return arrives, so
|
|
||||||
they never reach Claude Code inside the container.
|
|
||||||
|
|
||||||
This module sets the host terminal to raw mode, spawns the inner argv (the
|
|
||||||
container exec command), and restores the original terminal attributes on
|
|
||||||
exit. When stdin is not a TTY (piped invocations, CI) it falls through to a
|
|
||||||
bare subprocess.run so callers do not need to special-case non-interactive
|
|
||||||
contexts.
|
|
||||||
|
|
||||||
Usage (the `--` separator is the API contract — everything after it is the
|
|
||||||
inner command):
|
|
||||||
|
|
||||||
python pty_forward.py -- container exec --interactive --tty <name> <cmd>
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import os
|
|
||||||
import subprocess
|
|
||||||
import sys
|
|
||||||
import termios
|
|
||||||
import tty
|
|
||||||
|
|
||||||
|
|
||||||
def _inner_env() -> dict[str, str]:
|
|
||||||
env = dict(os.environ)
|
|
||||||
env.setdefault("TERM", "xterm-256color")
|
|
||||||
return env
|
|
||||||
|
|
||||||
|
|
||||||
def _run_inner(inner: list[str]) -> int:
|
|
||||||
return subprocess.run(inner, check=False, env=_inner_env()).returncode
|
|
||||||
|
|
||||||
|
|
||||||
def main(argv: list[str]) -> int:
|
|
||||||
"""Entry point. ``argv`` shape: ``-- <inner-argv...>``."""
|
|
||||||
if len(argv) < 2 or argv[0] != "--":
|
|
||||||
sys.stderr.write(
|
|
||||||
"usage: python pty_forward.py -- <container-exec-argv...>\n"
|
|
||||||
)
|
|
||||||
return 2
|
|
||||||
inner = argv[1:]
|
|
||||||
|
|
||||||
try:
|
|
||||||
fd = sys.stdin.fileno()
|
|
||||||
except OSError:
|
|
||||||
return _run_inner(inner)
|
|
||||||
|
|
||||||
if not os.isatty(fd):
|
|
||||||
return _run_inner(inner)
|
|
||||||
|
|
||||||
try:
|
|
||||||
old = termios.tcgetattr(fd)
|
|
||||||
except termios.error:
|
|
||||||
return _run_inner(inner)
|
|
||||||
|
|
||||||
try:
|
|
||||||
tty.setraw(fd)
|
|
||||||
return _run_inner(inner)
|
|
||||||
finally:
|
|
||||||
termios.tcsetattr(fd, termios.TCSADRAIN, old)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
sys.exit(main(sys.argv[1:]))
|
|
||||||
@@ -2,32 +2,26 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import sys
|
|
||||||
import unittest
|
import unittest
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
from bot_bottle.backend.macos_container import bottle as bottle_mod
|
from bot_bottle.backend.macos_container.bottle import MacosContainerBottle
|
||||||
from bot_bottle.backend.macos_container.bottle import MacosContainerBottle, _PTY_FORWARD_SCRIPT
|
|
||||||
|
|
||||||
|
|
||||||
class TestMacosContainerBottle(unittest.TestCase):
|
class TestMacosContainerBottle(unittest.TestCase):
|
||||||
def test_agent_argv_uses_pty_forward_and_container_exec(self):
|
def test_agent_argv_uses_container_exec(self):
|
||||||
bottle = MacosContainerBottle(
|
bottle = MacosContainerBottle(
|
||||||
"bot-bottle-dev-abc",
|
"bot-bottle-dev-abc",
|
||||||
lambda: None,
|
lambda: None,
|
||||||
None,
|
None,
|
||||||
agent_command="codex",
|
agent_command="codex",
|
||||||
)
|
)
|
||||||
with patch.dict(bottle_mod.os.environ, {}, clear=True):
|
|
||||||
argv = bottle.agent_argv(["run"])
|
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
[
|
[
|
||||||
sys.executable, _PTY_FORWARD_SCRIPT, "--",
|
|
||||||
"container", "exec", "--interactive", "--tty",
|
"container", "exec", "--interactive", "--tty",
|
||||||
"--env", "TERM",
|
|
||||||
"bot-bottle-dev-abc", "codex", "run",
|
"bot-bottle-dev-abc", "codex", "run",
|
||||||
],
|
],
|
||||||
argv,
|
bottle.agent_argv(["run"]),
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_agent_argv_includes_workdir(self):
|
def test_agent_argv_includes_workdir(self):
|
||||||
@@ -37,54 +31,15 @@ class TestMacosContainerBottle(unittest.TestCase):
|
|||||||
None,
|
None,
|
||||||
agent_workdir="/home/node/workspace",
|
agent_workdir="/home/node/workspace",
|
||||||
)
|
)
|
||||||
with patch.dict(bottle_mod.os.environ, {}, clear=True):
|
|
||||||
argv = bottle.agent_argv([])
|
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
[
|
[
|
||||||
sys.executable, _PTY_FORWARD_SCRIPT, "--",
|
|
||||||
"container", "exec", "--interactive", "--tty",
|
"container", "exec", "--interactive", "--tty",
|
||||||
"--env", "TERM",
|
|
||||||
"--workdir", "/home/node/workspace",
|
"--workdir", "/home/node/workspace",
|
||||||
"bot-bottle-dev-abc", "claude",
|
"bot-bottle-dev-abc", "claude",
|
||||||
],
|
],
|
||||||
argv,
|
bottle.agent_argv([]),
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_agent_argv_forwards_terminal_env_names_without_values(self):
|
|
||||||
bottle = MacosContainerBottle("bot-bottle-dev-abc", lambda: None, None)
|
|
||||||
with patch.dict(
|
|
||||||
bottle_mod.os.environ,
|
|
||||||
{
|
|
||||||
"TERM": "screen-256color",
|
|
||||||
"TERM_PROGRAM": "WezTerm",
|
|
||||||
"WEZTERM_PANE": "pane-id",
|
|
||||||
"SHELL": "/bin/zsh",
|
|
||||||
},
|
|
||||||
clear=True,
|
|
||||||
):
|
|
||||||
argv = bottle.agent_argv([])
|
|
||||||
self.assertIn("TERM", argv)
|
|
||||||
self.assertIn("TERM_PROGRAM", argv)
|
|
||||||
self.assertIn("WEZTERM_PANE", argv)
|
|
||||||
self.assertNotIn("SHELL", argv)
|
|
||||||
self.assertNotIn("TERM=screen-256color", argv)
|
|
||||||
self.assertNotIn("TERM_PROGRAM=WezTerm", argv)
|
|
||||||
self.assertNotIn("WEZTERM_PANE=pane-id", argv)
|
|
||||||
|
|
||||||
def test_agent_argv_always_forwards_term_name(self):
|
|
||||||
bottle = MacosContainerBottle("bot-bottle-dev-abc", lambda: None, None)
|
|
||||||
with patch.dict(bottle_mod.os.environ, {}, clear=True):
|
|
||||||
argv = bottle.agent_argv([])
|
|
||||||
self.assertIn("TERM", argv)
|
|
||||||
|
|
||||||
def test_agent_argv_no_tty_omits_wrapper_and_tty_flags(self):
|
|
||||||
bottle = MacosContainerBottle("bot-bottle-dev-abc", lambda: None, None)
|
|
||||||
argv = bottle.agent_argv([], tty=False)
|
|
||||||
self.assertNotIn("--tty", argv)
|
|
||||||
self.assertNotIn("--env", argv)
|
|
||||||
self.assertNotIn(_PTY_FORWARD_SCRIPT, argv)
|
|
||||||
self.assertEqual(["container", "exec", "bot-bottle-dev-abc", "claude"], argv)
|
|
||||||
|
|
||||||
def test_exec_pipes_script_to_shell(self):
|
def test_exec_pipes_script_to_shell(self):
|
||||||
bottle = MacosContainerBottle("bot-bottle-dev-abc", lambda: None, None)
|
bottle = MacosContainerBottle("bot-bottle-dev-abc", lambda: None, None)
|
||||||
with patch("bot_bottle.backend.macos_container.bottle.subprocess.run") as run:
|
with patch("bot_bottle.backend.macos_container.bottle.subprocess.run") as run:
|
||||||
|
|||||||
@@ -1,159 +0,0 @@
|
|||||||
"""Unit: macos-container pty_forward raw-mode wrapper (issue #245).
|
|
||||||
|
|
||||||
Tests argument parsing, non-TTY fallback, and the raw-mode
|
|
||||||
setup/restore sequence without requiring a real terminal.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import io
|
|
||||||
import termios
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import ANY, MagicMock, patch
|
|
||||||
|
|
||||||
from bot_bottle.backend.macos_container import pty_forward
|
|
||||||
|
|
||||||
|
|
||||||
def _fake_stdin(fd: int = 0) -> MagicMock:
|
|
||||||
"""Return a mock stdin whose fileno() returns *fd*."""
|
|
||||||
m = MagicMock()
|
|
||||||
m.fileno.return_value = fd
|
|
||||||
return m
|
|
||||||
|
|
||||||
|
|
||||||
class TestArgvParsing(unittest.TestCase):
|
|
||||||
def test_missing_separator_returns_error_exit_code(self):
|
|
||||||
with patch.object(pty_forward.sys, "stderr", new=io.StringIO()) as err:
|
|
||||||
rc = pty_forward.main(["container", "exec"])
|
|
||||||
self.assertEqual(2, rc)
|
|
||||||
self.assertIn("usage:", err.getvalue())
|
|
||||||
|
|
||||||
def test_too_few_args_returns_error_exit_code(self):
|
|
||||||
with patch.object(pty_forward.sys, "stderr", new=io.StringIO()):
|
|
||||||
self.assertEqual(2, pty_forward.main([]))
|
|
||||||
self.assertEqual(2, pty_forward.main(["--"]))
|
|
||||||
|
|
||||||
def test_separator_at_start_with_inner_is_valid(self):
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", _fake_stdin()),
|
|
||||||
patch.object(pty_forward.os, "isatty", return_value=False),
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 0
|
|
||||||
rc = pty_forward.main(["--", "container", "exec"])
|
|
||||||
self.assertEqual(0, rc)
|
|
||||||
run.assert_called_once()
|
|
||||||
self.assertEqual(["container", "exec"], run.call_args.args[0])
|
|
||||||
self.assertFalse(run.call_args.kwargs["check"])
|
|
||||||
|
|
||||||
|
|
||||||
class TestNonTtyFallback(unittest.TestCase):
|
|
||||||
def test_non_tty_stdin_runs_inner_directly(self):
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", _fake_stdin()),
|
|
||||||
patch.object(pty_forward.os, "isatty", return_value=False),
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 42
|
|
||||||
rc = pty_forward.main(
|
|
||||||
["--", "container", "exec", "--interactive", "--tty", "c", "claude"]
|
|
||||||
)
|
|
||||||
self.assertEqual(42, rc)
|
|
||||||
run.assert_called_once()
|
|
||||||
self.assertEqual(
|
|
||||||
["container", "exec", "--interactive", "--tty", "c", "claude"],
|
|
||||||
run.call_args.args[0],
|
|
||||||
)
|
|
||||||
self.assertFalse(run.call_args.kwargs["check"])
|
|
||||||
|
|
||||||
def test_fileno_error_runs_inner_directly(self):
|
|
||||||
bad_stdin = MagicMock()
|
|
||||||
bad_stdin.fileno.side_effect = OSError("pseudofile")
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", bad_stdin),
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 0
|
|
||||||
rc = pty_forward.main(["--", "container", "exec"])
|
|
||||||
run.assert_called_once()
|
|
||||||
self.assertEqual(["container", "exec"], run.call_args.args[0])
|
|
||||||
self.assertFalse(run.call_args.kwargs["check"])
|
|
||||||
self.assertEqual(0, rc)
|
|
||||||
|
|
||||||
|
|
||||||
class TestRawModeSetupAndRestore(unittest.TestCase):
|
|
||||||
def test_tty_stdin_sets_raw_mode_and_restores_on_exit(self):
|
|
||||||
saved_attrs = object()
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", _fake_stdin()),
|
|
||||||
patch.object(pty_forward.os, "isatty", return_value=True),
|
|
||||||
patch.object(pty_forward.termios, "tcgetattr", return_value=saved_attrs),
|
|
||||||
patch.object(pty_forward.tty, "setraw") as setraw,
|
|
||||||
patch.object(pty_forward.termios, "tcsetattr") as tcsetattr,
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 0
|
|
||||||
rc = pty_forward.main(["--", "container", "exec"])
|
|
||||||
|
|
||||||
self.assertEqual(0, rc)
|
|
||||||
setraw.assert_called_once()
|
|
||||||
tcsetattr.assert_called_once_with(
|
|
||||||
ANY, termios.TCSADRAIN, saved_attrs,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_tty_restores_on_subprocess_nonzero_exit(self):
|
|
||||||
saved_attrs = object()
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", _fake_stdin()),
|
|
||||||
patch.object(pty_forward.os, "isatty", return_value=True),
|
|
||||||
patch.object(pty_forward.termios, "tcgetattr", return_value=saved_attrs),
|
|
||||||
patch.object(pty_forward.tty, "setraw"),
|
|
||||||
patch.object(pty_forward.termios, "tcsetattr") as tcsetattr,
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 1
|
|
||||||
rc = pty_forward.main(["--", "container", "exec"])
|
|
||||||
|
|
||||||
self.assertEqual(1, rc)
|
|
||||||
tcsetattr.assert_called_once_with(
|
|
||||||
ANY, termios.TCSADRAIN, saved_attrs,
|
|
||||||
)
|
|
||||||
|
|
||||||
def test_tcgetattr_error_falls_back_to_bare_run(self):
|
|
||||||
with (
|
|
||||||
patch.object(pty_forward.sys, "stdin", _fake_stdin()),
|
|
||||||
patch.object(pty_forward.os, "isatty", return_value=True),
|
|
||||||
patch.object(
|
|
||||||
pty_forward.termios, "tcgetattr",
|
|
||||||
side_effect=termios.error("not a tty"),
|
|
||||||
),
|
|
||||||
patch.object(pty_forward.tty, "setraw") as setraw,
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 0
|
|
||||||
rc = pty_forward.main(["--", "container", "exec"])
|
|
||||||
|
|
||||||
setraw.assert_not_called()
|
|
||||||
run.assert_called_once()
|
|
||||||
self.assertEqual(["container", "exec"], run.call_args.args[0])
|
|
||||||
self.assertFalse(run.call_args.kwargs["check"])
|
|
||||||
self.assertEqual(0, rc)
|
|
||||||
|
|
||||||
def test_inner_run_sets_term_default_without_mutating_process_env(self):
|
|
||||||
with (
|
|
||||||
patch.dict(pty_forward.os.environ, {}, clear=True),
|
|
||||||
patch.object(pty_forward.subprocess, "run") as run,
|
|
||||||
):
|
|
||||||
run.return_value.returncode = 0
|
|
||||||
rc = pty_forward._run_inner(["container", "exec"])
|
|
||||||
|
|
||||||
self.assertNotIn("TERM", pty_forward.os.environ)
|
|
||||||
|
|
||||||
self.assertEqual(0, rc)
|
|
||||||
child_env = run.call_args.kwargs["env"]
|
|
||||||
self.assertEqual(["TERM"], sorted(child_env.keys()))
|
|
||||||
self.assertEqual("xterm-256color", child_env["TERM"])
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
||||||
Reference in New Issue
Block a user