From 0178e5c76ed25e85e7987c32f96c01beb6891afb Mon Sep 17 00:00:00 2001 From: Duncan McKinnon Date: Sun, 28 Jun 2026 10:58:55 -0700 Subject: [PATCH] make agent interactive --- src/thirdeye/agent/exec.py | 72 ++++++++++++++++++++++----- src/thirdeye/commands/agent.py | 23 ++++++++- tests/test_agent_command.py | 63 +++++++++++++++++++++++ tests/test_agent_exec.py | 91 +++++++++++++++++++++++++++++++--- 4 files changed, 229 insertions(+), 20 deletions(-) diff --git a/src/thirdeye/agent/exec.py b/src/thirdeye/agent/exec.py index 73ae624..452cc41 100644 --- a/src/thirdeye/agent/exec.py +++ b/src/thirdeye/agent/exec.py @@ -1,5 +1,6 @@ from __future__ import annotations +import os import shutil import subprocess import threading @@ -12,6 +13,16 @@ from thirdeye.agent.harness import AgentHarness +def _open_pty() -> tuple[int, int]: + import pty # deferred: not available on Windows + + return pty.openpty() + + +def _read_fd(fd: int, n: int) -> bytes: + return os.read(fd, n) + + def _list_sessions(thirdeye_home: Path, platform: str) -> set[str]: """Return the set of session IDs currently recorded for platform. Best-effort.""" try: @@ -47,6 +58,7 @@ def run_agent_streaming( *, output: Callable[[str], None] | None = None, thirdeye_home: Path | None = None, + use_pty: bool = False, ) -> tuple[int, int]: """Run the agent subprocess, streaming stdout to the terminal. @@ -54,11 +66,14 @@ def run_agent_streaming( harness: AgentHarness wrapping the desired adapter + mode. prompt: The composed prompt string to pass to the agent. cwd: Working directory for the subprocess. - output: Callable that receives each stdout line. Defaults to + output: Callable that receives each stdout chunk/line. Defaults to click.echo(line, nl=False) so callers can inject a mock for testing. thirdeye_home: If provided, any new sessions the agent spawns on its platform are tagged 'thirdeye-agent' after the run. + use_pty: If True, allocate a pseudo-terminal for stdout so the + subprocess sees a TTY and flushes output in real time. + If False (default), use a pipe. Returns: (returncode, duration_ms) @@ -74,6 +89,8 @@ def run_agent_streaming( def output(line: str) -> None: click.echo(line, nl=False) + _use_pty = use_pty + # Snapshot existing sessions before the run so we can detect new ones. pre_ids: set[str] = set() if thirdeye_home is not None: @@ -84,14 +101,26 @@ def output(line: str) -> None: stderr_lines: list[str] = [] - proc = subprocess.Popen( - cmd, - cwd=cwd, - stdin=subprocess.DEVNULL, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) + if _use_pty: + master_fd, slave_fd = _open_pty() + proc = subprocess.Popen( + cmd, + cwd=cwd, + stdin=subprocess.DEVNULL, + stdout=slave_fd, + stderr=subprocess.PIPE, + text=True, + ) + os.close(slave_fd) + else: + proc = subprocess.Popen( + cmd, + cwd=cwd, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) def _drain_stderr() -> None: assert proc.stderr is not None @@ -101,9 +130,28 @@ def _drain_stderr() -> None: stderr_thread = threading.Thread(target=_drain_stderr, daemon=True) stderr_thread.start() - assert proc.stdout is not None - for line in proc.stdout: - output(line) + if _use_pty: + buf = b"" + try: + while True: + try: + chunk = _read_fd(master_fd, 4096) + except OSError: + break + if not chunk: + break + buf += chunk + while b"\n" in buf: + line_bytes, buf = buf.split(b"\n", 1) + output(line_bytes.decode("utf-8", errors="replace") + "\n") + finally: + os.close(master_fd) + if buf: + output(buf.decode("utf-8", errors="replace")) + else: + assert proc.stdout is not None + for line in proc.stdout: + output(line) proc.wait() stderr_thread.join(timeout=5) diff --git a/src/thirdeye/commands/agent.py b/src/thirdeye/commands/agent.py index 41ed763..ac35bef 100644 --- a/src/thirdeye/commands/agent.py +++ b/src/thirdeye/commands/agent.py @@ -46,12 +46,20 @@ type=click.Path(file_okay=False, path_type=Path), help="Working directory context injected into the prompt (default: current dir).", ) +@click.option( + "--stream", + "stream", + is_flag=True, + default=False, + help="Allocate a pseudo-terminal so the agent streams output in real time (Unix only).", +) def agent_cmd( task: str, agent_name: str, fix_mode: bool, skills: tuple[str, ...], cwd: Path | None, + stream: bool, ) -> None: config = Config.load() @@ -81,9 +89,22 @@ def agent_cmd( mode = "fix" if fix_mode else "review" harness = AgentHarness(adapter, mode) + use_pty = stream + if stream and sys.platform == "win32": + click.echo( + "Warning: --stream uses a pseudo-terminal which is not supported on Windows; " + "falling back to standard pipe output.", + err=False, + ) + use_pty = False + try: returncode, _ = run_agent_streaming( - harness, prompt, cwd=cwd or Path.cwd(), thirdeye_home=config.root + harness, + prompt, + cwd=cwd or Path.cwd(), + thirdeye_home=config.root, + use_pty=use_pty, ) except FileNotFoundError as e: raise click.ClickException(str(e)) from e diff --git a/tests/test_agent_command.py b/tests/test_agent_command.py index 6606449..64abb37 100644 --- a/tests/test_agent_command.py +++ b/tests/test_agent_command.py @@ -9,6 +9,7 @@ # Patch targets for the two heavy dependencies. _PATCH_STREAM = "thirdeye.commands.agent.run_agent_streaming" _PATCH_PROMPT = "thirdeye.commands.agent.build_agent_prompt" +_PATCH_PLATFORM = "thirdeye.commands.agent.sys.platform" def _invoke(*args, **kwargs): @@ -202,3 +203,65 @@ def _fake_stream(harness, prompt, cwd, **kwargs): assert captured_prompt_cwd[0] == custom_dir assert len(captured_stream_cwd) == 1 assert captured_stream_cwd[0] == custom_dir + + +# --- --stream flag --- + + +def test_help_mentions_stream_flag(): + result = CliRunner().invoke(main, ["agent", "--help"]) + assert "--stream" in result.output + + +def test_stream_flag_passes_use_pty_true(): + """--stream forwards use_pty=True to run_agent_streaming.""" + captured_kwargs: list[dict] = [] + + def _fake_stream(harness, prompt, cwd, **kwargs): + captured_kwargs.append(kwargs) + return (0, 500) + + with ( + patch(_PATCH_STREAM, side_effect=_fake_stream), + patch(_PATCH_PROMPT, return_value="p"), + ): + CliRunner().invoke(main, ["agent", "x", "--stream"], catch_exceptions=False) + + assert captured_kwargs[0].get("use_pty") is True + + +def test_no_stream_flag_passes_use_pty_false(): + """Without --stream, use_pty=False is forwarded to run_agent_streaming.""" + captured_kwargs: list[dict] = [] + + def _fake_stream(harness, prompt, cwd, **kwargs): + captured_kwargs.append(kwargs) + return (0, 500) + + with ( + patch(_PATCH_STREAM, side_effect=_fake_stream), + patch(_PATCH_PROMPT, return_value="p"), + ): + CliRunner().invoke(main, ["agent", "x"], catch_exceptions=False) + + assert captured_kwargs[0].get("use_pty") is False + + +def test_stream_on_windows_prints_warning_and_falls_back_to_pipe(): + """On Windows, --stream emits a warning and falls back to use_pty=False.""" + captured_kwargs: list[dict] = [] + + def _fake_stream(harness, prompt, cwd, **kwargs): + captured_kwargs.append(kwargs) + return (0, 500) + + with ( + patch(_PATCH_STREAM, side_effect=_fake_stream), + patch(_PATCH_PROMPT, return_value="p"), + patch(_PATCH_PLATFORM, "win32"), + ): + result = CliRunner().invoke(main, ["agent", "x", "--stream"], catch_exceptions=False) + + assert result.exit_code == 0 + assert "warning" in result.output.lower() + assert captured_kwargs[0].get("use_pty") is False diff --git a/tests/test_agent_exec.py b/tests/test_agent_exec.py index f977a10..5f27fee 100644 --- a/tests/test_agent_exec.py +++ b/tests/test_agent_exec.py @@ -1,6 +1,7 @@ from __future__ import annotations import io +import subprocess from pathlib import Path from unittest.mock import MagicMock, patch @@ -39,7 +40,9 @@ def test_streams_stdout_to_output_callback(): patch("shutil.which", return_value="/usr/bin/claude"), patch("subprocess.Popen", return_value=mock_proc), ): - run_agent_streaming(_make_harness(), "task", Path("/tmp"), output=captured.append) + run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=False, output=captured.append + ) assert captured == ["line one\n", "line two\n"] @@ -52,7 +55,7 @@ def test_returns_returncode_zero_on_success(): patch("subprocess.Popen", return_value=mock_proc), ): rc, duration = run_agent_streaming( - _make_harness(), "task", Path("/tmp"), output=lambda _: None + _make_harness(), "task", Path("/tmp"), use_pty=False, output=lambda _: None ) assert rc == 0 @@ -65,7 +68,9 @@ def test_returns_nonzero_returncode_on_failure(): patch("shutil.which", return_value="/usr/bin/claude"), patch("subprocess.Popen", return_value=mock_proc), ): - rc, _ = run_agent_streaming(_make_harness(), "task", Path("/tmp"), output=lambda _: None) + rc, _ = run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=False, output=lambda _: None + ) assert rc == 1 @@ -78,7 +83,7 @@ def test_returns_duration_ms_as_int(): patch("subprocess.Popen", return_value=mock_proc), ): _, duration = run_agent_streaming( - _make_harness(), "task", Path("/tmp"), output=lambda _: None + _make_harness(), "task", Path("/tmp"), use_pty=False, output=lambda _: None ) assert isinstance(duration, int) @@ -93,7 +98,9 @@ def test_empty_stdout_produces_no_callback_calls(): patch("shutil.which", return_value="/usr/bin/claude"), patch("subprocess.Popen", return_value=mock_proc), ): - run_agent_streaming(_make_harness(), "task", Path("/tmp"), output=captured.append) + run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=False, output=captured.append + ) assert captured == [] @@ -105,7 +112,7 @@ def test_default_output_uses_click_echo(capsys): patch("shutil.which", return_value="/usr/bin/claude"), patch("subprocess.Popen", return_value=mock_proc), ): - run_agent_streaming(_make_harness(), "task", Path("/tmp")) + run_agent_streaming(_make_harness(), "task", Path("/tmp"), use_pty=False) out = capsys.readouterr().out assert "hello" in out @@ -126,6 +133,7 @@ def _fake_popen(cmd, **kwargs): _make_harness(), "my composed prompt", Path("/tmp"), + use_pty=False, output=lambda _: None, ) @@ -136,6 +144,69 @@ def _fake_popen(cmd, **kwargs): # --- tagging: new sessions are tagged 'thirdeye-agent' --- +# --- use_pty: configurable PTY vs pipe stdout --- + + +def test_use_pty_false_uses_subprocess_pipe(): + """use_pty=False forces pipe-based stdout regardless of platform.""" + captured_kwargs: list[dict] = [] + + def _fake_popen(cmd, **kwargs): + captured_kwargs.append(kwargs) + return _make_mock_proc(stdout="ok\n") + + with ( + patch("shutil.which", return_value="/usr/bin/claude"), + patch("subprocess.Popen", side_effect=_fake_popen), + ): + run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=False, output=lambda _: None + ) + + assert captured_kwargs[0]["stdout"] is subprocess.PIPE + + +def test_use_pty_true_calls_open_pty(): + """use_pty=True invokes _open_pty and passes the slave fd as stdout to Popen.""" + captured_kwargs: list[dict] = [] + + def _fake_popen(cmd, **kwargs): + captured_kwargs.append(kwargs) + return _make_mock_proc() + + with ( + patch("shutil.which", return_value="/usr/bin/claude"), + patch("subprocess.Popen", side_effect=_fake_popen), + patch("thirdeye.agent.exec._open_pty", return_value=(10, 11)) as mock_open_pty, + patch("thirdeye.agent.exec._read_fd", side_effect=OSError()), + patch("os.close"), + ): + run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=True, output=lambda _: None + ) + + mock_open_pty.assert_called_once() + assert captured_kwargs[0]["stdout"] == 11 + + +def test_pty_streams_output_to_callback(): + """PTY path decodes chunks and forwards complete lines to the output callback.""" + captured: list[str] = [] + + with ( + patch("shutil.which", return_value="/usr/bin/claude"), + patch("subprocess.Popen", return_value=_make_mock_proc()), + patch("thirdeye.agent.exec._open_pty", return_value=(10, 11)), + patch("thirdeye.agent.exec._read_fd", side_effect=[b"line one\nline two\n", OSError()]), + patch("os.close"), + ): + run_agent_streaming( + _make_harness(), "task", Path("/tmp"), use_pty=True, output=captured.append + ) + + assert captured == ["line one\n", "line two\n"] + + def test_no_thirdeye_home_skips_tagging(): """Without thirdeye_home, no session snapshot or tag write occurs.""" mock_proc = _make_mock_proc(stdout="ok\n") @@ -145,7 +216,9 @@ def test_no_thirdeye_home_skips_tagging(): patch("thirdeye.agent.exec._list_sessions") as mock_list, patch("thirdeye.agent.exec._tag_sessions") as mock_tag, ): - run_agent_streaming(_make_harness(), "x", Path("/tmp"), output=lambda _: None) + run_agent_streaming( + _make_harness(), "x", Path("/tmp"), use_pty=False, output=lambda _: None + ) mock_list.assert_not_called() mock_tag.assert_not_called() @@ -170,6 +243,7 @@ def _fake_list(home, platform): _make_harness(), "x", Path("/tmp"), + use_pty=False, output=lambda _: None, thirdeye_home=tmp_path, ) @@ -202,6 +276,7 @@ def _fake_tag(home, platform, sids): _make_harness(), "x", Path("/tmp"), + use_pty=False, output=lambda _: None, thirdeye_home=tmp_path, ) @@ -230,6 +305,7 @@ def _fake_tag(home, platform, sids): _make_harness(), "x", Path("/tmp"), + use_pty=False, output=lambda _: None, thirdeye_home=tmp_path, ) @@ -280,6 +356,7 @@ def _fake_popen(cmd, **kwargs): harness, "task text", cwd=Path("/proj/foo"), + use_pty=False, output=lambda _: None, thirdeye_home=tmp_path, )