Build a Background Coding Agent That Pushes a Branch for Review

#!/usr/bin/env bash
# Build the named snapshot every task starts from: a checkout of main in ~/repo.
# Usage: ./build-snapshot.sh <snapshot-name> <builder-sandbox-name>
set -euo pipefail
snap=$1
git -C mirror fetch -q origin
git -C mirror bundle create -q "$PWD/base.bundle" origin/main
id=$(sbx new --type small --no-env --name "$2" --json | jq -r 'select(.event == "ready") | .id')
trap 'sbx delete "$id" --yes >/dev/null; echo "deleted builder $id"' EXIT
echo "builder $id"
sbx scp base.bundle "$id:/home/user/" >/dev/null
sbx exec "$id" --timeout 120 "git init -q repo && cd repo && git fetch -q ~/base.bundle refs/remotes/origin/main:refs/remotes/origin/main && git checkout -q -B main origin/main && git log --oneline -1 && python3 -m unittest -q 2>&1 | tail -1"
sbx snapshot "$id" "$snap" >/dev/null
sbx snapshot info "$snap" --json | jq -c '{name: .snapshot.name, status: .snapshot.status, type: .snapshot.type}'

time ./build-snapshot.sh bgagent-base bgagent-1

builder sb_lapqk7usd6
1053b3b textkit: slugify
OK
{"name":"bgagent-base","status":"ready","type":"small"}
deleted builder sb_lapqk7usd6

real	0m8.360s

#!/usr/bin/env python3
"""agentd: a background coding agent service in one file (Python stdlib + sbx + git).

POST /tasks {"text": "..."}  queue a task (stands in for a chat message)
GET  /tasks/<id>             status, sandbox, branch, PR link, step timings
Each task: sandbox from a named snapshot -> agent step -> tests -> branch
-> the service pushes it -> optional PR (GH_REPO) -> sandbox deleted.
"""
import json, os, queue, subprocess, tempfile, threading, time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

SNAPSHOT = os.environ.get("SNAPSHOT", "agent-base")  # named snapshot with ~/repo
PREFIX = os.environ.get("PREFIX", "agent")           # sandbox names: PREFIX-<n>
WORKERS = int(os.environ.get("WORKERS", "2"))        # sandboxes at once
MIRROR = os.environ.get("MIRROR", "mirror")          # local clone of origin
GH_REPO = os.environ.get("GH_REPO")                  # owner/repo: open a real PR
AGENT_STEP = os.path.abspath("agent-step.sh")

tasks, jobs, lock, git_lock = {}, queue.Queue(), threading.Lock(), threading.Lock()


def sh(*cmd, timeout=900):
    p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if p.returncode != 0:
        tail = "\n".join((p.stdout + p.stderr).strip().splitlines()[-15:])
        raise RuntimeError(f"{cmd[0]} {cmd[1]} exited {p.returncode}:\n{tail}")
    return p.stdout


def update(t, status=None, **fields):
    with lock:
        if status:
            t["status"] = status
            t["steps"].append([round(time.time() - t["t0"], 1), status])
        t.update(fields)


def reply(t):
    """Stand-in for posting back to the chat thread the task came from.
    Fixed after the prod run and checked locally only (see step 5)."""
    if t["status"] == "done":
        msg = f"PR ready: {t['pr'] or t['branch'] + ' @ ' + t['commit']}"
    else:
        msg = f"Failed while {t['steps'][-2][1]}: {t['error'].splitlines()[-1]}"
    print(f"[chat] {t['id']}: {msg}", flush=True)


