Tutorial: a private coding agent

Build a coding agent that runs in a kosa8 microVM, uses a local llamay model through the MAQPNA gateway, is held to policies on file writes and shell commands, forks to try two fixes, and leaves a verifiable audit trail.

In this tutorial you build a small coding agent that fixes a bug, end to end, without any code or prompt leaving your machine. The agent runs in a kosa8 microVM that can reach only the MAQPNA gateway. It thinks with a local llamay model, through the gateway. Before every file write and every shell command it asks the gateway for a decision, and policies decide: writes only under /workspace, risky shell commands only after a human approves. Halfway through, you fork the running agent, so the parent and the child each try a different fix. At the end you read and verify the audit trail of both.

You start from an empty folder. It takes about 30 minutes, most of it the model download.

What you build#

flowchart LR
    subgraph vm["kosa8 microVM · egress: gateway only"]
        A["agent.py<br/>/workspace"]
    end
    subgraph host["Your machine"]
        G["MAQPNA gateway"]
        P["policy.json"]
        L["llamay<br/>local model"]
        AL[".maqpna/audit.jsonl"]
        K["kosa8d audit"]
    end
    A -->|"POST /llm/llamay/v1/chat/completions"| G
    A -->|"POST /v1/sessions/ID/tools:authorize"| G
    G --> L
    P --> G
    G --> AL
    G -->|"mirrored decisions"| K
    You["You: maqpna approvals approve"] --> G

• Model calls go to the gateway's model route llamay. The gateway checks the session token and the models:* scope, meters tokens, and gives each session its own llamay inference context.
• File writes and shell commands run inside the microVM, but only after the agent asks POST /v1/sessions/{id}/tools:authorize. The gateway evaluates the call as server maqpna-client, tool coder__write_file or coder__bash_run, runs the policy, and writes the decision to the audit ledger. The agent enforces the answer.
• The fork clones the running microVM. The clone becomes a child session with its own token, and its llamay context forks from the parent's.

Before you begin#

• An Apple silicon Mac, or x86-64 Linux with KVM. MAQPNA's kosa8 support has been run end to end on macOS; on Linux it uses the same kosa8 API.
• About 6 GB of free disk for the model and the sandbox images.
• curl and a terminal. You open two terminals in the same folder.

Step 1: Install the three tools#

Install the maqpna CLI, kosa8 and llamay with each product's own installer.

curl -fsSL https://maqpna.com/install.sh | sh
curl -fsSL https://kosa8.com/install.sh | sh
curl -fsSL https://llamay.com/install.sh | sh

curl -fsSL https://maqpna.com/install.sh | sh
curl -fsSL https://kosa8.com/install.sh | sh     # needs KVM, cloud-hypervisor and virtiofsd
curl -fsSL https://llamay.com/install.sh | sh

Start the kosa8 daemon and pull a model:

kosa8 daemon start
llamay pull qwen2.5:7b

qwen2.5:7b gives usable code. To try the flow faster, use qwen2.5:0.5b; its fixes are often wrong, which the tests then show.

Check it worked.

maqpna version
kosa8 run alpine echo hi      # prints: hi
llamay --version

Step 2: Create the project folder#

mkdir private-coder && cd private-coder

You create two files in it: policy.json (what the agent may do) and agent.py (the agent). maqpna dev keeps its own state in ./.maqpna.

Step 3: Write the policy#

The agent is named coder and runs in namespace dev (the maqpna dev run defaults). Its client-run tools are write_file and bash_run, which the gateway sees as coder__write_file and coder__bash_run on server maqpna-client.

Create policy.json:

{
  "policies": [
    {
      "name": "private-coder",
      "namespace": "dev",
      "agents": ["coder"],
      "defaultAction": "deny",
      "rules": [
        {
          "name": "no-path-traversal",
          "servers": ["maqpna-client"],
          "tools": ["coder__write_file"],
          "when": [{"arg": "path", "op": "regex", "value": "(^|/)\\.\\.(/|$)"}],
          "action": "deny",
          "reason": "paths with .. are not allowed"
        },
        {
          "name": "write-inside-workspace",
          "servers": ["maqpna-client"],
          "tools": ["coder__write_file"],
          "when": [{"arg": "path", "op": "regex", "value": "^/workspace/"}],
          "substrate": {"checks": ["kosa8"]},
          "action": "allow",
          "maxCallsPerMinute": 60
        },
        {
          "name": "no-writes-outside-workspace",
          "servers": ["maqpna-client"],
          "tools": ["coder__write_file"],
          "action": "deny",
          "reason": "the agent may write only under /workspace"
        },
        {
          "name": "risky-shell-needs-human",
          "servers": ["maqpna-client"],
          "tools": ["coder__bash_run"],
          "when": [
            {
              "arg": "command",
              "op": "regex",
              "value": "(^|[;&|(]\\s*)(sudo|rm\\s+-[A-Za-z]*[rRf]|git\\s+push|curl|wget|ssh|scp|chmod\\s+-R|dd)\\b"
            }
          ],
          "substrate": {"checks": ["kosa8"]},
          "action": "require_approval",
          "reason": "a shell command that deletes, escalates, pushes or reaches the network needs a human"
        },
        {
          "name": "shell-in-verified-sandbox",
          "servers": ["maqpna-client"],
          "tools": ["coder__bash_run"],
          "substrate": {"checks": ["kosa8"]},
          "action": "allow",
          "maxCallsPerMinute": 60
        }
      ]
    }
  ]
}

