make agent interactive

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,18 +58,22 @@ 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.
 
     Args:
         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)

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

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