#!/usr/bin/env python3 """ wakeloop.py — a minimal, complete harness for an autonomous LLM agent. One file, ~300 lines. Design principles (see The Wake Loop Handbook, ch. 3): * The harness is dumb; the agent is smart. This file does scheduling glue, context assembly, tool execution, budget enforcement, and a memory-write check. Nothing else. All "intelligence" lives in the model + its memory. * The process is stateless between wakes. Everything the agent knows lives in plain files under WORKSPACE. Kill the box, restore the files, nothing is lost. * Every wake must leave a written trace. If the agent forgets to journal, the harness writes a stub entry so the timeline never has silent gaps. Layout of WORKSPACE (created on first run): mandate.md — the agent's standing goal + rules (you write this) memory/core.md — long-term memory (the agent maintains this) memory/journal/ — one file per day, appended per wake inbox.txt — inbound messages queued between wakes outbox.txt — messages the agent wants delivered to the human state.json — wake counter, cumulative token spend wake.lock — prevents overlapping wakes Dependencies: python3.10+, `pip install anthropic`. Set ANTHROPIC_API_KEY. Run one wake: python3 wakeloop.py Schedule wakes: see the systemd timer in the README (or cron). License: MIT. Copy it, gut it, make it yours. """ import json import os import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path import anthropic # --------------------------------------------------------------------------- # Configuration. Deliberately a handful of constants, not a config framework. # --------------------------------------------------------------------------- WORKSPACE = Path(os.environ.get("WAKELOOP_WORKSPACE", "./workspace")).resolve() MODEL = os.environ.get("WAKELOOP_MODEL", "claude-sonnet-4-5") MAX_STEPS = 40 # tool-use round-trips per wake (hard stop) MAX_TOKENS_PER_WAKE = 200_000 # input+output tokens per wake (budget stop) MAX_OUTPUT_TOKENS = 4096 # per model call BASH_TIMEOUT = 120 # seconds per shell command JOURNAL_TAIL_DAYS = 3 # how many recent journal files to load into context TOOL_RESULT_CAP = 20_000 # chars of tool output returned to the model # --------------------------------------------------------------------------- # Tool definitions. Three tools are enough for a surprisingly capable agent: # a shell, a file writer, and a line to the human. Add more only when the # agent demonstrably needs them. # --------------------------------------------------------------------------- TOOLS = [ { "name": "bash", "description": ( "Run a shell command inside the workspace. stdout+stderr are " f"returned (truncated to {TOOL_RESULT_CAP} chars). Timeout " f"{BASH_TIMEOUT}s. The working directory is the workspace root." ), "input_schema": { "type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"], }, }, { "name": "write_file", "description": ( "Create or overwrite a file at a path relative to the workspace. " "Parent directories are created automatically." ), "input_schema": { "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"}, }, "required": ["path", "content"], }, }, { "name": "send_message", "description": ( "Queue a message for the human. Delivered from outbox.txt by " "whatever transport you wire up (Telegram, email, a cron job " "that mails the file). Use sparingly: one good message beats " "five fragments." ), "input_schema": { "type": "object", "properties": {"text": {"type": "string"}}, "required": ["text"], }, }, ] def run_tool(name: str, args: dict) -> str: """Execute one tool call and return its result as a string. Note what is NOT here: no retries, no cleverness, no interpretation. Errors are returned to the model as text — the model is the error handler. That is the whole point of an agent. """ if name == "bash": try: proc = subprocess.run( args["command"], shell=True, cwd=WORKSPACE, capture_output=True, text=True, timeout=BASH_TIMEOUT, ) out = (proc.stdout or "") + (proc.stderr or "") out = out.strip() or "(no output)" if proc.returncode != 0: out += f"\n(exit code {proc.returncode})" except subprocess.TimeoutExpired: out = f"(command timed out after {BASH_TIMEOUT}s)" return out[:TOOL_RESULT_CAP] if name == "write_file": # Confine writes to the workspace. This is a guardrail against # accidents, not a security boundary — the bash tool can already # write anywhere the process user can. Real isolation belongs at # the OS layer (dedicated user, container, or VM). See ch. 4. target = (WORKSPACE / args["path"]).resolve() if not str(target).startswith(str(WORKSPACE)): return "ERROR: path escapes workspace; write refused." target.parent.mkdir(parents=True, exist_ok=True) target.write_text(args["content"]) return f"wrote {len(args['content'])} chars to {target.relative_to(WORKSPACE)}" if name == "send_message": stamp = datetime.now(timezone.utc).isoformat(timespec="seconds") with open(WORKSPACE / "outbox.txt", "a") as f: f.write(f"--- {stamp}\n{args['text']}\n") return "queued for delivery." return f"ERROR: unknown tool {name!r}" # --------------------------------------------------------------------------- # Context assembly. The agent's entire worldview each wake is: mandate + # core memory + recent journal + queued messages + clock + wake number. # Keep this list short and stable; the agent relies on knowing what it gets. # --------------------------------------------------------------------------- def read_or(path: Path, fallback: str) -> str: return path.read_text() if path.exists() else fallback def drain_inbox() -> str: """Return queued human messages and clear the queue (atomically enough).""" inbox = WORKSPACE / "inbox.txt" if not inbox.exists(): return "(none)" text = inbox.read_text().strip() inbox.write_text("") # drained; the agent now owns the content return text or "(none)" def build_context(state: dict) -> tuple[str, str]: """Return (system_prompt, first_user_message).""" mandate = read_or(WORKSPACE / "mandate.md", "No mandate.md found. Ask the human what your goal is.") core = read_or(WORKSPACE / "memory" / "core.md", "(core memory is empty — this may be your first wake)") journal_dir = WORKSPACE / "memory" / "journal" journal_dir.mkdir(parents=True, exist_ok=True) recent = sorted(journal_dir.glob("*.md"))[-JOURNAL_TAIL_DAYS:] journal = "\n\n".join(f"--- {p.name} ---\n{p.read_text()}" for p in recent) now = datetime.now(timezone.utc) today = now.strftime("%Y-%m-%d") system = f"""You are an autonomous agent. Nobody watches this session live. You wake on a timer, work, and exit; your continuity exists only in your files. # Mandate (set by your human) {mandate} # Operating rules - Your workspace is {WORKSPACE}. Your memory lives in memory/. - Before ending the session you MUST append a journal entry to memory/journal/{today}.md (what you did / learned / plan next), and update memory/core.md if anything durable changed. Overwrite stale state; don't let core.md grow without bound. - Use send_message for anything the human must see. Don't spam. - You cannot spend money without explicit human approval via message. - When finished, stop calling tools and reply with a short closing note. # Core memory {core} # Recent journal {journal or '(no journal entries yet)'}""" user = (f"Wake #{state['wake']}. UTC time: {now.isoformat()}.\n" f"New messages from your human since last wake:\n{drain_inbox()}\n\n" f"Proceed with your work.") return system, user # --------------------------------------------------------------------------- # The loop itself: model ↔ tools until the model stops, or a cap trips. # --------------------------------------------------------------------------- def run_wake() -> None: state_path = WORKSPACE / "state.json" state = json.loads(read_or(state_path, '{"wake": 0, "tokens_lifetime": 0}')) state["wake"] += 1 # Snapshot today's journal size so we can verify a trace was written. today = datetime.now(timezone.utc).strftime("%Y-%m-%d") journal_file = WORKSPACE / "memory" / "journal" / f"{today}.md" journal_size_before = journal_file.stat().st_size if journal_file.exists() else 0 system, user = build_context(state) messages = [{"role": "user", "content": user}] client = anthropic.Anthropic() tokens_this_wake = 0 for step in range(MAX_STEPS): response = client.messages.create( model=MODEL, max_tokens=MAX_OUTPUT_TOKENS, system=system, tools=TOOLS, messages=messages, ) tokens_this_wake += response.usage.input_tokens + response.usage.output_tokens messages.append({"role": "assistant", "content": response.content}) tool_calls = [b for b in response.content if b.type == "tool_use"] if not tool_calls: break # the model chose to end the wake results = [] for call in tool_calls: print(f"[wake {state['wake']} step {step}] {call.name} " f"{json.dumps(call.input)[:200]}", flush=True) results.append({ "type": "tool_result", "tool_use_id": call.id, "content": run_tool(call.name, call.input), }) messages.append({"role": "user", "content": results}) if tokens_this_wake > MAX_TOKENS_PER_WAKE: # Tell the model the budget tripped so it can wrap up cleanly, # then give it exactly one more turn (the loop's next iteration # ends via the step cap or a no-tools reply). messages.append({"role": "user", "content": "BUDGET EXCEEDED. Write your journal " "entry now and stop."}) # --- post-wake enforcement: the trace must exist ----------------------- # Content-agnostic check: did today's journal file grow during this wake? # (Don't grep for phrases — agents word things unpredictably. This check # itself replaced a phrase-match that produced false stubs in testing.) journal_size_after = journal_file.stat().st_size if journal_file.exists() else 0 if journal_size_after <= journal_size_before: with open(journal_file, "a") as f: f.write(f"\n# Wake {state['wake']} (harness stub)\n" f"Agent exited without journaling. Steps used: see log. " f"Tokens: {tokens_this_wake}.\n") state["tokens_lifetime"] += tokens_this_wake state_path.write_text(json.dumps(state, indent=2)) print(f"[wake {state['wake']}] done. tokens={tokens_this_wake}", flush=True) def main() -> None: WORKSPACE.mkdir(parents=True, exist_ok=True) lock = WORKSPACE / "wake.lock" # A stale lock older than 2h is assumed dead (crashed wake) and replaced. if lock.exists() and time.time() - lock.stat().st_mtime < 7200: print("another wake is running; exiting.", flush=True) sys.exit(0) lock.write_text(str(os.getpid())) try: run_wake() finally: lock.unlink(missing_ok=True) if __name__ == "__main__": main()