How the gateway reads it:

• Rules run in order, and the first rule that matches decides. A call no rule matches gets defaultAction, deny.
• when conditions test the call's arguments. regex uses Go regular expressions.
• substrate: {"checks": ["kosa8"]} makes the allow rules apply only to sessions whose sandbox the identity broker verified with kosa8 before it minted the token. The same agent run outside a kosa8 microVM gets no writes and no shell. See identity checks in policy.
• require_approval holds the call until a person approves or denies it. The local gateway expires an approval after five minutes.

Check it worked. Test two calls offline. A policy test has no verified substrate, so even a write inside the workspace is denied there; that is the substrate rule at work:

maqpna policy lint policy.json
maqpna policy test --policy policy.json --ns dev --agent coder --server maqpna-client \
  --tool coder__write_file --args '{"path":"/etc/motd"}' --expect deny --explain
maqpna policy test --policy policy.json --ns dev --agent coder --server maqpna-client \
  --tool coder__write_file --args '{"path":"/workspace/calc.py"}' --expect deny --explain

Each prints the decision and the evaluation trace as JSON. The first is denied by private-coder/no-writes-outside-workspace. In the second, the trace says write-inside-workspace did not match, "why": "substrate check not in kosa8", so the call falls through to the same deny. Both commands exit with status 0 because the decision matched --expect.

Step 4: Write the agent#

The agent uses only the Python standard library, so it runs in the stock python:3.12-alpine image. It:

1. writes a small module with a bug, and its tests, into /workspace (each write authorized first);
2. asks the model what the bug is;
3. waits, so you can fork it;
4. reads its token claims to learn whether it is the parent or the fork, and picks a fix strategy;
5. asks the model for the fixed module, writes it, and runs the tests;
6. tries one write outside /workspace, to show the deny;
7. cleans up with rm -rf, which needs your approval.

Create agent.py:

"""A private coding agent for a kosa8 microVM, governed by MAQPNA."""
import base64
import hashlib
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.request

GATEWAY = os.environ["MAQPNA_GATEWAY_URL"]
MODEL_URL = os.environ["OPENAI_BASE_URL"] + "/chat/completions"
MODEL = os.environ.get("MAQPNA_MODEL_NAME", "llamay")
WORKSPACE = "/workspace"
HOLD = int(sys.argv[1]) if len(sys.argv) > 1 else 0

BUGGY = '''def median(values):
    """Return the median of a non-empty list of numbers."""
    s = sorted(values)
    return s[len(s) // 2]
'''

TESTS = '''import unittest
from calc import median


class MedianTest(unittest.TestCase):
    def test_odd(self):
        self.assertEqual(median([3, 1, 2]), 2)

    def test_even(self):
        self.assertEqual(median([4, 1, 3, 2]), 2.5)


if __name__ == "__main__":
    unittest.main()
'''

STRATEGIES = {
    "parent": "Make the smallest possible change to the existing code.",
    "fork": "Rewrite the function from scratch with clear variable names.",
}


def token():
    # Read the token on every call: after a fork, this file holds the child's token.
    with open(os.environ["MAQPNA_TOKEN_FILE"]) as f:
        return f.read().strip()


def claims():
    payload = token().split(".")[1]
    payload += "=" * (-len(payload) % 4)
    return json.loads(base64.urlsafe_b64decode(payload))


def post(url, body, headers=None, timeout=120):
    h = {"Authorization": "Bearer " + token(), "Content-Type": "application/json"}
    h.update(headers or {})
    req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=h, method="POST")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, json.loads(r.read() or b"{}")
    except urllib.error.HTTPError as e:
        raw = e.read()
        try:
            return e.code, json.loads(raw)
        except ValueError:
            return e.code, {"error": raw.decode(errors="replace")}


def log(msg):
    print(f"[{claims()['maqpna_session']}] {msg}", flush=True)