def run(t):
    sid, branch = None, f"agent/{t['id']}"
    work = tempfile.mkdtemp(prefix=f"{t['id']}-")
    try:
        update(t, "creating")
        out = sh("sbx", "new", "--from", SNAPSHOT, "--type", "small", "--no-env",
                 "--no-snapshots", "--ttl", "1800", "--name", f"{PREFIX}-{t['n']}", "--json")
        sid = next(e["id"] for e in map(json.loads, out.splitlines()) if e.get("event") == "ready")
        update(t, sandbox=sid)

        # Bring the snapshot's checkout up to today's main, then hand over the task.
        with git_lock:
            sh("git", "-C", MIRROR, "fetch", "-q", "origin")
            sh("git", "-C", MIRROR, "bundle", "create", "-q", f"{work}/base.bundle", "origin/main")
        with open(f"{work}/TASK.md", "w") as f:
            f.write(t["text"] + "\n")
        for name in ("base.bundle", "TASK.md"):
            sh("sbx", "scp", f"{work}/{name}", f"{sid}:/home/user/")
        sh("sbx", "scp", AGENT_STEP, f"{sid}:/home/user/")
        sh("sbx", "exec", sid, "--cwd", "repo", "--timeout", "120",
           "git fetch -q ~/base.bundle refs/remotes/origin/main:refs/remotes/origin/main"
           f" && git switch -q -C {branch} origin/main")

        update(t, "agent")
        sh("sbx", "exec", sid, "--cwd", "repo", "--timeout", "600", "bash ~/agent-step.sh")
        update(t, "testing")
        sh("sbx", "exec", sid, "--cwd", "repo", "--timeout", "600", "python3 -m unittest -q")

        # The sandbox commits and exports a bundle; it never holds a push credential.
        sh("sbx", "exec", sid, "--cwd", "repo", "--timeout", "120",
           "git add -A && git -c user.name=agentd -c user.email=agentd@example.com"
           f" commit -q -F ~/TASK.md && git bundle create -q ~/out.bundle origin/main..{branch}")
        sh("sbx", "scp", f"{sid}:/home/user/out.bundle", f"{work}/")
        update(t, "pushing")
        with git_lock:
            sh("git", "-C", MIRROR, "bundle", "verify", "-q", f"{work}/out.bundle")
            sh("git", "-C", MIRROR, "fetch", "-q", f"{work}/out.bundle", f"+{branch}:{branch}")
            sh("git", "-C", MIRROR, "push", "-q", "-f", "origin", branch)
            commit = sh("git", "-C", MIRROR, "rev-parse", "--short", branch).strip()
        pr = None
        if GH_REPO:  # needs GH_TOKEN in the service's environment, not the sandbox's
            pr = sh("gh", "pr", "create", "--repo", GH_REPO, "--head", branch,
                    "--title", t["text"][:72], "--body", f"Task {t['id']}: {t['text']}").strip()
        update(t, "done", branch=branch, commit=commit, pr=pr)
    except Exception as e:
        update(t, "failed", error=str(e))
    finally:
        if sid:
            subprocess.run(["sbx", "delete", sid, "--yes"], capture_output=True)
            gone = subprocess.run(["sbx", "info", sid], capture_output=True).returncode != 0
            with lock:
                t["deleted"] = gone
        reply(t)


def worker():
    while True:
        run(jobs.get())


class API(BaseHTTPRequestHandler):
    def send(self, code, body):
        data = json.dumps(body, indent=1).encode()
        self.send_response(code)
        self.send_header("content-type", "application/json")
        self.end_headers()
        self.wfile.write(data)

    def do_POST(self):
        if self.path != "/tasks":
            return self.send(404, {"error": "not found"})
        body = json.loads(self.rfile.read(int(self.headers["content-length"])))
        with lock:
            n = len(tasks) + 1
            t = {"id": f"t{n}", "n": n, "text": body["text"], "status": "queued",
                 "t0": time.time(), "steps": [[0, "queued"]], "sandbox": None,
                 "branch": None, "commit": None, "pr": None, "error": None}
            tasks[t["id"]] = t
        jobs.put(t)
        self.send(202, {"id": t["id"], "status": "queued", "url": f"/tasks/{t['id']}"})

    def do_GET(self):
        with lock:
            if self.path == "/tasks":
                return self.send(200, [{k: t[k] for k in ("id", "status")} for t in tasks.values()])
            t = tasks.get(self.path.removeprefix("/tasks/"))
            body = t and {k: v for k, v in t.items() if k not in ("n", "t0")}
        self.send(200, body) if body else self.send(404, {"error": "no such task"})

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    for _ in range(WORKERS):
        threading.Thread(target=worker, daemon=True).start()
    print(f"agentd: {WORKERS} workers, snapshot {SNAPSHOT}, on 127.0.0.1:8700", flush=True)
    ThreadingHTTPServer(("127.0.0.1", 8700), API).serve_forever()

#!/usr/bin/env bash
# Scripted stand-in for a coding agent: no model runs. It reads ~/TASK.md and
# edits the checkout in the current directory. Swap the body for your agent.
set -euo pipefail
task=$(cat ~/TASK.md)
case "$task" in
  *word*count*)
    cat >> textkit.py <<'PY'


def word_count(text: str) -> int:
    """Number of whitespace-separated words."""
    return len(text.split())
PY
    cat > test_word_count.py <<'PY'
import unittest

from textkit import word_count


class TestWordCount(unittest.TestCase):
    def test_counts_words(self):
        self.assertEqual(word_count("  two   words "), 2)


if __name__ == "__main__":
    unittest.main()
PY
    ;;
  *underscore*)
    # A plausible but wrong edit: it breaks an existing test.
    sed -i 's/\[^a-z0-9\]+/[^a-z0-9_]+/' textkit.py
    ;;
  *) echo "agent: don't know how to do: $task" >&2; exit 2 ;;
esac
git status --short