Shell output streams live as a scrolling message in the chat TUI and REPL; commands that stop on a prompt (sudo, ssh, y/n) run on a pty and ask you, with the wait excluded from the timeout

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 13:49:52 +02:00
parent 72f337f012
commit d030a4a1d3
3 changed files with 427 additions and 7 deletions
@@ -97,7 +97,7 @@ def _fromisoformat(text):
        return parsed.replace(tzinfo=tzinfo)
    raise ValueError("invalid datetime, use ISO like 2026-10-08T09:00")

-VERSION = "1.32.0"
+VERSION = "1.33.0"
 WORKER_STEPS = 12
 CREATE_SKILL_STEPS = 40
 FORK_MAX_DEPTH = 2
@@ -1069,6 +1069,14 @@ class Spinner:


 STREAM_HEIGHT = 4
+RUN_OUTPUT_CAP = 4000000
+RUN_PROMPT_IDLE = 1.0
+RUN_MAX_ASKS = 20
+RUN_ANSWER_CAP = 1000
+RUN_ROWS = 24
+RUN_COLS = 120
+RUN_CTRL_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]|\x1b\][^\x07]*\x07|[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
+RUN_PROMPT_RE = re.compile(r"(?:[:?>\]\)$#%]|\(y/n\)|\[y/n\])\s*$", re.I)


 def human_size(num):
@@ -1132,6 +1140,12 @@ class LiveStream:
                self.file.write("\r\033[K│ %s\n" % kept)
        self.file.flush()

+    def pause(self):
+        """The terminal is about to be used for a question: leave the window
+        where it is and start a fresh one below whatever gets printed."""
+        self.lines.clear()
+        self.shown = 0
+
    def close(self, status):
        if not self.enabled:
            return
@@ -1141,7 +1155,169 @@ class LiveStream:
        self.file.flush()


-def run_live(argv, timeout=SHELL_TIMEOUT, shell=False, cwd=None, env=None, stream=None):
+class PtyProc:
+    """A child on its own pty with the pty as controlling terminal, so ssh,
+    sudo and friends can prompt exactly as in a real terminal. One reader
+    (select), one writer; the caller owns the loop."""
+
+    @staticmethod
+    def available():
+        return pty is not None and termios is not None and fcntl is not None and select is not None
+
+    def __init__(self, argv, shell, cwd, env):
+        master, slave = pty.openpty()
+        try:
+            fcntl.ioctl(master, termios.TIOCSWINSZ, struct.pack("HHHH", RUN_ROWS, RUN_COLS, 0, 0))
+
+            def adopt():
+                os.setsid()
+                fcntl.ioctl(0, termios.TIOCSCTTY, 0)
+
+            self.proc = subprocess.Popen(argv, shell=shell, cwd=cwd, env=env, stdin=slave, stdout=slave, stderr=slave, preexec_fn=adopt, close_fds=True)
+        except BaseException:
+            os.close(master)
+            raise
+        finally:
+            os.close(slave)
+        self.master = master
+
+    def read(self, timeout):
+        """bytes, b"" when nothing arrived in time, None at end of output."""
+        if not select.select([self.master], [], [], max(0.0, timeout))[0]:
+            return b""
+        try:
+            return os.read(self.master, 65536) or None
+        except OSError:
+            return None
+
+    def write(self, text):
+        data = text.encode("utf-8", "replace")
+        while data:
+            sent = os.write(self.master, data)
+            data = data[sent:]
+
+    def echo_off(self):
+        try:
+            return not termios.tcgetattr(self.master)[3] & termios.ECHO
+        except termios.error:
+            return False
+
+    def kill(self):
+        try:
+            os.killpg(self.proc.pid, signal.SIGKILL)
+        except (OSError, ProcessLookupError):
+            self.proc.kill()
+
+    def close(self):
+        try:
+            os.close(self.master)
+        except OSError:
+            pass
+
+
+def run_live_pty(argv, timeout, shell, cwd, env, stream, ask):
+    """run_live on a pty. When the command sits on a prompt (partial last
+    line that looks like one, or echo switched off for a password) ask(text,
+    secret) gets the question; its answer is typed into the command. None
+    declines: the command gets end-of-file. Time spent waiting for the user
+    does not count against the timeout."""
+    started = time.time()
+    env = dict(env if env is not None else os.environ)
+    env.setdefault("PAGER", "cat")
+    env.setdefault("GIT_PAGER", "cat")
+    term = PtyProc(argv, shell, cwd, env)
+    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+    chunks = []
+    size = 0
+    partial = ""
+    asked = ""
+    asks = 0
+    declined = False
+    timed_out = False
+    deadline = started + timeout
+    last_output = time.time()
+
+    def emit(raw):
+        nonlocal size
+        line = RUN_CTRL_RE.sub("", raw.rstrip("\r").split("\r")[-1])
+        if size < RUN_OUTPUT_CAP:
+            chunks.append(line + "\n")
+            size += len(line) + 1
+        if stream is not None:
+            stream.feed(line + "\n")
+
+    def take(data):
+        nonlocal partial
+        partial += decoder.decode(data)
+        while "\n" in partial:
+            line, partial = partial.split("\n", 1)
+            emit(line)
+
+    try:
+        while True:
+            remaining = deadline - time.time()
+            if remaining <= 0:
+                timed_out = True
+                break
+            data = term.read(min(0.05, remaining))
+            if data is None:
+                break
+            if data:
+                take(data)
+                last_output = time.time()
+                continue
+            if term.proc.poll() is not None:
+                while True:
+                    data = term.read(0)
+                    if not data:
+                        break
+                    take(data)
+                break
+            text = RUN_CTRL_RE.sub("", partial).strip()
+            if declined or asks >= RUN_MAX_ASKS or not text or text == asked or time.time() - last_output < RUN_PROMPT_IDLE:
+                continue
+            secret = term.echo_off()
+            if not secret and (RUN_PROMPT_RE.search(text) is None or "\r" in partial):
+                continue
+            asked = text
+            asks += 1
+            if stream is not None and hasattr(stream, "pause"):
+                stream.pause()
+            waited = time.time()
+            answer = ask(text, secret)
+            deadline += time.time() - waited
+            last_output = time.time()
+            if answer is None:
+                declined = True
+                term.write("\x04")
+                note = "» %s (no answer)" % text
+            else:
+                answer = answer.replace("\r", "").split("\n")[0][:RUN_ANSWER_CAP]
+                term.write(answer + "\n")
+                note = "» %s %s" % (text, "(hidden)" if secret else answer)
+            if stream is not None:
+                stream.feed(note + "\n")
+    except BaseException:
+        term.kill()
+        term.proc.wait()
+        raise
+    finally:
+        if timed_out:
+            term.kill()
+        try:
+            term.proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            term.kill()
+            term.proc.wait()
+        term.close()
+    if partial.strip():
+        emit(partial)
+    return term.proc.returncode, "".join(chunks), time.time() - started, timed_out
+
+
+def run_live(argv, timeout=SHELL_TIMEOUT, shell=False, cwd=None, env=None, stream=None, ask=None):
+    if ask is not None and PtyProc.available():
+        return run_live_pty(argv, timeout, shell, cwd, env, stream, ask)
    started = time.time()
    proc = subprocess.Popen(argv, shell=shell, cwd=cwd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, errors="replace", bufsize=1)
    chunks = []
@@ -6480,14 +6656,53 @@ class Tools:
        env = dict(os.environ)
        for name, value in vault.items():
            env[secret_env_name(name)] = value
-        stream = LiveStream() if self.live and sys.stdout.isatty() else None
-        code, output, elapsed, timed_out = run_live(command, shell=True, cwd=workdir, env=env, stream=stream)
+        stream = self.shell_stream(command)
+        code, output, elapsed, timed_out = run_live(command, shell=True, cwd=workdir, env=env, stream=stream, ask=self.shell_ask(command))
        if stream is not None:
            stream.close(stream_status(code, output, elapsed, timed_out))
        if timed_out:
            return "error: timed out after %d seconds" % SHELL_TIMEOUT
        return "exit %d\n%s%s" % (code, spill_or_truncate(self.app.store, "output", "shell: " + command[:80], output.strip() or "(no output)", ["shell"], self.active_profile), curl_hint(command))

+    def shell_stream(self, command):
+        """Where a running command's output goes live: the chat scene's own
+        window when the app provides one, else the REPL's scrolling window,
+        else nowhere."""
+        factory = getattr(self.app, "stream_factory", None)
+        if factory is not None:
+            store = self.app.store
+            return factory(store.redact(command, self.active_profile) if store is not None else command)
+        return LiveStream() if self.live and sys.stdout.isatty() else None
+
+    def shell_ask(self, command):
+        """ask(text, secret) for a command that stops on a prompt (sudo,
+        ssh, a y/n), or None when nobody can answer: autonomous mode,
+        background workers and non-interactive stdin keep the old behavior
+        of a command that never gets input."""
+        app = self.app
+        if getattr(app, "auto", False) or not getattr(app, "persist", True) or getattr(app, "depth", 0) > 0:
+            return None
+        prompter = getattr(app, "prompter", None)
+
+        def answered(value):
+            if value is not None:
+                app.user_stepped_in = True
+            return value
+
+        if prompter is not None:
+            return lambda text, secret: answered(prompter("ask", "%s asks:\n%s" % (command[:60], text), []))
+        if not (sys.stdin.isatty() and sys.stdout.isatty()):
+            return None
+
+        def repl_ask(text, secret):
+            try:
+                line = getpass.getpass(paint("» %s " % text, Ansi.YELLOW)) if secret else input(paint("» %s " % text, Ansi.YELLOW))
+            except (EOFError, KeyboardInterrupt):
+                return None
+            return answered(line)
+
+        return repl_ask
+
    def run_shell_box(self, command, workdir, vault):
        engine, failure = self.box_engine()
        if not engine:
@@ -6496,8 +6711,9 @@ class Tools:
        for name, value in vault.items():
            extra += ["-e", "%s=%s" % (secret_env_name(name), value)]
        try:
-            stream = LiveStream() if self.live and sys.stdout.isatty() else None
-            code, output, elapsed, timed_out = run_live([engine, "exec", *extra, "-i", BOX_NAME, "sh", "-c", command], stream=stream)
+            stream = self.shell_stream(command)
+            ask = self.shell_ask(command)
+            code, output, elapsed, timed_out = run_live([engine, "exec", *extra, "-it" if ask else "-i", BOX_NAME, "sh", "-c", command], stream=stream, ask=ask)
        except (OSError, subprocess.SubprocessError) as exc:
            return "error: " + short_error(exc)
        if stream is not None:
@@ -14028,6 +14244,32 @@ def tui_chat_tool_row(name, args, width, base):
    return tui_pad_spans(spans, width, base)


+def tui_chat_live_rows(command, lines, status, width, base):
+    """A running command as a message: header chip with the command, the
+    last TUI_CHAT_LIVE_ROWS output lines in a fixed window, and a status row
+    once it ended. Wide lines are cut, narrow screens just get less text."""
+    width = max(10, int(width))
+    chip = TuiStyle(bg="role:accent2", fg="role:on_accent", bold=True)
+    head = TuiStyle(bg="role:tool_bg", fg="role:fg", bold=True)
+    body = TuiStyle(bg="role:tool_bg", fg="role:fg")
+    dim = TuiStyle(bg="role:tool_bg", fg="role:muted")
+    bar = TuiStyle(bg="role:tool_bg", fg="role:accent2")
+    title = command.replace("\n", " ")
+    room = max(0, width - 9)
+    if len(title) > room:
+        title = title[:max(0, room - 1)] + "…"
+    rows = [tui_pad_spans([TuiSpan(" ", base), TuiSpan(" ⌁ ", chip), TuiSpan(" %s " % title, head)], width, base)]
+    shown = list(lines)[-TUI_CHAT_LIVE_ROWS:]
+    if not shown:
+        shown = ["(waiting for output)"] if not status else ["(no output)"]
+    for text in shown:
+        text = text.expandtabs(4)[:max(0, width - 4)]
+        rows.append(tui_pad_spans([TuiSpan(" ", base), TuiSpan("│ ", bar), TuiSpan(text, body if lines else dim)], width, base))
+    if status:
+        rows.append(tui_pad_spans([TuiSpan(" ", base), TuiSpan("└ %s" % status[:max(0, width - 4)], dim)], width, base))
+    return rows
+
+
 def tui_chat_note_rows(text, width, side, base):
    """Command output and errors: accent bar plus plain wrapped text on the
    tool background, no markdown so aligned columns survive."""
@@ -14152,6 +14394,30 @@ class TuiChatState(dict):
        return super().get(key, default)


+class TuiLiveStream:
+    """Stream sink (feed/pause/close, like LiveStream) for a command's output
+    in the chat transcript. The journal block ["live", command, lines,
+    status] is the message; feeding mutates it from the turn thread and
+    asks the scene to repaint through `push`, which coalesces."""
+
+    def __init__(self, block, push):
+        self.block = block
+        self.push = push
+
+    def feed(self, line):
+        lines = self.block[2]
+        lines.append(tui_strip_ansi(line.rstrip("\n").split("\r")[-1]))
+        del lines[:-TUI_CHAT_LIVE_KEEP]
+        self.push()
+
+    def pause(self):
+        pass
+
+    def close(self, status):
+        self.block[3] = status or "done"
+        self.push()
+
+
 class TuiStdioGuard:
    """Counted stdout/stdin swap for turn threads. Several bots can run at
    once; a plain redirect_stdout per thread would restore in the wrong order
@@ -14183,6 +14449,9 @@ class TuiStdioGuard:

 TUI_STDIO_GUARD = TuiStdioGuard()
 TUI_CHAT_BLOCK_CAP = 300
+TUI_CHAT_LIVE_ROWS = 8
+TUI_CHAT_LIVE_KEEP = 200
+TUI_CHAT_LIVE_TICK = 0.05


 def tui_spawn_agent(base, name):
@@ -14279,6 +14548,8 @@ def tui_build_chat(app, agent, spawn=None):
            rows = tui_chat_welcome_rows(width, block[1])
        elif kind == "tool":
            rows = [tui_chat_tool_row(block[1], block[2], width, base)]
+        elif kind == "live":
+            rows = tui_chat_live_rows(block[1], block[2], block[3], width, base)
        elif kind == "note":
            rows = tui_chat_note_rows(block[2], width, block[1], base)
        else:
@@ -14626,6 +14897,7 @@ def tui_build_chat(app, agent, spawn=None):
        worker = session.agent
        worker.on_phase = lambda name, detail="": set_phase(session, name, detail)
        worker.prompter = lambda kind, ask, options: prompter(kind, ask, options, session)
+        worker.stream_factory = lambda command: live_stream(command, session)

        def turn():
            with TUI_STDIO_GUARD:
@@ -14658,6 +14930,29 @@ def tui_build_chat(app, agent, spawn=None):

        app.offload(turn, landed)

+    def live_stream(command, session):
+        """Called from a turn thread when a command starts: its window joins
+        the session journal as a message, then repaints at most every
+        TUI_CHAT_LIVE_TICK while output streams in."""
+        loop = app._loop
+        if loop is None:
+            return None
+        block = ["live", command[:300], [], ""]
+        flag = {"queued": False}
+
+        def flush():
+            flag["queued"] = False
+            if session is cur() and any(entry is block for entry in session.blocks):
+                replay()
+
+        def push():
+            if not flag["queued"]:
+                flag["queued"] = True
+                loop.call_soon_threadsafe(lambda: app.after(TUI_CHAT_LIVE_TICK, flush))
+
+        loop.call_soon_threadsafe(lambda: journal(block, session))
+        return TuiLiveStream(block, push)
+
    def finish_turn(session=None):
        session = session or cur()
        session.busy = False
@@ -14862,7 +15157,7 @@ def tui_build_chat(app, agent, spawn=None):
    render_history()
    paint_status()
    refs.update({"do_send": do_send, "run_pending": run_pending, "select_bot": select_bot, "toggle_sidebar": toggle_sidebar,
-                 "paint_status": paint_status, "prompter": prompter, "show": show, "replay": replay, "add_note": note,
+                 "paint_status": paint_status, "prompter": prompter, "live_stream": live_stream, "show": show, "replay": replay, "add_note": note,
                 "render_history": render_history})
    return root, refs

@@ -2890,6 +2890,60 @@ class StreamTests(unittest.TestCase):
        self.assertTrue(timed)
        self.assertLess(time.time() - started, 10)

+    def test_run_live_pty_answers_prompt(self):
+        asked = []
+
+        def ask(text, secret):
+            asked.append((text, secret))
+            return "yes"
+
+        cmd = "echo start; printf 'Continue? [y/n] '; read a; echo got:$a"
+        code, out, _elapsed, timed = tai.run_live(cmd, shell=True, ask=ask)
+        self.assertEqual((code, timed, asked), (0, False, [("Continue? [y/n]", False)]))
+        self.assertIn("start\n", out)
+        self.assertIn("got:yes\n", out)
+
+    def test_run_live_pty_secret_prompt_hidden(self):
+        asked = []
+        cmd = "stty -echo; printf 'Password: '; read p; stty echo; echo; echo len:${#p}"
+        _code, out, _elapsed, _timed = tai.run_live(cmd, shell=True, ask=lambda text, secret: asked.append((text, secret)) or "hunter2")
+        self.assertEqual(asked, [("Password:", True)])
+        self.assertIn("len:7", out)
+        self.assertNotIn("hunter2", out)
+
+    def test_run_live_pty_declined_sends_eof(self):
+        _code, out, _elapsed, timed = tai.run_live("printf 'name: '; read a; echo [$a]", shell=True, ask=lambda text, secret: None)
+        self.assertFalse(timed)
+        self.assertIn("[]", out)
+
+    def test_run_live_pty_wait_not_counted_in_timeout(self):
+        def slow(text, secret):
+            time.sleep(1.5)
+            return "ok"
+        code, out, _elapsed, timed = tai.run_live("printf 'go? '; read a; echo $a", shell=True, timeout=2, ask=slow)
+        self.assertEqual((code, timed), (0, False))
+        self.assertIn("ok", out)
+
+    def test_run_live_pty_progress_is_not_a_prompt(self):
+        asked = []
+        _code, out, _elapsed, _timed = tai.run_live("printf '50%%\\r'; sleep 1.5; printf '100%%\\n'", shell=True, ask=lambda text, secret: asked.append(text))
+        self.assertEqual(asked, [])
+        self.assertEqual(out, "100%\n")
+
+    def test_run_live_pty_timeout_kills_group(self):
+        started = time.time()
+        _code, _out, _elapsed, timed = tai.run_live("sleep 30; echo x", shell=True, timeout=1, ask=lambda text, secret: None)
+        self.assertTrue(timed)
+        self.assertLess(time.time() - started, 10)
+
+    def test_window_pause_starts_fresh_window(self):
+        buf = io.StringIO()
+        stream = tai.LiveStream(height=2, width=40, file=buf, enabled=True)
+        stream.feed("a\n")
+        stream.pause()
+        stream.feed("b\n")
+        self.assertNotIn("\033[", buf.getvalue())
+
    def test_shell_live_flag(self):
        app = FakeSecretApp(self.store)
        app.profile = "t"
@@ -6122,6 +6176,61 @@ class TuiChatTests(unittest.TestCase):
            app.top_modal()._run()
        self.assertEqual(self.run_prompt(pick_other, "ask", "Pick?", ("red", "green")), "mauve")

+    def test_live_block_renders_as_message_with_fixed_window(self):
+        app, _agent, _root, refs = self.make_chat()
+        lines = ["line %d" % n for n in range(30)]
+        refs["blocks"].append(["live", "make build", lines, ""])
+        refs["replay"]()
+        app.layout()
+        shown = app.render_text()
+        self.assertIn("make build", shown)
+        self.assertIn("line 29", shown)
+        self.assertNotIn("line 21\n", shown)
+        rows = tai.tui_chat_live_rows("make build", lines, "exit 0", 40, tai.TuiStyle())
+        self.assertEqual(len(rows), 1 + tai.TUI_CHAT_LIVE_ROWS + 1)
+
+    def test_live_stream_from_turn_thread_lands_in_transcript(self):
+        import asyncio
+        app, _agent, _root, refs = self.make_chat()
+        session = refs.session
+        done = []
+
+        def work():
+            stream = refs["live_stream"]("echo hi", session)
+            for n in range(5):
+                stream.feed("out %d\n" % n)
+            stream.close("exit 0 · 0.1s")
+            return True
+
+        async def go():
+            app._loop = asyncio.get_running_loop()
+            app.offload(work, lambda value, error: done.append((value, error)))
+            for _ in range(300):
+                await asyncio.sleep(0.01)
+                if done:
+                    await asyncio.sleep(0.2)
+                    break
+            app.layout()
+            self.shown = app.render_text()
+            app._loop = None
+
+        asyncio.run(go())
+        self.assertEqual(done, [(True, None)])
+        self.assertIn("echo hi", self.shown)
+        self.assertIn("out 4", self.shown)
+        self.assertIn("exit 0", self.shown)
+
+    def test_live_stream_feeds_block_and_caps(self):
+        block = ["live", "ls", [], ""]
+        pushed = []
+        stream = tai.TuiLiveStream(block, lambda: pushed.append(1))
+        for n in range(tai.TUI_CHAT_LIVE_KEEP + 50):
+            stream.feed("\033[31mrow %d\033[0m\n" % n)
+        stream.close("exit 0")
+        self.assertEqual(len(block[2]), tai.TUI_CHAT_LIVE_KEEP)
+        self.assertEqual(block[2][-1], "row %d" % (tai.TUI_CHAT_LIVE_KEEP + 49))
+        self.assertEqual(block[3], "exit 0")
+
    def test_prompter_without_loop_denies(self):
        app, _agent, _root, refs = self.make_chat()
        self.assertIsNone(refs["prompter"]("approve", "rm x", ["Allow once", "Deny"]))
@@ -313,6 +313,22 @@ then `online`. Thinking and typing animate: braille spinner in cycling
 accent colors, a shimmer sweeping the word, trailing dots lighting one
 by one, and `Ns` elapsed in the status bar.

+Shell commands (host and sandbox box) stream live as a message: the turn
+thread calls `agent.stream_factory(command)`, the scene journals a `live`
+block (header chip with the command, the last `TUI_CHAT_LIVE_ROWS` (8) lines
+in a fixed window, a status row at the end, `TUI_CHAT_LIVE_KEEP` (200) lines
+kept) and repaints it at most every `TUI_CHAT_LIVE_TICK` (50 ms). It stays in
+the transcript where it started, so you keep chatting while it scrolls. The
+REPL has the same thing as the 4-line `LiveStream` window. When a command
+stops on a prompt (sudo, ssh, a y/n; a partial last line ending like a prompt
+after 1 s of silence, or echo switched off), the command runs on a pty and
+the question goes through the same `agent.prompter("ask", ...)` dialog (REPL:
+`input`/`getpass`); the answer is typed into the command, a decline sends
+end-of-file, and the wait does not count against the 120 s timeout (max 20
+questions per command). Gap: the TUI dialog does not mask a typed password.
+Without an interactive user (autonomous mode, background workers, piped
+stdin) commands run on pipes with no input, as before.
+
 Approvals and `ask_user` are real dialogs. The agent calls
 `agent.prompter(kind, text, options)` from the turn thread; the chat
 scene opens a `TuiQuestion` (approve: Allow once / Always allow / Deny;