Shadow-Test Free Model Endpoint Changes Before You Rely on Them

Most endpoint regressions stay invisible until a client breaks.
A free model endpoint may change its timeout, envelope, or empty-response behavior without notice.
The fix is to replay saved prompts against a candidate endpoint.

This workflow uses MonkeyCode's free model access and free server option as the test target.
Disclosure: This article was prepared as part of MonkeyCode's product outreach.

You will build a local shadow harness.
It captures baseline responses in SQLite.
It replays the same prompts against a new endpoint.
It prints a pass or hold signal.

Stage 0: Define an envelope contract

Start with a small JSON file named contract.json.
It lists required response keys and hard limits.
The thresholds are yours to tune.

{
  "required_keys": ["text", "finish_reason"],
  "max_latency_ms": 20000,
  "max_bytes": 200000,
  "candidate_latency_ratio": 1.5
}

Verify the file before moving on.

python -m json.tool contract.json > /dev/null && echo ok

Stage 1: Store the harness

The script below has three modes.
The capture mode records live responses.
The replay mode sends the same prompts to a candidate endpoint.
The report mode compares the two sets.

The request body uses a generic input key.
Change it to match your provider.

import argparse
import json
import os
import sqlite3
import time
import urllib.error
import urllib.request

def envelope_ok(raw, required_keys):
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return 0
    return int(all(key in data for key in required_keys))

def call(endpoint, prompt, timeout_ms, key):
    body = json.dumps({'input': prompt}).encode()
    headers = {'Content-Type': 'application/json'}
    if key:
        headers['Authorization'] = 'Bearer ' + key
    req = urllib.request.Request(endpoint, data=body, headers=headers)
    start = time.perf_counter()
    try:
        with urllib.request.urlopen(req, timeout=timeout_ms / 1000) as resp:
            raw = resp.read().decode()
            return raw, resp.status, (time.perf_counter() - start) * 1000
    except Exception:
        return '', 0, (time.perf_counter() - start) * 1000

def init_db(db):
    conn = sqlite3.connect(db)
    conn.execute(
        'CREATE TABLE IF NOT EXISTS baseline (id INTEGER PRIMARY KEY, prompt TEXT UNIQUE, raw TEXT, status_code INTEGER, elapsed_ms REAL, envelope_ok INTEGER, captured_at TEXT DEFAULT CURRENT_TIMESTAMP)'
    )
    conn.execute(
        'CREATE TABLE IF NOT EXISTS candidate (id INTEGER PRIMARY KEY, prompt TEXT UNIQUE, raw TEXT, status_code INTEGER, elapsed_ms REAL, envelope_ok INTEGER, captured_at TEXT DEFAULT CURRENT_TIMESTAMP)'
    )
    return conn

