When the Same Model Name Starts Behaving Differently: A Startup Canary for Unattended Pipelines

import re
import json
import hashlib
 
def fingerprint(text: str) -> dict:
    """Extract only the structural fingerprint. Wording variation is dropped on purpose."""
    # 1) If it parses as JSON, take the set of top-level keys
    keys = []
    try:
        obj = json.loads(text)
        if isinstance(obj, dict):
            keys = sorted(obj.keys())
    except json.JSONDecodeError:
        pass
 
    # 2) Structural counts for prose-shaped output
    h2 = len(re.findall(r"^##\s+", text, re.MULTILINE))
    bullets = len(re.findall(r"^\s*[-*]\s+", text, re.MULTILINE))
 
    # 3) Register consistency: plain-form endings that should not appear in polite JA
    jotai = len(re.findall(r"[ぁ-んァ-ヶ一-龥](?:だ|である)[。．]", text))
 
    return {
        "json_keys": keys,
        "h2_buckets": min(h2, 12),       # round off count jitter lightly
        "bullet_buckets": round(bullets / 3),
        "register_violations": jotai,
        "is_json": bool(keys),
    }
 
def fingerprint_id(fp: dict) -> str:
    payload = json.dumps(fp, ensure_ascii=False, sort_keys=True)
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:12]

import os
import json
import time
from anthropic import Anthropic
 
client = Anthropic()  # API key comes from the environment
MODEL = "claude-opus-4-8"
BASELINE_PATH = "canary_baseline.json"
 
# Fixed prompts with unambiguous shape expectations (we read structure, not wording)
CANARY_CASES = [
    {
        "id": "json_schema",
        "prompt": "Return JSON only with exactly three keys: title, tags (array), summary. "
                  "Topic: 'API rate limits'. No preamble, no code fences.",
        "max_tokens": 300,
    },
    {
        "id": "structure",
        "prompt": "About Claude API prompt caching, use exactly three ## headings, "
                  "with exactly two bullets under each heading.",
        "max_tokens": 500,
    },
    {
        "id": "transform_shape",
        "prompt": "Return only the lowercased form of this input: HELLO-WORLD-123",
        "max_tokens": 80,
    },
]
 
def run_case(case: dict) -> str:
    resp = client.messages.create(
        model=MODEL,
        max_tokens=case["max_tokens"],
        temperature=0,                     # minimize jitter so the fingerprint is stable
        messages=[{"role": "user", "content": case["prompt"]}],
    )
    return resp.content[0].text
 
def collect() -> dict:
    out = {}
    for case in CANARY_CASES:
        text = run_case(case)
        fp = fingerprint(text)
        out[case["id"]] = {"fp": fp, "fid": fingerprint_id(fp)}
    return out
 
def save_baseline():
    snapshot = {"model": MODEL, "captured_at": time.time(), "cases": collect()}
    with open(BASELINE_PATH, "w", encoding="utf-8") as f:
        json.dump(snapshot, f, ensure_ascii=False, indent=2)
    print("baseline saved:", BASELINE_PATH)

def evaluate(case_id: str, current: dict, baseline: dict) -> list[str]:
    """Return only dangerous diffs. Harmless jitter is swallowed."""
    issues = []
    cur, base = current["fp"], baseline["fp"]
 
    # Schema: a JSON-expected case that stops being JSON is fatal
    if base["is_json"] and not cur["is_json"]:
        issues.append(f"[{case_id}] JSON output broke")
    if base["is_json"] and cur["is_json"] and cur["json_keys"] != base["json_keys"]:
        issues.append(f"[{case_id}] JSON keys changed {base['json_keys']} -> {cur['json_keys']}")
 
    # Register: any new plain-form ending is a fail (it feeds a quality gate)
    if cur["register_violations"] > base["register_violations"]:
        issues.append(f"[{case_id}] register drift {base['register_violations']} -> {cur['register_violations']}")
 
    # Structural counts: only flag a shift of two buckets or more
    if abs(cur["h2_buckets"] - base["h2_buckets"]) >= 2:
        issues.append(f"[{case_id}] heading count shifted {base['h2_buckets']} -> {cur['h2_buckets']}")
 
    return issues
 
def preflight() -> bool:
    """Call before the main batch. If it returns False, do not run the batch."""
    with open(BASELINE_PATH, encoding="utf-8") as f:
        baseline = json.load(f)
 
    current = collect()
    all_issues = []
    for case_id, cur in current.items():
        base = baseline["cases"].get(case_id)
        if base is None:
            continue
        all_issues += evaluate(case_id, cur, base)
 
    if all_issues:
        print("Drift detected — holding today's batch")
        for msg in all_issues:
            print("  -", msg)
        return False
 
    print("Canary passed — running the batch")
    return True

if __name__ == "__main__":
    if preflight():
        run_daily_batch()       # the real work, e.g. article generation
    else:
        notify("canary drift: manual review needed")  # email or push

# Before: trust the model name and run
def main_before():
    run_daily_batch()    # publishes even if the model silently changed
 
# After: pass through the instrument first
def main_after():
    if not preflight():               # ~6s, $0.01 of insurance
        notify("canary drift")
        return                        # do not publish on a risky day
    run_daily_batch()