agent_core.py

#!/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__)