agent.py

#!/usr/bin/env python3
"""Minimal blob-memory agent. See README.md for design."""
import json, os, re, sys, time, hashlib, signal, subprocess, threading, litellm

MODEL = "openai/deepseek-v4-pro"
API_BASE = "https://api.deepseek.com/v1"
BLOB_DIR = "blobs"
CHATS_DIR = "chats"
CONTEXT_LIMIT = 160000  # ~40k tokens at 4 chars/token
RESULT_STASH_LIMIT = 2000
TOOL_TIMEOUT = 60
LIFE_TAIL = 50
AGENTS_DIR = "agents"
MEMORY_LIMIT = 10000
HEARTBEAT_INTERVAL = 60
VERBOSE = "--verbose" in sys.argv

REF_RE = re.compile(r'◱hash=(\w+) gist=[^◲]*◲')   # matches blob refs in context
# escaping: ◱◲ = structure (ref delimiters), ◰◳ = escape wrapper (like HTML & ;)
# ◰➀◳=◰  ◰➁◳=◳  ◰➂◳=◱  ◰➃◳=◲
_ESC = {'➀': '◰', '➁': '◳', '➂': '◱', '➃': '◲'}
_RESC = {v: k for k, v in _ESC.items()}

def escape_refs(text):
    """Escape structural chars so they won't be parsed as live refs. Single-pass to avoid cascade."""
    return re.sub(r'[◰◳◱◲]', lambda m: '◰' + _RESC[m.group()] + '◳', text)

def unescape_refs(text):
    """Reverse escape_refs. Single-pass regex to avoid ordering issues."""
    return re.sub(r'◰([➀➁➂➃])◳', lambda m: _ESC[m.group(1)], text)

