Cap the Turns: A Weekend Call Envelope for Agent Side Projects

A weekend agent loop usually dies from unbounded turns, not from a weak prompt. A file-backed call envelope that refuses the next model or tool invocation after a local budget keeps the experiment from burning idle capacity overnight. The sketch below is small enough to finish in a sitting and strict enough to prove the stop with a test.

The failure this weekend actually needed to stop

Side-project agents retry, reflect, and call tools in a loop. The loop looks healthy. Logs stay green. Tokens still leave the process. Circuit breakers trip on errors. Tool allowlists block the wrong function. Neither construct stops a polite agent that never decides it is done.

Overnight runs make the gap obvious. A single missed max_steps in a toy orchestrator can spend a courtesy budget that was meant to last the week. The envelope treats remaining turns as a first-class resource, not as a comment in a README. Eval suites still matter, but they do not run while the laptop is closed. A hard local halt does.

Scope cut

The lab kept five jobs and dropped the rest. Scope was cut before any client code was written, because the weekend only pays for a working refusal.

In scope

1. Persist remaining call budget in a JSON file.
2. Wrap both model calls and tool calls behind one try_consume() gate.
3. Return a structured refusal the orchestrator can handle without a stack-trace dump.
4. Ship a CLI to inspect, reset, and dry-run the envelope.
5. Prove the halt with a unit test that does not talk to a network.

Out of scope for this weekend

• Distributed locks across several workers.
• Vendor-accurate token billing.
• Streaming usage callbacks from a provider.
• Multi-tenant quotas.
• A web dashboard.

The cut is the point. A finished envelope that stops is more useful than a half-built billing platform.

Envelope rules

The gate is deliberately boring. One process. One file. One lock. Model calls and tool calls share a single counter because weekend agents waste capacity on both.

Event	Remaining calls	Action
try_consume("model") and remaining > 0	decrement by 1	allow
try_consume("tool") and remaining > 0	decrement by 1	allow
remaining == 0	unchanged	refuse with EnvelopeEmpty
reset(n)	set to n	allow future consumes
crash mid-write	previous JSON	next start reloads last complete file

Splitting budgets by kind looks sophisticated. It also delays the halt test. That split was left for another sitting.

Working demo

The following Python module is a lab sketch. It is not a production quota service. Copy it into envelope.py and run it locally.

from __future__ import annotations

import argparse
import json
import os
import tempfile
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

Kind = Literal["model", "tool"]


class EnvelopeEmpty(RuntimeError):
    def __init__(self, kind: Kind, remaining: int) -> None:
        super().__init__(
            f"envelope empty; refused {kind} call; remaining={remaining}"
        )
        self.kind = kind
        self.remaining = remaining


