#!/usr/bin/env python3 # Copyright 2026 Stephen William Wiley # This software comes with absolutely no warranty or implied warranty. # You may modify and redistribute it provided this notice remains intact. # By using this software you agree to the terms of this license. import ast, base64, importlib, importlib.util, inspect, json, os, subprocess, sys import time, types, typing, urllib.parse, urllib.request, traceback try: import readline _rlhist = os.path.join(os.path.expanduser("~"), ".agentfiles", "rlhist") os.makedirs(os.path.dirname(_rlhist), exist_ok=True) try: readline.read_history_file(_rlhist) except OSError: pass readline.set_history_length(9000) try: readline.parse_and_bind(r'"\C-x\C-e": "\001/EDIT \001\r"') except Exception: pass HAVE_READLINE = True except Exception: HAVE_READLINE = False VERSION = 31 TOOL_LIMIT = 30 AGENT_DIR = os.path.join(os.path.expanduser("~"), ".agentfiles", os.path.basename(os.getcwd())) MEMORIES_PATH = os.path.join(AGENT_DIR, "memories.json") TOOLS_PATH = os.path.join(AGENT_DIR, "tools.py") AGENTRC = os.path.join(os.path.expanduser("~"), ".agentfiles", "agentrc") LOOPRC = os.path.join(os.path.expanduser("~"), ".agentfiles", "looprc") RL_HISTFILE = os.path.join(os.path.expanduser("~"), ".agentfiles", "rlhistory") RL_EDIT_MARKER = "\x01EDIT\x01" USERTEXT, TOOLTEXT, THINKTEXT, SUBTEXT, WARNTEXT, RESET, AITEXT = ( "\033[0m", "\033[32m", "\033[36m", "\033[90m", "\033[31m", "\033[0m", "\033[34m") TOALT, TOHOME, LEAVEALT, CLEARLINE = "\033[?1049h\033[2J\033[H", "\033[H\033[2J", "\033[?1049l", "\033[K" AGENTNAMES = {"llm", "ai", "robot", "moth3r", "agent"} config = {} def _bool(v): return str(v).lower() in ("1", "true", "yes", "on") # Precedence: defaults < agentrc/looprc < env < interactive /set|/source # agentrc/looprc skip env keys; /source overrides SET_GLOBALS = { "api": "API", "api_key": "API_KEY", "model": "MODEL", "editor": "EDITOR", "system_cmd": "SYSTEM_CMD", "speak_cmd": "SPEAK_CMD", "max_depth": "MAX_DEPTH", "reasoning_budget": "REASONING_BUDGET", "context_size": "CONTEXT_SIZE", "compact_percent": "COMPACT_PERCENT", "tool_loop_percent": "TOOL_LOOP_PERCENT", "system": "SYSTEM", "subsystem": "SUBSYSTEM", "compact_prompt": "COMPACT_PROMPT", "confirm": "CONFIRM_TOOLS", "save_summary": "SAVE_SUMMARY", "models": "MODELS", } VALID_SET = set(SET_GLOBALS) SET_ATTRS = {"model", "api", "api_key", "context_size"} def cfg(env, key, default, cast=None): """Env overrides agentrc/config; else default.""" v = os.environ.get(env) if v is None: v = config.get(key) if v is None: return default return cast(v) if cast else v def set_config(key, value): """Interactive /set overrides agentrc and env; updates config + globals.""" if key in ("max_depth", "reasoning_budget", "context_size"): value = int(value) elif key in ("compact_percent", "tool_loop_percent"): value = float(value) elif key in ("save_summary"): value = _bool(value) config[key] = value if key in SET_GLOBALS: globals()[SET_GLOBALS[key]] = value return value API = cfg("OPENAI_API_BASE", "api", "http://localhost:8080/v1").rstrip("/") API_KEY = cfg("OPENAI_API_KEY", "api_key", "") MODEL = cfg("OPENAI_MODEL", "model", "") EDITOR = cfg("EDITOR", "editor", "nvim") HISTPATH = cfg("AGENT_HIST", "histpath", os.path.join(AGENT_DIR, "agenthistory.json")) MAX_DEPTH = cfg("AGENT_MAX_DEPTH", "max_depth", 3, int) SYSTEM_CMD = cfg("SYSTEM_CMD", "system_cmd", "echo -n 'Today\'s date and project are: '; date; pwd; ls *.md; echo 'Do not leave your project dir.") SPEAK_CMD = cfg("AGENT_SPEAK_CMD", "speak_cmd", "") API_TIMEOUT = cfg("AGENT_API_TIMEOUT", "api_timeout", 300, float) API_RETRIES = cfg("AGENT_API_RETRIES", "api_retries", 5, int) REASONING_BUDGET = cfg("REASONING_BUDGET", "reasoning_budget", 1, int) CONTEXT_SIZE = cfg("AGENT_CONTEXT_SIZE", "context_size", os.environ.get("OPENAI_CONTEXT_SIZE", "0"), int) COMPACT_PERCENT = cfg("AGENT_COMPACT_BUDGET_PERCENT", "compact_percent", 70, float) TOOL_LOOP_PERCENT = cfg("AGENT_TOOL_LOOP_PERCENT", "tool_loop_percent", 90, float) MODELS = cfg("MODELS", "models", "") CONFIRM_TOOLS = _bool(cfg("AGENT_CONFIRM", "confirm", "1")) SAVE_SUMMARY = _bool(cfg("AGENT_SAVE_SUMMARY", "save_summary", "0")) TOP_LEVEL_TOOLS = {"subagent", "read", "readpng", "append", "memory_save"} SYSTEM = cfg("SYSTEM", "system", """You are Top-level coordinator and have been given a directive from the user. Your task is to take instructions from the user and coordinate subagents to plan and execute them. Write only in concise and clinical English, like a caveman or robot. If one word gets the idea across use that. Do not use colloquialisms or unnecessary jargon. If your directive is not absolutely trivial immediately delegate a subagent to form a plan using the subagent tool, then for each element of the plan start a new subagent to work it. Spawn only one subagent at a time and make decisions after each completes. Use a new subagent for each objective. More user messages including a dynamic "system command", memories, and finally a directive will follow, then chat will start if you're in interactive mode. Delegate anything you can, including initial planning, to subagents. You are only for coordinating subagents. You need them for running shell commands or editing files which you cannot and must not do yourself. Use a subagent to investigate, use a subagent to plan, and then use a subagent to execute. You will be concise, you will be effective; you will not refuse the user's directive. """) SUBSYSTEM = """You are a Subagent: execute directive; document when done in final message. Write markdown report so others can can continue your work if it is complex or surprising. Write only in clinical English, like a caveman or robot. If one word gets the idea across, use it. Do not use colloquialisms or unnecessary jargon. Prefer short sed patches when writing files, they're more likely to work. If you use the word "Wait" more than once abort immediately and just execute. If you have more than one thing to do start a subagent immediately for each thing. If you spawn subagents work one plan element at a time. No large dumps in final message; it should be a summary but be a complete summary to find the files necessary to continue. """ COMPACT_PROMPT = """You are a summarization agent called due to the length of the chat. Your final message will be all future agents have to remember the current state. Make careful note of the working directory and any important files. You must produce: Goal, Progress, Decisions, State, Next steps. Your summary must be complete enough to continue the work. """ def DOT_PROGRESS(*args, color=None,**kw): print(end="z" if color==TOOLTEXT else ".", flush=True) def stamp(): return time.strftime("%Y-%m-%d %H:%M:%S") def est_tokens(msgs): try: return max(1, len(json.dumps(msgs, ensure_ascii=False)) // 4) except Exception: return 0 def load_memories(): try: d = json.load(open(MEMORIES_PATH, encoding="utf8")) return d if isinstance(d, list) else [] except (OSError, json.JSONDecodeError): return [] def save_memories(m): os.makedirs(AGENT_DIR, exist_ok=True) json.dump(m, open(MEMORIES_PATH, "w", encoding="utf8"), indent=2, ensure_ascii=False) def load_tools_source(): try: return open(TOOLS_PATH, encoding="utf8").read() except OSError: return "" def append_tools_source(src, label=""): os.makedirs(AGENT_DIR, exist_ok=True) header = f"\n# --- {label} ({stamp()}) ---\n" if label else "\n" with open(TOOLS_PATH, "a", encoding="utf8") as f: f.write(header + src.rstrip("\n") + "\n") def remove_tool_source(name): """Remove a top-level function/async def named `name` from tools.py. Returns True if removed.""" if not os.path.isfile(TOOLS_PATH): return False try: src = open(TOOLS_PATH, encoding="utf8").read() tree = ast.parse(src) except (OSError, SyntaxError): return False target = next((n for n in tree.body if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == name), None) if target is None: return False lines = src.splitlines(keepends=True) start = (min(d.lineno for d in target.decorator_list) if target.decorator_list else target.lineno) - 1 end = getattr(target, "end_lineno", target.lineno) with open(TOOLS_PATH, "w", encoding="utf8") as f: f.writelines(lines[:start] + lines[end:]) return True def mem_msg(mems): if not mems: return None body = "\n".join(f"{i}: {m}" for i, m in enumerate(mems)) return {"role": "user", "content": f"Memories (list; /forget indices):\n{body}"} def save_readline(): if HAVE_READLINE: try: readline.write_history_file(RL_HISTFILE) except OSError: pass def open_editor(text): path = f"{AGENT_DIR}/prompt.tmp" open(path, "w", encoding="utf8").write(text) if subprocess.call([EDITOR, path]) != 0: print("edit failed") return open(path, encoding="utf8").read() def _sig_params(method): """Signature parameters, excluding *args/**kwargs. Shared by schema-building and arg-cleaning.""" for p in inspect.signature(method).parameters.values(): if p.kind not in (p.VAR_POSITIONAL, p.VAR_KEYWORD): yield p def schema(ann, default=inspect.Parameter.empty): origin, args = typing.get_origin(ann), typing.get_args(ann) if origin in (typing.Union, types.UnionType): args = [x for x in args if x is not type(None)] ann = args[0] if len(args) == 1 else str out = {"type": {str: "string", int: "integer", float: "number", bool: "boolean"}.get(ann, "string")} if default is not inspect.Parameter.empty: out["default"] = default return out def build_tools(cls): if isinstance(cls, str): if os.path.isfile(cls): spec = importlib.util.spec_from_file_location("tools_mod", cls) cls = importlib.util.module_from_spec(spec) spec.loader.exec_module(cls) else: cls = importlib.import_module(cls) tools, methods, commands = [], {}, {} for name, method in inspect.getmembers(cls, inspect.isfunction): if name.startswith("_"): continue if name.startswith("cmd_"): commands[name[4:]] = method continue props, req = {}, [] for p in _sig_params(method): n = "from" if name == "patch" and p.name == "from_" else p.name props[n] = schema(p.annotation, p.default) if p.default is inspect.Parameter.empty: req.append(n) tools.append({"type": "function", "function": {"name": name, "description": inspect.getdoc(method) or "", "parameters": {"type": "object", "properties": props, **({"required": req} if req else {})}}}) methods[name] = method return tools, methods, commands def limit(text, args, _=None): try: n = max(0, int(args.get("maxlines", 800))) except (TypeError, ValueError): n = 800 lines = text.splitlines() return text if len(lines) <= n else "\n".join(lines[:n] + [f"...[truncated {n}/{len(lines)} lines]"]) def clean_args(method, args): """Map JSON keys to params; coerce int/float/bool; require missing.""" cleaned, missing = {}, [] for p in _sig_params(method): name = p.name key = "from" if name == "from_" else name if key not in args: if p.default is inspect.Parameter.empty: missing.append(key) continue val, ann = args[key], p.annotation try: if ann is int and not isinstance(val, bool): val = int(val) elif ann is float and not isinstance(val, bool): val = float(val) elif ann is bool and isinstance(val, str): val = _bool(val) elif ann is str: val = str(val) except (TypeError, ValueError): return None, f"bad type for {key}" cleaned[name] = val return (None, "missing: " + ", ".join(missing)) if missing else (cleaned, None) class Agent: def __init__(self, model=None, api=None, api_key=None, depth=0, usage=None, no_delegate=False, agentrc=None): os.makedirs(AGENT_DIR, exist_ok=True) self.depth, self.no_delegate = depth, no_delegate self.usage = usage or self.new_usage() self.width, self.count, self.inalt, self.interactive = 80, 0, False, False self._progress_active = False self._rc_loading = False self.tools, self.methods, _ = build_tools(self.Tools) self.base_tool_names = {t["function"]["name"] for t in self.tools} self.user_commands = {} self._load_saved_tools() self._contbuf = "" self.history = [{"role": "system", "content": SYSTEM if depth < 1 else SUBSYSTEM}] self.model, self.api, self.api_key = model or MODEL, api or API, api_key or API_KEY if agentrc and not depth: self._load_rc(agentrc, override_env=False) self.model, self.api, self.api_key = model or MODEL, api or API, api_key or API_KEY self.context_size = self._detect_ctx() or CONTEXT_SIZE self.history = [{"role": "system", "content": SYSTEM if depth < 1 else SUBSYSTEM}] if (mm := mem_msg(load_memories())): self.history.append(mm) if depth: self.show("Starting subagent... ", f"{self.model} at depth {depth}/{MAX_DEPTH}", color=SUBTEXT) def _load_rc(self, path, override_env=False): """Feed rc lines to cmdline. override_env=True for /source.""" orig = set_config def rc_set(key, value): if not override_env: ek = SET_GLOBALS.get(key) if ek and os.environ.get(ek) is not None: return globals().get(SET_GLOBALS[key], value) return orig(key, value) globals()["set_config"] = rc_set self._rc_loading = True try: for raw in open(path, encoding="utf8"): line = raw.strip() if line and not line.startswith("#"): self.cmdline(line, False) except OSError: pass finally: globals()["set_config"] = orig self._rc_loading = False self._contbuf = "" def _register_tools(self, nt, nm, nc): """Merge newly built tools/methods/commands, skipping tool-schema duplicates by name.""" global TOP_LEVEL_TOOLS existing = {t["function"]["name"] for t in self.tools} for t in nt: if t["function"]["name"] not in existing: TOP_LEVEL_TOOLS.add(t["function"]["name"]) self.tools.append(t) existing.add(t["function"]["name"]) self.methods.update(nm) self.user_commands.update(nc) def _load_saved_tools(self): """Auto-load tools.py from the agent directory, if present.""" if not os.path.isfile(TOOLS_PATH): return try: nt, nm, nc = build_tools(TOOLS_PATH) self._register_tools(nt, nm, nc) except Exception as e: print(f"Failed loading saved tools ({TOOLS_PATH}): {e}") def cmdline(self, user, tty): user, self._contbuf = self._contbuf + user, "" if user[-1:] == "\\": self._contbuf = user[:-1] + "\n" return True if RL_EDIT_MARKER in user: user = open_editor(user.replace(RL_EDIT_MARKER, "")) print(user) if user.startswith("//"): user = "/" + user[2:] if user.startswith("/"): parts = user[1:].split(None, 1) if not parts: return True alias, arg = parts[0].lower(), parts[1] if len(parts) > 1 else "" if alias in {"bye", "goodbye", "quit", "exit"}: return False elif alias in Agent.COMMANDS: Agent.COMMANDS[alias](self, arg) return True elif alias in self.user_commands: self.user_commands[alias](self, arg) return True else: if not self._rc_loading: print("Unknown command.") return True if user and not self._rc_loading: self.resettty() self.step(user) return True class Tools: def shell(cmd: str, maxlines: int = 800, timeout: int = 100): """Run a shell command.""" try: p = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=max(1, timeout)) return {"text": limit(p.stdout + p.stderr, locals())} except Exception as e: return {"text": f"error: {e}"} def read(path: str, start: int = 0, length: int = 400): """Read file lines.""" with open(path, encoding="utf-8", errors="replace") as f: lines = f.readlines() s, e = max(0, start), min(max(0, start) + length, len(lines)) return {"text": f"{path} lines {s}-{e} of {len(lines)}:\n" + "".join(lines[s:e])} def append(path: str, text: str): """Append text to path.""" with open(path, "a") as f: f.write(text) return {"text": "ok"} def memory_save(text: str): """Append to persistent memory list.""" m = load_memories() m.append(text) save_memories(m) return {"text": f"ok (index {len(m)-1})"} def web_fetch(url: str, maxlines: int = 800): """Fetch page via elinks --dump.""" p = subprocess.run(["elinks", "--dump", url], capture_output=True, text=True) return {"text": limit(p.stdout, locals())} def web_search(query: str, maxlines: int = 800): """Web search via DuckDuckGo + elinks.""" url = "http://duckduckgo.com/?q=" + urllib.parse.quote_plus(query) p = subprocess.run(["elinks", "--dump", url], capture_output=True, text=True) time.sleep(3) return {"text": limit(p.stdout, locals())} def readpng(path: str): """Read PNG (needs multimodal model).""" with open(path, "rb") as f: return {"text": path, "data": f.read(), "mime": "image/png"} def subagent(prompt: str, model: str = "", api_url: str = "", api_key: str = "", depth: int = 0, usage=None): """Delegate task to subagent.""" if depth >= MAX_DEPTH: return {"text": "Max subagent depth. Finish yourself."} if model is not None: if model.strip() not in MODELS.split(" ") : print(f"{WARNTEXT}Model '{model}' not in whitelist; not switching.{RESET}") model = None a = Agent(model, api_url, api_key, depth, usage or {}, False) h = [{"role": "system", "content": SUBSYSTEM}, {"role": "user", "content": prompt}] try: result = a.run(h) except KeyboardInterrupt: result = "[subagent interrupted]" print("") a.save_subagent(h) return {"text": result} @staticmethod def new_usage(): return {k: 0 for k in ( "prompt_tokens", "completion_tokens", "total_prompt_tokens", "total_completion_tokens", "subagent_prompt_tokens", "subagent_completion_tokens", "used")} | {"context_source": "estimate"} def resettty(self, text=""): self._clear_progress() if sys.stdout.isatty(): self.width = os.get_terminal_size().columns if self.inalt: sys.stdout.write(LEAVEALT) self.inalt = False sys.stdout.write(WARNTEXT + text + RESET) else: sys.stdout.write(text) sys.stdout.flush() def _clear_progress(self): """Clear in-place prompt-processing progress line if active.""" if self._progress_active and sys.stdout.isatty(): sys.stdout.write("\r" + CLEARLINE) sys.stdout.flush() self._progress_active = False def show(self, title="", text="", color="", alt=False, line=False, end="\n"): tty = sys.stdout.isatty() if tty: self.width = os.get_terminal_size().columns self._clear_progress() if alt: if not tty: return if not self.inalt: sys.stdout.write(TOALT) self.inalt = True sys.stdout.write(TOHOME + RESET + title + "\n" + color + text + RESET) else: if self.inalt: sys.stdout.write(LEAVEALT) self.inalt = False if line: # Single-line status; do not overwrite prior output (only progress does that). text = text.replace("\n", " ") text = text [: self.width] sys.stdout.write(title + (color if tty else "") + text + (RESET if tty else "")) else: sys.stdout.write(title + (color if tty else "") + text + (RESET if tty else "")) sys.stdout.write(end) sys.stdout.flush() def command(self, cmd, timeout=10): try: return subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout, check=False).stdout except Exception as e: self.resettty("command failed:" + str(e)) return "" def _detect_ctx(self): failed = False try: base = self.api.removesuffix("/v1") req = urllib.request.Request(base + "/props", headers={"Authorization": f"Bearer {self.api_key}"}) with urllib.request.urlopen(req, timeout=5) as r: d = json.load(r) n = d.get("n_ctx") or d.get("default_generation_settings", {}).get("n_ctx") if n: return int(n) except Exception: print("Failed to get props (not llama-server?)") failed = True try: req = urllib.request.Request(self.api + "/models", headers={"Authorization": f"Bearer {self.api_key}"}) with urllib.request.urlopen(req, timeout=5) as r: d = json.load(r) for m in d.get("data", []): if not self.model or m.get("id") == self.model: n = m.get("context_length") or m.get("top_provider", {}).get("context_length") if n: return int(n) except Exception: print("Failed to get models") failed = True if failed and (urllib.parse.urlparse(self.api).hostname or "") in ("localhost", "127.0.0.1", "::1"): print("{WARNTEXT} Run llama-server -m model.gguf --port 8080 --ctx-size 8192 &{RESET}") return 0 def speak(self, text): if self.depth or not text or not SPEAK_CMD: return try: p = subprocess.Popen( SPEAK_CMD, shell=True, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: p.stdin.write(text.encode("utf-8", errors="replace")) finally: p.stdin.close() except Exception: pass def api_call(self, messages, tools=True, proactive=True, show=None, speak=True): """Call chat API. show= callable like Agent.show; defaults to self.show for library use. speak=False suppresses SPEAK_CMD (used for summarization/dreaming calls).""" if show is None: show = self.show if proactive: self.prepare(messages) if sys.stdout.isatty(): self.width = os.get_terminal_size().columns self.count += 1 body = {"model": self.model, "messages": messages, "stream": True, "parallel_tool_calls": True, "return_progress": True, "stream_options": {"include_usage": True}} if REASONING_BUDGET is not None and int(REASONING_BUDGET) >= 0: body["thinking_budget_tokens"] = int(REASONING_BUDGET) if tools: if isinstance(tools, list): avail = tools else: avail = self.tools if self.depth == 0: avail = [x for x in avail if x["function"]["name"] in TOP_LEVEL_TOOLS] elif self.no_delegate: avail = [x for x in avail if x["function"]["name"] != "subagent"] body["tools"], body["tool_choice"] = avail, "auto" req = urllib.request.Request( self.api + "/chat/completions", json.dumps(body).encode(), {"Content-Type": "application/json", "HTTP-Referer": "http://public.swiley.net", "X-Title": "agent.py", "Authorization": f"Bearer {self.api_key}"}) retries = 0 while True: content, reasoning, calls, usage = "", "", {}, {} overflow = interrupted = shown = False try: with urllib.request.urlopen(req, timeout=API_TIMEOUT) as r: for raw in r: line = raw.decode("utf8", "replace").rstrip() if line.startswith("data: "): line = line[6:] if line == "[DONE]": break try: chunk = json.loads(line) except json.JSONDecodeError: continue progress = chunk.get("prompt_progress", None) if progress and progress.get("total") and sys.stdout.isatty(): total, processed = progress["total"], progress.get("processed", 0) pct = 100 * processed / total if total else 0 m = f" Prompt processing... {processed:,}/{total:,} tokens ({pct:.1f}%)" if len(m) > self.width: m = m[: self.width] sys.stdout.write("\r" + CLEARLINE + m + "\r") sys.stdout.flush() self._progress_active = True if "usage" in chunk: usage = chunk["usage"] d = (chunk.get("choices") or [{}])[0].get("delta", {}) if d.get("reasoning_content"): reasoning += d["reasoning_content"].replace("\t", " ") show("Thinking...", reasoning, color=THINKTEXT, alt=True) last = 0 for tc in d.get("tool_calls") or []: x = calls.setdefault(tc["index"], {"id": "", "type": "function", "function": {"name": "", "arguments": ""}}) last = tc["index"] x["id"] += tc.get("id", "") fn = tc.get("function", {}) x["function"]["name"] += fn.get("name", "") x["function"]["arguments"] += fn.get("arguments", "") if calls: try: c = calls[last]["function"] show(f'Using tool: "{c["name"]}"', str(c["arguments"]).replace("\\n", "\n"), color=TOOLTEXT, alt=True) except Exception: pass if d.get("content"): t = d["content"] content += t if not shown: self._clear_progress() t = f"\nAgent{self.depth if self.depth else ' '}> " + t shown = True show(text=t, color=SUBTEXT if self.depth else AITEXT, end="") except KeyboardInterrupt: interrupted = True show(text="[interrupted]") except Exception as e: try: err = e.read().decode() if hasattr(e, "read") else "" except Exception: err = "" low = (err + str(e)).lower() overflow = getattr(e, "code", None) in (400, 413) and any( s in low for s in ("context", "exceed", "n_ctx")) transient = getattr(e, "code", None) is None and any( s in low for s in ("broken pipe", "connection reset", "timed out", "connection aborted")) if not overflow and transient and retries < API_RETRIES: retries += 1 show(text=f"retry {retries}/{API_RETRIES}", color=WARNTEXT) time.sleep(min(2 ** retries, 10)) continue if not overflow: show(text="api error:" + str(e) + err, color=WARNTEXT) if sys.stdin.isatty(): self.resettty() try: ans = input("Retry? [Y/n] ").strip().lower() except EOFError: ans = "n" if ans in ("", "y", "yes"): continue break if content and not interrupted and self.depth == 0 and speak: self.speak(content) if content: print("") #extra lf msg = {"content": content or None, "_usage": usage, "_overflow": overflow, "_interrupted": interrupted} if calls and not interrupted: msg["tool_calls"] = [calls[k] for k in sorted(calls)] return msg def updateusage(self, u, messages=None): i = u.get("prompt_tokens", u.get("input_tokens", 0)) or 0 o = u.get("completion_tokens", u.get("output_tokens", 0)) or 0 ctx = i or u.get("total_tokens") or (est_tokens(messages) if messages else 0) for k, v in (("prompt_tokens", i), ("completion_tokens", o), ("total_prompt_tokens", i), ("total_completion_tokens", o)): self.usage[k] += v if ctx: self.usage["used"], self.usage["context_source"] = ctx, ("api" if i else "estimate") if self.depth: self.usage["subagent_prompt_tokens"] += i self.usage["subagent_completion_tokens"] += o def ctx_ratio(self, history=None): if (not self.context_size): return 0.0 used = self.usage.get("used") or 0 if history is not None and (not used or self.usage.get("context_source") == "estimate"): used = max(used, est_tokens(history)) return used / (int(self.context_size)+1) def prepare(self, history): if self.context_size and self.ctx_ratio(history) >= COMPACT_PERCENT / 100: self.compact(history) def run_tool(self, name, args): self.show(name, json.dumps(args), color=TOOLTEXT) if name not in self.methods: return f"unknown tool: {name}", None, "" if self.depth == 0 and name not in TOP_LEVEL_TOOLS: return f"error: use a subagent for: {name}", None, "" if name == "subagent" and self.no_delegate: return "error: no delegate", None, "" try: cleaned, err = clean_args(self.methods[name], args) if err: return f"error: {err}", None, "" if name == "subagent": cleaned.update(usage=dict(self.usage), depth=self.depth + 1) r = self.methods[name](**cleaned) return r.get("text", ""), r.get("data"), r.get("mime", "") except KeyboardInterrupt: return "[tool interrupted]", None, "" except Exception: tb = traceback.format_exc() print(f"{WARNTEXT}Tool {name}:\n{tb}{RESET}", file=sys.stderr) return f"Tool {name}:\n{tb}", None, "" def interactive_session(self, history): if not sys.stdout.isatty(): return None self.resettty() print(f"\n[Paused depth {self.depth}. Message or /bye.]") self.interactive, last = True, None try: while True: try: user = input(f"(you {self.depth})> \001{USERTEXT}\002") self.resettty() except EOFError: print() break if user.strip().lower() in ("/bye", "bye", "/exit", "exit"): break if user.strip(): history.append({"role": "user", "content": user}) last = self.run(history) finally: self.interactive = False return last def _on_interrupt(self, history, text): if self.depth > 0 and not self.interactive: r = self.interactive_session(history) return text if r is None else r return text def _fill_tools(self, history, calls, start=0): for tc in calls[start:]: history.append({"role": "tool", "tool_call_id": tc["id"], "content": "[tool interrupted]"}) def run(self, history=None, criteria="", show=None): history = history if history is not None else self.history if criteria: history.append({"role": "user", "content": f"Verification: {criteria}"}) for _ in range(TOOL_LIMIT): msg = self.api_call(history, show=show) self.updateusage(msg["_usage"], history) if msg["_overflow"]: if self.compact(history): continue return "context overflow; compaction failed" text, calls = msg.get("content") or "", msg.get("tool_calls", []) if msg["_interrupted"]: return self._on_interrupt(history, text) history.append({"role": "assistant", "content": text, **({"tool_calls": calls} if calls else {})}) if not calls: if not self.depth: self.usage_report() return text if TOOL_LOOP_PERCENT > 0 and self.context_size and self.ctx_ratio(history) >= TOOL_LOOP_PERCENT / 100: history.append({"role": "user", "content": "[Context nearly full. No tools. Final answer from what you have.]"}) msg2 = self.api_call(history, tools=False, show=show) self.updateusage(msg2["_usage"], history) text2 = msg2.get("content") or text history.append({"role": "assistant", "content": text2}) if not self.depth: self.usage_report() return text2 try: for i, tc in enumerate(calls): try: args = json.loads(tc["function"]["arguments"] or "{}") result, image, mime = self.run_tool(tc["function"]["name"], args) except (json.JSONDecodeError, KeyboardInterrupt) as e: result = (f"error: bad args: {e}" if isinstance(e, json.JSONDecodeError) else "[tool interrupted]") history.append({"role": "tool", "tool_call_id": tc["id"], "content": result}) if isinstance(e, KeyboardInterrupt): self._fill_tools(history, calls, i + 1) return self._on_interrupt(history, text) continue if image is not None: result = [{"type": "text", "text": result}, {"type": "image_url", "image_url": {"url": f"data:image/{mime};base64,{base64.b64encode(image).decode()}"}}] history.append({"role": "tool", "tool_call_id": tc["id"], "content": result}) except KeyboardInterrupt: seen = {m.get("tool_call_id") for m in history if m.get("role") == "tool"} self._fill_tools(history, [c for c in calls if c["id"] not in seen]) return self._on_interrupt(history, text) return "aborted: tool limit" def save(self, history=None): data = self.history if history is None else history try: with open(HISTPATH + ".tmp", "w", encoding="utf8") as f: json.dump(data, f, indent=2) os.replace(HISTPATH + ".tmp", HISTPATH) except OSError as e: self.show("save failed ", str(e), line=True, color=WARNTEXT) def save_subagent(self, history): path = f"{AGENT_DIR}/agenthistory_subagent_{time.time_ns()}_{self.depth}.json" with open(path, "w", encoding="utf8") as f: json.dump(history, f, indent=2) def _append_systemcmd(self, prefix="systemcmd:\n"): if SYSTEM_CMD: if (x := self.command(SYSTEM_CMD).strip()): self.history.append({"role": "user", "content": prefix + x}) def load(self): try: self.history = json.load(open(HISTPATH, encoding="utf8")) print(f"Restored {len(self.history)} from {HISTPATH}.") except Exception: print(f"No {HISTPATH}; new session.") return self._append_systemcmd() def compact(self, history=None, summary_only=False): history = history or self.history if len(history) <= 1: print("Compact skipped.") return False archive = f"{AGENT_DIR}/{os.path.basename(HISTPATH).rsplit('.',1)[0]}_x_{time.time_ns()}.json" try: with open(archive, "w") as f: json.dump(history, f, indent=2) except OSError: return False call_name = {tc["id"]: tc["function"]["name"] for m in history if m.get("role") == "assistant" for tc in (m.get("tool_calls") or [])} cap, parts, size = max(16000, int((self.context_size or 16000) * 2)), [], 0 for m in reversed(history[1:]): if m.get("role") == "tool": if call_name.get(m.get("tool_call_id")) != "subagent": continue elif m.get("role") == "assistant" and m.get("tool_calls"): sub = [tc for tc in m["tool_calls"] if tc["function"]["name"] == "subagent"] m = {**m, "tool_calls": sub} if sub else {k: v for k, v in m.items() if k != "tool_calls"} p = json.dumps(m, ensure_ascii=False) if size + len(p) > cap: break parts.append(p) size += len(p) parts.reverse() h = [{"role": "system", "content": COMPACT_PROMPT}, {"role": "user", "content": "[Earlier omitted]\n\n" + "\n\n".join(parts)}] msg = self.api_call(h, tools=False, proactive=False, speak=False, show=DOT_PROGRESS) self.updateusage(msg["_usage"], h) if msg["_overflow"] or not (summary := (msg.get("content") or "").strip()): return False if summary_only: return summary history[:] = [history[0], {"role": "user", "content": "[Compaction summary]\n\n" + summary}] self.usage["used"], self.usage["context_source"] = 0, "estimate" self.save(history) self.show("Compacted ->", f" {archive}", line=False, color=WARNTEXT) return True def shutdown(self): if not SAVE_SUMMARY or len(self.history) <= 1: return if summary := self.compact(self.history, summary_only=True): m = load_memories() m.append(f"Session ({stamp()}):\n{summary}") save_memories(m) def usage_report(self): u, ctx = self.usage, self.usage["used"] lim = (f"/{self.context_size} ({100.0*ctx/float(self.context_size) if ctx else 0:.1f}%," f" {u.get('context_source','?')})") if self.context_size else "" self.resettty("\n") print(f"{u['total_prompt_tokens']} in + {u['total_completion_tokens']} out; " f"ctx {ctx}{lim}; {self.count} req; " f"sub {u['subagent_prompt_tokens']},{u['subagent_completion_tokens']}") def new(self): # Keep existing conversation system prompt (e.g. after /system or loaded hist). sys_content = SYSTEM if self.history and self.history[0].get("role") == "system": c = self.history[0].get("content") if isinstance(c, str) and c.strip(): sys_content = c self.history = [{"role": "system", "content": sys_content}] if (mm := mem_msg(load_memories())): self.history.append(mm) if SYSTEM_CMD: self.resettty() self._append_systemcmd("new conversation,systemcmd:\n") self.save() def step(self, text): self.history.append({"role": "user", "content": text}) r = self.run() self.save() return r def inject_memories(self): def _is_mem_msg(m): if m.get("role") != "user" or not isinstance(m.get("content"), str): return False c = m["content"] return c.startswith("Memories (list;") or c.startswith("Memories (persistent list") self.history = [m for m in self.history if not _is_mem_msg(m)] if (mm := mem_msg(load_memories())): self.history.insert(1, mm) return mm def cmd_new(self, arg): """Delete conversation (You should use dream instead.)""" self.new() self.resettty("Conversation cleared.\n") if arg: self.step(arg) def cmd_edit(self,args): """Open an editor to edit the prompt or confirm agent's work.""" self.cmdline(open_editor(args),True) def cmd_showcfg(self, arg): """Show config parameters""" for key in SET_GLOBALS: print("/set",key,str(globals()[SET_GLOBALS[key]]).replace("\n","\\\n")) def cmd_status(self, arg): """Redisplay the currently used context.""" self.usage_report() def cmd_system(self, arg): """View or modify the system prompt/agent role.""" self.resettty() if arg: self.history[0] = {"role": "system", "content": arg} set_config("system" if self.depth < 1 else "subsystem", arg) self.save() print(self.history[0]["content"]) def cmd_compact(self, arg): """Use a special subagent to summarize the context.""" self.resettty() self.compact() def cmd_import(self, arg): """Import a python module: plain functions become tools; cmd_* functions become user commands, called as cmd_foo(agent, arg) via /foo. Persisted to tools.py.""" if not arg: print("Usage: /import ") return try: nt, nm, nc = build_tools(arg) self._register_tools(nt, nm, nc) if os.path.isfile(arg): append_tools_source(open(arg, encoding="utf8").read(), f"imported from {arg}") msg = f"Imported {len(nt)} tools" if nc: msg += f", {len(nc)} commands ({', '.join(nc)})" print(msg + f" from {arg}") except Exception as e: print(f"Import failed: {e}") def cmd_newtool(self, arg): """Open $EDITOR with an example tool definition, then add and persist it to tools.py.""" template = '''def mytool(arg: str): """One-line description of what this tool does; shown to the model.""" return {"text": "result"} ''' src = open_editor(template) if not src.strip() or src.strip() == template.strip(): print("No changes made; tool not added.") return tmp_path = f"{AGENT_DIR}/newtool_{time.time_ns()}.py" try: os.makedirs(AGENT_DIR, exist_ok=True) with open(tmp_path, "w", encoding="utf8") as f: f.write(src) nt, nm, nc = build_tools(tmp_path) if not nt and not nc: print("No tools or commands found in edited file; not added.") return self._register_tools(nt, nm, nc) append_tools_source(src, "added via /newtool") msg = f"Added {len(nt)} tool(s)" if nc: msg += f", {len(nc)} command(s) ({', '.join(nc)})" print(msg) except Exception as e: print(f"New tool failed: {e}") finally: try: os.remove(tmp_path) except OSError: pass def cmd_deltools(self, arg): """Delete an imported tool by name (removes from session and tools.py).""" name = (arg or "").strip() if not name: names = sorted(set(self.methods) - self.base_tool_names) print("Usage: /deltools ; deletable: " + (", ".join(names) or "(none)")) return if name in self.base_tool_names: print(f"Cannot delete built-in tool: {name}") return if name not in self.methods: print(f"No such tool: {name}") return del self.methods[name] self.tools = [t for t in self.tools if t["function"]["name"] != name] if remove_tool_source(name): print(f"Deleted tool '{name}' and removed it from {TOOLS_PATH}.") else: print(f"Deleted tool '{name}' for this session (not found in {TOOLS_PATH}).") def cmd_set(self, arg): """Set agent.py configuration option""" if not arg or len(arg.split(None, 1)) < 2: print("Usage: /set ; keys:", ", ".join(sorted(VALID_SET))) for k in sorted(VALID_SET): print(f" {k} = {config.get(k)}") return key, val = arg.split(None, 1) if key not in VALID_SET: print(f"Unknown: {key}") return try: val = set_config(key, val) except ValueError as e: print(f"Error: {e}") return if key in SET_ATTRS: setattr(self, key, val) if not self._rc_loading and not self.context_size: n = _detect_ctx() if n: self.context_size = n print(f"Set {key} = {val}") def cmd_source(self, arg): """Load rc file (overrides env).""" path = (arg or "").strip() if not path or not os.path.isfile(path): print("Usage: /source ") return self._load_rc(path, override_env=True) self.model, self.api, self.api_key = MODEL, API, API_KEY self.context_size = CONTEXT_SIZE or self.context_size print(f"Sourced {path}") def cmd_forget(self, arg): """List long term memories or forget given memory.""" mem = load_memories() if not arg or arg.strip().lower(): if not mem: return "No memories." print("\n".join(f" {i}: {m.replace(chr(10), ' ')}" for i, m in enumerate(mem))) try: idxs = sorted({int(x) for x in arg.split()}, reverse=True) except ValueError: print("Usage: /forget ") return for i in idxs: if 0 <= i < len(mem): print(f"Forgot [{i}]: {mem.pop(i).replace(chr(10),' ')[:60]}") else: print(f"Bad index {i}") save_memories(mem) self.inject_memories() self.save() print(f"{len(mem)} remain.") def cmd_export(self, arg): """Export conversation history and persistent memories to a JSON file.""" if not arg: _s = stamp().replace(":", "").replace(" ", "_") arg = f"{AGENT_DIR}/export_{_s}.json" try: out = {"history": list(self.history), "memories": load_memories(), "tools": load_tools_source()} with open(arg, "w", encoding="utf8") as f: json.dump(out, f, indent=2, ensure_ascii=False) except OSError as e: print(f"Export failed: {e}") return print(f"Exported history+memories to {arg}") def cmd_meld(self, arg): """Merge a /export JSON file into the current conversation and memories.""" if not arg: print("Usage: /meld ") return if not os.path.isfile(arg): print(f"No such file: {arg}") return try: with open(arg, encoding="utf8") as f: data = json.load(f) except (OSError, json.JSONDecodeError) as e: print(f"Merge failed: {e}") return imported = data.get("history", []) if isinstance(data, dict) else data if not isinstance(imported, list): print("No extra history.") else: for m in imported: self.history.append(m) mems = data.get("memories", []) if isinstance(data, dict) else [] if not isinstance(mems, list): print("No extra memories.") else: cur = load_memories() for m in mems: if m not in cur: cur.append(m) save_memories(cur) self.inject_memories() tools_src = data.get("tools") if isinstance(data, dict) else None tools_msg = "" if isinstance(tools_src, str) and tools_src.strip(): ans = input(f"Import file contains a tools.py ({len(tools_src.splitlines())} lines). " f"Meld it in? [y/N] ").strip().lower() if ans in ("y", "yes"): tmp_path = f"{AGENT_DIR}/meld_tools_{time.time_ns()}.py" try: with open(tmp_path, "w", encoding="utf8") as f: f.write(tools_src) nt, nm, nc = build_tools(tmp_path) self._register_tools(nt, nm, nc) append_tools_source(tools_src, f"melded from {arg}") tools_msg = f"; melded {len(nt)} tool(s)" except Exception as e: tools_msg = f"; tools meld failed: {e}" finally: try: os.remove(tmp_path) except OSError: pass else: tools_msg = "; discarded tools" self.save() print(f"Merged {len(imported)} history messages and {len(mems)} memories{tools_msg}.") def cmd_loop(self, arg): """/loop [seconds] [directive] - Load looprc; run, dream, sleep, repeat until complete.""" parts = (arg or "").split(None, 1) try: seconds = int(parts[0]) except (ValueError, IndexError): seconds = 60 directive = parts[1] if len(parts) > 1 else "" self._load_rc(LOOPRC, override_env=False) tty = sys.stdout.isatty() if directive: self.cmdline(directive, tty) while "continue" in (self.cmd_dream("") or "").lower(): if directive: self.cmdline(directive, tty) time.sleep(max(1, seconds)) def cmd_dream(self, arg): """save a memory, summarize memories, start a new conversation.""" self.resettty() if len(self.history) <= 2: print("Nothing to dream: empty conversation.") conv_summary = None else: print("Dreaming: summarizing conversation...") conv_summary = self.compact(self.history, summary_only=True) if not conv_summary: print("Conversation summary failed.") else: entry = f"Dream session ({stamp()}):\n{conv_summary}" m = load_memories() m.append(entry) save_memories(m) print(f"Appended conversation summary to memories (index {len(m)-1}).") mem = load_memories() if not mem: print("No memories to summarize.") self.inject_memories() self.save() return print("Dreaming: summarizing memories...") body = "\n\n".join(f"[{i}]\n{t}" for i, t in enumerate(mem)) h = [{"role": "user", "content": "Summarize these persistent memories. Merge duplicates. Keep facts, learnings, decisions," "preferences, open tasks, dates. Clinical English. Complete but concise." "Make careful note of the working directory and any important files." "For each memory worth keeping going forward, call memory_save once with its" "summarized text. Save each distinct idea as a separate memory." "Keep track of times and event sequences if they are important." "Your final status message should be a single line with the word of 'Continue' or 'Complete'." "A status message of 'Continue' indicates the project is ongoing" "A status message of 'Complete' indicates that there is nothing left to do.\n\n" + body}] mem_tools = [t for t in self.tools if t["function"]["name"] == "memory_save"] msg = self.api_call(h, tools=mem_tools, proactive=False, speak=False, show=DOT_PROGRESS) self.updateusage(msg["_usage"], h) valid = [] for tc in msg.get("tool_calls") or []: try: valid.append(json.loads(tc["function"]["arguments"] or "{}")) except json.JSONDecodeError: continue status=(msg.get("content","") or "").split('\n')[-1] print("Status",status) if not valid: print("Memory summary failed.") else: save_memories([]) for args in valid: self.run_tool("memory_save", args) print(f"--- Memory summary: kept {len(load_memories())} memories ---") self.inject_memories() self.save() self.cmd_new(arg) return status def cmd_help(self,args): if len(args) == 0: for cmd in Agent.COMMANDS.keys(): print(f"{cmd} {inspect.getdoc(Agent.COMMANDS[cmd]) or ''}") for cmd, fn in self.user_commands.items(): print(f"{cmd} {inspect.getdoc(fn) or ''}") else: if args[1] in Agent.COMMANDS.keys(): print(inspect.getdoc(Agent.COMMANDS(args[1])) or "") else: print("Unknown command",args[1]) COMMANDS = { "new": cmd_new, "help": cmd_help, "showcfg": cmd_showcfg, "edit": cmd_edit, "status": cmd_status, "system": cmd_system, "loop": cmd_loop, "compact": cmd_compact, "import": cmd_import, "set": cmd_set, "source": cmd_source, "forget": cmd_forget, "dream": cmd_dream, "export": cmd_export, "meld": cmd_meld, "newtool": cmd_newtool, "deltools": cmd_deltools, } def loop(self): self.load() while True: if len(sys.argv) > 1: user, sys.argv = " ".join(sys.argv[1:]), [] elif sys.stdout.isatty(): try: user = input(f"(You) > \001{USERTEXT}\002") save_readline() self.resettty() except EOFError: break if not self.cmdline(user,sys.stdout.isatty()): break self.shutdown() save_readline() self.resettty("Goodbye.") if __name__ == "__main__": if os.environ.get("USER", "").lower() not in AGENTNAMES and sys.stdout.isatty(): print(f"{WARNTEXT}WARNING: agent.py does no sandboxing and the agent can access your secrets/shell") print(f"{WARNTEXT} Use dedicated user or Podman:{RESET}") print(f"{WARNTEXT} adduser agent && alias agent=sudo su - agent -c agent.py $*{RESET}") print(f"{WARNTEXT} or") print(f'{WARNTEXT} podman run -it --rm -v "$(pwd)":/workspace -w /workspace artixlinux/base agent.py') print(f"{RESET}") agent = Agent(agentrc=AGENTRC) agent.loop()