Weekend Build Log: An Idempotent Batch Worker for a Free-Model Side Project

A weekend batch job rarely fails because the model is bad. It fails because a retry re-sends work that already succeeded, and the second pass quietly doubles your cost, your labels, or both. This build log is about the boring machinery around one model call: an idempotency key, a bounded worker pool, and a retry table that decides which errors deserve another attempt.

The demo is small on purpose. It reads open issues from a single repository, asks a model for one of four labels, and writes a JSONL result file. No UI, no database of labels, no multi-repo support. The whole thing fits in one file and one SQLite table.

The scope cut, before any code

The weekend had roughly ten usable hours. That number drove the cut list more than any technical preference.

Shipped:

1. A CLI that takes a JSON file of issues and emits results.jsonl.
2. A content-hash idempotency key stored in SQLite, so a re-run skips finished work.
3. Bounded concurrency of two in-flight requests.
4. A dead-letter file for payloads that should never be retried.
5. A self-imposed character budget that stops the run instead of discovering the ceiling mid-demo.

Skipped, deliberately:

• A web UI. The demo shows the JSONL file in a terminal split pane.
• Vector search over past issues. Keyword dedupe was enough for forty items.
• Multi-provider routing. One endpoint, one model, one prompt.
• Label taxonomy learning. Four fixed labels beat a clustering step nobody asked for.

The skipped items are not future work promises. They were costs the demo did not need to pay.

The failure worth designing for

Two failure modes dominate a naive loop. The first is a rate-limited request that succeeds on attempt three after a long backoff. The second is a timeout where the server actually completed the work and only the response was lost.

The second one is the nasty case. Retrying it duplicates output. Skipping it loses work. There is no perfect answer without server-side idempotency tokens, so the worker picks a policy and records it: a timeout gets exactly one retry, and if that also times out the item goes to the dead-letter file for a human decision. Boring, auditable, and it does not silently double anything.

The artifact: one worker, four policies

The script below is a reference implementation for this post. It was written to be read and rerun, not benchmarked; no throughput numbers are claimed anywhere in it.

#!/usr/bin/env python3
"""Idempotent batch labeling against a free model endpoint."""
import asyncio, hashlib, json, os, random, sqlite3, sys
from pathlib import Path
import httpx

ENDPOINT = os.environ["MC_ENDPOINT"]        # OpenAI-compatible base URL
API_KEY  = os.environ["MC_API_KEY"]
MODEL    = os.environ.get("MC_MODEL", "default")

MAX_INFLIGHT = 2        # keep the pool small; free tiers punish bursts
MAX_ATTEMPTS = 4
BASE_DELAY   = 1.5      # seconds, doubled per attempt, plus jitter
CHAR_BUDGET  = 400_000  # my own ceiling, not a vendor quota

DB   = Path("state.sqlite3")
DEAD = Path("dead_letter.jsonl")

class Transient(Exception): ...
class Permanent(Exception): ...

def key_for(issue):
    """Key on id + text, so an edited issue is reprocessed."""
    h = hashlib.sha256()
    h.update(str(issue["id"]).encode())
    h.update(issue.get("title", "").encode())
    h.update(issue.get("body", "").encode())
    return h.hexdigest()[:16]

def open_state():
    con = sqlite3.connect(DB)
    con.execute("""CREATE TABLE IF NOT EXISTS done(
        key TEXT PRIMARY KEY, label TEXT, ts REAL)""")
    con.commit()
    return con

async def classify(client, issue):
    prompt = (
        "Label this issue as one of: bug, docs, question, other.\n"
        f"Title: {issue['title']}\n\nBody: {issue.get('body','')[:2000]}"
    )
    r = await client.post(
        f"{ENDPOINT}/chat/completions",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={"model": MODEL, "temperature": 0,
              "messages": [{"role": "user", "content": prompt}]},
        timeout=60.0,
    )
    if r.status_code in (429, 500, 502, 503, 504):
        raise Transient(f"http {r.status_code}")
    if r.status_code >= 400:
        raise Permanent(f"http {r.status_code}: {r.text[:200]}")
    return r.json()["choices"][0]["message"]["content"].strip().lower()