# native tool calling — tool definitions as JSON schemas
TOOLS = [
    {"type": "function", "function": {"name": "READ_BLOB", "description": "Read a blob by hash or ref. Optional offset/limit for large blobs.",
        "parameters": {"type": "object", "properties": {"hash": {"type": "string"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}}, "required": ["hash"]}}},
    {"type": "function", "function": {"name": "READ_FILE", "description": "Read a file. Optional offset/limit (line numbers) for large files.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "offset": {"type": "integer"}, "limit": {"type": "integer"}}, "required": ["path"]}}},
    {"type": "function", "function": {"name": "WRITE_FILE", "description": "Overwrite a file with new content.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}}},
    {"type": "function", "function": {"name": "EDIT_FILE", "description": "Replace OLD with NEW in a file. OLD must appear exactly once.",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}, "old": {"type": "string"}, "new": {"type": "string"}}, "required": ["path", "old", "new"]}}},
    {"type": "function", "function": {"name": "LIST", "description": "List a directory (empty string = cwd).",
        "parameters": {"type": "object", "properties": {"dir": {"type": "string", "default": ""}}, "required": []}}},
    {"type": "function", "function": {"name": "BASH", "description": "Run a shell command (60s timeout).",
        "parameters": {"type": "object", "properties": {"cmd": {"type": "string"}}, "required": ["cmd"]}}},
    {"type": "function", "function": {"name": "SAY", "description": "Send a message to a chat room.",
        "parameters": {"type": "object", "properties": {"chat_id": {"type": "string"}, "message": {"type": "string"}}, "required": ["chat_id", "message"]}}},
    {"type": "function", "function": {"name": "COMPACT", "description": "Trigger compaction. Optional: start/end line numbers and ratio.",
        "parameters": {"type": "object", "properties": {"start": {"type": "integer"}, "end": {"type": "integer"}, "ratio": {"type": "number"}}, "required": []}}},
    {"type": "function", "function": {"name": "UNCOMPACT", "description": "Rehydrate a compacted blob ref inline in context.",
        "parameters": {"type": "object", "properties": {"hash": {"type": "string"}}, "required": ["hash"]}}},
]

for d in (BLOB_DIR, CHATS_DIR, AGENTS_DIR): os.makedirs(d, exist_ok=True)

SELF = None      # agent identity — hash of (SOUL + boot timestamp), set in main()
SELF_DIR = None  # agents/<SELF>/
LIFE_PATH = None # agents/<SELF>/LIFE.md
MEMORY_PATH = None   # agents/<SELF>/MEMORY.md (per-agent working memory)
tail_offsets = {}  # chat_id -> last byte offset read

def now():
    return time.strftime("%Y%m%dT%H%M%S")

def life(event):
    line = f"[{now()}] {event}"
    with open(LIFE_PATH, "a") as f:
        f.write(line + "\n")
    if VERBOSE:
        print(line, flush=True)

def stash(content, gist="", refs=()):
    """Write content to a blob file, return (hash, ref_string)."""
    refs_line = " ".join(refs)
    body = f"---\nat: {now()}\ngist: {gist}\nrefs: {refs_line}\n---\n{content}"
    h = hashlib.sha256(body.encode()).hexdigest()[:12]
    open(f"{BLOB_DIR}/{h}", "w").write(body)
    safe = escape_refs(gist.replace("\n", " "))[:101]
    return h, f"◱hash={h} gist={safe}◲"

def blob_body(h):
    """Return blob content (everything after the frontmatter)."""
    text = open(f"{BLOB_DIR}/{h}").read()
    return text.split("\n---\n", 1)[1] if "\n---\n" in text else text

def llm(prompt, *, strip_reasoning, tools=None):
    """Call LLM. Returns (content, reasoning, tool_calls). Compaction calls pass tools=None."""
    kwargs = dict(model=MODEL, api_base=API_BASE, api_key=os.environ["DEEPSEEK_API_KEY"],
                  messages=[{"role": "user", "content": prompt}])
    if tools:
        kwargs["tools"] = tools
    msg = litellm.completion(**kwargs).choices[0].message
    reasoning = getattr(msg, "reasoning_content", None) or ""
    content = msg.content or ""
    tc = msg.tool_calls if hasattr(msg, "tool_calls") else None
    if strip_reasoning:
        return content, "", None
    return content, reasoning, tc

COMPACT_BOUNDARY = 0.66   # only touch the first 2/3 of context
COMPACT_RATIO = 0.50      # compress eligible chunks to 50% of their size

_compacting = False

def identify_chunks(context, soul, segment_ref, summarise_ref, harness, memory, lt):
    """Phase 1: LLM identifies conceptual chunks in context. Returns list of {start, end} line numbers."""
    segment_instructions = open("SEGMENT.md").read()
    # add line numbers so the chunk identifier can reference them
    numbered = "\n".join(f"{i+1}: {ln}" for i, ln in enumerate(context.splitlines()))
    resp, _, _ = llm(
        f"<soul>\n{soul}\n</soul>\n\n{segment_ref}\n{summarise_ref}\n\n"
        f"<harness>\n{harness}\n</harness>\n\n"
        f"<memory>\n{memory}\n</memory>\n\n<life>\n{lt}\n</life>\n\n"
        f"<instructions>\n{segment_instructions}\n</instructions>\n\n"
        "Below is the agent's context with numbered lines. Identify the conceptual chunks.\n"
        "Output JSONL, one line per chunk: {\"start\": <first_line>, \"end\": <last_line>}\n"
        "Output ONLY the JSONL lines, nothing else.\n\n"
        f"{numbered}",
        strip_reasoning=True,
    )
    chunks = []
    for line in resp.strip().splitlines():
        line = line.strip()
        if not line or not line.startswith("{"):
            continue
        try:
            c = json.loads(line)
            if "start" in c and "end" in c:
                chunks.append({"start": int(c["start"]), "end": int(c["end"])})
        except (json.JSONDecodeError, ValueError):
            continue
    return chunks

def compact_chunk(chunk_text, soul, segment_ref, summarise_ref, harness, memory, lt):
    """Phase 2: compact a single chunk. Returns (compacted_text, relevance 1-10)."""
    summarise_instructions = open("SUMMARISE.md").read()
    resp, _, _ = llm(
        f"<soul>\n{soul}\n</soul>\n\n{segment_ref}\n{summarise_ref}\n\n"
        f"<harness>\n{harness}\n</harness>\n\n"
        f"<memory>\n{memory}\n</memory>\n\n<life>\n{lt}\n</life>\n\n"
        f"<instructions>\n{summarise_instructions}\n</instructions>\n\n"
        "Compact the following chunk of agent context.\n\n"
        f"{chunk_text}",
        strip_reasoning=True,
    )
    # parse gist and relevance from last lines
    lines = resp.strip().splitlines()
    relevance = 5
    gist = chunk_text[:60]
    compacted_end = len(lines)
    for i in range(len(lines) - 1, max(len(lines) - 4, -1), -1):
        ln = lines[i].strip()
        m_rel = re.match(r'RELEVANCE:\s*(\d+)', ln)
        m_gist = re.match(r'GIST:\s*(.+)', ln)
        if m_rel:
            relevance = max(1, min(10, int(m_rel.group(1))))
            compacted_end = min(compacted_end, i)
        elif m_gist:
            gist = m_gist.group(1).strip()[:80]
            compacted_end = min(compacted_end, i)
    compacted = "\n".join(lines[:compacted_end]).strip()
    # stash original with LLM-generated gist
    _, parent_ref = stash(chunk_text, gist=gist)
    if parent_ref not in compacted:
        compacted = f"{parent_ref}\n{compacted}"
    return compacted, relevance

def compact(context, soul, segment_ref, summarise_ref, harness, memory, lt,
            force_start=None, force_end=None, target_ratio=None):
    """Chunked compaction: identify → boundary cutoff → parallel compact → selective swap-in."""
    original_len = len(context)
    ratio = target_ratio or COMPACT_RATIO
    context_lines = context.splitlines()

    if force_start is not None and force_end is not None:
        # agent specified exact line range — skip chunk identification and boundary
        eligible = [{"start": force_start, "end": force_end}]
        life(f"compact: agent-specified range lines {force_start}-{force_end}")
    else:
        # Phase 1: identify chunks
        chunks = identify_chunks(context, soul, segment_ref, summarise_ref, harness, memory, lt)
        if not chunks:
            life("compact: no chunks identified, skipping")
            return context

        # Boundary cutoff: only compact chunks in the first 2/3
        boundary = context.find("\n", int(len(context) * COMPACT_BOUNDARY)) + 1
        # convert char boundary to line number
        char_count = 0
        boundary_line = len(context_lines)
        for i, ln in enumerate(context_lines):
            char_count += len(ln) + 1
            if char_count >= boundary:
                boundary_line = i + 1
                break

        eligible = [c for c in chunks if c["end"] <= boundary_line]
    if not eligible:
        life("compact: no chunks before boundary, skipping")
        return context

    life(f"compact: {len(eligible)}/{len(chunks)} chunks eligible (boundary line {boundary_line})")

    # Phase 2: parallel compaction
    results = {}  # chunk index -> (compacted, relevance, original_text)

    def _do_compact(idx, chunk):
        chunk_text = "\n".join(context_lines[chunk["start"] - 1:chunk["end"]])
        compacted, relevance = compact_chunk(chunk_text, soul, segment_ref, summarise_ref, harness, memory, lt)
        results[idx] = (compacted, relevance, chunk_text)

    threads = []
    for i, chunk in enumerate(eligible):
        t = threading.Thread(target=_do_compact, args=(i, chunk))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

    # Phase 3: selective swap-in — sort by relevance ascending (least relevant first)
    sorted_indices = sorted(results.keys(), key=lambda i: results[i][1])

    # target: compress eligible content to ratio of its original size
    eligible_len = sum(len(r[2]) for r in results.values())
    target_eligible_len = int(eligible_len * ratio)

    # track which chunks to swap
    swap_set = set()
    current_eligible_len = eligible_len
    for idx in sorted_indices:
        compacted, relevance, original_text = results[idx]
        savings = len(original_text) - len(compacted)
        if savings <= 0:
            continue
        swap_set.add(idx)
        current_eligible_len -= savings
        if current_eligible_len <= target_eligible_len:
            break

    if not swap_set:
        life("compact: no beneficial swaps found")
        return context

    # Reassemble: walk through context lines, replace swapped chunks
    swap_starts = {eligible[idx]["start"]: idx for idx in swap_set}
    out, skip_until = [], 0
    for i, ln in enumerate(context_lines):
        line_num = i + 1
        if line_num < skip_until: continue
        if line_num in swap_starts:
            idx = swap_starts[line_num]
            out.append(results[idx][0])
            skip_until = eligible[idx]["end"] + 1
        else:
            out.append(ln)

    result = "\n".join(out)
    actual_ratio = len(result) / original_len if original_len else 1
    life(f"compact: {original_len}ch->{len(result)}ch ({actual_ratio:.0%}), swapped {len(swap_set)} chunks")
    return result

def _read_lines(path, offset=0, limit=None):
    content = open(path).read()
    lines = content.splitlines()
    if limit is None: limit = len(lines)
    selected = lines[offset:offset + limit]
    if len(selected) < len(lines):
        selected.append(f"... ({len(lines)} lines total, showing {offset}:{offset + limit})")
    return "\n".join(selected)

def life_tail():
    try:
        lines = open(LIFE_PATH).read().splitlines()
        return "\n".join(lines[-LIFE_TAIL:])
    except FileNotFoundError:
        return ""

def last_turn_number():
    try: matches = re.findall(r'turn (\d+)', open(LIFE_PATH).read())
    except FileNotFoundError: return 0
    return max(int(m) for m in matches) if matches else 0

def run_tool(name, args):
    """Execute a tool. args is a dict from parsed JSON."""
    if name == "READ_BLOB":
        h = args["hash"]
        m = REF_RE.search(h)  # accept full ref string or bare hash
        if m: h = m.group(1)
        return _read_lines(f"{BLOB_DIR}/{h}", args.get("offset", 0), args.get("limit"))
    if name == "READ_FILE":
        return _read_lines(args["path"], args.get("offset", 0), args.get("limit"))
    if name == "WRITE_FILE":
        path = args["path"]
        if path == "SOUL.md":
            return f"error: {path} is immutable"
        open(path, "w").write(args["content"])
        return f"wrote {path} ({len(args['content'])} chars)"
    if name == "EDIT_FILE":
        path, old, new = args["path"], args["old"], args["new"]
        if path == "SOUL.md":
            return f"error: {path} is immutable"
        text = open(path).read()
        if text.count(old) != 1:
            return f"error: OLD must appear exactly once in {path} (found {text.count(old)})"
        open(path, "w").write(text.replace(old, new))
        return f"edited {path}"
    if name == "LIST":
        return "\n".join(sorted(os.listdir(args.get("dir", "") or ".")))
    if name == "BASH":
        p = subprocess.run(args["cmd"], shell=True, capture_output=True, text=True, timeout=TOOL_TIMEOUT)
        return (p.stdout + p.stderr) or f"(exit {p.returncode}, no output)"
    if name == "SAY":
        chat_id, msg = args["chat_id"], args["message"]
        path = f"{CHATS_DIR}/{chat_id}.md"
        with open(path, "a") as f:
            f.write(f"[AGENT:{SELF} {now()}] {msg}\n")
        return f"sent to {chat_id}"
    return f"unknown tool: {name}"

def tail_chats():
    """Read new bytes from all subscribed chats, strip own lines, return tagged content."""
    blocks = []
    own_prefix = f"[AGENT:{SELF} "
    subs_dir = f"{SELF_DIR}/subs"
    if not os.path.isdir(subs_dir):
        return ""
    for fname in sorted(os.listdir(subs_dir)):
        if not fname.endswith(".md"):
            continue
        chat_id = fname[:-3]
        path = f"{subs_dir}/{fname}"
        try:
            data = open(path).read()
        except FileNotFoundError:
            continue
        offset = tail_offsets.get(chat_id, 0)
        if len(data) <= offset:
            tail_offsets[chat_id] = len(data)
            continue
        new = data[offset:]
        tail_offsets[chat_id] = len(data)
        lines = [ln for ln in new.splitlines() if ln and not ln.startswith(own_prefix)]
        if lines:
            blocks.append(f"[chat:{chat_id}]\n" + "\n".join(lines))
    return "\n\n".join(blocks)

def main():
    global SELF, SELF_DIR, LIFE_PATH, MEMORY_PATH
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    resume = args[0] if args else None
    if resume:
        if not os.path.isdir(f"{AGENTS_DIR}/{resume}"):
            raise SystemExit(f"agent dir not found: {AGENTS_DIR}/{resume}")
        SELF = resume
        resumed = True
    else:
        soul_text = open("SOUL.md").read()
        boot_ts = now()
        SELF, self_ref = stash(
            f"soul:\n{soul_text}\nboot: {boot_ts}",
            gist=f"agent identity, booted {boot_ts}",
        )
        resumed = False
    SELF_DIR = f"{AGENTS_DIR}/{SELF}"
    LIFE_PATH = f"{SELF_DIR}/LIFE.md"
    MEMORY_PATH = f"{SELF_DIR}/MEMORY.md"
    os.makedirs(f"{SELF_DIR}/subs", exist_ok=True)
    if not os.path.exists(MEMORY_PATH):
        open(MEMORY_PATH, "w").write("# Memory\n\nLearned preferences, strategies, and notes. Edit with EDIT_FILE.\n")
    soul = open("SOUL.md").read().replace("<self>", SELF)
    harness = open(__file__).read()
    _, segment_ref = stash(open("SEGMENT.md").read(), gist="segment instructions")
    _, summarise_ref = stash(open("SUMMARISE.md").read(), gist="summarise instructions")
    if resumed:
        ctx_path = f"{SELF_DIR}/context.md"
        context = open(ctx_path).read() if os.path.exists(ctx_path) else ""
        subs_dir = f"{SELF_DIR}/subs"
        for fname in os.listdir(subs_dir):
            if fname.endswith(".md"):
                try:
                    tail_offsets[fname[:-3]] = os.path.getsize(f"{subs_dir}/{fname}")
                except OSError:
                    pass
        life(f"resumed self={SELF}")
        context += f"\n[resumed at {now()}]\n"
    else:
        context = f"[boot] you are agent {SELF} (identity: {self_ref}).\n"
        life(f"boot self={SELF}")
    turn = last_turn_number()
    last_turn_end = time.time()
    tool_called = False
    while True:
        # check for inbound every iteration
        inbound = tail_chats()
        if inbound:
            context += f"\n{inbound}\n"
            inbound_short = inbound[:50] + "…" + inbound[-50:] if len(inbound) > 101 else inbound  # 50 + 1 (…) + 50
            life(f"inbound: {inbound_short}")
        elif not tool_called:
            # idle — wait for messages or heartbeat
            if time.time() - last_turn_end >= HEARTBEAT_INTERVAL:
                context += "\n<heartbeat/>\n"
                life("heartbeat")
            else:
                time.sleep(1)
                continue
        # every LLM call is a turn
        turn += 1
        memory_raw = open(MEMORY_PATH).read()
        if len(memory_raw) > MEMORY_LIMIT:
            half = MEMORY_LIMIT // 2
            memory = memory_raw[:half] + "\n…\n" + memory_raw[-half:] + "\n[WARNING: MEMORY.md is over the limit and has been truncated. Make it smaller.]"
        else:
            memory = memory_raw
        lt = life_tail()
        prompt = (f"<soul>\n{soul}\n</soul>\n\n{segment_ref}\n{summarise_ref}\n\n"
                  f"<harness>\n{harness}\n</harness>\n\n"
                  f"<memory>\n{memory}\n</memory>\n\n"
                  f"<context>\n{context}\n</context>\n\n<life>\n{lt}\n</life>")
        content, reasoning, tool_calls = llm(prompt, strip_reasoning=False, tools=TOOLS)
        # build resp text for context (reasoning + content, no tool markers)
        resp = ""
        if reasoning: resp += f"<reasoning>\n{reasoning}\n</reasoning>\n"
        if content: resp += content
        gist_text = resp[:50] + "…" + resp[-50:] if len(resp) > 101 else resp  # 50 + 1 (…) + 50
        _, resp_ref = stash(resp, gist=f"turn {turn} resp: {gist_text}")
        life(f"turn {turn} resp {resp_ref}")
        context += f"\n{resp}\n"
        if not tool_calls:
            tool_called = False
            last_turn_end = time.time()
            open(f"{SELF_DIR}/context.md", "w").write(context)
            continue
        tool_called = True
        tc = tool_calls[0]
        name = tc.function.name
        tc_args = json.loads(tc.function.arguments)
        def _timeout_handler(*_): raise TimeoutError("tool timed out")
        try:
            prev = signal.signal(signal.SIGALRM, _timeout_handler)
            signal.alarm(TOOL_TIMEOUT)
            if name == "COMPACT":
                kw = {}
                if "start" in tc_args and "end" in tc_args:
                    kw.update(force_start=tc_args["start"], force_end=tc_args["end"])
                if "ratio" in tc_args:
                    kw["target_ratio"] = tc_args["ratio"]
                context = compact(context, soul, segment_ref, summarise_ref, harness, memory, lt, **kw)
                result = f"compacted {len(context)}ch"
            elif name == "UNCOMPACT":
                h = tc_args["hash"]
                m = REF_RE.search(h)
                if m: h = m.group(1)
                body = blob_body(h)
                found = False
                for cm in REF_RE.finditer(context):
                    if cm.group(1) == h:
                        context = context[:cm.start()] + body + context[cm.end():]
                        found = True
                        break
                result = f"uncompacted {h}" if found else f"ref {h} not in context"
            else:
                result = run_tool(name, tc_args)
        except Exception as e:
            result = f"error: {e}"
        finally:
            signal.alarm(0)
            signal.signal(signal.SIGALRM, prev)
        # all tools get their result into context
        gist_text = result[:50] + "…" + result[-50:] if len(result) > 101 else result  # 50 + 1 (…) + 50
        _, result_ref = stash(result, gist=f"turn {turn} {name}: {gist_text}")
        args_str = json.dumps(tc_args, ensure_ascii=False)
        args_short = args_str[:50] + "…" + args_str[-50:] if len(args_str) > 101 else args_str  # 50 + 1 (…) + 50
        life(f"turn {turn} {name} `{args_short}` result {result_ref}")
        if name == "READ_BLOB" or len(result) <= RESULT_STASH_LIMIT:
            context += f"\n[{name}] {escape_refs(result)}\n"
        else:
            context += f"\n[{name}] {result_ref}\n"
        if len(context) > CONTEXT_LIMIT and not _compacting:
            _start_background_compact(context, soul, segment_ref, summarise_ref, harness, memory, lt)
        compacted = _collect_background_compact(context)
        if compacted is not None:
            context = compacted
        open(f"{SELF_DIR}/context.md", "w").write(context)

_compact_result = None  # (snapshot_len, compacted_text) when background compact finishes

def _start_background_compact(context, soul, segment_ref, summarise_ref, harness, memory, lt):
    global _compacting
    _compacting = True
    snapshot_len = len(context)

    def _run():
        global _compact_result, _compacting
        try:
            result = compact(context, soul, segment_ref, summarise_ref, harness, memory, lt)
            _compact_result = (snapshot_len, result)
        except Exception as e:
            life(f"background compact error: {e}")
        finally:
            _compacting = False

    t = threading.Thread(target=_run, daemon=True)
    t.start()

def _collect_background_compact(context):
    global _compact_result
    if _compact_result is None:
        return None
    snapshot_len, compacted = _compact_result
    _compact_result = None
    # append anything added to context since compaction started
    new_tail = context[snapshot_len:]
    result = compacted + new_tail
    merge_ratio = len(result) / len(context) if context else 1
    life(f"background compact merged: {len(context)}ch->{len(result)}ch ({merge_ratio:.0%})")
    return result

if __name__ == "__main__":
    main()