def p95(values):
    if not values:
        return 0.0
    ordered = sorted(values)
    idx = max(0, int(len(ordered) * 0.95) - 1)
    return ordered[idx]

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--mode', required=True, choices=['capture', 'replay', 'report'])
    parser.add_argument('--endpoint')
    parser.add_argument('--db', default='shadow.db')
    parser.add_argument('--contract', default='contract.json')
    args = parser.parse_args()

    contract = json.load(open(args.contract))
    required = contract['required_keys']
    timeout_ms = contract['max_latency_ms']
    key = os.environ.get('MODEL_KEY', '')

    conn = init_db(args.db)

    if args.mode == 'capture':
        if not args.endpoint:
            raise SystemExit('--endpoint is required for capture')
        count = 0
        for line in open('prompts.txt'):
            prompt = line.strip()
            if not prompt:
                continue
            raw, status, elapsed = call(args.endpoint, prompt, timeout_ms, key)
            conn.execute(
                'INSERT OR REPLACE INTO baseline (prompt, raw, status_code, elapsed_ms, envelope_ok) VALUES (?, ?, ?, ?, ?)',
                (prompt, raw, status, elapsed, envelope_ok(raw, required)),
            )
            count += 1
        conn.commit()
        print(f'captured {count} prompts')

    elif args.mode == 'replay':
        if not args.endpoint:
            raise SystemExit('--endpoint is required for replay')
        rows = conn.execute('SELECT id, prompt FROM baseline ORDER BY id').fetchall()
        count = 0
        for row_id, prompt in rows:
            raw, status, elapsed = call(args.endpoint, prompt, timeout_ms, key)
            conn.execute(
                'INSERT OR REPLACE INTO candidate (id, prompt, raw, status_code, elapsed_ms, envelope_ok) VALUES (?, ?, ?, ?, ?, ?)',
                (row_id, prompt, raw, status, elapsed, envelope_ok(raw, required)),
            )
            count += 1
        conn.commit()
        print(f'replayed {count} prompts')

    else:
        hard_failures = conn.execute(
            'SELECT COUNT(*) FROM candidate WHERE status_code = 0 OR status_code >= 400'
        ).fetchone()[0]
        schema_regressions = conn.execute(
            '''
            SELECT COUNT(*)
            FROM baseline b
            JOIN candidate c ON c.id = b.id
            WHERE b.envelope_ok = 1 AND c.envelope_ok = 0
            '''
        ).fetchone()[0]
        stable_baseline = conn.execute(
            'SELECT COUNT(*) FROM baseline WHERE envelope_ok = 1'
        ).fetchone()[0]
        base_latencies = [
            row[0] for row in conn.execute('SELECT elapsed_ms FROM baseline WHERE envelope_ok = 1')
        ]
        cand_latencies = [
            row[0] for row in conn.execute('SELECT elapsed_ms FROM candidate WHERE envelope_ok = 1')
        ]
        base_p95 = p95(base_latencies)
        cand_p95 = p95(cand_latencies)
        ratio = contract.get('candidate_latency_ratio', 1.5)
        latency_regressions = int(cand_p95 > base_p95 * ratio)
        print(f'stable_baseline_rows: {stable_baseline}')
        print(f'hard_failures: {hard_failures}')
        print(f'schema_regressions: {schema_regressions}')
        print(f'latency_regressions: {latency_regressions}')
        print(f'p95_baseline_ms: {base_p95:.1f}')
        print(f'p95_candidate_ms: {cand_p95:.1f}')
        if stable_baseline < 10 or hard_failures or schema_regressions or latency_regressions:
            print('promotion: HOLD')
        else:
            print('promotion: PASS')

if __name__ == '__main__':
    main()

Save this file as shadow_harness.py.
Then create a prompt list.
Keep it small and reproducible.

printf 'Explain idempotency.\nWrite a SQL migration.\nReturn a JSON error shape.\n' > prompts.txt

Verify the file has non-empty lines.

grep -c . prompts.txt

Stage 2: Capture a baseline

Run capture mode against the endpoint you trust today.

python shadow_harness.py --mode capture --endpoint "$BASELINE_ENDPOINT" --db shadow.db

Verify the baseline has enough stable rows.

sqlite3 shadow.db 'select count(*), sum(envelope_ok) from baseline;'

If the second value is below 10, your baseline is weak.
Add more prompts or fix the contract before you proceed.

Stage 3: Replay against a candidate

Point the harness at the new endpoint or route.

python shadow_harness.py --mode replay --endpoint "$CANDIDATE_ENDPOINT" --db shadow.db

Verify the candidate rows landed.

sqlite3 shadow.db 'select count(*), sum(envelope_ok) from candidate;'

Stage 4: Read the pass or hold report

python shadow_harness.py --mode report --db shadow.db

Example output:

stable_baseline_rows: 14
hard_failures: 0
schema_regressions: 0
latency_regressions: 0
p95_baseline_ms: 1840.2
p95_candidate_ms: 1920.7
promotion: PASS

Use this decision table.

Signal	Pass	Hold
Transport or HTTP failure	0	1 or more
Baseline-valid rows that fail candidate schema	0	1 or more
Stable baseline rows	10 or more	below 10
Candidate p95 vs baseline p95	within ratio	above ratio

A hold is not a failure.
It means you need more evidence before routing traffic.

Limitations

Replay only covers prompts you captured.
It cannot predict novel prompts or semantic quality.
It does not load-test concurrent traffic.
A shadow pass is a release gate, not a correctness proof.

Skip this approach when your prompt set changes every hour.
Skip it when the endpoint's value is mostly creative text.
Skip it when you already have a canary and metrics pipeline.

If you use MonkeyCode's free server option, keep a shadow DB next to your proxy.
Replay the saved prompts before every routing change.