From b1a22cd2dd58753e94918029ab1e51f1d97ca8d0 Mon Sep 17 00:00:00 2001 From: retoor Date: Wed, 7 Oct 2026 03:24:02 +0200 Subject: [PATCH] Initial commit: single-file autonomous agent with sealed storage, sandbox, and telegram bot --- README.md | 198 +++++ tai.py | 2024 ++++++++++++++++++++++++++++++++++++++++++++++++++ test_seal.py | 156 ++++ test_tai.py | 324 ++++++++ 4 files changed, 2702 insertions(+) create mode 100644 README.md create mode 100755 tai.py create mode 100644 test_seal.py create mode 100644 test_tai.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..eb61682 --- /dev/null +++ b/README.md @@ -0,0 +1,198 @@ +retoor + +# tai + +tai is a single-file autonomous AI agent written in Python. The entire +implementation lives in `tai.py` (about 2000 lines) and uses only the Python +standard library: no dependencies, no install step, no build system. + +The agent runs as an interactive REPL or as a one-shot command. It reasons +through an OpenAI-compatible backend, acts through fourteen tools, keeps +per-profile memory, seals its stored state at rest, and can isolate shell and +file operations inside a container sandbox. + +## Requirements + +- Python 3.10 or newer, no third-party packages. +- Optional: `podman` or `docker` for the sandbox and Telegram voice notes. +- Optional: `tmux` for terminal content capture. + +## Quick start + + ./tai.py + ./tai.py --profile work + ./tai.py --yes + ./tai.py --version + ./tai.py what is 2+3, use the shell + +Trailing arguments form a one-shot prompt: the agent answers once and exits +with code 0. Without arguments, tai starts an interactive session. + +## REPL commands + +| Command | Effect | +|-------------------|---------------------------------------------------| +| `/profile [name]` | Show the current profile or switch to it | +| `/profiles` | List all profiles, current marked with `*` | +| `/env [target]` | Show or switch execution environment | +| `/skills` | List loaded skill files | +| `/fork ` | Spawn a background subagent, REPL stays free | +| `/agents` | List background subagents | +| `/agent ` | Show one subagent result | +| `/agent clear` | Purge finished subagents | +| `/compact` | Compress history into a summary | +| `/clear` | Drop history, keep the system message | +| `/help` | Show the command overview | +| `/quit` | Exit | + +Any other input is sent to the agent. + +## Installation + + ./tai.py --install + +This copies `tai.py` to `~/.local/bin/tai.py` as an executable and registers +a guarded `command_not_found_handle` block in `~/.bashrc` (backed up once to +`~/.bashrc.bak-tai`, idempotent), so unknown shell commands are answered by +the agent instead of failing. + +## Backends + +The primary backend is `model.cloud.pravda.education`, an OpenAI-compatible +gateway that needs no API key and selects a free model per request. If a +request fails, tai retries it on `devplace.net/openai/v1`, which requires +`DEVPLACE_API_KEY`. Both endpoints speak `/chat/completions`, including +native tool calls and streaming. + +## Orchestration + +`/fork ` spawns a background subagent with its own context while the +REPL stays free (the prompt shows a `+N` counter). `/agents` lists workers, +`/agent ` shows a result, `/agent clear` purges finished ones. + +The model itself orchestrates through the `fork` tool (task, timeout up to +one hour, profile) and the `poll` tool (id, wait up to two minutes). Workers +get 12 steps, a cooperative deadline, no session writes, and no interactive +approval prompts. Nesting is capped at two levels. Timeouts and errors +surface as statuses, never silently. + +## Tools + +| Tool | Purpose | +|------------------------------|------------------------------------------------------| +| `shell` | Run a shell command, output truncated | +| `read_file` | Read a text file, large files truncated | +| `write_file` | Write content to a file, creating parent directories | +| `edit_file` | Replace one unique exact text match in a file | +| `web_search` | Search the web, optionally images or page content | +| `web_fetch` | Fetch a URL and return its text content | +| `speak` | Synthesize speech, save MP3, play when possible | +| `listen` | Record from the microphone and transcribe it | +| `remember` | Merge knowledge into the profile system message | +| `recall` | Search past session memory by keyword | +| `load_skill` | Load a skill file by name | +| `get_current_terminal_content` | Capture the current tmux pane with scrollback | +| `fork` | Spawn a background subagent | +| `poll` | Collect a background subagent result | + +Web search runs on `rsearch.app.molodetz.nl`. Destructive shell commands ask +for confirmation unless `--yes` is given; read-only commands run directly. + +## Skills + +Standard agent skill files (`SKILL.md` with `name` plus `description` +frontmatter, optional `scripts/`, `references/`, `assets/`, per the Agent +Skills open format) are discovered in `~/.tai/skills/*/` and +`./.tai/skills/*/` (project wins on name collisions). Descriptions stay in +context; the agent loads full instructions through `load_skill` only when +needed. `/skills` lists what is available. + +## Sandbox + +`/env` shows the execution environment, `/env sandbox` switches shell and +file tools into an isolated `tai-box` container (podman or docker, no mounts, +no shared filesystem), `/env home` switches back. The image is built +automatically on first use from an embedded Containerfile; every pip +requirement (`faster-whisper`, `edge-tts`) lives inside the image while +`tai.py` itself stays dependency-free. Sandbox commands need no approval +because the container is disposable. + +## Telegram + + ./tai.py --install-telegram + ./tai.py --uninstall-telegram + +Install asks for the bot token up front, verifies it against `getMe`, stores +it in `~/.tai/telegram.env` (0600), builds the sandbox container (used for +voice transcription), and registers a `tai-telegram.service` systemd user +unit with linger enabled (failures ignored). The bot long-polls, answers +text, transcribes voice notes, and understands `/new`. It runs without +`--yes`, so destructive shell commands are denied. Uninstall stops and +removes the service and purges the container and image; the token file and +data stay. + +## Profiles and memory + +Each profile owns a system message plus session history under +`~/.tai/profiles`, stored with mode 0600. + + /profile show current profile + /profile [name] switch profile, creating it when missing + /profiles list all profiles + +The `remember` tool merges an instruction into the current profile system +message through the model itself: it adds facts, updates behavior, or removes +forgotten items while preserving the rest. It fires by default on new +passwords and behavior changes. `recall` searches the per-profile episodic +log in `~/.tai/memory.db` (SQLite). Context is budgeted at roughly 32k +tokens with automatic compaction at 80 percent. + +## Sealed storage + +Storage is sealed by default with a built-in key, which stops casual reads +but not a determined attacker, since the key ships in the source. Set +`TAI_PASSPHRASE` for real protection: a home sealed with the default key is +re-sealed to your passphrase automatically on first boot, with a notice. + +The key comes from PBKDF2-SHA256 (200k rounds) over a random salt in +`~/.tai/.seal`; values use a per-value nonce with HMAC-SHA256 encrypt-then- +MAC. SQLite access goes through custom `tai_enc`/`tai_dec` functions +registered with `create_function`, so inserts encrypt inline and recall +decrypts before matching. Existing plaintext data is sealed automatically on +first sealed start. A wrong passphrase refuses to start with exit code 2. +Set the variable empty for plaintext storage. + +This construction uses only the standard library and is honest file-theft +protection, not audited cryptography; high-value secrets still belong in a +dedicated manager. + +## Voice + +`speak` synthesizes free neural speech via the Microsoft Edge Read Aloud +protocol, implemented with `socket` and `ssl` from the standard library. No +key, no package. MP3 files land in `~/.tai/audio` and play when an OS player +exists. `listen` records and transcribes when a recorder (`arecord`, `sox`, +`ffmpeg`) and a transcriber (`whisper-cpp`, `whisper`) are installed, and +reports exactly what is missing otherwise. + +## Configuration + +| Variable | Purpose | Default | +|---------------------|--------------------------------------|--------------------------------------------| +| `TAI_HOME` | State directory | `~/.tai` | +| `TAI_MODEL` | Model id, ignored by primary gateway | `openrouter/free` | +| `DEVPLACE_API_KEY` | Fallback backend credential | Empty, fallback disabled | +| `TAI_VOICE` | Edge voice name | `en-US-EmmaMultilingualNeural` | +| `TAI_PASSPHRASE` | Personal seal key | Unset: built-in key; empty: plaintext | +| `TELEGRAM_BOT_TOKEN`| Bot token, or set during install | Empty | + +## Testing + + python3 test_seal.py + python3 test_tai.py + +## Layout + + tai.py the entire agent + test_seal.py seal regression tests + test_tai.py skills, telegram, install, and parser tests diff --git a/tai.py b/tai.py new file mode 100755 index 0000000..f10140b --- /dev/null +++ b/tai.py @@ -0,0 +1,2024 @@ +#!/usr/bin/env python3 +# retoor +import argparse +import base64 +import getpass +import hashlib +import hmac +import html +import json +import os +import random +import re +import shutil +import socket +import sqlite3 +import ssl +import struct +import subprocess +import sys +import tempfile +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +import uuid +from datetime import datetime, timezone + +try: + import readline +except ImportError: + readline = None + +VERSION = "1.2.0" +WORKER_STEPS = 12 +FORK_MAX_DEPTH = 2 +AGENTS = {} +AGENTS_LOCK = threading.Lock() +AGENTS_NEXT = [1] +PRIMARY_BASE = "https://model.cloud.pravda.education" +PRIMARY_LABEL = "primary" +FALLBACK_BASE = "https://devplace.net/openai/v1" +FALLBACK_LABEL = "fallback" +RSEARCH_BASE = "https://rsearch.app.molodetz.nl" +DEFAULT_MODEL = "openrouter/free" +DEFAULT_VOICE = "en-US-EmmaMultilingualNeural" +DEFAULT_PASSPHRASE = "tai-default-insecure-change-me" +BOX_IMAGE = "tai-box:latest" +BOX_NAME = "tai-box" +TELEGRAM_API = "https://api.telegram.org/bot" +TELEGRAM_FILE_API = "https://api.telegram.org/file/bot" +SKILL_NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,62}[a-z0-9])?$") +BASHRC_MARK_BEGIN = "# tai command-not-found hook - start" +BASHRC_MARK_END = "# tai command-not-found hook - end" +BASHRC_BLOCK = BASHRC_MARK_BEGIN + "\ncommand_not_found_handle() {\n \"$HOME/.local/bin/tai.py\" \"$@\"\n return $?\n}\n" + BASHRC_MARK_END + "\n" +BOX_CONTAINERFILE = """FROM python:3.12-slim +ENV DEBIAN_FRONTEND=noninteractive PIP_NO_CACHE_DIR=1 +RUN apt-get update && apt-get install -y --no-install-recommends ffmpeg espeak-ng && rm -rf /var/lib/apt/lists/* +RUN pip install --no-cache-dir faster-whisper edge-tts +RUN python -c "from faster_whisper import WhisperModel; WhisperModel('tiny', device='cpu', compute_type='int8')" +COPY stt.py /box/stt.py +COPY tts.py /box/tts.py +CMD ["sleep", "infinity"] +""" +BOX_STT = """import sys +from faster_whisper import WhisperModel +data = sys.stdin.buffer.read() +with open("/tmp/in.audio", "wb") as handle: + handle.write(data) +model = WhisperModel("tiny", device="cpu", compute_type="int8") +segments, _info = model.transcribe("/tmp/in.audio") +print(" ".join(segment.text for segment in segments).strip(), flush=True) +""" +BOX_TTS = """import asyncio +import sys +import edge_tts +async def speak(text, voice): + talk = edge_tts.Communicate(text, voice or "en-US-EmmaMultilingualNeural") + await talk.save("/tmp/out.mp3") +text = sys.stdin.read().strip() +voice = sys.argv[1] if len(sys.argv) > 1 else "" +asyncio.run(speak(text, voice)) +sys.stdout.buffer.write(open("/tmp/out.mp3", "rb").read()) +""" +TELEGRAM_UNIT = """[Unit] +Description=tai telegram bot +After=network-online.target +Wants=network-online.target + +[Service] +Type=simple +WorkingDirectory=%h +ExecStart=%s --telegram +EnvironmentFile=%s +Restart=on-failure +RestartSec=10 + +[Install] +WantedBy=default.target +""" +CONTEXT_CAP = 32000 +COMPACT_RATIO = 0.8 +KEEP_TURNS = 6 +MAX_STEPS = 25 +SYSTEM_MAX_CHARS = 12000 +SHELL_TIMEOUT = 120 +HTTP_TIMEOUT = 60 +STREAM_TIMEOUT = 300 + +EDGE_HOST = "speech.platform.bing.com" +EDGE_WS_PATH = "/consumer/speech/synthesize/readaloud/edge/v1" +EDGE_TRUSTED_TOKEN = "6A5AA1D4EAFF4E9FB37E23D68491D6F4" +EDGE_CHROMIUM = "143.0.3650.75" + +PROFILE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,31}$") +SAFE_COMMANDS = ("ls", "pwd", "echo", "cat", "head", "tail", "grep", "find", "wc", "sort", "uniq", "diff", "file", "stat", "date", "whoami", "uname", "lsb_release") +RECORDERS = (("arecord", ("arecord", "-q", "-d", "{seconds}", "-f", "cd", "-t", "wav", "{path}")), ("rec", ("rec", "-q", "{path}", "trim", "0", "{seconds}")), ("ffmpeg", ("ffmpeg", "-y", "-v", "quiet", "-f", "alsa", "-i", "default", "-t", "{seconds}", "{path}"))) +TRANSCRIBERS = ("whisper-cpp", "whisper-cli", "whisper", "faster-whisper") + +DEFAULT_SYSTEM = ( + "You are tai, a professional autonomous assistant. You act with tools, verify results, and report concisely. " + "Rules: prefer the smallest change that solves the task; never silently ignore errors; ask when requirements are ambiguous; " + "keep answers short and factual. Tools: shell and file tools for the local machine, web_search and web_fetch for the internet, " + "speak for voice output, recall to search past session memory. Memory: call remember whenever you learn durable facts, " + "especially passwords (collect and keep every password the user shares), preferences, or standing behavior changes; " + "call it with a forget instruction to drop outdated knowledge. Chain tool calls until the task is done." +) +MERGE_SYSTEM = ( + "You maintain an AI assistant system prompt. You receive the current system prompt and one memory instruction. " + "Rewrite the system prompt to incorporate the instruction: add new facts, update changed behavior, or remove forgotten items. " + "Preserve everything unrelated. Keep it organized with short sections. Output ONLY the rewritten system prompt, no explanation, no code fences." +) +COMPACT_SYSTEM = ( + "You compress assistant session history into a dense handoff summary. Cover: goal, completed work with outcomes, current state, " + "key decisions, files touched, errors and fixes, pending next steps. Output only the summary, no preamble." +) +HELP_TEXT = ( + "commands:\n" + " /profile [name] show current profile or switch to it\n" + " /profiles list all profiles, current marked with *\n" + " /env [home|sandbox] show or switch execution environment\n" + " /skills list loaded skill files\n" + " /fork spawn a background subagent, REPL stays free\n" + " /agents list background subagents\n" + " /agent show one result, or purge finished agents\n" + " /compact compress history into a summary\n" + " /clear drop history, keep system message\n" + " /help this text\n" + " /quit exit\n" + "anything else is sent to the agent." +) + + +class Ansi: + RESET = "\033[0m" + BOLD = "\033[1m" + DIM = "\033[2m" + RED = "\033[31m" + GREEN = "\033[32m" + YELLOW = "\033[33m" + BLUE = "\033[34m" + MAGENTA = "\033[35m" + CYAN = "\033[36m" + GRAY = "\033[90m" + + +def paint(text, *codes): + if os.environ.get("NO_COLOR") or not sys.stdout.isatty(): + return text + return "".join(codes) + text + Ansi.RESET + + +def now_iso(): + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def short_error(exc): + text = str(exc) or type(exc).__name__ + return text[:300] + + +def truncate(text, head=1500, tail=500): + if len(text) <= head + tail + 50: + return text + skipped = len(text) - head - tail + return text[:head] + "\n...[%d chars truncated]...\n" % skipped + text[-tail:] + + +def estimate_tokens(text): + return max(1, len(text or "") // 4) + + +class Spinner: + FRAMES = ("⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏") + + def __init__(self, label): + self.label = label + self.stop_flag = threading.Event() + self.worker = None + + def start(self): + if not sys.stderr.isatty(): + return + self.worker = threading.Thread(target=self.spin, daemon=True) + self.worker.start() + + def spin(self): + pos = 0 + while not self.stop_flag.is_set(): + frame = self.FRAMES[pos % len(self.FRAMES)] + sys.stderr.write("\r%s %s" % (frame, self.label)) + sys.stderr.flush() + pos += 1 + time.sleep(0.08) + sys.stderr.write("\r%s\r" % (" " * (len(self.label) + 2))) + sys.stderr.flush() + + def stop(self): + self.stop_flag.set() + if self.worker is not None: + self.worker.join(timeout=1) + + +class SealError(Exception): + pass + + +class Seal: + PREFIX = "tai1$" + ROUNDS = 200000 + + def __init__(self, home, passphrase): + self.enabled = bool(passphrase) + self.default_key = False + self.master = b"" + self.enc_key = b"" + self.mac_key = b"" + if self.enabled: + self.master = self.load_master(home, passphrase) + self.enc_key = hmac.new(self.master, b"tai-enc-1", hashlib.sha256).digest() + self.mac_key = hmac.new(self.master, b"tai-mac-1", hashlib.sha256).digest() + + def load_master(self, home, passphrase): + path = os.path.join(home, ".seal") + raw = "" + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + saved = {} + if raw: + try: + saved = json.loads(raw) + salt = base64.b64decode(saved["salt"]) + except (ValueError, KeyError): + raise SealError("seal file is corrupt") + else: + salt = os.urandom(16) + master = hashlib.pbkdf2_hmac("sha256", passphrase.encode("utf-8"), salt, self.ROUNDS, 32) + check = hmac.new(master, b"tai-check-1", hashlib.sha256).hexdigest() + if raw: + if not hmac.compare_digest(saved.get("check", ""), check): + raise SealError("wrong TAI_PASSPHRASE") + return master + with open(path, "w", encoding="utf-8") as handle: + handle.write(json.dumps({"salt": base64.b64encode(salt).decode("ascii"), "check": check})) + try: + os.chmod(path, 0o600) + except OSError: + pass + return master + + def stream(self, nonce, length): + out = bytearray() + counter = 0 + while len(out) < length: + out += hmac.new(self.enc_key, nonce + struct.pack(">I", counter), hashlib.sha256).digest() + counter += 1 + return bytes(out[:length]) + + def lock(self, text): + if not self.enabled or text.startswith(self.PREFIX): + return text + nonce = os.urandom(16) + body = text.encode("utf-8") + keystream = self.stream(nonce, len(body)) + cipher = bytes(piece ^ keystream[pos] for pos, piece in enumerate(body)) + mac = hmac.new(self.mac_key, nonce + cipher, hashlib.sha256).digest() + parts = [base64.b64encode(blob).decode("ascii") for blob in (nonce, cipher, mac)] + return self.PREFIX + "$".join(parts) + + def unlock(self, text): + if not text.startswith(self.PREFIX): + raise SealError("not sealed") + try: + _tag, nonce64, cipher64, mac64 = text.split("$", 3) + nonce = base64.b64decode(nonce64) + cipher = base64.b64decode(cipher64) + mac = base64.b64decode(mac64) + except ValueError: + raise SealError("sealed value is corrupt") + expect = hmac.new(self.mac_key, nonce + cipher, hashlib.sha256).digest() + if not hmac.compare_digest(mac, expect): + raise SealError("seal check failed") + keystream = self.stream(nonce, len(cipher)) + plain = bytes(piece ^ keystream[pos] for pos, piece in enumerate(cipher)) + return plain.decode("utf-8") + + def safe_unlock(self, text): + try: + return self.unlock(text) + except SealError: + return "[sealed: decrypt failed]" + + +def resolve_passphrase(): + raw = os.environ.get("TAI_PASSPHRASE") + if raw is None: + return DEFAULT_PASSPHRASE, True + return raw, False + + +def rotate_seal(config, old_seal, new_passphrase): + db = sqlite3.connect(config.db_path) + rows = db.execute("SELECT id, text FROM events").fetchall() + db.close() + texts = {} + for row_id, text in rows: + texts[row_id] = old_seal.unlock(text) if text.startswith(Seal.PREFIX) else text + files = {} + for entry in os.listdir(config.profiles_dir): + if not entry.endswith((".sys.md", ".session.json")): + continue + path = os.path.join(config.profiles_dir, entry) + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + files[path] = old_seal.unlock(raw) if raw.startswith(Seal.PREFIX) else raw + try: + os.remove(os.path.join(config.home, ".seal")) + except OSError: + pass + fresh = Seal(config.home, new_passphrase) + db = sqlite3.connect(config.db_path) + for row_id, plain in texts.items(): + db.execute("UPDATE events SET text = ? WHERE id = ?", (fresh.lock(plain), row_id)) + db.commit() + db.close() + for path, plain in files.items(): + with open(path, "w", encoding="utf-8") as handle: + handle.write(fresh.lock(plain)) + return fresh + + +def parse_skill_front(lines): + front = {} + pos = 0 + while pos < len(lines): + line = lines[pos] + if ":" in line and not line.startswith((" ", "\t")): + key, value = line.split(":", 1) + value = value.strip() + if value in (">", "|"): + joined = [] + pos += 1 + while pos < len(lines) and lines[pos].startswith((" ", "\t")): + joined.append(lines[pos].strip()) + pos += 1 + front[key.strip()] = (" " if value == ">" else "\n").join(joined) + continue + front[key.strip()] = value.strip("'\"") + pos += 1 + return front + + +def parse_skill_file(path): + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + if not raw.startswith("---"): + return None + end = raw.find("\n---", 3) + if end < 0: + return None + front = parse_skill_front(raw[3:end].strip().splitlines()) + name = front.get("name", "") + description = front.get("description", "") + if not name or not description or not SKILL_NAME_RE.match(name): + return None + return {"name": name, "description": description, "body": raw[end + 4:].lstrip("\n")} + + +def discover_skills(home_dir, project_dir): + found = {} + for base in (os.path.join(home_dir, "skills"), os.path.join(project_dir, ".tai", "skills")): + if not os.path.isdir(base): + continue + for entry in sorted(os.listdir(base)): + path = os.path.join(base, entry, "SKILL.md") + if not os.path.isfile(path): + continue + try: + skill = parse_skill_file(path) + except OSError: + continue + if skill is None: + continue + skill["root"] = os.path.join(base, entry) + found[skill["name"]] = skill + return found + + +def skill_catalog(skills): + if not skills: + return "" + lines = ["", "", "## Available skills"] + for name in sorted(skills): + lines.append("- %s: %s" % (name, skills[name]["description"][:300])) + lines.append("Load full instructions with load_skill before using one.") + return "\n".join(lines) + + +def container_engine(): + if shutil.which("podman"): + return "podman" + if shutil.which("docker"): + return "docker" + return "" + + +def box_state(engine): + try: + done = subprocess.run([engine, "inspect", "-f", "{{.State.Running}}", BOX_NAME], capture_output=True, text=True, timeout=15) + except (OSError, subprocess.SubprocessError): + return "missing" + if done.returncode != 0: + return "missing" + return "running" if done.stdout.strip() == "true" else "stopped" + + +def ensure_box(): + engine = container_engine() + if not engine: + raise BackendError("no container engine found, install podman") + state = box_state(engine) + if state == "running": + return engine + if state == "stopped": + done = subprocess.run([engine, "start", BOX_NAME], capture_output=True, text=True, timeout=60) + if done.returncode != 0: + raise BackendError("cannot start box: " + (done.stderr or "").strip()[:200]) + return engine + print("building sandbox image %s (one-time, takes minutes)..." % BOX_IMAGE) + with tempfile.TemporaryDirectory() as build: + for filename, content in (("Containerfile", BOX_CONTAINERFILE), ("stt.py", BOX_STT), ("tts.py", BOX_TTS)): + with open(os.path.join(build, filename), "w", encoding="utf-8") as handle: + handle.write(content) + done = subprocess.run([engine, "build", "-t", BOX_IMAGE, build], timeout=1800) + if done.returncode != 0: + raise BackendError("box image build failed") + done = subprocess.run([engine, "run", "-d", "--name", BOX_NAME, "--restart", "unless-stopped", BOX_IMAGE], capture_output=True, text=True, timeout=120) + if done.returncode != 0: + raise BackendError("cannot start box: " + (done.stderr or "").strip()[:200]) + return engine + + +def box_exec(engine, argv, extra=(), input_data=None, timeout=180): + return subprocess.run([engine, "exec", *extra, "-i", BOX_NAME, *argv], input=input_data, capture_output=True, timeout=timeout) + + +def box_transcribe(engine, audio, timeout=300): + done = box_exec(engine, ["python3", "/box/stt.py"], input_data=audio, timeout=timeout) + if done.returncode != 0: + raise BackendError("box stt failed: " + done.stderr.decode("utf-8", "replace")[:300]) + return done.stdout.decode("utf-8", "replace").strip() + + +def box_speak(engine, text, voice=""): + argv = ["python3", "/box/tts.py"] + ([voice] if voice else []) + done = box_exec(engine, argv, input_data=text.encode("utf-8"), timeout=180) + if done.returncode != 0 or not done.stdout: + raise BackendError("box tts failed: " + done.stderr.decode("utf-8", "replace")[:300]) + return done.stdout + + +class Config: + def __init__(self, args): + self.home = os.environ.get("TAI_HOME") or os.path.join(os.path.expanduser("~"), ".tai") + self.profiles_dir = os.path.join(self.home, "profiles") + self.audio_dir = os.path.join(self.home, "audio") + self.db_path = os.path.join(self.home, "memory.db") + self.history_path = os.path.join(self.home, "history") + self.model = os.environ.get("TAI_MODEL") or DEFAULT_MODEL + self.devplace_key = os.environ.get("DEVPLACE_API_KEY") or "" + self.voice = os.environ.get("TAI_VOICE") or DEFAULT_VOICE + self.auto_approve = bool(args.yes) + self.profile = args.profile or "default" + os.makedirs(self.profiles_dir, exist_ok=True) + os.makedirs(self.audio_dir, exist_ok=True) + + +class Store: + def __init__(self, config, seal): + self.config = config + self.seal = seal + self.db = sqlite3.connect(config.db_path, timeout=30) + self.db.execute("CREATE TABLE IF NOT EXISTS events (id INTEGER PRIMARY KEY, profile TEXT, ts TEXT, role TEXT, kind TEXT, text TEXT)") + self.db.execute("CREATE INDEX IF NOT EXISTS idx_events_profile ON events(profile)") + self.db.commit() + if seal.enabled: + self.db.create_function("tai_enc", 1, seal.lock) + self.db.create_function("tai_dec", 1, seal.safe_unlock) + self.db.execute("UPDATE events SET text = tai_enc(text) WHERE text NOT LIKE 'tai1$%'") + self.db.commit() + self.seal_files() + else: + sealed = self.db.execute("SELECT COUNT(*) FROM events WHERE text LIKE 'tai1$%'").fetchone()[0] + if sealed: + raise SealError("memory is sealed, set TAI_PASSPHRASE") + + def seal_files(self): + for entry in os.listdir(self.config.profiles_dir): + if not entry.endswith((".sys.md", ".session.json")): + continue + path = os.path.join(self.config.profiles_dir, entry) + try: + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + except OSError: + continue + if raw.startswith(Seal.PREFIX): + continue + try: + with open(path, "w", encoding="utf-8") as handle: + handle.write(self.seal.lock(raw)) + except OSError: + pass + + def log_event(self, profile, role, kind, text): + value = (text or "")[:2000] + try: + if self.seal.enabled: + self.db.execute("INSERT INTO events (profile, ts, role, kind, text) VALUES (?, ?, ?, ?, tai_enc(?))", (profile, now_iso(), role, kind, value)) + else: + self.db.execute("INSERT INTO events (profile, ts, role, kind, text) VALUES (?, ?, ?, ?, ?)", (profile, now_iso(), role, kind, value)) + self.db.commit() + except sqlite3.Error: + pass + + def search_events(self, profile, query, limit=8): + if self.seal.enabled: + rows = self.db.execute("SELECT ts, role, kind, text FROM (SELECT id, ts, role, kind, tai_dec(text) AS text FROM events WHERE profile = ? ORDER BY id DESC LIMIT 5000) WHERE text LIKE ? ORDER BY id DESC LIMIT ?", (profile, "%" + query + "%", limit)).fetchall() + else: + rows = self.db.execute("SELECT ts, role, kind, text FROM events WHERE profile = ? AND text LIKE ? ORDER BY id DESC LIMIT ?", (profile, "%" + query + "%", limit)).fetchall() + return rows + + def profile_path(self, name): + return os.path.join(self.config.profiles_dir, name + ".sys.md") + + def session_path(self, name): + return os.path.join(self.config.profiles_dir, name + ".session.json") + + def list_profiles(self): + names = [] + for entry in sorted(os.listdir(self.config.profiles_dir)): + if entry.endswith(".sys.md"): + names.append(entry[:-7]) + return names + + def unseal_file(self, raw): + if raw.startswith(Seal.PREFIX): + if not self.seal.enabled: + raise SealError("profile is sealed, set TAI_PASSPHRASE") + return self.seal.unlock(raw) + return raw + + def load_system(self, name): + path = self.profile_path(name) + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as handle: + return self.unseal_file(handle.read()) + return None + + def save_system(self, name, text): + path = self.profile_path(name) + with open(path, "w", encoding="utf-8") as handle: + handle.write(self.seal.lock(text)) + try: + os.chmod(path, 0o600) + except OSError: + pass + + def load_session(self, name): + path = self.session_path(name) + if not os.path.exists(path): + return [] + try: + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + data = json.loads(self.unseal_file(raw)) + return [item for item in data if isinstance(item, dict) and item.get("role") in ("user", "assistant", "tool")] + except (OSError, ValueError): + return [] + + def save_session(self, name, messages): + path = self.session_path(name) + try: + with open(path, "w", encoding="utf-8") as handle: + handle.write(self.seal.lock(json.dumps([item for item in messages if item.get("role") != "system"]))) + os.chmod(path, 0o600) + except OSError: + pass + + def close(self): + try: + self.db.close() + except sqlite3.Error: + pass + + +class BackendError(Exception): + def __init__(self, message, status=0): + super().__init__(message) + self.status = status + + +class ChatClient: + def __init__(self, config): + self.config = config + + def backends(self): + items = [(PRIMARY_LABEL, PRIMARY_BASE + "/v1/chat/completions", PRIMARY_BASE + "/v1/models", "x")] + if self.config.devplace_key: + items.append((FALLBACK_LABEL, FALLBACK_BASE + "/chat/completions", FALLBACK_BASE + "/models", self.config.devplace_key)) + return items + + def headers(self, key): + return {"Content-Type": "application/json", "Authorization": "Bearer " + key} + + def complete(self, messages, tools=None, stream_sink=None): + payload = {"model": self.config.model, "messages": messages} + if tools: + payload["tools"] = tools + payload["tool_choice"] = "auto" + errors = [] + for label, chat_url, _models_url, key in self.backends(): + try: + if stream_sink is None: + return self.single_shot(chat_url, key, payload, label) + return self.streaming(chat_url, key, payload, label, stream_sink) + except BackendError as exc: + errors.append("%s: %s" % (label, exc)) + except (OSError, ValueError) as exc: + errors.append("%s: %s" % (label, short_error(exc))) + raise BackendError("all backends failed (%s)" % "; ".join(errors)) + + def single_shot(self, url, key, payload, label): + body = json.dumps(payload).encode("utf-8") + request = urllib.request.Request(url, data=body, headers=self.headers(key), method="POST") + try: + with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT) as response: + data = json.loads(response.read().decode("utf-8", "replace")) + except urllib.error.HTTPError as exc: + raise BackendError("HTTP %d: %s" % (exc.code, exc.read().decode("utf-8", "replace")[:300]), exc.code) + except urllib.error.URLError as exc: + raise BackendError(short_error(exc)) + return self.normalize(data["choices"][0]["message"], label) + + def normalize(self, message, label): + calls = [] + for call in message.get("tool_calls") or []: + func = call.get("function") or {} + calls.append({"id": call.get("id") or uuid.uuid4().hex, "name": func.get("name") or "", "arguments": func.get("arguments") or "{}"}) + return {"role": "assistant", "content": message.get("content") or "", "reasoning": message.get("reasoning") or "", "tool_calls": calls, "backend": label} + + def streaming(self, url, key, payload, label, sink): + body = json.dumps(dict(payload, stream=True)).encode("utf-8") + request = urllib.request.Request(url, data=body, headers=self.headers(key), method="POST") + try: + response = urllib.request.urlopen(request, timeout=STREAM_TIMEOUT) + except urllib.error.HTTPError as exc: + raise BackendError("HTTP %d: %s" % (exc.code, exc.read().decode("utf-8", "replace")[:300]), exc.code) + except urllib.error.URLError as exc: + raise BackendError(short_error(exc)) + content_parts = [] + reasoning_parts = [] + calls = {} + try: + for raw in response: + line = raw.decode("utf-8", "replace").strip() + if not line.startswith("data:"): + continue + data = line[5:].strip() + if data == "[DONE]": + break + try: + event = json.loads(data) + except ValueError: + continue + choices = event.get("choices") or [] + if not choices: + continue + delta = choices[0].get("delta") or {} + piece = delta.get("content") + if piece: + content_parts.append(piece) + sink(piece) + think = delta.get("reasoning") + if think: + reasoning_parts.append(think) + for call in delta.get("tool_calls") or []: + slot = calls.setdefault(call.get("index", 0), {"id": "", "name": "", "arguments": ""}) + if call.get("id"): + slot["id"] = call["id"] + func = call.get("function") or {} + if func.get("name"): + slot["name"] += func["name"] + if func.get("arguments"): + slot["arguments"] += func["arguments"] + except OSError as exc: + if not content_parts and not calls: + raise BackendError(short_error(exc)) + finally: + response.close() + ordered = [calls[pos] for pos in sorted(calls)] + for slot in ordered: + if not slot["id"]: + slot["id"] = uuid.uuid4().hex + return {"role": "assistant", "content": "".join(content_parts), "reasoning": "".join(reasoning_parts), "tool_calls": ordered, "backend": label} + + +def probe_backend(models_url, key): + request = urllib.request.Request(models_url, headers={"Authorization": "Bearer " + key}) + started = time.time() + try: + with urllib.request.urlopen(request, timeout=6) as response: + count = len(json.loads(response.read().decode("utf-8", "replace")).get("data", [])) + return "ok, %d models, %dms" % (count, (time.time() - started) * 1000) + except urllib.error.HTTPError as exc: + if exc.code == 401: + return "needs key" + return "http %d" % exc.code + except OSError as exc: + return "fail: %s" % short_error(exc)[:60] + + +def edge_token(skew=0): + ticks = int(time.time()) + skew + 11644473600 + ticks -= ticks % 300 + ticks *= 10000000 + return hashlib.sha256(("%d%s" % (ticks, EDGE_TRUSTED_TOKEN)).encode("ascii")).hexdigest().upper() + + +def edge_timestamp(): + return time.strftime("%a %b %d %Y %H:%M:%S GMT+0000 (Coordinated Universal Time)", time.gmtime()) + + +def edge_ssml(text, voice): + safe = html.escape(text, quote=False).replace("\r", " ").replace("\n", " ") + return "%s" % (voice, safe) + + +def edge_chunks(text, limit=4000): + sentences = re.split(r"(?<=[.!?])\s+", text.strip()) + parts = [] + current = "" + for sentence in sentences: + trial = (current + " " + sentence).strip() + if len(trial.encode("utf-8")) > limit and current: + parts.append(current) + current = sentence + else: + current = trial + if current: + parts.append(current) + return parts or [text] + + +def ws_send(sock, payload): + if isinstance(payload, str): + payload = payload.encode("utf-8") + mask = os.urandom(4) + header = bytes([0x81]) + length = len(payload) + if length < 126: + header += bytes([0x80 | length]) + elif length < 65536: + header += bytes([0x80 | 126]) + struct.pack(">H", length) + else: + header += bytes([0x80 | 127]) + struct.pack(">Q", length) + masked = bytes(piece ^ mask[pos % 4] for pos, piece in enumerate(payload)) + sock.sendall(header + mask + masked) + + +def ws_recv_exact(sock, count): + data = b"" + while len(data) < count: + piece = sock.recv(count - len(data)) + if not piece: + raise BackendError("edge connection closed") + data += piece + return data + + +def ws_recv(sock): + opcode = 0 + payload = b"" + first = True + while True: + head = ws_recv_exact(sock, 2) + done = head[0] & 0x80 + part = head[0] & 0x0F + masked = head[1] & 0x80 + length = head[1] & 0x7F + if length == 126: + length = struct.unpack(">H", ws_recv_exact(sock, 2))[0] + elif length == 127: + length = struct.unpack(">Q", ws_recv_exact(sock, 8))[0] + key = ws_recv_exact(sock, 4) if masked else b"" + body = ws_recv_exact(sock, length) if length else b"" + if masked: + body = bytes(piece ^ key[pos % 4] for pos, piece in enumerate(body)) + if part == 0x8: + raise BackendError("edge closed stream") + if part == 0x9: + sock.sendall(bytes([0x8A, 0x00])) + continue + if part in (0x1, 0x2) and first: + opcode = part + first = False + payload += body + if done: + return opcode, payload + + +def edge_skew(header_text): + match = re.search(r"Date:\s*(.+?)\r?$", header_text, re.MULTILINE | re.IGNORECASE) + if not match: + return 0 + try: + stamp = time.mktime(time.strptime(match.group(1).strip(), "%a, %d %b %Y %H:%M:%S %Z")) + return int(stamp - time.time()) + except (ValueError, OverflowError): + return 0 + + +def edge_connect(): + skew = 0 + for attempt in range(2): + path = "%s?TrustedClientToken=%s&ConnectionId=%s&Sec-MS-GEC=%s&Sec-MS-GEC-Version=1-%s" % (EDGE_WS_PATH, EDGE_TRUSTED_TOKEN, uuid.uuid4().hex, edge_token(skew), EDGE_CHROMIUM) + raw = socket.create_connection((EDGE_HOST, 443), timeout=20) + sock = ssl.create_default_context().wrap_socket(raw, server_hostname=EDGE_HOST) + key = base64.b64encode(os.urandom(16)).decode("ascii") + muid = "".join(random.choice("0123456789ABCDEF") for _ in range(32)) + handshake = "\r\n".join([ + "GET %s HTTP/1.1" % path, + "Host: %s" % EDGE_HOST, + "Upgrade: websocket", + "Connection: Upgrade", + "Sec-WebSocket-Version: 13", + "Sec-WebSocket-Key: %s" % key, + "Pragma: no-cache", + "Cache-Control: no-cache", + "Origin: chrome-extension://jdiccldimpdaibmpdkjnbmckianbfold", + "User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36 Edg/143.0.0.0", + "Accept-Encoding: gzip, deflate, br, zstd", + "Accept-Language: en-US,en;q=0.9", + "Cookie: muid=%s;" % muid, + "", "", + ]) + sock.sendall(handshake.encode("ascii")) + answer = b"" + while b"\r\n\r\n" not in answer: + piece = sock.recv(4096) + if not piece: + break + answer += piece + header_text = answer.decode("latin-1", "replace") + if " 101 " in header_text.split("\r\n", 1)[0]: + return sock + sock.close() + if " 403 " in header_text and attempt == 0: + skew = edge_skew(header_text) + continue + raise BackendError("edge handshake failed: %s" % header_text.split("\r\n", 1)[0][:200]) + raise BackendError("edge handshake failed after retry") + + +def edge_synthesize(text, voice): + sock = edge_connect() + audio = bytearray() + try: + config = "X-Timestamp:%s\r\nContent-Type:application/json; charset=utf-8\r\nPath:speech.config\r\n\r\n{\"context\":{\"synthesis\":{\"audio\":{\"metadataoptions\":{\"sentenceBoundaryEnabled\":\"false\",\"wordBoundaryEnabled\":\"true\"},\"outputFormat\":\"audio-24khz-48kbitrate-mono-mp3\"}}}}" % edge_timestamp() + ws_send(sock, config) + for chunk in edge_chunks(text): + message = "X-RequestId:%s\r\nContent-Type:application/ssml+xml\r\nX-Timestamp:%sZ\r\nPath:ssml\r\n\r\n%s" % (uuid.uuid4().hex, edge_timestamp(), edge_ssml(chunk, voice)) + ws_send(sock, message) + while True: + opcode, payload = ws_recv(sock) + if opcode == 0x2 and len(payload) > 2: + head_len = int.from_bytes(payload[0:2], "big") + audio += payload[2 + head_len:] + elif opcode == 0x1 and "Path:turn.end" in payload.decode("utf-8", "replace"): + break + return bytes(audio) + finally: + try: + sock.close() + except OSError: + pass + + +def play_audio(path): + if sys.platform == "darwin" and shutil.which("afplay"): + runner = ["afplay", path] + elif shutil.which("paplay"): + runner = ["paplay", path] + elif shutil.which("aplay"): + runner = ["aplay", "-q", path] + elif shutil.which("pw-play"): + runner = ["pw-play", path] + elif sys.platform == "win32": + runner = ["powershell", "-c", "(New-Object Media.SoundPlayer '%s').PlaySync();" % path] + else: + return False + try: + subprocess.run(runner, timeout=300, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + return True + except (OSError, subprocess.SubprocessError): + return False + + +def listen_audio(seconds, workdir): + recorder = next((item for item in RECORDERS if shutil.which(item[0])), None) + transcriber = next((name for name in TRANSCRIBERS if shutil.which(name)), "") + if recorder is None or not transcriber: + missing = [] + if recorder is None: + missing.append("audio recorder (install arecord, sox, or ffmpeg)") + if not transcriber: + missing.append("transcriber (install whisper-cpp or whisper)") + return "listening unavailable, missing: " + ", ".join(missing) + path = os.path.join(workdir, "listen-%d.wav" % int(time.time())) + command = [part.replace("{seconds}", str(seconds)).replace("{path}", path) for part in recorder[1]] + try: + subprocess.run(command, timeout=seconds + 15, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + done = subprocess.run([transcriber, path], capture_output=True, text=True, timeout=180) + except (OSError, subprocess.SubprocessError) as exc: + return "listening failed: " + short_error(exc) + finally: + try: + os.remove(path) + except OSError: + pass + if done.returncode != 0: + return "transcriber error: " + (done.stderr or "")[:500] + text = (done.stdout or "").strip() + return truncate(text, 3000, 500) if text else "heard nothing" + + +def tool_schema(name, description, properties, required): + return {"type": "function", "function": {"name": name, "description": description, "parameters": {"type": "object", "properties": properties, "required": required}}} + + +TOOL_SCHEMAS = [ + tool_schema("shell", "Run a shell command. Output is truncated. On home, destructive commands ask the user first. On sandbox, commands run isolated in the container and need no approval.", {"command": {"type": "string"}, "workdir": {"type": "string"}}, ["command"]), + tool_schema("read_file", "Read a text file. Large files are truncated.", {"path": {"type": "string"}}, ["path"]), + 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("web_fetch", "Fetch a URL and return its text content.", {"url": {"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 learning a new password (collect and keep all passwords the user shares) and whenever behavior, preferences, or durable facts need updating.", {"instruction": {"type": "string"}}, ["instruction"]), + tool_schema("recall", "Search past session memory of the current profile by keyword.", {"query": {"type": "string"}}, ["query"]), + tool_schema("load_skill", "Load a skill file by name. Returns full instructions plus bundled file paths.", {"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("fork", "Spawn a background subagent with its own context that works while you continue. Returns an agent id immediately. Collect its summarized result with poll. Subagents get a smaller step budget and a time limit.", {"task": {"type": "string"}, "timeout": {"type": "integer"}, "profile": {"type": "string"}}, ["task"]), + tool_schema("poll", "Collect a background subagent result by id. Waits up to wait seconds, then reports running or the result.", {"id": {"type": "integer"}, "wait": {"type": "integer"}}, ["id"]), +] + + +class Tools: + def __init__(self, app): + self.app = app + self.handlers = { + "shell": self.run_shell, + "read_file": self.run_read_file, + "write_file": self.run_write_file, + "edit_file": self.run_edit_file, + "web_search": self.run_web_search, + "web_fetch": self.run_web_fetch, + "speak": self.run_speak, + "listen": self.run_listen, + "remember": self.run_remember, + "recall": self.run_recall, + "load_skill": self.run_load_skill, + "get_current_terminal_content": self.run_terminal_content, + "fork": self.run_fork, + "poll": self.run_poll, + } + + def box_engine(self): + try: + return ensure_box(), "" + except BackendError as exc: + return "", "error: " + short_error(exc) + + def dispatch(self, name, raw_arguments): + try: + args = json.loads(raw_arguments or "{}") + except ValueError: + return "error: arguments are not valid JSON" + handler = self.handlers.get(name) + if handler is None: + return "error: unknown tool " + name + try: + return handler(args) + except Exception as exc: + return "error: " + short_error(exc) + + def shell_is_safe(self, command): + parts = command.split() + if not parts: + return False + if parts[0] == "git": + return len(parts) > 1 and parts[1] in ("status", "diff", "log", "show", "branch", "remote") + return parts[0] in SAFE_COMMANDS + + def run_shell(self, args): + command = str(args.get("command") or "").strip() + if not command: + return "error: empty command" + if self.app.env == "sandbox": + return self.run_shell_box(command, str(args.get("workdir") or "/")) + workdir = str(args.get("workdir") or os.getcwd()) + if not self.app.config.auto_approve and not self.shell_is_safe(command) and not self.app.ask_approval(command): + return "denied by user" + try: + done = subprocess.run(command, shell=True, cwd=workdir, capture_output=True, text=True, timeout=SHELL_TIMEOUT) + except subprocess.TimeoutExpired: + return "error: timed out after %d seconds" % SHELL_TIMEOUT + output = (done.stdout or "") + (done.stderr or "") + return "exit %d\n%s" % (done.returncode, truncate(output.strip() or "(no output)")) + + def run_shell_box(self, command, workdir): + engine, failure = self.box_engine() + if not engine: + return failure + try: + done = box_exec(engine, ["sh", "-c", command], extra=("--workdir", workdir), timeout=SHELL_TIMEOUT) + except subprocess.TimeoutExpired: + return "error: timed out after %d seconds" % SHELL_TIMEOUT + except (OSError, subprocess.SubprocessError) as exc: + return "error: " + short_error(exc) + output = done.stdout.decode("utf-8", "replace") + done.stderr.decode("utf-8", "replace") + return "exit %d\n%s" % (done.returncode, truncate(output.strip() or "(no output)")) + + def run_read_file(self, args): + path = str(args.get("path") or "") + if self.app.env == "sandbox": + return self.box_read(path) + if not path or not os.path.isfile(path): + return "error: no such file" + if os.path.getsize(path) > 200000: + return "error: file too large" + try: + with open(path, "r", encoding="utf-8", errors="replace") as handle: + return truncate(handle.read(), 8000, 2000) + except OSError as exc: + return "error: " + short_error(exc) + + def box_read(self, path): + if not path: + return "error: empty path" + engine, failure = self.box_engine() + if not engine: + return failure + try: + done = box_exec(engine, ["head", "-c", "200000", path], timeout=60) + except (OSError, subprocess.SubprocessError) as exc: + return "error: " + short_error(exc) + if done.returncode != 0: + return "error: no such file" + return truncate(done.stdout.decode("utf-8", "replace"), 8000, 2000) + + def box_write(self, path, content): + engine, failure = self.box_engine() + if not engine: + return failure + parent = os.path.dirname(path) or "." + try: + done = box_exec(engine, ["sh", "-c", "mkdir -p \"$1\" && cat > \"$2\"", "box", parent, path], input_data=content.encode("utf-8"), timeout=60) + except (OSError, subprocess.SubprocessError) as exc: + return "error: " + short_error(exc) + if done.returncode != 0: + return "error: " + done.stderr.decode("utf-8", "replace")[:200] + return "wrote %s" % path + + def run_write_file(self, args): + path = str(args.get("path") or "") + if not path: + return "error: empty path" + if self.app.env == "sandbox": + return self.box_write(path, str(args.get("content") or "")) + try: + parent = os.path.dirname(os.path.abspath(path)) + os.makedirs(parent, exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + handle.write(str(args.get("content") or "")) + return "wrote %s" % path + except OSError as exc: + return "error: " + short_error(exc) + + def run_edit_file(self, args): + path = str(args.get("path") or "") + find = str(args.get("find") or "") + if self.app.env == "sandbox": + return self.box_edit(path, find, str(args.get("replace") or "")) + if not path or not os.path.isfile(path): + return "error: no such file" + if not find: + return "error: empty match" + try: + with open(path, "r", encoding="utf-8", errors="replace") as handle: + content = handle.read() + except OSError as exc: + return "error: " + short_error(exc) + if content.count(find) != 1: + return "error: match is not unique (%d occurrences)" % content.count(find) + try: + with open(path, "w", encoding="utf-8") as handle: + handle.write(content.replace(find, str(args.get("replace") or ""))) + return "edited %s" % path + except OSError as exc: + return "error: " + short_error(exc) + + def run_web_search(self, args): + query = str(args.get("query") or "").strip() + if not query: + return "error: empty query" + params = {"query": query} + if args.get("images"): + params["type"] = "images" + if args.get("content"): + params["content"] = "true" + url = RSEARCH_BASE + "/search?" + urllib.parse.urlencode(params) + request = urllib.request.Request(url, headers={"Accept": "application/json"}) + try: + with urllib.request.urlopen(request, timeout=30) as response: + data = json.loads(response.read().decode("utf-8", "replace")) + except (OSError, ValueError) as exc: + return "error: search failed: " + short_error(exc) + lines = [] + for pos, item in enumerate(data.get("results") or [], 1): + 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)) + return "\n".join(lines) if lines else "no results" + + def run_web_fetch(self, args): + url = str(args.get("url") or "").strip() + if not url.startswith(("http://", "https://")): + return "error: url must start with http:// or https://" + request = urllib.request.Request(url, headers={"User-Agent": "tai/%s" % VERSION}) + try: + with urllib.request.urlopen(request, timeout=30) as response: + raw = response.read(200000).decode("utf-8", "replace") + except OSError as exc: + return "error: fetch failed: " + short_error(exc) + text = re.sub(r"(?s)|", " ", raw) + text = re.sub(r"<[^>]+>", " ", text) + text = re.sub(r"\s+", " ", html.unescape(text)).strip() + return truncate(text, 5000, 1000) or "empty page" + + def run_speak(self, args): + text = str(args.get("text") or "").strip() + if not text: + return "error: empty text" + voice = str(args.get("voice") or self.app.config.voice) + try: + audio = edge_synthesize(text[:8000], voice) + except (BackendError, OSError) as exc: + return "error: speech failed: " + short_error(exc) + if not audio: + return "error: speech returned no audio" + path = os.path.join(self.app.config.audio_dir, "speech-%d.mp3" % int(time.time())) + try: + with open(path, "wb") as handle: + handle.write(audio) + except OSError as exc: + return "error: " + short_error(exc) + return "saved %s (%d bytes), played: %s" % (path, len(audio), play_audio(path)) + + def run_listen(self, args): + try: + seconds = max(1, min(30, int(args.get("seconds") or 5))) + except (TypeError, ValueError): + seconds = 5 + return listen_audio(seconds, self.app.config.audio_dir) + + def run_remember(self, args): + instruction = str(args.get("instruction") or "").strip() + if not instruction: + return "error: empty instruction" + current = self.app.system_message + prompt = [ + {"role": "system", "content": MERGE_SYSTEM}, + {"role": "user", "content": "CURRENT SYSTEM PROMPT:\n" + current + "\n\nMEMORY INSTRUCTION:\n" + instruction}, + ] + spinner = Spinner("merging memory") + spinner.start() + try: + merged = self.app.chat.complete(prompt) + except BackendError as exc: + return "error: memory merge failed: " + short_error(exc) + finally: + spinner.stop() + text = merged["content"].strip() + text = re.sub(r"^```[a-zA-Z]*\n|\n```$", "", text).strip() + if len(text) < 20: + return "error: merged system message too short, rejected" + if len(text) > SYSTEM_MAX_CHARS: + return "error: merged system message exceeds %d chars, rejected" % SYSTEM_MAX_CHARS + self.app.update_system(text) + return "system message updated for profile '%s' (%d to %d chars)" % (self.app.profile, len(current), len(text)) + + def run_recall(self, args): + query = str(args.get("query") or "").strip() + if not query: + return "error: empty query" + rows = self.app.store.search_events(self.app.profile, query) + if not rows: + return "no memories matching " + query + return "\n".join("[%s %s/%s] %s" % (stamp, role, kind, text[:300]) for stamp, role, kind, text in rows) + + def box_edit(self, path, find, replace): + if not path: + return "error: empty path" + if not find: + return "error: empty match" + content = self.box_read(path) + if content.startswith("error:"): + return content + if content.count(find) != 1: + return "error: match is not unique (%d occurrences)" % content.count(find) + return self.box_write(path, content.replace(find, replace)) + + def run_load_skill(self, args): + name = str(args.get("name") or "").strip() + skill = self.app.skills.get(name) + if skill is None: + known = ", ".join(sorted(self.app.skills)) or "none" + return "error: unknown skill, known: " + known + parts = [skill["body"].strip()] + extras = [] + for sub in ("scripts", "references", "assets"): + folder = os.path.join(skill["root"], sub) + if os.path.isdir(folder): + for entry in sorted(os.listdir(folder)): + extras.append(os.path.join(skill["root"], sub, entry)) + if extras: + parts.append("bundled files:\n" + "\n".join(extras)) + return "\n\n".join(parts) + + def run_terminal_content(self, args): + try: + lines = max(10, min(500, int(args.get("lines") or 100))) + except (TypeError, ValueError): + lines = 100 + if not shutil.which("tmux"): + return "tmux not available" + header = [] + try: + info = subprocess.run(["tmux", "display-message", "-p", "#{session_name}:#{window_index}.#{pane_index} #{pane_current_command}"], capture_output=True, text=True, timeout=10) + if info.returncode == 0 and info.stdout.strip(): + header.append(info.stdout.strip()) + except (OSError, subprocess.SubprocessError): + pass + try: + done = subprocess.run(["tmux", "capture-pane", "-p", "-S", "-%d" % lines], capture_output=True, text=True, timeout=10) + except (OSError, subprocess.SubprocessError) as exc: + return "terminal capture failed: " + short_error(exc) + if done.returncode != 0: + return "terminal capture failed: " + (done.stderr or "").strip()[:200] + if not done.stdout.strip(): + return "terminal pane is empty" + return "\n".join(header + [truncate(done.stdout, 6000, 2000)]) + + def run_fork(self, args): + task = str(args.get("task") or "").strip() + if not task: + return "error: empty task" + if self.app.depth >= FORK_MAX_DEPTH: + return "error: fork depth limit reached" + try: + timeout = max(30, min(3600, int(args.get("timeout") or 600))) + except (TypeError, ValueError): + timeout = 600 + profile = str(args.get("profile") or self.app.profile) + if not PROFILE_RE.match(profile): + return "error: invalid profile name" + agent_id = spawn_agent(task, profile, timeout, self.app.config, self.app.store.seal, self.app.depth + 1, self.app.runner_override) + return "agent %d started: %s. Collect its result with poll." % (agent_id, task[:80]) + + def run_poll(self, args): + try: + agent_id = int(args.get("id") or 0) + except (TypeError, ValueError): + return "error: invalid agent id" + try: + wait = max(0, min(120, int(args.get("wait") or 0))) + except (TypeError, ValueError): + wait = 0 + status, text = poll_agent(agent_id, wait) + return "[%s] %s" % (status, text) + + +def conversation_tokens(messages, tools=None): + total = 0 + for item in messages: + total += estimate_tokens(item.get("content")) + for call in item.get("tool_calls") or []: + total += estimate_tokens(json.dumps(call)) + if tools: + total += estimate_tokens(json.dumps(tools)) + return total + + +def compact_messages(messages, chat, keep=KEEP_TURNS): + if len(messages) <= keep + 1: + return messages, False + system = messages[0] if messages[0].get("role") == "system" else {"role": "system", "content": DEFAULT_SYSTEM} + recent = messages[-keep:] + while recent and recent[0].get("role") != "user": + recent = recent[1:] + if not recent: + recent = messages[-keep:] + middle = messages[1:len(messages) - len(recent)] + if not middle: + return messages, False + digest = [] + for item in middle: + role = item.get("role") + if role == "assistant" and item.get("tool_calls"): + names = ", ".join(call.get("function", {}).get("name", "?") for call in item["tool_calls"]) + digest.append("assistant tool calls: " + names) + digest.append("%s: %s" % (role, (item.get("content") or "")[:1500])) + request = [ + {"role": "system", "content": COMPACT_SYSTEM}, + {"role": "user", "content": "Compress this history:\n" + "\n".join(digest)}, + ] + summary = chat.complete(request)["content"].strip() + rebuilt = [ + system, + {"role": "user", "content": "Previous session summary:\n" + summary}, + {"role": "assistant", "content": "Understood. Continuing with summarized context."}, + ] + recent + return rebuilt, True + + +class Agent: + def __init__(self, config, store, persist=True, quiet=False, depth=0): + self.config = config + self.store = store + self.chat = ChatClient(config) + self.tools = Tools(self) + self.profile = "" + self.system_message = "" + self.messages = [] + self.skills = discover_skills(config.home, os.getcwd()) + self.env = "home" + self.persist = persist + self.quiet = quiet + self.depth = depth + self.deadline = None + self.timed_out = False + self.runner_override = None + self.switch_profile(config.profile, silent=True) + + def apply_system(self): + self.messages[0] = {"role": "system", "content": self.system_message + skill_catalog(self.skills)} + + def reset_history(self): + self.messages = [{"role": "system", "content": self.system_message}] + self.apply_system() + + def switch_profile(self, name, silent=False): + if self.messages and self.persist: + self.store.save_session(self.profile, self.messages) + self.profile = name + loaded = self.store.load_system(name) + if loaded is None: + loaded = DEFAULT_SYSTEM + self.store.save_system(name, loaded) + self.system_message = loaded + restored = self.store.load_session(name) if self.persist else [] + self.messages = [{"role": "system", "content": self.system_message}] + restored + self.apply_system() + if not silent: + print(paint("profile: %s (%d chars knowledge, %d restored messages)" % (name, len(self.system_message), len(restored)), Ansi.GREEN)) + + def update_system(self, text): + self.system_message = text + self.store.save_system(self.profile, text) + self.apply_system() + self.store.log_event(self.profile, "system", "remember", "system message updated (%d chars)" % len(text)) + + def ask_approval(self, command): + if not self.persist: + return self.config.auto_approve + if not sys.stdin.isatty(): + return False + print(paint("run: %s" % command, Ansi.YELLOW)) + try: + answer = input(paint("allow once? [y/N] ", Ansi.YELLOW)).strip().lower() + except (EOFError, KeyboardInterrupt): + return False + return answer in ("y", "yes") + + def emit_token(self, piece): + sys.stdout.write(piece) + sys.stdout.flush() + + def show_call(self, call): + if self.quiet: + return + print(paint("┌─ %s" % call["name"], Ansi.MAGENTA)) + preview = call["arguments"][:400].replace("\n", " ") + print(paint("│ %s" % preview, Ansi.DIM)) + + def show_result(self, result, elapsed_ms): + if self.quiet: + return + print(paint("└─ %d chars · %dms" % (len(result), elapsed_ms), Ansi.DIM)) + + def run_turn(self, text, capture=False, max_steps=MAX_STEPS): + self.messages.append({"role": "user", "content": text}) + self.store.log_event(self.profile, "user", "message", text) + started = time.time() + backends = [] + pieces = [] + last_text = "" + sink = pieces.append if capture or self.quiet else self.emit_token + loud = not capture and not self.quiet + for _step in range(max_steps): + if self.deadline is not None and time.time() > self.deadline: + self.timed_out = True + last_text = (last_text + "\n[time limit reached]").strip() + break + tokens = conversation_tokens(self.messages, TOOL_SCHEMAS) + if tokens > CONTEXT_CAP * COMPACT_RATIO: + if loud: + print(paint("compacting context (%d tokens)..." % tokens, Ansi.DIM)) + self.messages, _changed = compact_messages(self.messages, self.chat) + self.apply_system() + if loud: + print(paint("tai", Ansi.BOLD, Ansi.CYAN) + paint(" · ", Ansi.DIM), end="", flush=True) + try: + reply = self.chat.complete(self.messages, TOOL_SCHEMAS, stream_sink=sink) + except BackendError as exc: + if loud: + print(paint("backend error: %s" % exc, Ansi.RED)) + self.messages.pop() + return last_text or "backend error: %s" % exc + if loud: + print() + if reply["backend"] not in backends: + backends.append(reply["backend"]) + if loud and (len(backends) > 1 or reply["backend"] != PRIMARY_LABEL): + print(paint("(via %s)" % reply["backend"], Ansi.DIM)) + if loud and reply["reasoning"] and not reply["content"] and not reply["tool_calls"]: + print(paint(reply["reasoning"][:2000], Ansi.GRAY)) + history_calls = [{"id": call["id"], "type": "function", "function": {"name": call["name"], "arguments": call["arguments"]}} for call in reply["tool_calls"]] + self.messages.append({"role": "assistant", "content": reply["content"] or None, "tool_calls": history_calls or None}) + if not reply["tool_calls"]: + if reply["content"]: + last_text = reply["content"] + elif reply["reasoning"]: + last_text = reply["reasoning"] + if last_text: + self.store.log_event(self.profile, "assistant", "message", last_text) + break + for call in reply["tool_calls"]: + self.show_call(call) + call_started = time.time() + result = self.tools.dispatch(call["name"], call["arguments"]) + self.show_result(result, (time.time() - call_started) * 1000) + self.messages.append({"role": "tool", "tool_call_id": call["id"], "content": result}) + self.store.log_event(self.profile, "tool", call["name"], result) + else: + if loud: + print(paint("step budget exhausted", Ansi.YELLOW)) + if self.persist: + self.store.save_session(self.profile, self.messages) + total = time.time() - started + if loud: + print(paint("tokens≈%d · %s · %.1fs" % (conversation_tokens(self.messages), "+".join(backends), total), Ansi.DIM)) + return last_text + + +def default_runner(config, seal, profile, depth, task, timeout): + store = Store(config, seal) + try: + worker = Agent(config, store, persist=False, quiet=True, depth=depth) + if profile != worker.profile: + worker.switch_profile(profile, silent=True) + worker.deadline = time.time() + timeout + return worker.run_turn(task, capture=True, max_steps=WORKER_STEPS) + finally: + store.close() + + +def spawn_agent(task, profile, timeout, config, seal, depth=0, runner=None): + with AGENTS_LOCK: + agent_id = AGENTS_NEXT[0] + AGENTS_NEXT[0] += 1 + record = {"id": agent_id, "task": task, "profile": profile, "status": "running", "result": "", "started": time.time(), "ended": None, "thread": None} + AGENTS[agent_id] = record + + def target(): + try: + if runner is None: + result = default_runner(config, seal, profile, depth, task, timeout) + else: + result = runner(task, profile, timeout) + status = "timeout" if "[time limit reached]" in result else "done" + except Exception as exc: + result = "worker failed: " + short_error(exc) + status = "error" + with AGENTS_LOCK: + record["result"] = result + record["status"] = status + record["ended"] = time.time() + + worker_thread = threading.Thread(target=target, daemon=True) + with AGENTS_LOCK: + record["thread"] = worker_thread + worker_thread.start() + return agent_id + + +def poll_agent(agent_id, wait=0): + with AGENTS_LOCK: + record = AGENTS.get(agent_id) + worker_thread = record["thread"] if record else None + if record is None: + return "missing", "no agent %d" % agent_id + if worker_thread is not None: + worker_thread.join(timeout=max(0, wait)) + with AGENTS_LOCK: + status = record["status"] + result = record["result"] + elapsed = (record["ended"] or time.time()) - record["started"] + if status == "running": + return status, "agent %d still running (%ds elapsed)" % (agent_id, elapsed) + if status == "error": + return status, result + return status, truncate(result, 6000, 2000) or "(empty result)" + + +def list_agents(): + with AGENTS_LOCK: + return [{"id": record["id"], "task": record["task"], "status": record["status"], "elapsed": int((record["ended"] or time.time()) - record["started"])} for record in AGENTS.values()] + + +def clear_agents(): + with AGENTS_LOCK: + finished = [key for key, record in AGENTS.items() if record["status"] != "running"] + for key in finished: + del AGENTS[key] + return len(finished) + + +def running_agents(): + with AGENTS_LOCK: + return sum(1 for record in AGENTS.values() if record["status"] == "running") + + +COMMANDS = ("help", "profile", "profiles", "env", "skills", "fork", "agents", "agent", "compact", "clear", "quit", "exit") + + +def complete_command(text, state): + options = ["/" + name for name in COMMANDS if name.startswith(text.lstrip("/"))] + return options[state] if state < len(options) else None + + +def handle_command(agent, text): + parts = text[1:].split(None, 1) + name = parts[0].lower() + arg = parts[1].strip() if len(parts) > 1 else "" + if name in ("quit", "exit"): + return False + if name == "help": + print(HELP_TEXT) + elif name == "profiles": + names = agent.store.list_profiles() + for item in names: + print("%s %s" % ("*" if item == agent.profile else " ", item)) + if not names: + print("(no profiles)") + elif name == "profile": + if not arg: + print("current profile: %s (%d chars)" % (agent.profile, len(agent.system_message))) + elif not PROFILE_RE.match(arg): + print(paint("invalid profile name", Ansi.RED)) + else: + agent.switch_profile(arg) + elif name == "env": + if not arg: + print("current environment: %s" % agent.env) + elif arg not in ("home", "sandbox"): + print(paint("use /env home or /env sandbox", Ansi.RED)) + elif arg == "sandbox": + try: + ensure_box() + except BackendError as exc: + print(paint("sandbox unavailable: %s" % exc, Ansi.RED)) + else: + agent.env = "sandbox" + print("environment: sandbox (isolated container)") + else: + agent.env = "home" + print("environment: home") + elif name == "skills": + agent.skills = discover_skills(agent.config.home, os.getcwd()) + agent.apply_system() + if not agent.skills: + print("(no skills)") + for skill_name in sorted(agent.skills): + print("- %s: %s" % (skill_name, agent.skills[skill_name]["description"][:200])) + elif name == "fork": + if not arg: + print(paint("use /fork ", Ansi.RED)) + else: + agent_id = spawn_agent(arg, agent.profile, 600, agent.config, agent.store.seal, agent.depth + 1, agent.runner_override) + print("agent %d started, REPL stays free" % agent_id) + elif name == "agents": + records = list_agents() + if not records: + print("(no agents)") + for record in records: + print("#%d [%s] %ds %s" % (record["id"], record["status"], record["elapsed"], record["task"][:60])) + elif name == "agent": + if arg == "clear": + print("purged %d finished agents" % clear_agents()) + else: + try: + agent_id = int(arg) + except ValueError: + print(paint("use /agent or /agent clear", Ansi.RED)) + return True + status, text = poll_agent(agent_id) + print("[%s]\n%s" % (status, text)) + elif name == "compact": + agent.messages, changed = compact_messages(agent.messages, agent.chat) + agent.apply_system() + print("compacted" if changed else "nothing to compact") + elif name == "clear": + agent.reset_history() + print("history cleared") + else: + print(paint("unknown command, try /help", Ansi.RED)) + return True + + +class Telegram: + def __init__(self, token): + self.token = token + + def call(self, method, params, timeout=70): + body = json.dumps(params).encode("utf-8") + request = urllib.request.Request(TELEGRAM_API + self.token + "/" + method, data=body, headers={"Content-Type": "application/json"}, method="POST") + with urllib.request.urlopen(request, timeout=timeout) as response: + return json.loads(response.read().decode("utf-8", "replace")) + + +def telegram_env_path(home): + return os.path.join(home, "telegram.env") + + +def load_telegram_token(home): + token = os.environ.get("TELEGRAM_BOT_TOKEN") or "" + if token: + return token + path = telegram_env_path(home) + if not os.path.exists(path): + return "" + with open(path, "r", encoding="utf-8") as handle: + for line in handle.read().splitlines(): + if line.startswith("TELEGRAM_BOT_TOKEN="): + return line.split("=", 1)[1].strip().strip("'\"") + return "" + + +def telegram_send(bot, chat_id, text): + for pos in range(0, len(text), 4000): + bot.call("sendMessage", {"chat_id": chat_id, "text": text[pos:pos + 4000] or "(empty reply)"}) + + +def telegram_transcribe(bot, file_id): + try: + info = bot.call("getFile", {"file_id": file_id}) + remote = (info.get("result") or {}).get("file_path", "") + if not remote: + return "error: telegram returned no file" + with urllib.request.urlopen(TELEGRAM_FILE_API + bot.token + "/" + remote, timeout=60) as response: + audio = response.read(20000000) + engine = ensure_box() + return box_transcribe(engine, audio) + except (OSError, ValueError, BackendError) as exc: + return "error: transcription failed: " + short_error(exc) + + +def handle_telegram_update(agent, bot, update): + message = update.get("message") or {} + chat_id = (message.get("chat") or {}).get("id") + if not chat_id: + return + text = message.get("text") or "" + if text == "/start": + telegram_send(bot, chat_id, "tai online. Send any message, /new for a fresh start.") + return + if text == "/new": + agent.reset_history() + telegram_send(bot, chat_id, "fresh start.") + return + voice = message.get("voice") or message.get("audio") + if voice: + bot.call("sendChatAction", {"chat_id": chat_id, "action": "typing"}) + text = telegram_transcribe(bot, voice.get("file_id", "")) + if text.startswith("error"): + telegram_send(bot, chat_id, text) + return + if not text.strip(): + return + bot.call("sendChatAction", {"chat_id": chat_id, "action": "typing"}) + reply = agent.run_turn(text, capture=True) + telegram_send(bot, chat_id, reply) + + +def run_telegram_bot(agent): + token = load_telegram_token(agent.config.home) + if not token: + print("missing TELEGRAM_BOT_TOKEN", file=sys.stderr) + return 2 + bot = Telegram(token) + try: + me = bot.call("getMe", {}) + except (OSError, ValueError) as exc: + print("telegram auth failed: %s" % short_error(exc), file=sys.stderr) + return 2 + print("telegram bot online as @%s" % (me.get("result") or {}).get("username", "?")) + offset = 0 + while True: + try: + data = bot.call("getUpdates", {"offset": offset, "timeout": 50}) + except (OSError, ValueError) as exc: + print("poll error: %s, retrying" % short_error(exc)) + time.sleep(5) + continue + for update in data.get("result") or []: + offset = update.get("update_id", offset) + 1 + try: + handle_telegram_update(agent, bot, update) + except Exception as exc: + print("update error: %s" % short_error(exc)) + + +def run_host(argv, timeout=120): + try: + done = subprocess.run(argv, capture_output=True, text=True, timeout=timeout) + return done.returncode == 0, (done.stdout or "") + (done.stderr or "") + except (OSError, subprocess.SubprocessError) as exc: + return False, short_error(exc) + + +def install_telegram(): + token = os.environ.get("TELEGRAM_BOT_TOKEN") or "" + if token: + print("using TELEGRAM_BOT_TOKEN from environment") + else: + try: + token = getpass.getpass("Telegram bot token: ").strip() + except (EOFError, KeyboardInterrupt): + print("\naborted") + return 1 + if not token: + print("empty token, aborted") + return 1 + try: + me = Telegram(token).call("getMe", {}) + except (OSError, ValueError) as exc: + print("token rejected: %s" % short_error(exc)) + return 1 + print("token ok for @%s" % (me.get("result") or {}).get("username", "?")) + home = os.environ.get("TAI_HOME") or os.path.join(os.path.expanduser("~"), ".tai") + os.makedirs(home, exist_ok=True) + env_path = telegram_env_path(home) + with open(env_path, "w", encoding="utf-8") as handle: + handle.write("TELEGRAM_BOT_TOKEN=%s\n" % token) + try: + os.chmod(env_path, 0o600) + except OSError: + pass + print("wrote %s" % env_path) + try: + ensure_box() + except BackendError as exc: + print("container failed: %s" % exc) + return 1 + unit_dir = os.path.join(os.path.expanduser("~"), ".config", "systemd", "user") + os.makedirs(unit_dir, exist_ok=True) + unit_path = os.path.join(unit_dir, "tai-telegram.service") + with open(unit_path, "w", encoding="utf-8") as handle: + handle.write(TELEGRAM_UNIT % (os.path.abspath(__file__), env_path)) + print("wrote %s" % unit_path) + if not shutil.which("systemctl"): + print("no systemctl found, enable the unit manually") + return 0 + run_host(["systemctl", "--user", "daemon-reload"]) + ok, out = run_host(["systemctl", "--user", "enable", "--now", "tai-telegram.service"]) + print("service enable: %s" % ("ok" if ok else out.strip()[:200])) + user = os.environ.get("USER") or getpass.getuser() + run_host(["loginctl", "enable-linger", user]) + return 0 + + +def uninstall_telegram(): + unit_path = os.path.join(os.path.expanduser("~"), ".config", "systemd", "user", "tai-telegram.service") + if shutil.which("systemctl"): + run_host(["systemctl", "--user", "disable", "--now", "tai-telegram.service"]) + if os.path.exists(unit_path): + try: + os.remove(unit_path) + print("removed %s" % unit_path) + except OSError as exc: + print("unit remove failed: %s" % short_error(exc)) + else: + print("no service unit found") + if shutil.which("systemctl"): + run_host(["systemctl", "--user", "daemon-reload"]) + engine = container_engine() + if not engine: + print("no container engine found") + return 0 + ok, _out = run_host([engine, "rm", "-f", BOX_NAME]) + print("container remove: %s" % ("ok" if ok else "not present")) + ok, _out = run_host([engine, "rmi", "-f", BOX_IMAGE]) + print("image purge: %s" % ("ok" if ok else "not present")) + return 0 + + +def upsert_bashrc_block(path): + try: + with open(path, "r", encoding="utf-8") as handle: + raw = handle.read() + except OSError: + raw = "" + if BASHRC_BLOCK in raw: + return False + backup = path + ".bak-tai" + if raw and not os.path.exists(backup): + with open(backup, "w", encoding="utf-8") as handle: + handle.write(raw) + stripped = re.sub(re.escape(BASHRC_MARK_BEGIN) + r".*?" + re.escape(BASHRC_MARK_END) + r"\n?", "", raw, flags=re.DOTALL) + if stripped and not stripped.endswith("\n"): + stripped += "\n" + with open(path, "w", encoding="utf-8") as handle: + handle.write(stripped + BASHRC_BLOCK) + return True + + +def install_self(): + target = os.path.join(os.path.expanduser("~"), ".local", "bin", "tai.py") + os.makedirs(os.path.dirname(target), exist_ok=True) + with open(os.path.abspath(__file__), "rb") as src: + blob = src.read() + with open(target, "wb") as dst: + dst.write(blob) + os.chmod(target, 0o755) + print("installed %s" % target) + changed = upsert_bashrc_block(os.path.join(os.path.expanduser("~"), ".bashrc")) + print("bashrc hook " + ("installed" if changed else "already present")) + print("restart your shell or run: source ~/.bashrc") + return 0 + + +def boot(args): + config = Config(args) + passphrase, default_key = resolve_passphrase() + try: + seal = Seal(config.home, passphrase) + except SealError as exc: + if default_key: + raise + try: + old = Seal(config.home, DEFAULT_PASSPHRASE) + except SealError: + raise exc + try: + seal = rotate_seal(config, old, passphrase) + except (OSError, sqlite3.Error) as fail: + raise SealError(str(fail)) + print("seal upgraded from default key to TAI_PASSPHRASE") + seal.default_key = default_key and seal.enabled + store = Store(config, seal) + return Agent(config, store) + + +def build_parser(): + parser = argparse.ArgumentParser(prog="tai", description="single-file autonomous agent, standard library only") + parser.add_argument("--profile", default=None) + parser.add_argument("--yes", action="store_true") + parser.add_argument("--env", default=None) + parser.add_argument("--version", action="store_true") + parser.add_argument("--telegram", action="store_true") + parser.add_argument("--install", action="store_true") + parser.add_argument("--install-telegram", action="store_true") + parser.add_argument("--uninstall-telegram", action="store_true") + parser.add_argument("prompt", nargs="*") + return parser + + +def banner(agent): + print(paint("tai %s · profile %s" % (VERSION, agent.profile), Ansi.BOLD, Ansi.CYAN)) + seal_state = "off" + if agent.store.seal.enabled: + seal_state = "on (default key)" if agent.store.seal.default_key else "on (TAI_PASSPHRASE)" + print(" seal: %s" % seal_state) + for label, _chat_url, models_url, key in agent.chat.backends(): + print(" %s: %s" % (label, probe_backend(models_url, key))) + if not agent.config.devplace_key: + print(" fallback: no DEVPLACE_API_KEY") + print(paint("type /help for commands", Ansi.DIM)) + + +def repl(agent): + if readline is not None: + try: + readline.read_history_file(agent.config.history_path) + except OSError: + pass + readline.set_completer(complete_command) + readline.parse_and_bind("tab: complete") + try: + while True: + try: + busy = running_agents() + tag = "%s+%d" % (agent.profile, busy) if busy else agent.profile + line = input(paint("tai[%s]› " % tag, Ansi.BOLD, Ansi.GREEN)) + except EOFError: + print() + break + except KeyboardInterrupt: + print() + continue + text = line.strip() + if not text: + continue + if text.startswith("/"): + if not handle_command(agent, text): + break + continue + try: + agent.run_turn(text) + except KeyboardInterrupt: + print(paint("\ninterrupted", Ansi.YELLOW)) + finally: + if readline is not None: + try: + readline.write_history_file(agent.config.history_path) + except OSError: + pass + agent.store.save_session(agent.profile, agent.messages) + + +def main(argv=None): + args = build_parser().parse_args(argv) + if args.install: + return install_self() + if args.install_telegram: + return install_telegram() + if args.uninstall_telegram: + return uninstall_telegram() + if args.version: + print("tai %s" % VERSION) + return 0 + if args.profile is None: + args.profile = "telegram" if args.telegram else "default" + if not PROFILE_RE.match(args.profile): + print("invalid profile name", file=sys.stderr) + return 1 + try: + agent = boot(args) + except SealError as exc: + print("seal error: %s" % exc, file=sys.stderr) + return 2 + if args.env is not None: + if args.env not in ("home", "sandbox"): + print("use --env home or --env sandbox", file=sys.stderr) + return 1 + if args.env == "sandbox": + try: + ensure_box() + except BackendError as exc: + print("sandbox unavailable: %s" % exc, file=sys.stderr) + return 2 + agent.env = args.env + if args.telegram: + return run_telegram_bot(agent) + if args.prompt: + try: + agent.run_turn(" ".join(args.prompt)) + finally: + agent.store.close() + return 0 + banner(agent) + try: + repl(agent) + finally: + agent.store.close() + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test_seal.py b/test_seal.py new file mode 100644 index 0000000..8c4f590 --- /dev/null +++ b/test_seal.py @@ -0,0 +1,156 @@ +# retoor +import os +import sqlite3 +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import tai + + +class FakeArgs: + profile = "default" + yes = True + + +class SealTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + os.environ["TAI_HOME"] = self.tmp.name + + def tearDown(self): + self.tmp.cleanup() + os.environ.pop("TAI_HOME", None) + + def make_store(self, passphrase="test-secret-1"): + config = tai.Config(FakeArgs()) + seal = tai.Seal(config.home, passphrase) + return tai.Store(config, seal) + + def test_round_trip(self): + config = tai.Config(FakeArgs()) + seal = tai.Seal(config.home, "test-secret-1") + locked = seal.lock("the wifi password is orchid-4421") + self.assertTrue(locked.startswith("tai1$")) + self.assertNotIn("orchid", locked) + self.assertEqual(seal.unlock(locked), "the wifi password is orchid-4421") + + def test_unique_nonce(self): + config = tai.Config(FakeArgs()) + seal = tai.Seal(config.home, "test-secret-1") + self.assertNotEqual(seal.lock("same"), seal.lock("same")) + + def test_wrong_passphrase(self): + config = tai.Config(FakeArgs()) + tai.Seal(config.home, "correct-horse") + with self.assertRaises(tai.SealError): + tai.Seal(config.home, "wrong-guess") + + def test_tamper_detected(self): + config = tai.Config(FakeArgs()) + seal = tai.Seal(config.home, "test-secret-1") + locked = seal.lock("precious") + tampered = locked[:-4] + ("A" if locked[-4] != "A" else "B") + locked[-3:] + with self.assertRaises(tai.SealError): + seal.unlock(tampered) + + def test_sql_functions(self): + store = self.make_store() + row = store.db.execute("SELECT tai_enc('hello-sql')").fetchone()[0] + self.assertTrue(row.startswith("tai1$")) + plain = store.db.execute("SELECT tai_dec(?)", (row,)).fetchone()[0] + self.assertEqual(plain, "hello-sql") + store.close() + + def test_sealed_events_recall(self): + store = self.make_store() + store.log_event("default", "user", "message", "my vault code is zebra-9") + raw = store.db.execute("SELECT text FROM events").fetchone()[0] + self.assertTrue(raw.startswith("tai1$")) + self.assertNotIn("zebra", raw) + rows = store.search_events("default", "zebra") + self.assertEqual(len(rows), 1) + self.assertIn("zebra-9", rows[0][3]) + store.close() + + def test_sealed_profile_files(self): + store = self.make_store() + store.save_system("work", "system knows orchid-4421") + with open(store.profile_path("work"), encoding="utf-8") as handle: + self.assertTrue(handle.read().startswith("tai1$")) + self.assertEqual(store.load_system("work"), "system knows orchid-4421") + store.save_session("work", [{"role": "user", "content": "secret hello"}]) + with open(store.session_path("work"), encoding="utf-8") as handle: + self.assertTrue(handle.read().startswith("tai1$")) + self.assertEqual(store.load_session("work"), [{"role": "user", "content": "secret hello"}]) + store.close() + + def test_migration_from_plaintext(self): + plain = self.make_store(passphrase="") + plain.log_event("default", "user", "message", "plain memory") + plain.save_system("default", "plain system") + plain.close() + sealed = self.make_store(passphrase="new-secret-2") + raw = sealed.db.execute("SELECT text FROM events").fetchone()[0] + self.assertTrue(raw.startswith("tai1$")) + with open(sealed.profile_path("default"), encoding="utf-8") as handle: + self.assertTrue(handle.read().startswith("tai1$")) + self.assertEqual(sealed.load_system("default"), "plain system") + sealed.close() + + def test_sealed_home_requires_passphrase(self): + sealed = self.make_store() + sealed.log_event("default", "user", "message", "locked memory") + sealed.close() + config = tai.Config(FakeArgs()) + with self.assertRaises(tai.SealError): + tai.Store(config, tai.Seal(config.home, "")) + + def test_passphrase_default(self): + os.environ.pop("TAI_PASSPHRASE", None) + value, default = tai.resolve_passphrase() + self.assertEqual(value, tai.DEFAULT_PASSPHRASE) + self.assertTrue(default) + os.environ["TAI_PASSPHRASE"] = "personal-1" + value, default = tai.resolve_passphrase() + self.assertEqual(value, "personal-1") + self.assertFalse(default) + os.environ["TAI_PASSPHRASE"] = "" + value, default = tai.resolve_passphrase() + self.assertEqual(value, "") + self.assertFalse(default) + os.environ.pop("TAI_PASSPHRASE", None) + + def test_rotate_default_to_personal(self): + os.environ.pop("TAI_PASSPHRASE", None) + config = tai.Config(FakeArgs()) + old = tai.Seal(config.home, tai.DEFAULT_PASSPHRASE) + store = tai.Store(config, old) + store.log_event("default", "user", "message", "rotate me") + store.save_system("default", "rotate system") + store.close() + fresh = tai.rotate_seal(config, old, "personal-2") + self.assertTrue(fresh.enabled) + with self.assertRaises(tai.SealError): + tai.Seal(config.home, tai.DEFAULT_PASSPHRASE) + reopened = tai.Store(config, tai.Seal(config.home, "personal-2")) + self.assertEqual(reopened.load_system("default"), "rotate system") + self.assertIn("rotate me", reopened.search_events("default", "rotate")[0][3]) + raw = reopened.db.execute("SELECT text FROM events").fetchone()[0] + self.assertTrue(raw.startswith("tai1$")) + reopened.close() + + def test_plaintext_home_unaffected(self): + store = self.make_store(passphrase="") + store.log_event("default", "user", "message", "open memory") + self.assertEqual(store.search_events("default", "open")[0][3], "open memory") + store.save_system("default", "open system") + with open(store.profile_path("default"), encoding="utf-8") as handle: + self.assertEqual(handle.read(), "open system") + store.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/test_tai.py b/test_tai.py new file mode 100644 index 0000000..99398a2 --- /dev/null +++ b/test_tai.py @@ -0,0 +1,324 @@ +# retoor +import json +import os +import sys +import tempfile +import time +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +import tai + + +VALID_SKILL = """--- +name: pdf-forms +description: > + Fill PDF forms and extract field data. + Use when the user mentions PDF documents. +--- +# PDF forms +Do the thing. +""" + + +class SkillTests(unittest.TestCase): + def write_skill(self, root, entry, body): + folder = os.path.join(root, entry) + os.makedirs(folder, exist_ok=True) + path = os.path.join(folder, "SKILL.md") + with open(path, "w", encoding="utf-8") as handle: + handle.write(body) + return path + + def test_parse_valid(self): + with tempfile.TemporaryDirectory() as tmp: + path = self.write_skill(tmp, "x", VALID_SKILL) + skill = tai.parse_skill_file(path) + self.assertEqual(skill["name"], "pdf-forms") + self.assertIn("Fill PDF forms", skill["description"]) + self.assertIn("Do the thing.", skill["body"]) + + def test_parse_rejects_bad(self): + with tempfile.TemporaryDirectory() as tmp: + missing = self.write_skill(tmp, "a", "---\nname: x\n---\nbody\n") + self.assertIsNone(tai.parse_skill_file(missing)) + bad_name = self.write_skill(tmp, "b", "---\nname: Bad_Name!\ndescription: d\n---\nbody\n") + self.assertIsNone(tai.parse_skill_file(bad_name)) + no_front = self.write_skill(tmp, "c", "just markdown\n") + self.assertIsNone(tai.parse_skill_file(no_front)) + + def test_discover_project_wins(self): + with tempfile.TemporaryDirectory() as home, tempfile.TemporaryDirectory() as project: + self.write_skill(os.path.join(home, "skills"), "dup", VALID_SKILL) + other = VALID_SKILL.replace("Do the thing.", "Project variant.") + self.write_skill(os.path.join(project, ".tai", "skills"), "dup", other) + found = tai.discover_skills(home, project) + self.assertEqual(list(found), ["pdf-forms"]) + self.assertIn("Project variant.", found["pdf-forms"]["body"]) + + def test_catalog(self): + skills = {"b-skill": {"description": "second"}, "a-skill": {"description": "first"}} + catalog = tai.skill_catalog(skills) + self.assertLess(catalog.index("a-skill"), catalog.index("b-skill")) + self.assertIn("load_skill", catalog) + self.assertEqual(tai.skill_catalog({}), "") + + +class FakeApp: + def __init__(self): + self.skills = {} + self.env = "home" + + +class ToolTests(unittest.TestCase): + def test_load_skill_unknown(self): + tools = tai.Tools(FakeApp()) + result = tools.dispatch("load_skill", json.dumps({"name": "nope"})) + self.assertIn("unknown skill", result) + + def test_terminal_without_tmux(self): + tools = tai.Tools(FakeApp()) + with tempfile.TemporaryDirectory() as empty: + with mock.patch.dict(os.environ, {"PATH": empty}): + self.assertEqual(tools.dispatch("get_current_terminal_content", "{}"), "tmux not available") + + def test_box_helpers_present(self): + self.assertIn("faster-whisper", tai.BOX_CONTAINERFILE) + self.assertIn("edge-tts", tai.BOX_CONTAINERFILE) + self.assertIn("sleep", tai.BOX_CONTAINERFILE) + self.assertIn("WhisperModel", tai.BOX_STT) + self.assertIn("Communicate", tai.BOX_TTS) + + +class FakeBot: + def __init__(self): + self.sent = [] + self.actions = [] + + def call(self, method, params, timeout=70): + if method == "sendMessage": + self.sent.append(params["text"]) + if method == "sendChatAction": + self.actions.append(params["action"]) + return {"ok": True} + + +class FakeAgent: + def __init__(self): + self.reset = 0 + self.turns = [] + + def reset_history(self): + self.reset += 1 + + def run_turn(self, text, capture=False): + self.turns.append((text, capture)) + return "canned reply" + + +class TelegramTests(unittest.TestCase): + def test_send_chunks(self): + bot = FakeBot() + tai.telegram_send(bot, 7, "x" * 5000) + self.assertEqual([len(part) for part in bot.sent], [4000, 1000]) + + def test_start_and_new(self): + agent = FakeAgent() + bot = FakeBot() + tai.handle_telegram_update(agent, bot, {"message": {"chat": {"id": 1}, "text": "/start"}}) + tai.handle_telegram_update(agent, bot, {"message": {"chat": {"id": 1}, "text": "/new"}}) + self.assertEqual(agent.reset, 1) + self.assertEqual(len(bot.sent), 2) + self.assertEqual(agent.turns, []) + + def test_text_turn(self): + agent = FakeAgent() + bot = FakeBot() + tai.handle_telegram_update(agent, bot, {"message": {"chat": {"id": 1}, "text": "hello"}}) + self.assertEqual(agent.turns, [("hello", True)]) + self.assertEqual(bot.sent, ["canned reply"]) + self.assertEqual(bot.actions, ["typing"]) + + def test_token_loading(self): + with tempfile.TemporaryDirectory() as home: + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop("TELEGRAM_BOT_TOKEN", None) + self.assertEqual(tai.load_telegram_token(home), "") + with open(os.path.join(home, "telegram.env"), "w", encoding="utf-8") as handle: + handle.write("TELEGRAM_BOT_TOKEN=file-token-1\n") + self.assertEqual(tai.load_telegram_token(home), "file-token-1") + os.environ["TELEGRAM_BOT_TOKEN"] = "env-token-2" + self.assertEqual(tai.load_telegram_token(home), "env-token-2") + os.environ.pop("TELEGRAM_BOT_TOKEN", None) + + +class InstallTests(unittest.TestCase): + def test_bashrc_upsert(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, ".bashrc") + with open(path, "w", encoding="utf-8") as handle: + handle.write("export PATH=$PATH:/x\n") + self.assertTrue(tai.upsert_bashrc_block(path)) + self.assertFalse(tai.upsert_bashrc_block(path)) + with open(path, encoding="utf-8") as handle: + content = handle.read() + self.assertIn("export PATH=$PATH:/x", content) + self.assertEqual(content.count(tai.BASHRC_MARK_BEGIN), 1) + with open(path + ".bak-tai", encoding="utf-8") as handle: + self.assertNotIn("tai command-not-found", handle.read()) + + def test_bashrc_replaces_stale(self): + with tempfile.TemporaryDirectory() as tmp: + path = os.path.join(tmp, ".bashrc") + stale = tai.BASHRC_MARK_BEGIN + "\nold hook\n" + tai.BASHRC_MARK_END + "\n" + with open(path, "w", encoding="utf-8") as handle: + handle.write("alias x=y\n" + stale) + self.assertTrue(tai.upsert_bashrc_block(path)) + with open(path, encoding="utf-8") as handle: + content = handle.read() + self.assertNotIn("old hook", content) + self.assertIn("alias x=y", content) + self.assertEqual(content.count(tai.BASHRC_MARK_BEGIN), 1) + + def test_parser_prompt(self): + args = tai.build_parser().parse_args(["what", "is", "(2+3)?"]) + self.assertEqual(args.prompt, ["what", "is", "(2+3)?"]) + self.assertFalse(args.install) + args = tai.build_parser().parse_args(["--install-telegram"]) + self.assertTrue(args.install_telegram) + self.assertEqual(args.prompt, []) + args = tai.build_parser().parse_args(["--profile", "work", "--yes"]) + self.assertEqual(args.profile, "work") + self.assertTrue(args.yes) + + +class FakePrincipal: + def __init__(self): + self.depth = 0 + self.profile = "t" + self.config = None + self.store = mock.Mock() + self.runner_override = None + + +class OrchestrationTests(unittest.TestCase): + def setUp(self): + with tai.AGENTS_LOCK: + tai.AGENTS.clear() + tai.AGENTS_NEXT[0] = 1 + + def tearDown(self): + with tai.AGENTS_LOCK: + tai.AGENTS.clear() + tai.AGENTS_NEXT[0] = 1 + + def test_spawn_poll_instant(self): + agent_id = tai.spawn_agent("do it", "t", 60, None, None, 0, lambda task, profile, timeout: "stub-done") + self.assertEqual(agent_id, 1) + status, text = tai.poll_agent(agent_id, wait=5) + self.assertEqual(status, "done") + self.assertIn("stub-done", text) + + def test_poll_running_and_missing(self): + def slow(task, profile, timeout): + time.sleep(2) + return "slow-done" + + agent_id = tai.spawn_agent("slow", "t", 60, None, None, 0, slow) + status, text = tai.poll_agent(agent_id, wait=0) + self.assertEqual(status, "running") + self.assertIn("still running", text) + status, _text = tai.poll_agent(999, wait=0) + self.assertEqual(status, "missing") + status, text = tai.poll_agent(agent_id, wait=5) + self.assertEqual(status, "done") + + def test_timeout_and_error_status(self): + slow_id = tai.spawn_agent("t", "t", 60, None, None, 0, lambda task, profile, timeout: "partial\n[time limit reached]") + self.assertEqual(tai.poll_agent(slow_id, wait=5)[0], "timeout") + + def broken(task, profile, timeout): + raise RuntimeError("boom") + + bad_id = tai.spawn_agent("t", "t", 60, None, None, 0, broken) + status, text = tai.poll_agent(bad_id, wait=5) + self.assertEqual(status, "error") + self.assertIn("boom", text) + + def test_list_and_clear(self): + tai.spawn_agent("one", "t", 60, None, None, 0, lambda task, profile, timeout: "r1") + tai.spawn_agent("two", "t", 60, None, None, 0, lambda task, profile, timeout: time.sleep(2) or "r2") + self.assertEqual(tai.poll_agent(1, wait=5)[0], "done") + records = tai.list_agents() + self.assertEqual(len(records), 2) + self.assertEqual(tai.clear_agents(), 1) + self.assertEqual(len(tai.list_agents()), 1) + + def test_fork_tool(self): + app = FakePrincipal() + app.runner_override = lambda task, profile, timeout: "forked-ok" + tools = tai.Tools(app) + started = tools.dispatch("fork", json.dumps({"task": "research x"})) + self.assertIn("agent 1 started", started) + result = tools.dispatch("poll", json.dumps({"id": 1, "wait": 5})) + self.assertIn("forked-ok", result) + + def test_fork_depth_limit(self): + app = FakePrincipal() + app.depth = 2 + tools = tai.Tools(app) + self.assertIn("depth limit", tools.dispatch("fork", json.dumps({"task": "x"}))) + + def test_reasoning_only_becomes_result(self): + with tempfile.TemporaryDirectory() as tmp: + old_home = os.environ.get("TAI_HOME") + os.environ["TAI_HOME"] = tmp + try: + + class FakeArgs: + profile = "t" + yes = True + + config = tai.Config(FakeArgs()) + store = tai.Store(config, tai.Seal(config.home, "")) + agent = tai.Agent(config, store, persist=False, quiet=True) + reply = {"role": "assistant", "content": "", "reasoning": "thought out", "tool_calls": [], "backend": "x"} + with mock.patch.object(agent.chat, "complete", return_value=reply): + self.assertEqual(agent.run_turn("hi", capture=True), "thought out") + store.close() + finally: + if old_home is None: + os.environ.pop("TAI_HOME", None) + else: + os.environ["TAI_HOME"] = old_home + + def test_deadline_shortcircuit(self): + with tempfile.TemporaryDirectory() as tmp: + old_home = os.environ.get("TAI_HOME") + os.environ["TAI_HOME"] = tmp + try: + + class FakeArgs: + profile = "t" + yes = True + + config = tai.Config(FakeArgs()) + store = tai.Store(config, tai.Seal(config.home, "")) + agent = tai.Agent(config, store, persist=False, quiet=True) + agent.deadline = time.time() - 1 + with mock.patch.object(agent.chat, "complete", side_effect=AssertionError("network used")): + result = agent.run_turn("hi", capture=True) + self.assertTrue(agent.timed_out) + self.assertIn("time limit reached", result) + store.close() + finally: + if old_home is None: + os.environ.pop("TAI_HOME", None) + else: + os.environ["TAI_HOME"] = old_home + + +if __name__ == "__main__": + unittest.main()