@dataclass
class EnvelopeState:
    remaining: int
    consumed_model: int
    consumed_tool: int

    def to_dict(self) -> dict:
        return {
            "remaining": self.remaining,
            "consumed_model": self.consumed_model,
            "consumed_tool": self.consumed_tool,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "EnvelopeState":
        return cls(
            remaining=int(data["remaining"]),
            consumed_model=int(data.get("consumed_model", 0)),
            consumed_tool=int(data.get("consumed_tool", 0)),
        )


class CallEnvelope:
    def __init__(self, path: Path) -> None:
        self.path = path
        self._lock = threading.Lock()
        self.path.parent.mkdir(parents=True, exist_ok=True)
        if not self.path.exists():
            self._write(
                EnvelopeState(remaining=0, consumed_model=0, consumed_tool=0)
            )

    def _read(self) -> EnvelopeState:
        raw = json.loads(self.path.read_text(encoding="utf-8"))
        return EnvelopeState.from_dict(raw)

    def _write(self, state: EnvelopeState) -> None:
        fd, tmp_name = tempfile.mkstemp(dir=str(self.path.parent), suffix=".tmp")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(state.to_dict(), handle, indent=2)
                handle.write("\n")
            os.replace(tmp_name, self.path)
        except Exception:
            if os.path.exists(tmp_name):
                os.remove(tmp_name)
            raise

    def reset(self, remaining: int) -> EnvelopeState:
        if remaining < 0:
            raise ValueError("remaining must be >= 0")
        with self._lock:
            state = EnvelopeState(
                remaining=remaining, consumed_model=0, consumed_tool=0
            )
            self._write(state)
            return state

    def snapshot(self) -> EnvelopeState:
        with self._lock:
            return self._read()

    def try_consume(self, kind: Kind) -> EnvelopeState:
        with self._lock:
            state = self._read()
            if state.remaining <= 0:
                raise EnvelopeEmpty(kind, state.remaining)
            state.remaining -= 1
            if kind == "model":
                state.consumed_model += 1
            else:
                state.consumed_tool += 1
            self._write(state)
            return state


def guarded_model_call(envelope: CallEnvelope, prompt: str) -> str:
    envelope.try_consume("model")
    # Lab stub: replace with the project's real client after tests pass.
    return f"stub-response:{len(prompt)}"


def guarded_tool_call(envelope: CallEnvelope, name: str, payload: dict) -> dict:
    envelope.try_consume("tool")
    return {"tool": name, "ok": True, "echo": payload}


def run_agent_loop(
    envelope: CallEnvelope, steps: list[tuple[Kind, str]]
) -> list[str]:
    log: list[str] = []
    for kind, payload in steps:
        try:
            if kind == "model":
                text = guarded_model_call(envelope, payload)
                log.append(f"model ok: {text}")
            else:
                result = guarded_tool_call(envelope, payload, {"src": "lab"})
                log.append(f"tool ok: {result['tool']}")
        except EnvelopeEmpty as exc:
            log.append(f"halt: {exc}")
            break
    return log


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Weekend call envelope")
    parser.add_argument("--path", default=".agent-envelope.json")
    sub = parser.add_subparsers(dest="cmd", required=True)
    reset = sub.add_parser("reset")
    reset.add_argument("remaining", type=int)
    sub.add_parser("status")
    consume = sub.add_parser("consume")
    consume.add_argument("kind", choices=["model", "tool"])
    demo = sub.add_parser("demo")
    demo.add_argument("--budget", type=int, default=3)
    return parser


def main() -> None:
    args = build_parser().parse_args()
    envelope = CallEnvelope(Path(args.path))
    if args.cmd == "reset":
        state = envelope.reset(args.remaining)
        print(json.dumps(state.to_dict()))
        return
    if args.cmd == "status":
        print(json.dumps(envelope.snapshot().to_dict(), indent=2))
        return
    if args.cmd == "consume":
        try:
            state = envelope.try_consume(args.kind)
            print(json.dumps(state.to_dict()))
        except EnvelopeEmpty as exc:
            print(json.dumps({"error": str(exc), "remaining": exc.remaining}))
            raise SystemExit(2)
        return
    if args.cmd == "demo":
        envelope.reset(args.budget)
        steps: list[tuple[Kind, str]] = [
            ("model", "plan the edit"),
            ("tool", "read_file"),
            ("model", "apply the edit"),
            ("tool", "run_tests"),
            ("model", "summarize"),
        ]
        for line in run_agent_loop(envelope, steps):
            print(line)
        print(json.dumps(envelope.snapshot().to_dict(), indent=2))


if __name__ == "__main__":
    main()

The stub model function is intentional. The envelope must be testable without a vendor key. Swap guarded_model_call for the project's real client only after the halt behavior is proven.

Commands to run the demo

python envelope.py reset 3
python envelope.py status
python envelope.py consume model
python envelope.py consume tool
python envelope.py demo --budget 3

Expected demo shape: three successful consumes, then a halt line, then a snapshot with remaining at 0. The fourth and fifth planned steps never run. That refusal is the whole product of the weekend.

Inspect the file after the demo. The JSON is the audit trail this sitting can afford.

cat .agent-envelope.json

A typical snapshot looks like this:

{
  "remaining": 0,
  "consumed_model": 2,
  "consumed_tool": 1
}

If remaining is still positive after five planned steps, the gate never wrapped the loop. Fix that before adding retries, reflection, or a second tool.

Test plan

Save this beside the module as test_envelope.py. The tests never open a socket. If they fail, the envelope is wrong. If they pass, the agent loop can be wired next.

from pathlib import Path

import pytest

from envelope import CallEnvelope, EnvelopeEmpty, run_agent_loop


def test_halt_before_extra_model_call(tmp_path: Path) -> None:
    path = tmp_path / "env.json"
    envelope = CallEnvelope(path)
    envelope.reset(2)
    log = run_agent_loop(
        envelope,
        [
            ("model", "one"),
            ("tool", "two"),
            ("model", "three"),
        ],
    )
    assert log[0].startswith("model ok")
    assert log[1].startswith("tool ok")
    assert log[2].startswith("halt")
    snap = envelope.snapshot()
    assert snap.remaining == 0
    assert snap.consumed_model == 1
    assert snap.consumed_tool == 1


def test_empty_reset_refuses_immediately(tmp_path: Path) -> None:
    envelope = CallEnvelope(tmp_path / "env.json")
    envelope.reset(0)
    with pytest.raises(EnvelopeEmpty):
        envelope.try_consume("model")

pytest test_envelope.py -q

Add one more check when the real client lands: a dry-run flag that records intended consumes without calling the network. The envelope already returns state. Logging that state is cheaper than debugging a surprise bill later.

Parking the loop off the laptop

A local envelope still needs somewhere to run the process. Laptops sleep. SSH sessions drop. A long-lived worker that only has to stay awake is enough for a weekend lab.

MonkeyCode is one lane for that kind of sitting. Disclosure: This article was prepared as part of MonkeyCode's product outreach. The project is open source and, per the operator, offers free model access plus a free server option. The envelope does not depend on that lane. The same JSON file works in front of any client the side project already uses.

Keep the integration boring. Export the envelope path as an environment variable. Run the same python envelope.py demo as a process that outlives a closed lid. Point the real model client at whatever endpoint the lab already trusts. Do not teach the envelope to parse vendor invoices. Call counts are the only signal this weekend believes.

Readers who already keep agent loops on a laptop can park that same process on the free server option if they want the envelope to survive sleep. No other product step is required for the demo to be useful.

What this weekend skipped

Several tempting features were cut on purpose. Each skip kept the demo honest.

• Token estimators. Character-length heuristics lie. They create false confidence.
• Per-kind budgets. Two counters look neat and delay the halt test.
• Redis. A side project with one worker does not need a network lock.
• Auto-reset at midnight. Silent refills hide leaks.
• Prompt-cache awareness. The envelope counts turns, not billed tokens.
• A pretty status page. python envelope.py status already prints JSON.

A green pytest run at the end of the sitting beat a larger design that still called the model "just once more." The skipped list is also a backlog filter: nothing on it is required to stop an unbounded loop.

Limitations and who should skip this

The file is not a ledger. Two processes on one file will race even with os.replace. The lock is in-process only. Crash recovery keeps the last complete JSON, which can lose an in-flight decrement or fail to decrement if the process dies before _write. Neither case is acceptable for paid production traffic.

Do not use this envelope when any of the following is true:

• Several workers share one budget.
• The agent spends real money and needs vendor-accurate usage.
• Compliance requires an immutable audit log.
• The orchestrator already has a job queue with a dead-letter path and a max-attempts field that is actually enforced.

Teams with those needs want a real quota service, not a weekend JSON file. The sketch is for a single operator, a single process, and a budget measured in turns rather than currency.

What the sitting actually produced

The working demo is a counter, a refusal, and a test. That is enough to stop a chatty agent before Monday. Wire the project's real model client only after pytest is green. Leave billing, dashboards, and multi-worker locks for a weekend that has already earned them.