Portability Before Spill: A Two-Path Conformance Harness for Local-First Agents

A small team ships an agent that reads incident notes and emits a structured triage record. On a workstation the pipeline is fast and the payload never leaves the building, but the laptop sleeps, the local model is occasionally absent after an upgrade, and one long backfill job takes far too long. Someone proposes a one-line change that points the client at a hosted endpoint whenever the local run looks slow. The change ships, and nobody can say afterwards which job classes quietly started leaving the host.

That failure is rarely a model-quality problem. It is a policy placement problem: the decision to spill to a network path gets encoded as a runtime heuristic instead of a per-job-class contract that a test can enforce. A local-first system needs two things kept separate — a measured answer to "which jobs are portable" and a hard guard for "which jobs may never leave". The harness below produces the first and enforces the second.

The three constraints that bracket every job class

Before comparing endpoints, the interesting question is which constraints a job class actually carries. Secrets are binary and non-negotiable: if the payload contains a credential, a customer identifier, or an internal hostname, the server path is out of scope regardless of latency. Offline tolerance is a durability question rather than a speed one, and it decides whether a path failure can be retried later or must fail the job. Latency budget is the only constraint that both paths can genuinely compete on, which is why it should be measured rather than assumed.

A useful side effect of writing these three constraints down is that most job classes become boring. In a typical agent codebase, a large majority of classes are secret-bearing or offline-critical, and only a thin slice of bulk, non-sensitive, latency-tolerant work is actually a candidate for a network path. The harness exists to confirm that slice with numbers instead of intuition.

Step 1 — Write the case file with structural specs

Exact-match diffing does not work across two endpoints, because the same valid answer can be phrased differently. Assertions should therefore describe the shape of a correct record: required keys, enumerations, and minimum list lengths. Each case also carries a local_only flag, which is the policy statement the harness will enforce in code.

{"id": "triage-short", "local_only": false, "prompt": "Return JSON with keys severity(low|medium|high), owner, summary for: disk full on build-07, fixed by pruning cache", "spec": {"require_keys": ["severity", "owner", "summary"], "enum": {"severity": ["low", "medium", "high"]}}}
{"id": "triage-secret", "local_only": true, "prompt": "Summarise the rotation diff for the internal token, JSON keys action, risk", "spec": {"require_keys": ["action", "risk"], "enum": {"risk": ["low", "high"]}}}
{"id": "backfill-merge", "local_only": false, "prompt": "Merge the following 40 incident lines into JSON with keys clusters and count", "spec": {"require_keys": ["clusters", "count"], "min_items": {"clusters": 2}}}

The second case is the important one. It is not a performance test and it must never produce a server measurement, because a measurement requires sending the payload. The flag makes that structurally impossible rather than culturally discouraged.

Step 2 — Run the same cases against both paths

The harness only needs the standard library, because both endpoints speak the OpenAI-style chat completions shape. Endpoints are supplied through environment variables so that no credential is committed alongside the code.

#!/usr/bin/env python3
"""pathcheck.py - two-path conformance harness for local-first agents."""
import json, os, re, statistics, time, urllib.request

LOCAL = {"name": "local", "base": os.environ.get("LOCAL_BASE", "http://127.0.0.1:8080/v1"),
         "key": os.environ.get("LOCAL_KEY", "not-needed"), "model": os.environ.get("LOCAL_MODEL", "local-default")}
SERVER = {"name": "server", "base": os.environ.get("SERVER_BASE", ""),
          "key": os.environ.get("SERVER_KEY", ""), "model": os.environ.get("SERVER_MODEL", "free-default")}
TRIALS = int(os.environ.get("TRIALS", "3"))
TIMEOUT = float(os.environ.get("TIMEOUT", "60"))