async def handle(sem, client, con, issue, stats):
    key = key_for(issue)
    if con.execute("SELECT 1 FROM done WHERE key=?", (key,)).fetchone():
        stats["skipped"] += 1
        return
    approx = len(issue.get("title", "")) + len(issue.get("body", ""))
    if stats["chars"] + approx > CHAR_BUDGET:
        stats["budget_stop"] = True
        return
    async with sem:
        for attempt in range(1, MAX_ATTEMPTS + 1):
            try:
                label = await classify(client, issue)
            except Transient as e:
                if attempt == MAX_ATTEMPTS:
                    DEAD.write_text(DEAD.read_text() + json.dumps(
                        {"key": key, "reason": str(e)}) + "\n" if DEAD.exists()
                        else json.dumps({"key": key, "reason": str(e)}) + "\n")
                    stats["dead"] += 1
                    return
                delay = BASE_DELAY * 2 ** (attempt - 1) + random.uniform(0, 0.5)
                await asyncio.sleep(delay)
                continue
            except Permanent as e:
                with DEAD.open("a") as f:
                    f.write(json.dumps({"key": key, "reason": str(e)}) + "\n")
                stats["dead"] += 1
                return
            con.execute("INSERT OR REPLACE INTO done VALUES (?,?,?)",
                        (key, label, __import__("time").time()))
            con.commit()
            stats["chars"] += approx
            stats["ok"] += 1
            print(json.dumps({"key": key, "label": label}))
            return

async def main(path, dry_run=False):
    issues = json.loads(Path(path).read_text())
    if dry_run:
        for i in issues:
            print(key_for(i), "|", i["title"][:60])
        return 0
    con, stats = open_state(), {"ok": 0, "skipped": 0, "dead": 0,
                                "chars": 0, "budget_stop": False}
    sem = asyncio.Semaphore(MAX_INFLIGHT)
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*(handle(sem, client, con, i, stats) for i in issues))
    print(json.dumps(stats), file=sys.stderr)
    if stats["budget_stop"]:
        return 3
    return 2 if stats["dead"] else 0

if __name__ == "__main__":
    raise SystemExit(asyncio.run(main(sys.argv[1], "--dry-run" in sys.argv)))

Why the key is a content hash

Keying on the issue id alone means an edited issue is never relabeled. Keying on the body alone means two issues with identical text collapse into one. Hashing id, title, and body gives the property the demo needs: identical input is skipped, changed input is reprocessed.

The trade-off is visible. Any whitespace-only edit forces a re-run. For a forty-issue demo that is cheap; for a large backlog it is not, and a normalized body hash would be the next cut to make.

A retry decision table

Condition	Action	Reason
429, 500, 502, 503, 504	Retry with exponential backoff plus jitter	Transient by definition
400, 401, 403, 422	Dead-letter, no retry	The payload or key is wrong; retrying repeats the bug
Client timeout	One retry, then dead-letter	Outcome is ambiguous; cap the duplicate risk
Key already in SQLite	Skip	Idempotency
Character budget exceeded	Stop, exit code 3	Fail on a known boundary, not a surprise one
Empty model response	Dead-letter	A malformed answer is not a network problem

Exit codes matter for the demo: 0 is clean, 2 means dead letters exist and someone should look, 3 means the budget stopped the run. A shell set -e around the script then does something useful.

Where the runs actually live

The worker was written against an OpenAI-compatible endpoint, which keeps the client portable. MonkeyCode provides free model access and a free server option (per the operator; current limits and terms should be checked before planning around them), so the API key and base URL come from environment variables rather than being baked into the script.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

Putting the worker on the free server option is the part that makes it a weekend project rather than a laptop babysitting job: the script keeps running after the terminal closes. Uptime was not measured for this post, and the idempotency table is what makes restarting safe either way.

# plan first: no network calls, just keys and titles
export MC_ENDPOINT="$MC_ENDPOINT" MC_API_KEY="$MC_API_KEY"
python worker.py issues.json --dry-run

# then a real run; re-running is safe because of the key table
python worker.py issues.json; echo "exit=$?"

What this approach does not give you

• Exact token accounting. CHAR_BUDGET counts characters, which is a crude proxy. It is wrong for code-heavy or non-Latin text, and it does not know how a given endpoint tokenizes.
• Cross-machine idempotency. SQLite is local. Two workers on two hosts will both process the same issue.
• Ordering. asyncio.gather finishes in whatever order the network allows.
• Correct labels. The prompt is four lines. It is a demo of plumbing, not a classifier.

Skip this pattern if you need exactly-once semantics, if labels must be reviewed before they touch a real repository, or if your issue bodies are large enough that a character budget is the wrong control entirely.

The part worth carrying to the next project is the table, not the code. Decide which errors earn a retry before the first request goes out, write the decision down, and let the script enforce it. If you want to try the free model access and free server option yourself, the two environment variables above are the only integration points the worker needs.