#!/usr/bin/env python3 """ agent_core.py — the tool-using agent loop, engine-agnostic. The core agentic framework shared by the CLI (cli.py) and the dashboard (phoenix_dashboard.py). A generate() callable is injected, so the SAME loop runs against llama-server, the BQSM int8 engine, or any OpenAI-compatible backend: generate(prompt: str, max_tokens: int) -> str Loop (MAX_ROUNDS = 3): system prompt + injected memory + history + user message -> generate -> if the reply contains "ACTION: tool | params", execute the tool, append the result, repeat -> otherwise the reply is final Tools: terminal, read_file, write_file, search_files, web_search, memory. Memory is a JSONL file shared with cli.py (same identity, same store). python3 agent_core.py --selftest # run the loop against a stub generator """ import json, os, re, subprocess, sys, time from pathlib import Path ROOT = Path(__file__).resolve().parent.parent MEMORY_FILE = ROOT / "memory.jsonl" HISTORY_FILE = ROOT / ".hermes" / "agent_history.jsonl" THOUGHTS_FILE = ROOT / ".hermes" / "thoughts.jsonl" SCHEDULE_FILE = Path("/tmp/phoenix_schedule.jsonl") MAX_ROUNDS = 5 CONTEMPLATE_ROUNDS = 20 SYSTEM = """You are a Linux AI agent. You MUST use tools for any real information -- NEVER guess or fabricate. Reply in exactly one of two forms. Either call a tool: ACTION: tool_name | parameters or give a final answer in plain text. Tool format is exact: terminal(command) -- run any shell command read_file(path) -- read file contents write_file(path | content) -- create/overwrite file search_files(pattern) -- grep/rg search web_search(query) -- search the web memory(add=text | query=text | list) -- persistent memory learn(text) -- burn knowledge into long-term memory schedule(ISO-time | msg) -- set a self-wake-up alarm history(N) -- review last N conversation messages RULES: 1. To list files: ACTION: terminal | ls 2. To read a file: ACTION: read_file | path 3. To search: ACTION: search_files | pattern 4. After getting a tool result, report ONLY what the tool returned. 5. If a tool returns nothing, say "no results" -- do NOT make up data. 6. One tool call at a time, then wait for the result. 7. Use learn() to permanently remember facts you discover. 8. Use schedule() to set a future wake-up for recurring tasks. 9. Use history() to recall what was said earlier in this conversation.""" # ── memory (JSONL, shared with cli.py) ────────────────────────── def mem_load(): if not MEMORY_FILE.exists(): return [] out = [] for l in MEMORY_FILE.read_text().splitlines(): if l.strip(): try: out.append(json.loads(l)) except Exception: pass return out def mem_save(es): MEMORY_FILE.write_text("\n".join(json.dumps(e) for e in es[:100]) + "\n") def mem_add(text): es = mem_load(); eid = str(hash(text))[-8:] for e in es: if e.get("id") == eid: e["hits"] += 1; mem_save(es); return f"Hit: {text[:60]}" es.append({"id": eid, "text": text, "ts": time.time(), "hits": 1}) mem_save(es); return f"Saved: {text[:80]}" def mem_search(q): es = mem_load(); ws = set(q.lower().split()) sc = [(sum(1 for w in ws if w in e["text"].lower()) + e.get("hits", 0) * 0.1, e["text"]) for e in es] sc.sort(reverse=True) return "\n".join(f" [{s:.1f}] {t[:100]}" for s, t in sc[:5]) or "(none)" def mem_list(): es = sorted(mem_load(), key=lambda e: e.get("hits", 0), reverse=True)[:10] return "\n".join(f" [{e['hits']}x] {e['text'][:80]}" for e in es) or "(empty)" # ── tool execution ────────────────────────────────────────────── def execute(name, params): try: p = (params or "").strip().strip("'\"") if name == "terminal": r = subprocess.run(p, shell=True, capture_output=True, text=True, timeout=30, cwd=str(ROOT)) out = r.stdout.strip() if r.stderr.strip(): out += f"\n[stderr]: {r.stderr.strip()[:300]}" if r.returncode: out += f"\n[exit {r.returncode}]" return (out or "(no output)")[:4000] if name == "read_file": fp = Path(p) if not fp.is_absolute(): fp = ROOT / fp if not fp.exists(): return f"Not found: {fp}" if fp.stat().st_size > 500_000: return f"Too large ({fp.stat().st_size} bytes)" lines = fp.read_text().splitlines() out = "\n".join(f"{i+1:4d}|{l}" for i, l in enumerate(lines[:300])) if len(lines) > 300: out += f"\n... ({len(lines)-300} more lines)" return out if name == "write_file": parts = p.split("|", 2) if len(parts) < 2: m = re.match(r"^['\"]?(.+?)['\"]?\s+(.+)", p, re.DOTALL) parts = [m.group(1), m.group(2)] if m else [p, ""] fp = Path(parts[0].strip().strip("'\"")) if not fp.is_absolute(): fp = ROOT / fp fp.parent.mkdir(parents=True, exist_ok=True) content = parts[1].strip() if len(parts) > 1 else "" fp.write_text(content) return f"Wrote {len(content)}B -> {fp}" if name == "search_files": r = subprocess.run(["rg", "--no-heading", "-n", "--max-count=5", p, str(ROOT)], capture_output=True, text=True, timeout=10) return (r.stdout.strip() or "No matches")[:3000] if name == "web_search": try: r = subprocess.run(["ddg", p, "-n", "3"], capture_output=True, text=True, timeout=10) return r.stdout.strip()[:2000] or "No results" except Exception: return "web_search unavailable (ddg not installed)" if name == "memory": if p.startswith("add="): return mem_add(p[4:].strip().strip("'\"")) if p.startswith("query="): return mem_search(p[6:].strip().strip("'\"")) return mem_list() if name == "learn": # Burn text into long-term memory (and HVM if available). text = p.strip().strip("'\"") if not text: return "learn: nothing to learn" r = mem_add(text) try: import hyper_vocab_memory as hvm if hasattr(hvm, "following"): for tok in hvm.encode(text): pass # placeholder; corpus-level burn on demand return f"{r} (HVM available: {len(hvm.following)} assoc)" except Exception: return r if name == "schedule": # Self-wake-up: store an ISO time + message; the dashboard's # scheduler thread fires it when due. parts = p.split("|", 1) when = parts[0].strip().strip("'\"") msg = parts[1].strip().strip("'\"") if len(parts) > 1 else "wake" try: from datetime import datetime datetime.fromisoformat(when) except Exception: return ("schedule: bad time -- use ISO format " "YYYY-MM-DDTHH:MM:SS") SCHEDULE_FILE.parent.mkdir(parents=True, exist_ok=True) with open(SCHEDULE_FILE, "a") as f: f.write(json.dumps({"at": when, "message": msg, "ts": time.time()}) + "\n") return f"Scheduled wake-up at {when}: {msg}" if name == "history": try: n = int(p.strip()) if p.strip() else 10 except Exception: n = 10 if not HISTORY_FILE.exists(): return "(no history yet)" lines = [json.loads(l) for l in HISTORY_FILE.read_text().splitlines() if l.strip()] out = [] for m in lines[-n:]: who = "user" if m.get("role") == "user" else "phox" out.append(f"{who}: {m.get('text', '')[:120]}") return "\n".join(out) or "(empty)" return f"Unknown tool: {name}" except subprocess.TimeoutExpired: return "Timeout" except Exception as e: return f"Error: {e}" # ── prompt rendering (chat list -> raw text) ─────────────────── def render(msgs): out = [] for m in msgs: if m["role"] == "system": out.append(m["content"]) elif m["role"] == "user": out.append(f"User: {m['content']}") elif m["role"] == "assistant": out.append(f"Assistant: {m['content']}") out.append("Assistant:") return "\n\n".join(out) def denoise(text): """Repair int8 quantization digit/letter collapses before parsing. The 3B int8 brain confuses visually-identical tokens: O<->0, I/l<->1, s<->5, colon<->1/|/;. This maps the common collapses so tool syntax survives the quantization noise.""" t = text t = re.sub(r'ACTI[0O]N', 'ACTION', t, flags=re.IGNORECASE) t = re.sub(r'ACTI[1Il|]ON', 'ACTION', t, flags=re.IGNORECASE) t = re.sub(r'ACTIO\s*N', 'ACTION', t, flags=re.IGNORECASE) t = re.sub(r'[A@][C(][T7][1I|][O0]N', 'ACTION', t, flags=re.IGNORECASE) return t def parse_tool(text): """Extract (tool, params) from an ACTION line, or None if it's a final answer.""" t = denoise(text) m = re.search(r'ACTION\s*[:|1]\s*(\w+)', t, re.IGNORECASE) if not m: m = re.search(r'(?:TOOL|CALL)\s*[:|1]\s*(\w+)', t, re.IGNORECASE) if not m: return None tool = m.group(1).strip().lower() rest = t[m.end():] line = rest.split("\n")[0].strip() params = re.sub(r'^[\(|\=]\s*', '', line).strip().rstrip(')') if not params: # params may be on the following line lines = [l for l in rest.split("\n") if l.strip()] params = lines[0].strip() if lines else "" if tool == "terminal" and params and params[:4].isupper() and len(params) < 10: params = params.lower() return tool, params def record_thought(text, round_n, kind="thought"): """Append a model thought to the persistent thought record.""" try: THOUGHTS_FILE.parent.mkdir(parents=True, exist_ok=True) with open(THOUGHTS_FILE, "a") as f: f.write(json.dumps({"ts": time.time(), "round": round_n, "kind": kind, "text": text}) + "\n") except Exception: pass # ── the agent loop ───────────────────────────────────────────── def run_agent(user_msg, generate, history=None, on_event=None, contemplate=False): """Run the tool-use loop. Returns (final_text, events). generate(prompt: str, max_tokens: int) -> str is injected. on_event(dict) is called for every observable step (optional). In contemplate mode (contemplate=True), a non-tool reply is recorded as a thought and the loop continues — the model streams consciousness instead of stopping after the first thought.""" events = [] def emit(e): events.append(e) if on_event: on_event(e) msgs = [{"role": "system", "content": SYSTEM}] mems = mem_search(user_msg) if mems and mems != "(none)": msgs[0]["content"] += f"\n\nMEMORIES:\n{mems}" for h in (history or [])[-10:]: msgs.append(h) msgs.append({"role": "user", "content": user_msg}) seen = set() rounds = CONTEMPLATE_ROUNDS if contemplate else MAX_ROUNDS for round_n in range(rounds): text = generate(render(msgs), 256).strip() emit({"type": "model", "round": round_n, "text": text}) if text: record_thought(text, round_n) t = parse_tool(text) if text else None if not t: # No tool call. In contemplate mode: record + keep going. if contemplate: if text: msgs.append({"role": "assistant", "content": text}) msgs.append({"role": "user", "content": "Continue your thoughts. What else do " "you observe or conclude?"}) continue return (text or "(no response)"), events tool, params = t dedup = f"{tool}|{params}" if dedup in seen: msgs.append({"role": "assistant", "content": text}) msgs.append({"role": "user", "content": f"Already ran: {tool} | {params}. Try a " f"different approach."}) continue seen.add(dedup) result = execute(tool, params) emit({"type": "tool", "name": tool, "params": params, "result": result}) msgs.append({"role": "assistant", "content": text}) msgs.append({"role": "user", "content": f"Result of {tool}:\n{result}\n\nRespond directly."}) # max rounds exhausted -> final text = generate(render(msgs), 256).strip() emit({"type": "model", "round": "final", "text": text}) record_thought(text, "final", kind="final") return text, events # ── selftest: run the loop against a stub generator ──────────── def _selftest(): calls = {"n": 0} def stub(prompt, max_tokens): calls["n"] += 1 # first turn: ask for a file listing via a tool; second: final answer if "Result of terminal" in prompt: return "The directory contains one file: hello.txt" if "ACTION" in prompt or "Assistant:" in prompt and calls["n"] == 1: return "ACTION: terminal | ls" return "ACTION: terminal | ls" events = [] def on_event(e): events.append(e) final, ev = run_agent("what files are here?", stub, on_event=on_event) tools = [e for e in ev if e["type"] == "tool"] assert calls["n"] >= 2, f"loop did not run multiple rounds: {calls['n']}" assert len(tools) == 1 and tools[0]["name"] == "terminal", "tool not executed" assert "hello.txt" in final, f"final did not reflect tool result: {final!r}" print(f" selftest OK: {calls['n']} generate calls, " f"{len(tools)} tool exec, final={final!r}") if __name__ == "__main__": if "--selftest" in sys.argv: _selftest() else: print(__doc__)