diff --git a/tai.py b/tai.py index bb248d5..9892b3c 100755 --- a/tai.py +++ b/tai.py @@ -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 diff --git a/test_tai.py b/test_tai.py index 0240662..406ae85 100644 --- a/test_tai.py +++ b/test_tai.py @@ -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"])) diff --git a/tui.md b/tui.md index 7f0078e..fdf8e74 100644 --- a/tui.md +++ b/tui.md @@ -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;