FENCE = re.compile(r"```

(?:json)?\s*(\{.*?\})\s*

```", re.S)


def extract_json(text):
    if not text:
        return None
    m = FENCE.search(text)
    raw = m.group(1) if m else None
    if raw is None:
        start, end = text.find("{"), text.rfind("}")
        if start == -1 or end <= start:
            return None
        raw = text[start:end + 1]
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None


def post_chat(ep, prompt):
    body = json.dumps({"model": ep["model"], "temperature": 0,
                       "messages": [{"role": "user", "content": prompt}]}).encode()
    req = urllib.request.Request(ep["base"].rstrip("/") + "/chat/completions", data=body,
                                 headers={"Content-Type": "application/json",
                                          "Authorization": "Bearer " + ep["key"]})
    t0 = time.perf_counter()
    with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
        payload = json.loads(r.read())
    return payload["choices"][0]["message"]["content"], time.perf_counter() - t0


def satisfies(obj, spec):
    for key in spec.get("require_keys", []):
        if key not in obj:
            return False, "missing key " + key
    for field, allowed in spec.get("enum", {}).items():
        val = obj.get(field)
        if not isinstance(val, str) or val.strip().lower() not in allowed:
            return False, "enum " + field + "=" + repr(val)
    for field, n in spec.get("min_items", {}).items():
        val = obj.get(field)
        if not isinstance(val, list) or len(val) < n:
            return False, "min_items " + field
    return True, "ok"


def may_send(ep, case):
    return not (ep["name"] == "server" and case.get("local_only"))


def run_case(ep, case):
    passes, times, notes = 0, [], []
    for _ in range(TRIALS):
        try:
            text, dt = post_chat(ep, case["prompt"])
        except Exception as exc:
            notes.append("transport: " + type(exc).__name__)
            continue
        times.append(dt)
        obj = extract_json(text)
        if obj is None:
            notes.append("no JSON object in reply")
            continue
        ok, why = satisfies(obj, case.get("spec", {}))
        passes += int(ok)
        if not ok:
            notes.append(why)
    return {"endpoint": ep["name"], "pass": passes, "trials": TRIALS,
            "p50_s": round(statistics.median(times), 3) if times else None,
            "notes": sorted(set(notes))[:3]}


def main(cases_path):
    with open(cases_path, encoding="utf-8") as fh:
        cases = [json.loads(line) for line in fh if line.strip()]
    results = []
    for case in cases:
        for ep in (LOCAL, SERVER):
            if not may_send(ep, case):
                results.append({"case": case["id"], "endpoint": ep["name"], "skipped": "local_only"})
                continue
            row = run_case(ep, case)
            row["case"] = case["id"]
            results.append(row)
    with open("portability.json", "w", encoding="utf-8") as fh:
        json.dump(results, fh, indent=2)
    for row in results:
        print(row)


if __name__ == "__main__":
    import sys
    main(sys.argv[1] if len(sys.argv) > 1 else "cases.jsonl")

Run it with three trials per case and per path, which is enough to catch structural failures and wildly wrong latency assumptions while staying inside a coffee break.

LOCAL_BASE=http://127.0.0.1:8080/v1 \
SERVER_BASE=$FREE_SERVER_BASE SERVER_KEY=$FREE_SERVER_KEY \
TRIALS=3 TIMEOUT=45 python3 pathcheck.py cases.jsonl

Step 3 — Read the matrix as a policy, not a scoreboard

Two numbers per cell matter, and neither is a leaderboard rank. Pass rate below trials means the path cannot satisfy the structural contract reliably, which disqualifies that job class from the path no matter how fast it is. Median elapsed time is a planning figure only, and it includes network variance, so it should be treated as a loose bound rather than a service-level target.

A useful convention is to classify each non-secret case into one of three outcomes. Portable means both paths pass every trial, which makes spill a legitimate fallback under load. Local-only in practice means the server path fails the structural spec, so the case joins the guarded set even though it carries no secret. Unstable means at least one path fails intermittently, which is a prompt or spec defect to fix before any routing work continues.

Job class	Secrets in payload	Offline required	Local median in budget	Default path
Short note triage	no	no	yes	local, server as fallback
Credential rotation diff	yes	n/a	n/a	local, guarded, never spills
Bulk incident backfill	no	no	no	server, batched, chunked
Timeline merge	no	yes	yes	local, queued if host sleeps
Ad-hoc exploration	no	no	sometimes	server first, local for secrets

The table is the deliverable. It is reviewable in a pull request, it survives staff turnover, and it gives the guard in may_send something meaningful to protect.

Where a network path genuinely wins

Measured results tend to concentrate the server advantage in three places: long-context bulk work that would thrash a workstation, cold starts on machines where the local model is not resident, and burst handling when several jobs queue behind one local process. When those dominate a workload, a free server option is a pragmatic second rung rather than a compromise, and the open-source MonkeyCode project offers exactly that shape of setup — a client you can run locally, a free model tier, and a free server option. The project quotes 10 million free tokens, and that figure should be re-checked on the project's own pages, since free-tier terms change and this article does not audit them.

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

In practice the free server fits the rung above a local model for non-sensitive, latency-tolerant jobs: it removes the cold-start cliff and absorbs the long tail without a procurement conversation. It does not fit inside an air-gapped workflow, it does not make a secret-bearing payload safe, and it does not remove the need to re-run the harness whenever the case file or the endpoint changes.

Limitations and who should skip this

Three trials per case is a smoke test, not a statistical claim; a path that fails once in three is unstable, while a path that passes three times is merely not yet disproven. The assertions check structure rather than meaning, so a record can satisfy every key and still be wrong, which means a small human-reviewed sample remains necessary. Latency figures include the network and the host's mood, and they say nothing about throughput under concurrency. Finally, may_send is a policy guard inside one script, not a security boundary; production code needs a payload-level assertion before serialization so that a new code path cannot bypass the flag.

Some teams should not adopt this pattern at all. If the local model cannot satisfy the structural specs on any trial, the correct next step is model or prompt work, not routing. If the payloads are regulated and no review process exists for the comparison path, the harness will only document a compliance problem faster. If the workflow is a single script run by one person, a manual spot check is cheaper than a committed matrix.

A reproducible close

Commit cases.jsonl, pathcheck.py, and the generated portability.json together, and make the table the source of truth for placement decisions. Re-run the harness whenever a model, a prompt, or an endpoint changes, and diff the matrix instead of arguing about it. Teams that want a concrete second rung to compare against can point SERVER_BASE at the free MonkeyCode server option and let three trials answer the question that intuition keeps getting wrong.