From 80bb245be7a60918611148a72d4f064d096c2722 Mon Sep 17 00:00:00 2001 From: retoor Date: Mon, 12 Oct 2026 12:45:50 +0200 Subject: [PATCH] TUI: honest phases (thinking/running/typing/online) with animation, real approval and ask_user dialogs instead of silent denial, 18x faster frames, paint on input, fd fixes for turn-thread stdout swap Co-Authored-By: Claude Sonnet 5.5 --- tai.py | 369 ++++++++++++++++++++++++++++++++++++++++++++-------- test_tai.py | 187 +++++++++++++++++++++++++- tui.md | 44 +++++-- 3 files changed, 534 insertions(+), 66 deletions(-) diff --git a/tai.py b/tai.py index eb11f83..4869462 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.28.0" +VERSION = "1.29.0" WORKER_STEPS = 12 CREATE_SKILL_STEPS = 40 FORK_MAX_DEPTH = 2 @@ -7076,18 +7076,24 @@ class Tools: return "ask_user unavailable in autonomous mode: decide yourself and state the assumption" if not getattr(self.app, "persist", True) or getattr(self.app, "depth", 0) > 0: return "ask_user unavailable to background workers: decide yourself and state the assumption" - if not sys.stdin.isatty(): - return "error: no interactive user on this stdin, decide yourself and state the assumption" - print(paint("?- " + question, Ansi.CYAN, Ansi.BOLD)) - for pos, label in enumerate(options, 1): - print(paint(" %d) %s" % (pos, label), Ansi.DIM)) - try: - if options: - raw = input(paint("answer [1-%d, option text, or free input] " % len(options), Ansi.YELLOW)) - else: - raw = input(paint("answer (free input, empty skips) ", Ansi.YELLOW)) - except (EOFError, KeyboardInterrupt): - return "user gave no answer (input closed): proceed with your best assumption" + prompter = getattr(self.app, "prompter", None) + if prompter is not None: + raw = prompter("ask", question, options) + if raw is None: + return "user gave no answer (input closed): proceed with your best assumption" + else: + if not sys.stdin.isatty(): + return "error: no interactive user on this stdin, decide yourself and state the assumption" + print(paint("?- " + question, Ansi.CYAN, Ansi.BOLD)) + for pos, label in enumerate(options, 1): + print(paint(" %d) %s" % (pos, label), Ansi.DIM)) + try: + if options: + raw = input(paint("answer [1-%d, option text, or free input] " % len(options), Ansi.YELLOW)) + else: + raw = input(paint("answer (free input, empty skips) ", Ansi.YELLOW)) + except (EOFError, KeyboardInterrupt): + return "user gave no answer (input closed): proceed with your best assumption" self.app.user_stepped_in = True answer = raw.strip() if not answer: @@ -7974,6 +7980,18 @@ class Agent: return True if not self.persist: return self.config.auto_approve + prompter = getattr(self, "prompter", None) + if prompter is not None: + choice = prompter("approve", command, ["Allow once", "Always allow (approvals off)", "Deny"]) + if choice == "Always allow (approvals off)": + self.user_stepped_in = True + self.yolo = True + self.config.auto_approve = True + return True + if choice == "Allow once": + self.user_stepped_in = True + return True + return False if not sys.stdin.isatty(): return False print(paint("run: %s" % command, Ansi.YELLOW)) @@ -8161,6 +8179,17 @@ class Agent: return None, self.run_mention(selectors["bot"], prompt, text, capture, max_steps) return prompt, None + def phase(self, name, detail=""): + """Report what the turn is really doing (thinking, tool, typing) to + an optional on_phase(name, detail) listener; listener errors never + reach the turn.""" + listener = getattr(self, "on_phase", None) + if listener is not None: + try: + listener(name, detail) + except Exception: + pass + def run_turn(self, text, capture=False, max_steps=MAX_STEPS, _routed=False): self.apply_system() if not _routed: @@ -8203,6 +8232,7 @@ class Agent: if loud: spinner = Spinner("thinking") spinner.start() + self.phase("thinking") try: reply = self.chat.complete(self.messages, payload, stream_sink=sink, deadline=self.deadline) except BackendError as exc: @@ -8231,12 +8261,14 @@ class Agent: elif reply["reasoning"]: last_text = reply["reasoning"] if last_text: + self.phase("typing") self.store.log_event(self.profile, "assistant", "message", last_text) break aborted = False step_pairs = [] for pos, call in enumerate(reply["tool_calls"]): self.show_call(call) + self.phase("tool", call["name"]) call_started = time.time() try: result = self.tools.dispatch(call["name"], call["arguments"]) @@ -9703,6 +9735,8 @@ def boot(args): TUI_FPS = 30 +TUI_INPUT_PAINT_GAP = 0.008 + TUI_MIN_ROWS = 8 TUI_MIN_COLS = 20 @@ -10023,27 +10057,65 @@ def tui_flatten(spans): def tui_runs(cells): + """Merge neighbor cells with equal style into runs. Cells mostly share + one style object, so identity short-circuits the key comparison.""" runs = [] + last = last_key = None for char, style in cells: - if runs and runs[-1][1].key() == style.key(): + if style is last: runs[-1][0].append(char) - else: - runs.append([[char], style]) + continue + key = style.key() + last = style + if runs and key == last_key: + runs[-1][0].append(char) + continue + last_key = key + runs.append([[char], style]) return [TuiSpan("".join(chars), style) for chars, style in runs] def tui_pad_spans(spans, width, fill_style): - cells = tui_flatten(spans)[:max(0, width)] - while len(cells) < width: - cells.append((" ", fill_style)) - return tui_runs(cells) + return tui_slice_spans(spans, 0, width, fill_style) def tui_slice_spans(spans, start, length, fill_style): - cells = tui_flatten(spans)[max(0, start):max(0, start) + max(0, length)] - while len(cells) < max(0, length): - cells.append((" ", fill_style)) - return tui_runs(cells) + """Columns [start, start+length) of a span row, padded with fill_style + to exactly length. Works on whole spans, never per character; neighbors + sharing a style object (or an equal key) merge into one run.""" + start = max(0, start) + length = max(0, length) + end = start + length + out = [] + last = last_key = None + + def put(text, style): + nonlocal last, last_key + if style is last: + out[-1].text += text + return + key = style.key() + if out and key == last_key: + out[-1].text += text + else: + out.append(TuiSpan(text, style)) + last_key = key + last = style + + pos = 0 + for span in spans: + if pos >= end: + break + nxt = pos + len(span.text) + if nxt > start: + piece = span.text[max(start, pos) - pos:min(end, nxt) - pos] + if piece: + put(piece, span.style) + pos = nxt + covered = min(pos, end) - start if pos > start else 0 + if covered < length: + put(" " * (length - covered), fill_style) + return out class TuiPanel: @@ -10647,6 +10719,7 @@ class TuiCompositor: runs_by_row.setdefault(ay, []).append((max(panel.ax, x0), min(panel.ax + panel.aw, x1), panel.ax, spans)) frame = [[(" ", (None, None, False, False, False, False, False)) for _x in range(cols)] for _y in range(rows)] styles = {} + keys = {} for ay, runs in runs_by_row.items(): cuts = set() for x0, x1, _origin, _spans in runs: @@ -10664,11 +10737,23 @@ class TuiCompositor: if winner is None: continue _x0, _x1, origin, spans = winner - cells = tui_flatten(spans)[start - origin:end - origin] - for offset, (char, style) in enumerate(cells): - key = style.key() - styles[key] = style - frame[ay][start + offset] = (char, key) + out_row = frame[ay] + col = origin + for span in spans: + text = span.text + nxt = col + len(text) + lo = max(col, start) + hi = min(nxt, end) + if lo < hi: + style = span.style + key = keys.get(id(style)) + if key is None: + key = keys[id(style)] = style.key() + styles[key] = style + out_row[lo:hi] = [(char, key) for char in text[lo - col:hi - col]] + col = nxt + if col >= end: + break self.styles = styles return frame @@ -12756,6 +12841,8 @@ class TuiApp: self._loop = None self._failure = None self._input_seen = False + self._last_frame = 0.0 + self._stdout_fd = None self._readers = {} self._writers = {} self._eof = set() @@ -13065,7 +13152,7 @@ class TuiApp: rows, cols = 24, 80 if fcntl is not None and termios is not None: try: - packed = fcntl.ioctl(sys.stdout.fileno(), termios.TIOCGWINSZ, b"\0" * 8) + packed = fcntl.ioctl(self._stdout_fd if self._stdout_fd is not None else sys.stdout.fileno(), termios.TIOCGWINSZ, b"\0" * 8) rows, cols = struct.unpack("HH", packed[:4]) except OSError: pass @@ -13083,6 +13170,7 @@ class TuiApp: if not sys.stdin.isatty() or not sys.stdout.isatty(): raise TuiError("tui needs an interactive terminal") self.rows, self.cols = self._winsize() + self._stdout_fd = sys.stdout.fileno() fd = sys.stdin.fileno() self._saved_termios = termios.tcgetattr(fd) tty.setraw(fd) @@ -13096,7 +13184,7 @@ class TuiApp: def _teardown(self): try: - out = sys.stdout.fileno() + out = self._stdout_fd if self._stdout_fd is not None else sys.stdout.fileno() os.write(out, (TUI_PASTE_DISABLE + TUI_KITTY_DISABLE + TUI_MOUSE_DISABLE + TUI_ALT_LEAVE + "\x1b[?25h").encode("utf-8")) except OSError: pass @@ -13184,6 +13272,20 @@ class TuiApp: self._input_seen = True for event in self.input.feed(data): self.dispatch(event) + if time.time() - self._last_frame >= TUI_INPUT_PAINT_GAP: + self._paint_frame() + + def _paint_frame(self): + """Write one frame if anything is dirty. False when stdout is gone.""" + if not (self._need_paint or self._need_layout): + return True + try: + TuiApp.write_all(self._stdout_fd, self.frame_bytes().encode("utf-8")) + except OSError: + self._running = False + return False + self._last_frame = time.time() + return True def _on_term_readable(self, fd, term): if not term.feed_output(): @@ -13234,7 +13336,6 @@ class TuiApp: self._loop = loop loop.set_exception_handler(self._loop_failed) stdin_fd = sys.stdin.fileno() - stdout_fd = sys.stdout.fileno() self.dirty_layout() self._running = True loop.add_reader(stdin_fd, self._on_stdin, stdin_fd) @@ -13260,11 +13361,8 @@ class TuiApp: self.tick_widgets(tick_now - last, tick_now) last = tick_now self.ensure_focus() - if self._need_paint or self._need_layout: - try: - TuiApp.write_all(stdout_fd, self.frame_bytes().encode("utf-8")) - except OSError: - break + if not self._paint_frame(): + break await asyncio.sleep(max(0.0, frame - (time.time() - started))) finally: for fd in list(self._readers): @@ -13500,6 +13598,12 @@ def tui_chat_sidebar_w(cols): TUI_CHAT_BUBBLE_MAX = 0.7 TUI_PREFERRED_MODEL = "molodetz" TUI_CHAT_SPINNER = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏" +TUI_CHAT_ONLINE_GLYPH = "●" +TUI_CHAT_TYPE_ANSWER = object() +TUI_CHAT_PROMPT_TIMEOUT = 600 +TUI_CHAT_PROMPT_CHARS = 600 +TUI_CHAT_REVEAL_TICK = 0.04 +TUI_CHAT_REVEAL_STEPS = 40 TUI_CHAT_TICK_SEND = "◷" TUI_CHAT_TICK_SENT = "✓" TUI_CHAT_TICK_READ = "✓✓" @@ -13836,7 +13940,7 @@ def tui_build_chat(app, agent): paints first and the turn blocks the loop like the telegram bot does.""" refs = {"drafts": {}, "pending": None, "note": "", "sidebar_auto": False, "sidebar_hide": False, "last_key": None, "bots": [], "pending_tick": None, - "blocks": [], "render_w": 0, "busy": False, "spin": 0, "since": 0.0} + "blocks": [], "render_w": 0, "busy": False, "spin": 0, "since": 0.0, "phase": "idle", "detail": "", "reveal": None} panel_style = TuiStyle(bg="role:bg", fg="role:fg") dim_style = TuiStyle(bg="role:bg", fg="role:muted") err_style = TuiStyle(bg="role:bg", fg="role:error") @@ -13859,7 +13963,7 @@ def tui_build_chat(app, agent): head_name = TuiText("", h=1, wrap=False, style=name_style, id="head_name") status_row = TuiCols(h=1, style=bar_style, id="status_row") head_dot = TuiText("● ", w=2, h=1, wrap=False, style=TuiStyle(bg="role:panel", fg="role:success"), id="head_dot") - head_status = TuiText("", w="1fr", h=1, wrap=False, style=TuiStyle(bg="role:panel", fg="role:muted"), id="head_status") + head_status = TuiRich(w="1fr", h=1, style=TuiStyle(bg="role:panel", fg="role:muted"), id="head_status") status_row.add(head_dot, head_status) info.add(head_name, status_row) head_meta = TuiRich(w=24, h=3, style=bar_style, id="head_meta") @@ -13989,6 +14093,40 @@ def tui_build_chat(app, agent): def chip(text, style): return TuiSpan(" %s " % text, style) + def phase_word(): + """What the turn is really doing right now: thinking while the model + or the loop works, running inside a tool call, typing once + the reply exists and is being laid out.""" + if refs["phase"] == "tool": + return "running %s" % (refs["detail"] or "tool") + if refs["phase"] == "typing": + return "typing" + if refs["phase"] == "ask": + return "waiting for you" + return "thinking" + + def status_spans(): + """Header status line. Idle is a calm green dot; a live phase gets a + spinner in cycling accent colors and a shimmer sweeping the word, + with trailing dots that light up one after another.""" + panel = "role:panel" + if not (refs["busy"] or refs["pending"] is not None): + return "online", TUI_CHAT_ONLINE_GLYPH, TuiStyle(bg=panel, fg="role:success"), [TuiSpan("online", TuiStyle(bg=panel, fg="role:muted"))] + word = phase_word() + frame = refs["spin"] + tint = ("role:accent", "role:accent2", "role:accent3")[(frame // 3) % 3] + glyph = TUI_CHAT_SPINNER[frame % len(TUI_CHAT_SPINNER)] if refs["phase"] != "typing" else ("✎" if frame % 4 < 2 else "✐") + head = frame % (len(word) + 6) - 3 + spans = [] + for pos, char in enumerate(word): + reach = abs(pos - head) + fg = "role:fg" if reach == 0 else (tint if reach == 1 else "role:muted") + spans.append(TuiSpan(char, TuiStyle(bg=panel, fg=fg, bold=reach <= 1))) + lit = (frame // 4) % 4 + for pos in range(3): + spans.append(TuiSpan(".", TuiStyle(bg=panel, fg=tint if pos < lit else "role:border", bold=pos < lit))) + return word + "...", glyph, TuiStyle(bg=panel, fg=tint, bold=True), spans + def paint_status(): width = max(20, app.cols) key_style = TuiStyle(bg="role:service_bg", fg="role:service_fg", bold=True) @@ -13996,8 +14134,8 @@ def tui_build_chat(app, agent): gap = TuiSpan(" ", bar_style) left = [] if refs["note"]: - warn = refs["note"].startswith("thinking") - glyph = TUI_CHAT_SPINNER[refs["spin"] % len(TUI_CHAT_SPINNER)] if warn else "●" + warn = refs["busy"] or refs["pending"] is not None + glyph = status_spans()[1] if warn else "●" left.append([TuiSpan(" %s " % glyph, TuiStyle(bg="role:panel", fg="role:warning" if warn else "role:success", bold=True)), TuiSpan(refs["note"], TuiStyle(bg="role:panel", fg="role:fg", bold=True))]) for keycap, label in (("F2", "Bots"), ("F4", "Theme"), ("Tab", "Focus"), ("i", "Input"), ("/help", "Commands")): @@ -14030,9 +14168,11 @@ def tui_build_chat(app, agent): av = TuiStyle(bg=tint, fg="role:on_accent", bold=True) avatar.set([[TuiSpan(" " * 7, av)], [TuiSpan((" %s " % (agent.bot[:1].upper() or "?")).center(7), av)], [TuiSpan(" " * 7, av)]]) head_name.text = agent.bot - typing = refs["pending"] is not None or refs["busy"] - head_status.text = "typing..." if typing else "online" - head_dot.style = TuiStyle(bg="role:panel", fg="role:warning" if typing else "role:success") + _text, glyph, dot_style, spans = status_spans() + head_status.set(spans) + head_status.text = _text + head_dot.text = glyph + " " + head_dot.style = dot_style brand.set([[], [TuiSpan(" ✦ ", TuiStyle(bg="role:panel", fg="role:accent", bold=True)), TuiSpan("tai", TuiStyle(bg="role:panel", fg="role:fg", bold=True)), TuiSpan(" %s" % VERSION, TuiStyle(bg="role:panel", fg="role:muted"))], @@ -14130,6 +14270,7 @@ def tui_build_chat(app, agent): refs["pending_tick"] = bubble("out", text, stamp, TUI_CHAT_TICK_SEND) refs["pending"] = text refs["note"] = "thinking..." + refs["phase"], refs["detail"] = "thinking", "" refresh_bots() paint_status() app.after(0.05, run_pending) @@ -14141,6 +14282,8 @@ def tui_build_chat(app, agent): return refs["busy"] = True refs["since"] = time.time() + agent.on_phase = on_phase + agent.prompter = prompter def turn(): old_stdin = sys.stdin @@ -14158,23 +14301,145 @@ def tui_build_chat(app, agent): sys.stdin = old_stdin def landed(answer, error): - refs["busy"] = False if error is not None: answer = "error: %s" % short_error(error) settle_pending_tick() + full = tui_strip_ansi(answer or "[no reply]") stamp = datetime.now().strftime("%H:%M") - bubble("in", tui_strip_ansi(answer or "[no reply]"), stamp, sender=agent.bot) - refs["note"] = "" - refresh_bots() - paint_status() + animate = app._loop is not None and len(full) > 12 + block = bubble("in", "" if animate else full, stamp, sender=agent.bot) + if not animate: + finish_turn() + return + refs["phase"] = "typing" + refs["reveal"] = {"block": block, "full": full, "shown": 0, + "step": max(6, -(-len(full) // TUI_CHAT_REVEAL_STEPS))} + app.after(TUI_CHAT_REVEAL_TICK, reveal_step) app.offload(turn, landed) + def finish_turn(): + refs["busy"] = False + refs["phase"], refs["detail"], refs["reveal"] = "idle", "", None + refs["note"] = "" + refresh_bots() + paint_status() + + def reveal_step(): + """Lay the reply out word by word: real typing, not a fake wait. + Bounded to TUI_CHAT_REVEAL_STEPS ticks however long the reply is.""" + state = refs["reveal"] + if state is None: + return + full = state["full"] + end = min(len(full), state["shown"] + state["step"]) + while end < len(full) and not full[end - 1].isspace(): + end += 1 + state["shown"] = end + block = state["block"] + if any(entry is block for entry in refs["blocks"]): + block[2] = full[:end] + replay() + if end >= len(full): + finish_turn() + else: + app.after(TUI_CHAT_REVEAL_TICK, reveal_step) + + def on_phase(name, detail=""): + refs["phase"], refs["detail"] = name, detail + + def style_dialog(dialog): + """Dress a stock dialog in the chat theme: surface fill, accent + rounded border, accent buttons.""" + dialog.style = TuiStyle(bg="role:surface", fg="role:fg") + dialog.border_style = TuiStyle(bg="role:bg", fg="role:accent") + dialog.title_style = TuiStyle(bg="role:bg", fg="role:accent2", bold=True) + dialog.rounded = True + dialog.padding = TuiPanel.norm_padding((0, 1)) + for node in dialog.walk(): + if node is dialog: + continue + if isinstance(node, TuiButton): + node.style = TuiStyle(bg="role:accent", fg="role:on_accent", bold=True) + node.rounded = True + node.border_style = TuiStyle(bg="role:surface", fg="role:accent") + node.focus_border_style = TuiStyle(bg="role:surface", fg="role:accent2") + elif isinstance(node, TuiTextBox): + node.style = TuiStyle(bg="role:bg", fg="role:fg") + elif not isinstance(node, (TuiRows, TuiCols)): + node.style = TuiStyle(bg="role:surface", fg="role:fg") + return dialog + + def prompter(kind, text, options): + """Called from the turn thread: show a real dialog on the loop and + wait for the answer. Returns the chosen option / typed text, or None + when the dialog was dismissed or TUI_CHAT_PROMPT_TIMEOUT passed (so a + walked-away user means deny, never an open-ended hang).""" + loop = app._loop + if loop is None: + return None + answer = {"value": None} + done = threading.Event() + shown = [] + text = str(text)[:TUI_CHAT_PROMPT_CHARS] + + def finish(event=None): + done.set() + + def pick(value): + answer["value"] = value + + def ask_text(): + dialog = style_dialog(TuiPromptDialog(text if kind == "ask" else "Reply:", box_h=3, on_execute=pick, + title="Question", w=min(72, max(30, app.cols - 6)))) + dialog.on_close = finish + shown.append(dialog) + app.root.add(dialog) + app.dirty_layout() + + def chosen(value): + if value is TUI_CHAT_TYPE_ANSWER: + answer["typing"] = True + return + pick(value) + + def open_dialog(): + if kind == "approve": + dialog = TuiQuestion("Run this?\n\n%s" % text, options, on_done=pick, title="Approval needed", + w=min(72, max(30, app.cols - 6))) + elif options: + dialog = TuiQuestion(text, list(options) + [("Type an answer...", TUI_CHAT_TYPE_ANSWER)], on_done=chosen, + title="Question", w=min(72, max(30, app.cols - 6))) + else: + ask_text() + return + + def closed(event): + if answer.get("typing"): + answer["typing"] = False + ask_text() + return + finish() + + dialog.on_close = closed + shown.append(style_dialog(dialog)) + app.root.add(dialog) + app.dirty_layout() + + before = refs["phase"], refs["detail"] + refs["phase"], refs["detail"] = "ask", "" + loop.call_soon_threadsafe(open_dialog) + if not done.wait(TUI_CHAT_PROMPT_TIMEOUT): + for dialog in shown: + loop.call_soon_threadsafe(dialog.close) + refs["phase"], refs["detail"] = before + return answer["value"] + def spin(): if not refs["busy"]: return refs["spin"] += 1 - refs["note"] = "thinking... %ds" % int(time.time() - refs["since"]) + refs["note"] = "%s... %ds" % (phase_word(), int(time.time() - refs["since"])) paint_status() def toggle_sidebar(): @@ -14232,13 +14497,13 @@ def tui_build_chat(app, agent): root.on_resize = on_resize app.every(0.2, tick_input) app.every(1.0, paint_status) - app.every(0.12, spin) + app.every(0.08, spin) on_resize(None) refresh_bots() render_history() paint_status() refs.update({"do_send": do_send, "run_pending": run_pending, "toggle_sidebar": toggle_sidebar, - "paint_status": paint_status, "show": show, "replay": replay, "add_note": note, + "paint_status": paint_status, "prompter": prompter, "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 2a81955..dd9828d 100644 --- a/test_tai.py +++ b/test_tai.py @@ -5967,13 +5967,196 @@ class TuiChatTests(unittest.TestCase): def test_busy_turn_blocks_second_send_and_bot_switch(self): app, agent, _root, refs = self.make_chat() refs["do_send"]("one") - self.assertEqual(refs["head_status"].text, "typing...") + self.assertEqual(refs["head_status"].text, "thinking...") refs["do_send"]("two") self.assertEqual(agent.turns, []) refs["run_pending"]() self.assertEqual(agent.turns, ["one"]) self.assertFalse(refs["busy"]) + def test_status_follows_real_phases(self): + app, agent, _root, refs = self.make_chat() + self.assertEqual(refs["head_status"].text, "online") + refs["do_send"]("go") + agent.on_phase = None + phases = [] + original = agent.run_turn + + def turn(text, capture=False): + agent.on_phase("tool", "shell") + refs["paint_status"]() + phases.append(refs["head_status"].text) + agent.on_phase("typing") + refs["paint_status"]() + phases.append(refs["head_status"].text) + return original(text, capture) + + agent.run_turn = turn + refs["run_pending"]() + self.assertEqual(phases, ["running shell...", "typing..."]) + self.assertEqual(refs["head_status"].text, "online") + + def test_typing_is_shown_only_while_reply_reveals(self): + import asyncio + app, agent, _root, refs = self.make_chat() + long_reply = "word " * 80 + agent.run_turn = lambda text, capture=False: long_reply + seen = [] + + async def go(): + app._loop = asyncio.get_running_loop() + refs["do_send"]("hello") + seen.append(refs["head_status"].text) + refs["run_pending"]() + for _ in range(200): + app._fire_timers(time.time() + 1) + await asyncio.sleep(0.01) + seen.append(refs["head_status"].text) + if not refs["busy"]: + break + app._loop = None + + asyncio.run(go()) + self.assertEqual(seen[0], "thinking...") + self.assertIn("typing...", seen) + self.assertEqual(seen[-1], "online") + self.assertLess(seen.index("typing..."), len(seen) - 1) + self.assertTrue(all(item in ("thinking...", "typing...", "online") for item in seen)) + shown = " ".join(line for line, _spans in refs["transcript"].lines) + self.assertIn("word word word", shown) + self.assertEqual(refs["blocks"][-1][2], long_reply) + + def test_status_shimmer_changes_each_frame_but_text_is_stable(self): + app, _agent, _root, refs = self.make_chat() + refs["do_send"]("go") + refs["paint_status"]() + first = [(span.text, span.style.key()) for span in refs["head_status"].rows[0]] + refs["spin"] += 2 + refs["paint_status"]() + second = [(span.text, span.style.key()) for span in refs["head_status"].rows[0]] + self.assertNotEqual(first, second) + self.assertEqual("".join(x for x, _k in first), "".join(x for x, _k in second)) + + def test_frame_stays_inside_budget_on_large_screen(self): + app = tai.TuiApp() + app.rows, app.cols = 50, 200 + agent = FakeChatAgent() + for n in range(40): + agent.messages += [{"role": "user", "content": "q %d" % n}, + {"role": "assistant", "content": "answer " + "word " * 60}] + root, refs = tai.tui_build_chat(app, agent) + app.mount(root) + refs["replay"]() + app.frame_bytes() + started = time.perf_counter() + for _ in range(20): + app.dirty_paint() + app.frame_bytes() + per_frame = (time.perf_counter() - started) / 20 + self.assertLess(per_frame, 1.0 / tai.TUI_FPS, "frame %.1fms over the %.1fms budget" % (per_frame * 1000, 1000.0 / tai.TUI_FPS)) + + def run_prompt(self, drive, kind="approve", text="ls -la", options=("Allow once", "Always allow (approvals off)", "Deny")): + """Drive prompter() from a worker thread against a live asyncio loop; + drive(app, refs) answers the dialog. Returns the prompter result.""" + import asyncio + app, _agent, _root, refs = self.make_chat() + result = [] + + async def go(): + app._loop = asyncio.get_running_loop() + app.offload(lambda: refs["prompter"](kind, text, list(options)), lambda value, error: result.append((value, error))) + for _ in range(300): + await asyncio.sleep(0.01) + if app.top_modal() is not None: + break + app.layout() + self.assertIsNotNone(app.top_modal(), "dialog never opened") + self.shown = app.render_text() + drive(app, refs) + for _ in range(300): + await asyncio.sleep(0.01) + if result: + break + app._loop = None + + asyncio.run(go()) + self.assertEqual(len(result), 1, "prompter never returned") + self.assertIsNone(result[0][1]) + return result[0][0] + + def test_approval_dialog_allows_once(self): + value = self.run_prompt(lambda app, refs: self.key(app, "enter")) + self.assertEqual(value, "Allow once") + self.assertIn("Approval needed", self.shown) + self.assertIn("ls -la", self.shown) + + def test_approval_dialog_deny_and_escape(self): + def deny(app, refs): + self.key(app, "down") + self.key(app, "down") + self.key(app, "enter") + self.assertEqual(self.run_prompt(deny), "Deny") + self.assertIsNone(self.run_prompt(lambda app, refs: self.key(app, "escape"))) + + def test_ask_dialog_free_text_and_typed_answer(self): + def type_it(app, refs): + for char in "blue": + self.key(app, char, char) + app.top_modal()._run() + self.assertEqual(self.run_prompt(type_it, "ask", "Favorite color?", ()), "blue") + + def pick_other(app, refs): + self.key(app, "down") + self.key(app, "down") + self.key(app, "enter") + app.layout() + for char in "mauve": + self.key(app, char, char) + app.top_modal()._run() + self.assertEqual(self.run_prompt(pick_other, "ask", "Pick?", ("red", "green")), "mauve") + + def test_prompter_without_loop_denies(self): + app, _agent, _root, refs = self.make_chat() + self.assertIsNone(refs["prompter"]("approve", "rm x", ["Allow once", "Deny"])) + + def test_agent_hooks_use_prompter_for_approval_and_questions(self): + import types + calls = [] + stub = types.SimpleNamespace(yolo=False, auto=False, persist=True, user_stepped_in=False, + config=types.SimpleNamespace(auto_approve=False)) + stub.prompter = lambda kind, text, options: calls.append(kind) or "Allow once" + self.assertTrue(tai.Agent.ask_approval(stub, "ls")) + stub.prompter = lambda kind, text, options: "Deny" + self.assertFalse(tai.Agent.ask_approval(stub, "ls")) + stub.prompter = lambda kind, text, options: "Always allow (approvals off)" + self.assertTrue(tai.Agent.ask_approval(stub, "ls")) + self.assertTrue(stub.yolo) + self.assertTrue(stub.config.auto_approve) + stub.yolo = False + stub.prompter = lambda kind, text, options: None + self.assertFalse(tai.Agent.ask_approval(stub, "ls")) + + def test_paint_and_winsize_survive_redirected_stdout(self): + """A turn thread swaps sys.stdout for a StringIO; the loop must keep + painting to the real terminal fd and keep reading its size.""" + app, _agent, _root, _refs = self.make_chat() + read_fd, write_fd = os.pipe() + app._stdout_fd = write_fd + app._running = True + app.dirty_paint() + old = sys.stdout + sys.stdout = io.StringIO() + try: + self.assertTrue(app._paint_frame()) + app._winsize() + finally: + sys.stdout = old + os.close(write_fd) + painted = os.read(read_fd, 1 << 20) + os.close(read_fd) + self.assertTrue(app._running) + self.assertIn(b"\x1b[", painted) + def test_hidden_sidebar_frees_its_columns(self): app, _agent, _root, refs = self.make_chat(cols=100) right = refs["transcript"].parent @@ -6187,7 +6370,7 @@ class TuiChatTests(unittest.TestCase): shown = " ".join(line for line, _spans in refs["transcript"].lines) self.assertIn(tai.TUI_CHAT_TICK_SEND, shown) self.assertNotIn(tai.TUI_CHAT_TICK_READ, shown) - self.assertEqual(refs["head_status"].text, "typing...") + self.assertEqual(refs["head_status"].text, "thinking...") refs["run_pending"]() shown = " ".join(line for line, _spans in refs["transcript"].lines) self.assertIn(tai.TUI_CHAT_TICK_READ, shown) diff --git a/tui.md b/tui.md index dfe15cf..315908f 100644 --- a/tui.md +++ b/tui.md @@ -272,18 +272,38 @@ quits from outside the input. Below 80 columns the sidebar auto-hides (`TUI_CHAT_NARROW_COLS`); a manual F2 hide/show wins over the automatic rule until toggled again. -Turns run through `app.offload`: your message and a spinner -`thinking... Ns` status paint first, the turn runs on its own thread, -and the reply lands back on the loop, so the UI keeps rendering, themes -switch and the sidebar scrolls while the model thinks. While a turn is -busy a second send and a bot switch are ignored (the thread owns the -agent and the vault); the store connection is opened with -`check_same_thread=False` for that hand-off. `ask_user` cannot prompt mid-turn here (stdin -is swapped for a non-tty during the turn, so the tool declines and -the agent decides instead). Approvals cannot prompt mid-turn either -for the same reason, so gated tools are denied unless the session -pre-approved them (`--yes` / `--yolo`); allowlisted read-only -commands still run. +Turns run through `app.offload`: your message paints first, the turn +runs on its own thread, and the reply lands back on the loop, so the UI +keeps rendering, themes switch, resizes reflow and the sidebar scrolls +while the model works. While a turn is busy a second send and a bot +switch are ignored (the thread owns the agent and the vault); the store +connection is opened with `check_same_thread=False` for that hand-off. +The loop keeps the real stdout fd from setup, because the turn thread +swaps `sys.stdout`. + +The header status only says what is true. The agent reports its real +phase through `agent.on_phase(name, detail)`: `thinking` while the model +or the loop works, `running ` inside a tool call, `waiting for you` +while a dialog is open, `typing...` once the final reply exists and is +being laid out word by word (bounded to `TUI_CHAT_REVEAL_STEPS` ticks), +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. + +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; +ask: the options plus "Type an answer...") on the loop and the thread +waits. Esc, Cancel, or no answer within `TUI_CHAT_PROMPT_TIMEOUT` (600 +s) means deny / no answer, never a hang. Without a `prompter` (CLI, +Telegram) behavior is unchanged. + +Measured (this machine, 40-exchange transcript, repaint of a dirty +frame): 1.9 ms at 80x24, 5.0 ms at 200x50, 7.8 ms at 300x70 against a +33 ms budget (guarded by a unit test at 200x50). Live in a pty: a +resize reflows to a correct frame in 18-60 ms, a burst of 30 resizes +ends on the right frame, key to screen averages 23 ms (input paints +immediately, no longer waiting for the next tick). ## Terminal widget notes