def authorize(tool, args):
    """Ask the gateway before running a tool. Waits while a human decides."""
    url = f"{GATEWAY}/v1/sessions/{claims()['maqpna_session']}/tools:authorize"
    body = {"client": "coder", "tool": tool, "args": args}
    status, out = post(url, body, {"X-Maqpna-Async": "1"})
    approval = out.get("approvalId")
    if status == 202:
        log(f"{tool} held for approval {approval}: {out.get('reason', '')}")
    deadline = time.time() + 330
    while status == 202 and time.time() < deadline:
        time.sleep(3)
        status, out = post(url, body, {"X-Maqpna-Async": "1", "X-Maqpna-Approval-Id": approval})
    if status == 200 and out.get("decision") == "allow":
        return True
    log(f"{tool} {args} not allowed: HTTP {status} {out.get('error') or out.get('reason')}")
    return False


def write_file(path, content):
    args = {"path": path, "sha256": hashlib.sha256(content.encode()).hexdigest()}
    if not authorize("write_file", args):
        return False
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w") as f:
        f.write(content)
    log(f"wrote {path}")
    return True


def bash_run(command):
    if not authorize("bash_run", {"command": command}):
        return None
    r = subprocess.run(["/bin/sh", "-c", command], capture_output=True, text=True, timeout=120)
    return r.returncode, (r.stdout + r.stderr).strip()


def chat(messages):
    status, out = post(MODEL_URL, {"model": MODEL, "max_tokens": 400, "messages": messages}, timeout=300)
    if status != 200:
        raise SystemExit(f"model call failed: HTTP {status} {out}")
    return out["choices"][0]["message"]["content"]


def main():
    log(f"started in {os.environ.get('MAQPNA_RUNTIME')} with model {MODEL}")
    write_file(f"{WORKSPACE}/calc.py", BUGGY)
    write_file(f"{WORKSPACE}/test_calc.py", TESTS)
    code, output = bash_run(f"cd {WORKSPACE} && python -m unittest -q test_calc") or (None, "not run")
    log(f"tests before the fix: exit {code}\n{output}")

    messages = [
        {"role": "system", "content": "You are a careful Python developer."},
        {"role": "user", "content": f"This module fails its tests.\n\n{BUGGY}\n{TESTS}\n"
                                    "In one sentence, what is the bug?"},
    ]
    analysis = chat(messages)
    log(f"model: {analysis}")
    messages.append({"role": "assistant", "content": analysis})

    if HOLD:
        log(f"holding {HOLD} s: fork me now with `maqpna dev fork`")
        time.sleep(HOLD)

    role = "fork" if claims().get("maqpna_parent_session") else "parent"
    log(f"I am the {role}; strategy: {STRATEGIES[role]}")
    messages.append({"role": "user", "content": STRATEGIES[role] + " Reply with the complete fixed calc.py "
                                                "in one ```python code block and nothing else."})
    reply = chat(messages)
    m = re.search(r"```(?:python)?\n(.*?)```", reply, re.S)
    if not m:
        log("the model returned no code block; stopping")
        return 1
    write_file(f"{WORKSPACE}/calc.py", m.group(1))
    code, output = bash_run(f"cd {WORKSPACE} && python -m unittest -q test_calc") or (None, "not run")
    log(f"tests after the {role}'s fix: {'PASS' if code == 0 else 'FAIL'}\n{output}")

    write_file("/etc/motd", "the agent was here\n")  # outside /workspace: denied by policy

    bash_run(f"rm -rf {WORKSPACE}/__pycache__")  # needs a human
    return 0 if code == 0 else 1


if __name__ == "__main__":
    sys.exit(main())

Three details matter in a sandbox:

• The token is a file. In a kosa8 sandbox the session token is only in MAQPNA_TOKEN_FILE, never in the environment. token() reads it on every call, so after the fork the clone uses the child's token at once.
• The session comes from the token. MAQPNA_SESSION in the clone still names the parent. claims() decodes the token's payload and reads maqpna_session and maqpna_parent_session. The agent only reads the claims; the gateway verifies the signature.
• Approvals are polled. With X-Maqpna-Async: 1, a held call answers 202 with an approvalId. The agent retries with X-Maqpna-Approval-Id until the gateway answers 200 (approved) or 403 (denied or expired). An approval is single use and bound to these exact arguments.

Step 5: Start MAQPNA with kosa8, llamay and the policy#

maqpna dev up --kosa8 --llamay qwen2.5:7b --policy policy.json --no-echo

This starts, on your machine:

• llamay serving qwen2.5:7b on 127.0.0.1:11435, as model route llamay, with its request audit anchored in the ledger;
• the identity broker on 127.0.0.1:8081, with the kosa8 identity check;
• the MAQPNA gateway on your LAN address, port 8080, with your policy and the kosa8 audit mirror and anchor.

--no-echo leaves out the test MCP server; this agent needs none.

Check it worked. The output lists the gateway on your LAN address (for example http://192.168.1.20:8080) and model llamay -> http://192.168.1.20:8080/llm/llamay/v1 (OpenAI base URL). maqpna dev status shows the components running.

Step 6: Run the agent in a kosa8 microVM#

In the first terminal:

maqpna dev run --session fix-median \
  --scope tools:maqpna-client --scope 'models:*' \
  --kosa8-image python:3.12-alpine --kosa8-keep \
  -- python -c "$(cat agent.py)" 60

• --scope tools:maqpna-client lets the session use tools:authorize; models:* lets it call the model.
• python -c "$(cat agent.py)" 60 passes the agent's source as one argument, so no image build is needed. 60 is the hold time in seconds.
• --kosa8-keep keeps the sandbox running so you can fork it.

MAQPNA boots sandbox maqpna-fix-median, has the broker verify it, delivers the token, and streams the agent's output:

• the two writes and the first test run are allowed (the tests fail: that is the bug);
• the model explains the bug;
• the agent prints holding 60 s: fork me now with `maqpna dev fork`.

Step 7: Fork the agent to try a second fix#

While the agent holds, in the second terminal in the same folder:

maqpna dev fork --session fix-median

kosa8 clones the running microVM, memory and disk, in about a second. MAQPNA mints a token for child session fix-median-fork-1 (verified by the broker, with fix-median as its parent), writes it into the clone, and streams the clone's output in this terminal.

When the hold ends, both copies continue from the same point:

• the parent, in terminal 1, makes the smallest change;
• the fork, in terminal 2, rewrites the function.

Each asks the model in its own llamay context; the fork's context is a fork of the parent's, so it does not prefill the conversation again. Each writes its own /workspace/calc.py (the clone has its own disk copy) and runs the tests. Which fix passes depends on the model.

Then each copy tries /etc/motd and prints write_file … not allowed: HTTP 403 …, and each holds rm -rf /workspace/__pycache__ for approval.

Step 8: Approve the clean-up#

In a third terminal, or the second one once the fork has printed its approval, load the local admin token and list the pending approvals:

eval "$(maqpna dev env)"
maqpna approvals list

You see two approvals, one for fix-median and one for fix-median-fork-1, both for maqpna-client/coder__bash_run with reason a shell command that deletes, escalates, pushes or reaches the network needs a human. Approve each one as a different person than the session's user:

maqpna approvals approve <approval-id> --approver reviewer@localhost --note "cache clean-up"

Within three seconds the agent retries, gets allow, runs the command and exits. Run maqpna approvals deny <approval-id> --approver reviewer@localhost instead, and the agent prints the denial and skips the command.

Step 9: Read and verify the audit trail#

The timelines. One per session:

maqpna dev timeline --session fix-median
maqpna dev timeline --session fix-median-fork-1

Each lists the model calls and every maqpna-client decision with its rule: allowed writes by private-coder/write-inside-workspace, the denied /etc/motd by private-coder/no-writes-outside-workspace, and the approved rm -rf. The fork's records carry parentSession fix-median. Add -o json for every field.

Where it ran. Check what kosa8 reports for the sandbox and that its egress is the gateway only:

maqpna dev attest --session fix-median

The ledger and the anchors. Verify the hash chain, then look for the kosa8 anchors the gateway wrote every 30 seconds:

maqpna audit verify .maqpna/audit.jsonl
grep -c '"kosa8/audit-anchor"' .maqpna/audit.jsonl
maqpna llamay verify-audit .maqpna/llamay-audit.jsonl
kosa8 audit verify

• maqpna audit verify prints OK with the record count.
• The anchor count is at least 1.
• maqpna llamay verify-audit verifies llamay's own request audit.
• kosa8 audit verify verifies kosa8's chain, which now also holds an mcp.decision record for each of the agent's decisions, pointing at maqpna-ledger-<seq>.

Step 10: Clean up#

kosa8 sandbox rm maqpna-fix-median
kosa8 sandbox rm maqpna-fix-median-fork-1
maqpna dev down

The ledger, the llamay audit and the logs stay in .maqpna/.

What can go wrong#

For the kosa8 and llamay side, see kosa8 and Troubleshooting.

Next steps#

• Tutorial: fork and explore: fork an agent into several branches and compare them.
• Write policies and Human approvals.
• kosa8 reference and llamay with MAQPNA.
• Security model of the stack.