From f9c48c6b93e4a91a859a0a5b7f8f28a5ea6e7649 Mon Sep 17 00:00:00 2001 From: retoor Date: Mon, 12 Oct 2026 03:10:25 +0200 Subject: [PATCH] Update --- README.md | 2 +- tai.py | 262 +++++++++++++++++++++++++++++++++++++++++++--------- test_tai.py | 173 ++++++++++++++++++++++++++++++++++ 3 files changed, 393 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index 7ab4706..12ea1d4 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,7 @@ A chunk hit from `search` returns **the actual chunk text**, not just FTS's high `export_terminal_log` is only ever offered inside a live tmux session (see the environment-gating section above — this is the tool `TOOL_ENV_GATES` exists for). When it's available, it runs `tmux capture-pane -p -S -` — the **entire** pane scrollback, not just what's currently visible — writes it verbatim to `~/.tai/termlogs//-.log`, saves it as a `termlog`-kind vault record for provenance and full-text paging, and immediately chunks and indexes it exactly like everything else above. From that point on, "what was that error three hundred lines back" is a `search` call, not a scrollback hunt — the parent record stays for full context, the chunks stay for precision, and the whole thing cost one tmux subprocess call and zero network requests. -Its quieter cousin, `get_current_terminal_content`, grabs just the visible pane (configurable line count) for a quick glance without writing or indexing anything — the difference between a peek and a capture is deliberate. +Its quieter cousin, `get_current_terminal_content`, grabs just the visible pane (configurable line count) for a quick glance without writing or indexing anything, the difference between a peek and a capture is deliberate. It also appends recent shell history, takes a `target` pane to peek at other terminals, and has two siblings: `list_terminals` for counts and ids of every open terminal, `tail_terminals` for the last 3 lines of each at once. --- diff --git a/tai.py b/tai.py index 1675b98..5cfc288 100755 --- a/tai.py +++ b/tai.py @@ -539,6 +539,14 @@ AUDIT_MAX_CHARS = 20000 FILE_RECORD_MAX = 50000 SHELL_SNAPSHOT_MAX_FILES = 25 BACKUP_KEEP = 10 +SHELL_HISTORY_LINES = 50 +SHELL_HISTORY_CHARS = 4000 +SHELL_HISTORY_READ_BYTES = 32768 +TERMINAL_LIST_MAX_PANES = 50 +TAIL_TERMINALS_DEFAULT_LINES = 3 +TAIL_TERMINALS_MAX_LINES = 10 +TAIL_TERMINALS_MAX_PANES = 20 +TAIL_PER_PANE_CHARS = 800 _UNSET = object() @@ -752,6 +760,105 @@ def tmux_pane_header(): return "" +ZSH_HISTORY_RE = re.compile(r"^: \d+:\d+;(.*)$") + + +def shell_history_paths(): + """Shell history files for terminal context, most specific first.""" + paths = [] + histfile = os.environ.get("HISTFILE") + if histfile: + paths.append(histfile) + home = os.path.expanduser("~") + for cand in (os.path.join(home, ".bash_history"), + os.path.join(home, ".zsh_history"), + os.path.join(home, ".local/share/fish/fish_history")): + if cand not in paths: + paths.append(cand) + return paths + + +def read_history_file(path): + """Tail one shell history file, oldest command first. Never raises.""" + try: + size = os.path.getsize(path) + except OSError: + return [] + try: + with open(path, "rb") as handle: + if size > SHELL_HISTORY_READ_BYTES: + handle.seek(size - SHELL_HISTORY_READ_BYTES) + handle.readline() + raw = handle.read() + except OSError: + return [] + fish = os.path.basename(path) == "fish_history" + entries = [] + for line in raw.decode("utf-8", errors="replace").splitlines(): + stripped = line.strip() + if not stripped: + continue + if fish: + if not stripped.startswith("- cmd: "): + continue + stripped = stripped[len("- cmd: "):].strip() + if stripped: + entries.append(stripped) + continue + match = ZSH_HISTORY_RE.match(stripped) + if match: + stripped = match.group(1).strip() + if not stripped: + continue + entries.append(stripped) + return entries + + +def shell_history_section(): + """Capped recent-commands block for terminal context, or '' when none.""" + merged = [] + for path in shell_history_paths(): + merged.extend(read_history_file(path)) + entries = [] + for line in merged: + if not entries or line != entries[-1]: + entries.append(line) + entries = entries[-SHELL_HISTORY_LINES:] if SHELL_HISTORY_LINES > 0 else [] + text = "\n".join(entries) + if len(text) > SHELL_HISTORY_CHARS: + text = text[-SHELL_HISTORY_CHARS:] + if "\n" in text: + text = text.split("\n", 1)[1] + if not text.strip(): + return "" + return "\n\n## Recent shell history\n(oldest first, newest last)\n```\n%s\n```" % text + + +def tmux_panes(): + """Every pane in every tmux session. Returns (panes, error).""" + if not shutil.which("tmux"): + return [], "tmux not available" + fmt = "#{session_name}\t#{window_index}\t#{pane_index}\t#{pane_id}\t#{pane_active}\t#{pane_current_command}\t#{pane_current_path}" + try: + done = subprocess.run(["tmux", "list-panes", "-a", "-F", fmt], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=10) + except (OSError, subprocess.SubprocessError) as exc: + return [], "terminal list failed: " + short_error(exc) + if done.returncode != 0: + err = (done.stderr or "").strip()[:200] + if "no server running" in err: + return [], "no open terminals (tmux server not running)" + return [], "terminal list failed: " + err + panes = [] + for line in done.stdout.splitlines(): + parts = line.split("\t") + if len(parts) != 7: + continue + session, window, index, pane_id, active, command, path = parts + panes.append({"ref": "%s:%s.%s" % (session, window, index), "id": pane_id, + "active": active == "1", "command": command or "?", "path": path or "?"}) + return panes, "" + + VERSION_RE = re.compile(r"^VERSION = \"(\d+)\.(\d+)\.(\d+)\"$", re.MULTILINE) @@ -4043,6 +4150,7 @@ class ChatClient: self.fixed_key = None self.provider_pin = None self.model_pin = None + self.prefer = None self.health = ModelHealth(getattr(config, "home", "") or "") def muse_key(self): @@ -4189,6 +4297,13 @@ class ChatClient: cands = [cand for cand in cands if (cand["provider"] or "").lower() == self.provider_pin] if self.model_pin: cands = [cand for cand in cands if cand["model"] == self.model_pin] + if self.prefer and not self.provider_pin and not self.model_pin: + label, model = self.prefer + first = [cand for cand in cands if cand["provider"] == label and cand["model"] == model] + if not first and label == DEVPLACE_LABEL and getattr(self.config, "devplace_key", ""): + base = self.config.devplace_base + first = [{"provider": DEVPLACE_LABEL, "chat_url": base + "/chat/completions", "models_url": base + "/models", "key": self.config.devplace_key, "model": model, "profile": ""}] + cands = first + [cand for cand in cands if not any(cand is item for item in first)] return cands def pin_note(self): @@ -4943,13 +5058,16 @@ TOOL_SCHEMAS = [ tool_schema("write_file", "Write content to a file, creating parent directories. Overwrites existing files.", {"path": {"type": "string"}, "content": {"type": "string"}}, ["path", "content"]), tool_schema("edit_file", "Replace one exact text match in a file. Fails unless the match is unique.", {"path": {"type": "string"}, "find": {"type": "string"}, "replace": {"type": "string"}}, ["path", "find", "replace"]), tool_schema("web_search", "Search the web. Set images for image search, content to include fetched page text.", {"query": {"type": "string"}, "images": {"type": "boolean"}, "content": {"type": "boolean"}}, ["query"]), + tool_schema("describe_image", "Describe an image with a vision model (rsearch). Pass an http(s) url or a local file path (PNG, JPEG, GIF, WebP, BMP, max 10 MB).", {"source": {"type": "string"}}, ["source"]), tool_schema("web_fetch", "HTTP client: fetch a URL with any method and return status, content type, size, and content. GET by default; pass method (POST, PUT, PATCH, DELETE, HEAD, OPTIONS), headers, and a string body for APIs. HTML becomes readable text, JSON stays raw, raw forces the raw body. Use this for all HTTP(S), never curl or wget via shell. Pass auth_secret to send a vault secret as an Authorization header; the value stays sealed and secret use asks the user first.", {"url": {"type": "string"}, "method": {"type": "string"}, "headers": {"type": "object"}, "body": {"type": "string"}, "timeout": {"type": "integer"}, "raw": {"type": "boolean"}, "auth_secret": {"type": "string"}, "auth_header": {"type": "string"}, "auth_prefix": {"type": "string"}}, ["url"]), tool_schema("speak", "Synthesize text to speech with a free voice, save the MP3, and play it when a player exists.", {"text": {"type": "string"}, "voice": {"type": "string"}}, ["text"]), tool_schema("listen", "Record from the microphone for some seconds and transcribe it. Needs OS audio tools.", {"seconds": {"type": "integer"}}, []), tool_schema("remember", "Update your own system message with new knowledge or behavior. The instruction is merged into the existing system message, unrelated parts stay intact. Use it to forget as well by instructing what to remove. Execute by default when behavior, preferences, or durable facts need updating. Never pass passwords, tokens, or secret values here; store those with store_secret instead.", {"instruction": {"type": "string"}}, ["instruction"]), tool_schema("recall", "Search past session memory of the current profile by keyword, optionally filtered by tags.", {"query": {"type": "string"}, "tags": {"type": "array"}}, ["query"]), tool_schema("load_skill", "Load a skill by name. Returns full instructions plus bundled file paths. Missing skills with a blueprint are researched and built on demand.", {"name": {"type": "string"}}, ["name"]), - tool_schema("get_current_terminal_content", "Capture visible text of the current tmux pane including scrollback. Works inside tmux or against a running tmux server.", {"lines": {"type": "integer"}}, []), + tool_schema("get_current_terminal_content", "Capture visible text of the current tmux pane including scrollback, plus recent shell history. Pass target (a pane id like %3 or session:window.pane from list_terminals) to capture another open terminal instead. Works inside tmux or against a running tmux server.", {"lines": {"type": "integer"}, "target": {"type": "string"}}, []), + tool_schema("list_terminals", "List every open tmux terminal: all panes in all sessions with ids, current commands, and working directories, plus session/window/pane counts. Use a pane id as target for get_current_terminal_content. Needs a running tmux server.", {}, []), + tool_schema("tail_terminals", "One overview of all open tmux terminals: the last lines (default 3) of each at once, labeled with pane ids so you can spot which terminal holds what, then capture it fully with get_current_terminal_content. Needs a running tmux server.", {"lines": {"type": "integer"}}, []), tool_schema("export_terminal_log", "Only available inside a live tmux session. Exports the full pane scrollback (tmux capture-pane -pS -, the entire history, not just what's visible) to a log file under tai's home dir, then chunks it and indexes the chunks for retrieval with SQLite FTS5 + bm25 (lexical only, no embeddings). From then on, use the search tool (kind chunk) to answer questions about anything that happened in this terminal; it returns the actual chunk text, not just a snippet.", {"tags": {"type": "array"}}, []), tool_schema("plexer_send", "Only available when a tai-managed terminal session is active (either tai's own native wrapper, or an active tmux pane). Sends literal keystrokes into that session's shell, as if typed. Set enter to also press Enter and actually run it. No output is returned by this call; follow up with plexer_read.", {"text": {"type": "string"}, "enter": {"type": "boolean"}}, ["text"]), tool_schema("plexer_read", "Only available when a tai-managed terminal session is active. Reads back the last N lines the session's shell has produced (its actual rendered output, not a tool-call summary).", {"lines": {"type": "integer"}}, []), @@ -5011,13 +5129,16 @@ TOOL_TAGS = { "write_file": ("write", "create", "overwrite"), "edit_file": ("edit", "change", "modify", "replace", "patch", "fix", "alter"), "web_search": ("search", "web", "internet", "lookup", "google"), + "describe_image": ("image", "describe", "vision", "picture", "photo", "screenshot"), "web_fetch": ("fetch", "download", "url", "link", "page", "curl", "wget"), "speak": ("speak", "say", "voice", "talk", "tts"), "listen": ("listen", "hear", "microphone", "mic", "transcribe", "stt", "dictate"), "remember": ("remember", "memorize", "preference"), "recall": ("recall", "memory", "history", "past"), "load_skill": ("skill", "capability"), - "get_current_terminal_content": ("terminal", "tmux", "pane", "scrollback", "screen"), + "get_current_terminal_content": ("terminal", "tmux", "pane", "scrollback", "screen", "history", "bash"), + "list_terminals": ("terminal", "tmux", "panes", "sessions", "windows", "list", "open", "count"), + "tail_terminals": ("terminal", "tmux", "panes", "tail", "peek", "overview", "every"), "export_terminal_log": ("terminal", "tmux", "export", "log", "capture", "history", "rag", "index"), "plexer_send": ("plexer", "terminal", "tmux", "keys", "send-keys", "type", "keystroke"), "plexer_read": ("plexer", "terminal", "tmux", "pane", "read", "capture"), @@ -5058,7 +5179,6 @@ def tool_catalog(): name = schema["function"]["name"] if not tool_env_ok(name): continue - tool_schema("describe_image", "Describe an image with a vision model (rsearch). Pass an http(s) url or a local file path (PNG, JPEG, GIF, WebP, BMP, max 10 MB).", {"source": {"type": "string"}}, ["source"]), if name in CORE_TOOLS: continue first = schema["function"]["description"].split(".")[0][:80] @@ -5129,7 +5249,6 @@ def singular_noun(word): if word.endswith(("ses", "xes", "zes", "ches", "shes")): return word[:-2] if word.endswith("s") and not word.endswith(("ss", "us", "is", "os")): - "describe_image": ("image", "describe", "vision", "picture", "photo", "screenshot"), return word[:-1] return word @@ -5765,6 +5884,7 @@ class Tools: "write_file": self.run_write_file, "edit_file": self.run_edit_file, "web_search": self.run_web_search, + "describe_image": self.run_describe_image, "web_fetch": self.run_web_fetch, "speak": self.run_speak, "listen": self.run_listen, @@ -5772,6 +5892,8 @@ class Tools: "recall": self.run_recall, "load_skill": self.run_load_skill, "get_current_terminal_content": self.run_terminal_content, + "list_terminals": self.run_list_terminals, + "tail_terminals": self.run_tail_terminals, "export_terminal_log": self.run_export_terminal_log, "plexer_send": self.run_plexer_send, "plexer_read": self.run_plexer_read, @@ -5884,7 +6006,6 @@ class Tools: def run_shell(self, args): command = str(args.get("command") or "").strip() if not command: - "describe_image": self.run_describe_image, return "error: empty command" secret_names = args.get("secrets") or [] if not isinstance(secret_names, list) or any(not isinstance(item, str) for item in secret_names): @@ -6126,8 +6247,35 @@ class Tools: lines.append("%d. %s\n %s\n %s" % (pos, item.get("title"), item.get("url"), (item.get("description") or "")[:400])) if item.get("content"): lines.append(" content: " + truncate(str(item["content"]), 2000, 500)) + self.result_note = "%d results" % len(data.get("results") or []) return "\n".join(lines) if lines else "no results" + def run_describe_image(self, args): + source = str(args.get("source") or "").strip() + if not source: + return "error: empty source" + if source.startswith(("http://", "https://")): + url = RSEARCH_BASE + "/describe?" + urllib.parse.urlencode({"url": source}) + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + else: + path = os.path.expanduser(source) + try: + if os.path.getsize(path) > 10 * 1024 * 1024: + return "error: image larger than 10 MB" + with open(path, "rb") as handle: + body = handle.read() + except OSError as exc: + return "error: " + short_error(exc) + request = urllib.request.Request(RSEARCH_BASE + "/describe", data=body, headers={"Accept": "application/json", "Content-Type": "application/octet-stream"}, method="POST") + try: + with urllib.request.urlopen(request, timeout=60) as response: + data = json.loads(response.read().decode("utf-8", "replace")) + except (OSError, ValueError) as exc: + return "error: describe failed: " + short_error(exc) + text = str(data.get("description") or data.get("error") or "no description") + self.result_note = text + return text + def run_web_fetch(self, args): url = str(args.get("url") or "").strip() if not url.startswith(("http://", "https://")): @@ -6247,35 +6395,8 @@ class Tools: self.app.update_system(text) return "system message updated for profile '%s' (%d to %d chars)" % (self.app.profile, len(current), len(text)) - self.result_note = "%d results" % len(data.get("results") or []) def run_recall(self, args): query = str(args.get("query") or "").strip() - def run_describe_image(self, args): - source = str(args.get("source") or "").strip() - if not source: - return "error: empty source" - if source.startswith(("http://", "https://")): - url = RSEARCH_BASE + "/describe?" + urllib.parse.urlencode({"url": source}) - request = urllib.request.Request(url, headers={"Accept": "application/json"}) - else: - path = os.path.expanduser(source) - try: - if os.path.getsize(path) > 10 * 1024 * 1024: - return "error: image larger than 10 MB" - with open(path, "rb") as handle: - body = handle.read() - except OSError as exc: - return "error: " + short_error(exc) - request = urllib.request.Request(RSEARCH_BASE + "/describe", data=body, headers={"Accept": "application/json", "Content-Type": "application/octet-stream"}, method="POST") - try: - with urllib.request.urlopen(request, timeout=60) as response: - data = json.loads(response.read().decode("utf-8", "replace")) - except (OSError, ValueError) as exc: - return "error: describe failed: " + short_error(exc) - text = str(data.get("description") or data.get("error") or "no description") - self.result_note = text - return text - if not query: return "error: empty query" tags = args.get("tags") or [] @@ -6325,19 +6446,71 @@ class Tools: lines = max(10, min(500, int(args.get("lines") or 100))) except (TypeError, ValueError): lines = 100 + target = str(args.get("target") or "").strip()[:128] if not shutil.which("tmux"): return "tmux not available" - pane_header = tmux_pane_header() - header = [pane_header] if pane_header else [] + history = shell_history_section() + if target: + header = ["pane %s" % target] + else: + pane_header = tmux_pane_header() + header = [pane_header] if pane_header else [] + argv = ["tmux", "capture-pane", "-p", "-S", "-%d" % lines] + if target: + argv += ["-t", target] try: - done = subprocess.run(["tmux", "capture-pane", "-p", "-S", "-%d" % lines], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=10) + done = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=10) except (OSError, subprocess.SubprocessError) as exc: - return "terminal capture failed: " + short_error(exc) + return "terminal capture failed: " + short_error(exc) + history if done.returncode != 0: - return "terminal capture failed: " + (done.stderr or "").strip()[:200] + return "terminal capture failed: " + (done.stderr or "").strip()[:200] + history if not done.stdout.strip(): - return "terminal pane is empty" - return "\n".join(header + [truncate(done.stdout, 6000, 2000)]) + return "terminal pane is empty" + history + return "\n".join(header + [truncate(done.stdout, 6000, 2000)]) + history + + def run_list_terminals(self, args): + panes, error = tmux_panes() + if error: + return error + if not panes: + return "no open terminals" + sessions = len({pane["ref"].split(":")[0] for pane in panes}) + windows = len({pane["ref"].rsplit(".", 1)[0] for pane in panes}) + lines = ["%d session%s · %d window%s · %d pane%s" % ( + sessions, "" if sessions == 1 else "s", windows, "" if windows == 1 else "s", + len(panes), "" if len(panes) == 1 else "s")] + for pane in panes[:TERMINAL_LIST_MAX_PANES]: + lines.append("%s %s %s %s%s" % (pane["ref"], pane["id"], pane["command"], pane["path"], + " *" if pane["active"] else "")) + if len(panes) > TERMINAL_LIST_MAX_PANES: + lines.append("... +%d more" % (len(panes) - TERMINAL_LIST_MAX_PANES)) + return "\n".join(lines) + + def run_tail_terminals(self, args): + try: + lines = max(1, min(TAIL_TERMINALS_MAX_LINES, int(args.get("lines") or TAIL_TERMINALS_DEFAULT_LINES))) + except (TypeError, ValueError): + lines = TAIL_TERMINALS_DEFAULT_LINES + panes, error = tmux_panes() + if error: + return error + if not panes: + return "no open terminals" + blocks = [] + for pane in panes[:TAIL_TERMINALS_MAX_PANES]: + try: + done = subprocess.run(["tmux", "capture-pane", "-p", "-t", pane["id"], "-S", "-%d" % lines], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, timeout=10) + except (OSError, subprocess.SubprocessError): + blocks.append("=== %s %s (%s) ===\n(capture failed)" % (pane["ref"], pane["id"], pane["command"])) + continue + if done.returncode != 0 or not done.stdout.strip(): + blocks.append("=== %s %s (%s) ===\n(empty)" % (pane["ref"], pane["id"], pane["command"])) + continue + blocks.append("=== %s %s (%s) ===\n%s" % (pane["ref"], pane["id"], pane["command"], done.stdout.strip()[:TAIL_PER_PANE_CHARS])) + if len(panes) > TAIL_TERMINALS_MAX_PANES: + blocks.append("... +%d more panes, capture one with get_current_terminal_content" % (len(panes) - TAIL_TERMINALS_MAX_PANES)) + return truncate("\n\n".join(blocks), 6000, 2000) def run_export_terminal_log(self, args): if not os.environ.get("TMUX"): @@ -7431,6 +7604,10 @@ class Agent: if self.quiet: return self.show_diff() + note = getattr(self.tools, "result_note", None) + if note: + self.tools.result_note = None + print(paint(note, Ansi.DIM)) print(paint("└─ %d chars · %dms" % (len(result), elapsed_ms), Ansi.DIM)) def show_diff(self): @@ -7604,10 +7781,6 @@ class Agent: for pos, call in enumerate(reply["tool_calls"]): self.show_call(call) call_started = time.time() - note = getattr(self.tools, "result_note", None) - if note: - self.tools.result_note = None - print(paint(note, Ansi.DIM)) try: result = self.tools.dispatch(call["name"], call["arguments"]) except Denied as denied: @@ -12692,6 +12865,7 @@ def tui_chat_sidebar_w(cols): TUI_CHAT_BUBBLE_MAX = 0.7 +TUI_PREFERRED_MODEL = "molodetz" TUI_CHAT_TICK_SEND = "◷" TUI_CHAT_TICK_SENT = "✓" TUI_CHAT_TICK_READ = "✓✓" @@ -13090,7 +13264,9 @@ def tui_build_chat(app, agent): def tui_chat(agent, theme_name): tui_set_theme(theme_name) agent.quiet = True + agent.chat.prefer = (DEVPLACE_LABEL, TUI_PREFERRED_MODEL) app = TuiApp() + app.rows, app.cols = app._winsize() scheduler_stop = start_scheduler(agent.config, agent.store.seal) try: root, refs = tui_build_chat(app, agent) diff --git a/test_tai.py b/test_tai.py index e5274b2..94da4af 100644 --- a/test_tai.py +++ b/test_tai.py @@ -103,6 +103,169 @@ class ToolTests(unittest.TestCase): self.assertIn("Communicate", tai.BOX_TTS) +class TerminalContextTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.hist = os.path.join(self.tmp.name, "hist") + with open(self.hist, "w") as handle: + handle.write("git status\ngit push\n") + self.paths = mock.patch.object(tai, "shell_history_paths", return_value=[self.hist]) + self.paths.start() + self.which = mock.patch.object(tai.shutil, "which", return_value="/usr/bin/tmux") + self.which.start() + self.calls = [] + + def tearDown(self): + self.which.stop() + self.paths.stop() + self.tmp.cleanup() + + def tmux(self, panes=(), captures=None, list_error=None): + captures = captures or {} + + def side(argv, **kwargs): + self.calls.append(list(argv)) + sub = argv[1] if len(argv) > 1 else "" + if sub == "display-message": + return mock.Mock(returncode=0, stdout="main:0.0 bash", stderr="") + if sub == "list-panes": + if list_error is not None: + return mock.Mock(returncode=1, stdout="", stderr=list_error) + return mock.Mock(returncode=0, stdout="".join(line + "\n" for line in panes), stderr="") + if sub == "capture-pane": + key = argv[argv.index("-t") + 1] if "-t" in argv else "" + if key in captures: + out = captures[key] + if isinstance(out, tuple): + return mock.Mock(returncode=out[0], stdout=out[1], stderr=out[2]) + return mock.Mock(returncode=0, stdout=out, stderr="") + return mock.Mock(returncode=1, stdout="", stderr="can't find pane: %s" % key) + raise AssertionError("unexpected tmux call %r" % (argv,)) + + return mock.patch.object(tai.subprocess, "run", side_effect=side) + + def pane(self, session, window, index, pid, active, command, path): + return "\t".join([session, str(window), str(index), pid, "1" if active else "0", command, path]) + + def test_history_section_bash_order(self): + section = tai.shell_history_section() + self.assertIn("## Recent shell history", section) + self.assertLess(section.index("git status"), section.index("git push")) + + def test_history_section_missing_files(self): + with mock.patch.object(tai, "shell_history_paths", return_value=[os.path.join(self.tmp.name, "nope")]): + self.assertEqual(tai.shell_history_section(), "") + + def test_history_zsh_extended_stripped(self): + with open(self.hist, "w") as handle: + handle.write(": 1690000000:0;vim foo\n: 1690000001:5;make\n") + section = tai.shell_history_section() + self.assertIn("vim foo", section) + self.assertNotIn("1690000000", section) + + def test_history_fish_cmds(self): + fish = os.path.join(self.tmp.name, "fish_history") + with open(fish, "w") as handle: + handle.write("- cmd: ls -la\n when: 1690000000\n- cmd: pwd\n when: 1690000001\n") + with mock.patch.object(tai, "shell_history_paths", return_value=[fish]): + section = tai.shell_history_section() + self.assertIn("ls -la", section) + self.assertIn("pwd", section) + self.assertNotIn("when:", section) + + def test_history_caps_keep_newest(self): + with open(self.hist, "w") as handle: + for num in range(200): + handle.write("cmd-%03d %s\n" % (num, "x" * 100)) + section = tai.shell_history_section() + self.assertLessEqual(len(section), tai.SHELL_HISTORY_CHARS + 200) + self.assertIn("cmd-199", section) + self.assertNotIn("cmd-000", section) + + def test_content_appends_history(self): + tools = tai.Tools(FakeApp()) + with self.tmux(captures={"": "visible pane output"}): + result = tools.dispatch("get_current_terminal_content", "{}") + self.assertIn("visible pane output", result) + self.assertIn("## Recent shell history", result) + self.assertIn("git push", result) + + def test_content_target_other_pane(self): + tools = tai.Tools(FakeApp()) + with self.tmux(captures={"%5": "other pane output"}): + result = tools.dispatch("get_current_terminal_content", json.dumps({"target": "%5"})) + self.assertIn("other pane output", result) + self.assertIn("pane %5", result) + self.assertTrue(any("-t" in call and "%5" in call for call in self.calls)) + + def test_content_bad_target_keeps_history(self): + tools = tai.Tools(FakeApp()) + with self.tmux(captures={}): + result = tools.dispatch("get_current_terminal_content", json.dumps({"target": "%9"})) + self.assertTrue(result.startswith("terminal capture failed")) + self.assertIn("git push", result) + + def test_list_terminals_counts_and_marks_active(self): + tools = tai.Tools(FakeApp()) + panes = (self.pane("main", 0, 0, "%1", True, "bash", "/home/u"), + self.pane("main", 0, 1, "%2", False, "vim", "/home/u"), + self.pane("other", 1, 0, "%3", False, "python", "/tmp")) + with self.tmux(panes=panes): + result = tools.dispatch("list_terminals", "{}") + self.assertIn("2 sessions", result) + self.assertIn("2 windows", result) + self.assertIn("3 panes", result) + self.assertIn("main:0.0 %1 bash /home/u *", result) + self.assertIn("other:1.0 %3 python /tmp", result) + + def test_list_terminals_no_server(self): + tools = tai.Tools(FakeApp()) + with self.tmux(list_error="no server running on /tmp/tmux-1000/default"): + self.assertEqual(tools.dispatch("list_terminals", "{}"), "no open terminals (tmux server not running)") + + def test_tail_terminals_labels_each_pane(self): + tools = tai.Tools(FakeApp()) + panes = (self.pane("main", 0, 0, "%1", True, "bash", "/home/u"), + self.pane("main", 0, 1, "%2", False, "vim", "/home/u")) + with self.tmux(panes=panes, captures={"%1": "one\ntwo\nthree", "%2": "alpha\nbeta"}): + result = tools.dispatch("tail_terminals", "{}") + self.assertIn("=== main:0.0 %1 (bash) ===", result) + self.assertIn("three", result) + self.assertIn("=== main:0.1 %2 (vim) ===", result) + self.assertIn("beta", result) + self.assertTrue(any("-3" in call for call in self.calls)) + + def test_tail_terminals_clamps_lines(self): + tools = tai.Tools(FakeApp()) + panes = (self.pane("main", 0, 0, "%1", True, "bash", "/home/u"),) + with self.tmux(panes=panes, captures={"%1": "x"}): + tools.dispatch("tail_terminals", json.dumps({"lines": 99})) + self.assertTrue(any("-10" in call for call in self.calls)) + + def test_tail_terminals_dead_pane_skipped_loudly(self): + tools = tai.Tools(FakeApp()) + panes = (self.pane("main", 0, 0, "%1", True, "bash", "/home/u"), + self.pane("main", 0, 1, "%2", False, "vim", "/home/u")) + with self.tmux(panes=panes, captures={"%1": "alive", "%2": (1, "", "can't find pane")}): + result = tools.dispatch("tail_terminals", "{}") + self.assertIn("alive", result) + self.assertIn("(empty)", result) + + def test_tools_lazy_load_on_keywords(self): + names = {schema["function"]["name"] for schema in tai.select_tools("list all open terminal sessions")} + self.assertIn("list_terminals", names) + names = {schema["function"]["name"] for schema in tai.select_tools("tail every terminal pane")} + self.assertIn("tail_terminals", names) + names = {schema["function"]["name"] for schema in tai.select_tools("what is in my bash history")} + self.assertIn("get_current_terminal_content", names) + + def test_schemas_registered(self): + schemas = {schema["function"]["name"]: schema["function"] for schema in tai.TOOL_SCHEMAS} + self.assertIn("target", schemas["get_current_terminal_content"]["parameters"]["properties"]) + self.assertIn("list_terminals", schemas) + self.assertIn("lines", schemas["tail_terminals"]["parameters"]["properties"]) + + class AskUserTests(unittest.TestCase): def tools(self, **attrs): app = FakeApp() @@ -5242,6 +5405,16 @@ class TuiChatTests(unittest.TestCase): for char in text: self.key(app, char, char) + def test_prefer_puts_devplace_molodetz_first(self): + chat = tai.ChatClient.__new__(tai.ChatClient) + chat.provider_pin = chat.model_pin = None + chat.prefer = (tai.DEVPLACE_LABEL, "molodetz") + chat.config = type("C", (), {"devplace_key": "k", "devplace_base": "https://x/v1"})() + other = {"provider": "openrouter", "model": "a", "chat_url": "u", "models_url": "m", "key": "", "profile": ""} + out = chat._apply_pins([other]) + self.assertEqual((out[0]["provider"], out[0]["model"]), (tai.DEVPLACE_LABEL, "molodetz")) + self.assertIs(out[1], other) + def test_sidebar_width_clamps(self): self.assertEqual(tai.tui_chat_sidebar_w(100), 30) self.assertEqual(tai.tui_chat_sidebar_w(200), 36)