#!/usr/bin/env python3 """ generate.py -- build TOOLCALL-300, a conformance suite for the layer that turns a language model's tool call into a call that actually matches the declared schema. Standard library only. No network. Deterministic: same seed, byte-identical corpus. Public domain (CC0). python3 generate.py --out . --seed 20260819 python3 generate.py --check-only --out . # re-run every integrity check python3 generate.py --list-checks GROUND TRUTH IS PRODUCED BY CONSTRUCTION. For every recoverable case the correct call -- name plus a schema-valid argument object -- is built FIRST, from the declared schema. The malformed text is then derived from it by a single named mutation. No parser, validator or model is ever asked what the answer is, so the labels cannot inherit anyone's bug. INTEGRITY ASSERTIONS BLOCK THE WRITE. Nothing is written unless every check in CHECKS passes. `--list-checks` prints them. The one that matters most: no recoverable case may already validate against its own declared schema. A suite with freebies in it inflates every score ever run on it. """ import argparse import hashlib import io import json import os import random import sys SEED_DEFAULT = 20260819 N_PER_CATEGORY = 25 CATEGORIES = [ "wrong_tool_name", "missing_required_arg", "extra_undeclared_arg", "type_coercion", "enum_violation", "nested_flattened", "array_vs_scalar", "args_as_string", "multiple_calls", "hallucinated_tool", "truncated", "unrecoverable", ] REFUSAL_CATEGORIES = {"hallucinated_tool", "unrecoverable"} # --------------------------------------------------------------- tool catalogue def S(t, **kw): d = {"type": t} d.update(kw) return d TOOLS = [ {"name": "get_weather", "description": "Return a forecast for one location.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "location": S("string"), "units": S("string", enum=["celsius", "fahrenheit"], default="celsius"), "days": S("integer"), "include_hourly": S("boolean")}, "required": ["location", "units", "days"]}}, {"name": "send_email", "description": "Send one message to one or more recipients.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "to": S("array", items=S("string")), "subject": S("string"), "body": S("string"), "priority": S("string", enum=["low", "normal", "high"], default="normal"), "cc": S("array", items=S("string"))}, "required": ["to", "subject", "body", "priority"]}}, {"name": "create_ticket", "description": "Open a work item on the tracker.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "title": S("string"), "priority": S("string", enum=["low", "medium", "high", "urgent"], default="medium"), "labels": S("array", items=S("string")), "assignee": S("string"), "estimate_hours": S("number")}, "required": ["title", "priority"]}}, {"name": "search_products", "description": "Search the catalogue.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "query": S("string"), "filters": {"type": "object", "additionalProperties": False, "properties": {"category": S("string"), "max_price": S("number"), "in_stock": S("boolean")}, "required": ["category"]}, "limit": S("integer", default=10), "sort": S("string", enum=["relevance", "price_asc", "price_desc"], default="relevance")}, "required": ["query", "filters", "limit"]}}, {"name": "book_flight", "description": "Reserve one itinerary.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "origin": S("string"), "destination": S("string"), "date": S("string"), "passengers": S("integer", default=1), "cabin": S("string", enum=["economy", "premium", "business", "first"], default="economy"), "seat_prefs": S("array", items=S("string"))}, "required": ["origin", "destination", "date", "passengers", "cabin"]}}, {"name": "schedule_meeting", "description": "Put one meeting on the calendar.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "title": S("string"), "attendees": S("array", items=S("string")), "start_iso": S("string"), "duration_minutes": S("integer", default=30), "room": S("string"), "agenda": S("array", items=S("string"))}, "required": ["title", "attendees", "start_iso", "duration_minutes"]}}, {"name": "translate_text", "description": "Translate a passage.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "text": S("string"), "target_lang": S("string", enum=["en", "fr", "de", "es", "ja"]), "formal": S("boolean", default=False), "glossary": {"type": "object", "additionalProperties": False, "properties": {"domain": S("string"), "strict": S("boolean")}, "required": ["domain"]}}, "required": ["text", "target_lang", "formal"]}}, {"name": "run_query", "description": "Run one read-only statement.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "sql": S("string"), "database": S("string", enum=["analytics", "billing", "crm"]), "timeout_seconds": S("integer", default=30), "dry_run": S("boolean", default=False), "tags": S("array", items=S("string"))}, "required": ["sql", "database", "timeout_seconds"]}}, {"name": "convert_currency", "description": "Convert an amount between currencies.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "amount": S("number"), "source_currency": S("string", enum=["EUR", "USD", "GBP", "JPY"]), "target_currency": S("string", enum=["EUR", "USD", "GBP", "JPY"]), "rounding": S("string", enum=["none", "nearest", "bankers"], default="none"), "on_date": S("string")}, "required": ["amount", "source_currency", "target_currency", "rounding"]}}, {"name": "update_record", "description": "Write one row.", "parameters": {"type": "object", "additionalProperties": False, "properties": { "table": S("string", enum=["customers", "orders", "invoices"]), "record_id": S("integer"), "fields": {"type": "object", "additionalProperties": False, "properties": {"status": S("string"), "note": S("string"), "amount": S("number")}, "required": ["status"]}, "upsert": S("boolean", default=False), "reason_code": S("string")}, "required": ["table", "record_id", "fields", "upsert"]}}, ] BY_NAME = {t["name"]: t for t in TOOLS} TOOL_NAMES = [t["name"] for t in TOOLS] # names that exist in no catalogue above -- checked by CHECK 12 HALLUCINATED = [ "get_forecast", "delete_customer", "fetch_invoice_pdf", "post_message", "list_calendars", "cancel_subscription", "summarise_thread", "open_browser", "read_file", "charge_card", "resize_image", "lookup_address", "train_model", "archive_project", "send_sms", "rotate_key", ] POOL = { "city": ["Lisbon", "Tallinn", "Osaka", "Bergen", "Cusco", "Malmo", "Utrecht", "Cork", "Graz", "Split", "Antwerp", "Bilbao", "Turku", "Brno", "Zadar", "Aarhus", "Lund", "Ghent", "Pau", "Trieste"], "email": ["priya@northwind.example", "sam@lattice.example", "lena@quaystone.example", "omar@driftbyte.example", "ines@parallax.example", "tomas@kelvinlabs.example", "yuki@harborline.example", "dana@meridian.example", "noor@brightfen.example", "ivo@stonegate.example"], "subject": ["Quarterly usage report", "Access review", "Renewal window", "Migration plan", "Incident postmortem", "Invoice correction", "Onboarding checklist", "Capacity forecast"], "body": ["The attached figures cover the last full billing period.", "Please confirm the seat count before the renewal date.", "The export finished; nothing further is needed from you.", "Two records were merged during the cleanup and are listed below.", "The change window closes at the end of the week."], "title": ["Retry logic drops the last batch", "Timeout on cold start", "Duplicate rows after import", "Stale cache on rename", "Pagination skips the final page", "Rounding drift on totals", "Webhook fires twice", "Search misses hyphenated terms"], "label": ["backend", "urgent-review", "data", "regression", "billing", "infra", "ux"], "person": ["priya", "sam", "lena", "omar", "ines", "tomas", "yuki", "dana"], "query": ["waterproof duffel", "standing desk", "usb-c hub", "noise cancelling headset", "mechanical keyboard", "carbon steel pan", "merino base layer"], "category": ["luggage", "furniture", "peripherals", "audio", "kitchen", "apparel"], "sql": ["SELECT id, total FROM orders WHERE total > 100", "SELECT count(*) FROM customers WHERE created_at > '2026-01-01'", "SELECT sku, stock FROM inventory WHERE stock < 5", "SELECT region, sum(amount) FROM invoices GROUP BY region", "SELECT email FROM subscriptions WHERE status = 'past_due'"], "text": ["The maintenance window has been moved to Sunday.", "Your order has shipped and will arrive within three days.", "Please read the terms before continuing.", "The report is ready for download.", "Two seats remain on your current plan."], "domain": ["legal", "medical-devices", "aviation", "retail", "finance"], "room": ["B12", "Atrium", "Lab-3", "North Annex", "Room 214"], "date": ["2026-09-14", "2026-10-02", "2026-11-30", "2026-12-24", "2027-01-08"], "iso": ["2026-09-14T09:00:00Z", "2026-10-02T14:30:00Z", "2026-11-30T08:15:00Z", "2026-12-24T11:00:00Z", "2027-01-08T16:45:00Z"], "status": ["shipped", "pending", "closed", "refunded", "on_hold"], "note": ["adjusted after audit", "customer requested change", "duplicate of 4471", "manual correction", "reopened by support"], "junk_key": ["reason", "confidence", "_thought", "explanation", "step", "source", "chain_of_thought", "why", "note_to_self", "tool_choice_reason"], "prose": ["I do not have a tool that can do that.", "No function call is needed for this request.", "Let me think about which tool applies here.", "Sorry - I cannot help with that one.", "The user has not given me enough to go on.", "First I should ask a clarifying question.", "None of the available tools match.", "I will answer directly instead of calling a tool."], } def args_for(r, name, opts=()): """A schema-valid argument object, built from the schema, before any mutation.""" if name == "get_weather": a = {"location": r.choice(POOL["city"]), "units": r.choice(["celsius", "fahrenheit"]), "days": r.randint(1, 14)} if "include_hourly" in opts: a["include_hourly"] = r.choice([True, False]) return a if name == "send_email": a = {"to": [r.choice(POOL["email"])], "subject": "%s %d" % (r.choice(POOL["subject"]), r.randint(100, 999)), "body": r.choice(POOL["body"]), "priority": r.choice(["low", "normal", "high"])} if "cc" in opts: a["cc"] = [r.choice(POOL["email"])] return a if name == "create_ticket": a = {"title": "%s (%d)" % (r.choice(POOL["title"]), r.randint(100, 999)), "priority": r.choice(["low", "medium", "high", "urgent"])} if "labels" in opts: a["labels"] = [r.choice(POOL["label"])] if "assignee" in opts: a["assignee"] = r.choice(POOL["person"]) if "estimate_hours" in opts: a["estimate_hours"] = r.choice([0.5, 1.5, 3.0, 8.0]) return a if name == "search_products": a = {"query": "%s %d" % (r.choice(POOL["query"]), r.randint(10, 99)), "filters": {"category": r.choice(POOL["category"]), "max_price": float(r.randrange(20, 400, 5)), "in_stock": r.choice([True, False])}, "limit": r.choice([5, 10, 20, 25, 50])} if "sort" in opts: a["sort"] = r.choice(["relevance", "price_asc", "price_desc"]) return a if name == "book_flight": o, d = r.sample(POOL["city"], 2) a = {"origin": o, "destination": d, "date": r.choice(POOL["date"]), "passengers": r.randint(1, 6), "cabin": r.choice(["economy", "premium", "business", "first"])} if "seat_prefs" in opts: a["seat_prefs"] = [r.choice(["aisle", "window", "exit_row", "forward"])] return a if name == "schedule_meeting": a = {"title": "%s %d" % (r.choice(POOL["subject"]), r.randint(10, 99)), "attendees": [r.choice(POOL["email"])], "start_iso": r.choice(POOL["iso"]), "duration_minutes": r.choice([15, 30, 45, 60, 90])} if "room" in opts: a["room"] = r.choice(POOL["room"]) if "agenda" in opts: a["agenda"] = [r.choice(POOL["subject"]), r.choice(POOL["subject"])] return a if name == "translate_text": a = {"text": "%s (%d)" % (r.choice(POOL["text"]), r.randint(100, 999)), "target_lang": r.choice(["en", "fr", "de", "es", "ja"]), "formal": r.choice([True, False])} if "glossary" in opts: a["glossary"] = {"domain": r.choice(POOL["domain"]), "strict": r.choice([True, False])} return a if name == "run_query": a = {"sql": "%s LIMIT %d" % (r.choice(POOL["sql"]), r.randint(10, 999)), "database": r.choice(["analytics", "billing", "crm"]), "timeout_seconds": r.choice([5, 15, 30, 60, 120])} if "dry_run" in opts: a["dry_run"] = r.choice([True, False]) if "tags" in opts: a["tags"] = [r.choice(POOL["label"])] return a if name == "convert_currency": s, t = r.sample(["EUR", "USD", "GBP", "JPY"], 2) a = {"amount": round(r.uniform(5, 5000), 2), "source_currency": s, "target_currency": t, "rounding": r.choice(["none", "nearest", "bankers"])} if "on_date" in opts: a["on_date"] = r.choice(POOL["date"]) return a if name == "update_record": a = {"table": r.choice(["customers", "orders", "invoices"]), "record_id": r.randint(1000, 99999), "fields": {"status": r.choice(POOL["status"])}, "upsert": r.choice([True, False])} if "fields_note" in opts: a["fields"]["note"] = r.choice(POOL["note"]) if "reason_code" in opts: a["reason_code"] = "RC-%d" % r.randint(10, 99) return a raise AssertionError("no argument factory for %s" % name) OPTIONAL_ARGS = { "get_weather": ["include_hourly"], "send_email": ["cc"], "create_ticket": ["labels", "assignee", "estimate_hours"], "search_products": ["sort"], "book_flight": ["seat_prefs"], "schedule_meeting": ["room", "agenda"], "translate_text": ["glossary"], "run_query": ["dry_run", "tags"], "convert_currency": ["on_date"], "update_record": ["fields_note", "reason_code"], } # optional PROPERTIES (schema level) that can be truncated, per tool TRUNCATABLE = { "get_weather": [], "send_email": ["cc"], "create_ticket": ["labels", "assignee"], "search_products": ["sort"], "book_flight": ["seat_prefs"], "schedule_meeting": ["room", "agenda"], "translate_text": [], "run_query": ["tags"], "convert_currency": ["on_date"], "update_record": ["reason_code"], } # ------------------------------------------------------------------- validation def validate(v, schema, path="args"): """A JSON Schema subset: type, enum, properties, required, items, additionalProperties:false. Returns a list of error strings.""" errs = [] t = schema.get("type") if t == "object": if not isinstance(v, dict): return ["%s: expected object, got %s" % (path, type(v).__name__)] props = schema.get("properties", {}) for k in schema.get("required", []): if k not in v: errs.append("%s.%s: required property missing" % (path, k)) if schema.get("additionalProperties") is False: for k in v: if k not in props: errs.append("%s.%s: undeclared property" % (path, k)) for k, val in v.items(): if k in props: errs += validate(val, props[k], "%s.%s" % (path, k)) elif t == "array": if not isinstance(v, list): return ["%s: expected array, got %s" % (path, type(v).__name__)] if "items" in schema: for i, x in enumerate(v): errs += validate(x, schema["items"], "%s[%d]" % (path, i)) elif t == "string": if not isinstance(v, str): errs.append("%s: expected string" % path) elif t == "integer": if isinstance(v, bool) or not isinstance(v, int): errs.append("%s: expected integer" % path) elif t == "number": if isinstance(v, bool) or not isinstance(v, (int, float)): errs.append("%s: expected number" % path) elif t == "boolean": if not isinstance(v, bool): errs.append("%s: expected boolean" % path) if "enum" in schema and v not in schema["enum"]: errs.append("%s: %s is not one of %s" % (path, json.dumps(v), schema["enum"])) return errs def norm_name(x): """Namespace, separators, case and call parentheses are noise; identity is not.""" if not isinstance(x, str): return None s = x.strip() if s.endswith("()"): s = s[:-2] for sep in (".", ":", "/"): if sep in s: s = s.rsplit(sep, 1)[1] return s.strip().lower().replace("_", "").replace("-", "").replace(" ", "") def norm_enum(x): if not isinstance(x, str): return None return x.strip().lower().replace("-", "_").replace(" ", "_") def already_valid(text, tools): """True if the raw model text is ALREADY a schema-valid call: a freebie.""" try: d = json.loads(text) except Exception: return False if not isinstance(d, dict) or "name" not in d: return False tool = next((t for t in tools if t["name"] == d.get("name")), None) if tool is None: return False a = d.get("arguments", {}) if not isinstance(a, dict): return False return not validate(a, tool["parameters"]) def canon(v): return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False) # ------------------------------------------------------------------- generation def declare(r, target, extra_avoid=()): others = [n for n in TOOL_NAMES if n != target and n not in extra_avoid] r.shuffle(others) names = [target] + others[:r.choice([0, 1, 2])] r.shuffle(names) return [BY_NAME[n] for n in names] def dumps_call(r, name, args, indent="auto", args_first=False): if indent == "auto": indent = r.choice([None, 2]) d = {"arguments": args, "name": name} if args_first else {"name": name, "arguments": args} return json.dumps(d, indent=indent, ensure_ascii=False) MANGLERS = [ ("upper case name", lambda n: n.upper()), ("pascal case name", lambda n: "".join(w.capitalize() for w in n.split("_"))), ("namespace prefix", lambda n: "functions." + n), ("hyphens for underscores", lambda n: n.replace("_", "-")), ("call parentheses appended", lambda n: n + "()"), ("surrounding whitespace", lambda n: " " + n + " "), ("colon namespace", lambda n: "tools:" + n), ("camel case name", lambda n: n.split("_")[0] + "".join(w.capitalize() for w in n.split("_")[1:])), ("spaces for underscores", lambda n: n.replace("_", " ")), ] def g_wrong_tool_name(r): name = r.choice(TOOL_NAMES) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) variant, f = r.choice(MANGLERS) called = f(name) tools = declare(r, name) text = dumps_call(r, called, args) return dict(variant=variant, tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("The model wrote the tool name as %r. Under the identity rule " "(namespace, separators, case and trailing parentheses are noise) " "that resolves to exactly one declared tool, %r, and to no other. " "The arguments were already correct, so nothing else changes." % (called, name))) def g_missing_required_arg(r): cands = [(n, p) for n in TOOL_NAMES for p in BY_NAME[n]["parameters"]["required"] if "default" in BY_NAME[n]["parameters"]["properties"][p]] name, prop = r.choice(cands) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) default = BY_NAME[name]["parameters"]["properties"][prop]["default"] sent = {k: v for k, v in args.items() if k != prop} expected_args = dict(sent) expected_args[prop] = default tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant="required property omitted, declared default fills it", tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": expected_args}, rationale=("%r is required and was not sent. The schema declares " "default %s for it, so the call is completable without " "inventing anything: the default is the schema's own answer, " "not the repairer's guess. A required property with no " "declared default is a different case and lives in " "`unrecoverable`." % (prop, json.dumps(default)))) def g_extra_undeclared_arg(r): name = r.choice(TOOL_NAMES) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) props = BY_NAME[name]["parameters"]["properties"] junk = [k for k in POOL["junk_key"] if k not in props] r.shuffle(junk) k = r.choice([1, 1, 2]) sent = dict(args) added = junk[:k] for j in added: sent[j] = r.choice([r.choice(POOL["prose"]), r.randint(1, 99), round(r.uniform(0, 1), 2), r.choice([True, False])]) tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant="undeclared properties added", tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("The model added %s, which the schema does not declare and " "which `additionalProperties: false` forbids. The repair is to " "drop them; every declared value was already correct. Passing " "them through is what breaks a strict server." % ", ".join(repr(x) for x in added))) COERCIONS = [ ("integer as a decimal string", "integer", lambda v, r: str(v)), ("number as a decimal string", "number", lambda v, r: str(v)), ("boolean as a lowercase string", "boolean", lambda v, r: "true" if v else "false"), ("boolean as a python literal string", "boolean", lambda v, r: "True" if v else "False"), ("integer as a whole float", "integer", lambda v, r: float(v)), ] def g_type_coercion(r): while True: name = r.choice(TOOL_NAMES) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.4] args = args_for(r, name, opts) props = BY_NAME[name]["parameters"]["properties"] variant, want, f = r.choice(COERCIONS) targets = [k for k, v in args.items() if k in props and props[k].get("type") == want and "enum" not in props[k]] if targets: break prop = r.choice(targets) sent = dict(args) sent[prop] = f(args[prop], r) tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant=variant, tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("%r is declared %s and arrived as %s. The conversion is " "lossless and reversible -- the JSON literal of the sent value " "reads back as exactly one value of the declared type -- so the " "repair is defined. A value that does not round-trip (\"soon\" " "for an integer) is not coercible and lives in `unrecoverable`." % (prop, want, json.dumps(sent[prop])))) def g_enum_violation(r): cands = [(n, p) for n in TOOL_NAMES for p, s in BY_NAME[n]["parameters"]["properties"].items() if "enum" in s] while True: name, prop = r.choice(cands) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) if prop in args: break args[prop] = r.choice(BY_NAME[name]["parameters"]["properties"][prop]["enum"]) break member = args[prop] forms = [("upper case enum member", member.upper()), ("title case enum member", member.title()), ("spaces for underscores", member.replace("_", " ")), ("hyphens for underscores", member.replace("_", "-")), ("padded enum member", " %s " % member), ("lower case enum member", member.lower())] forms = [(v, s) for v, s in forms if s != member] variant, sent_val = r.choice(forms) sent = dict(args) sent[prop] = sent_val tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant=variant, tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("%r must be one of %s. The model wrote %s, which after " "case-folding, trimming and treating '-' and ' ' as '_' matches " "exactly one member, %s, and no other. A value that matches none " "of them is not repairable and lives in `unrecoverable`." % (prop, BY_NAME[name]["parameters"]["properties"][prop]["enum"], json.dumps(sent_val), json.dumps(member)))) NESTED = {"search_products": ("filters", "sort"), "translate_text": ("glossary", None), "update_record": ("fields", "reason_code")} def g_nested_flattened(r): name = r.choice(sorted(NESTED)) prop = NESTED[name][0] opts = list(OPTIONAL_ARGS[name]) args = args_for(r, name, opts) if prop == "fields": args["fields"]["note"] = r.choice(POOL["note"]) nested = args[prop] top_props = set(BY_NAME[name]["parameters"]["properties"]) dotted = r.choice([True, False]) sent = {k: v for k, v in args.items() if k != prop} for k, v in nested.items(): sent["%s.%s" % (prop, k) if dotted else k] = v tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant="nested object flattened with dotted keys" if dotted else "nested object flattened onto the top level", tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("%r is declared as a nested object holding %s. The model wrote " "those keys %s. Each one belongs to exactly one nested property " "and to no declared top-level property, so re-nesting them is " "determined by the schema alone." % (prop, sorted(nested), "as dotted paths" if dotted else "flat, at the top level")) + (" No flattened key collides with %s." % sorted(top_props - {prop}))) ARRAY_PROPS = [("send_email", "to"), ("send_email", "cc"), ("create_ticket", "labels"), ("schedule_meeting", "attendees"), ("schedule_meeting", "agenda"), ("run_query", "tags"), ("book_flight", "seat_prefs")] SCALAR_PROPS = [("get_weather", "location"), ("get_weather", "days"), ("create_ticket", "title"), ("book_flight", "date"), ("run_query", "sql"), ("convert_currency", "amount"), ("update_record", "record_id"), ("translate_text", "text")] def g_array_vs_scalar(r): wrap = r.choice([True, False]) if wrap: name, prop = r.choice(ARRAY_PROPS) args = args_for(r, name, OPTIONAL_ARGS[name]) if prop not in args: args[prop] = [r.choice(POOL["label"])] args[prop] = [args[prop][0]] sent = dict(args) sent[prop] = args[prop][0] variant = "single value where an array is declared" why = ("%r is declared an array of %s and the model sent one bare %s. Wrapping it " "in a one-element array changes no data and is the only reading that " "satisfies the schema." % (prop, "strings", "string")) else: name, prop = r.choice(SCALAR_PROPS) args = args_for(r, name, [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3]) sent = dict(args) sent[prop] = [args[prop]] variant = "one-element array where a scalar is declared" why = ("%r is declared a scalar and the model wrapped it in a one-element array. " "Unwrapping loses nothing. An array of two or more here would be ambiguous " "and is not in this category." % prop) tools = declare(r, name) text = dumps_call(r, name, sent) return dict(variant=variant, tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=why) def g_args_as_string(r): name = r.choice(TOOL_NAMES) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) style = r.choice(["compact", "indented", "double encoded", "escaped newlines"]) if style == "compact": s = json.dumps(args, ensure_ascii=False) elif style == "indented": s = json.dumps(args, indent=2, ensure_ascii=False) elif style == "escaped newlines": s = json.dumps(args, ensure_ascii=False, separators=(",\n", ": ")) else: s = json.dumps(json.dumps(args, ensure_ascii=False), ensure_ascii=False) tools = declare(r, name) text = json.dumps({"name": name, "arguments": s}, ensure_ascii=False, indent=r.choice([None, 2])) return dict(variant="arguments delivered as a JSON string (%s)" % style, tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("`arguments` is a string, not an object -- the single most common " "shape in the wild, because several APIs specify it that way. " "Decoding it %s yields an object that validates as-is; nothing is " "guessed." % ("twice" if style == "double encoded" else "once"))) def g_multiple_calls(r): name = r.choice(TOOL_NAMES) opts = [o for o in OPTIONAL_ARGS[name] if r.random() < 0.3] args = args_for(r, name, opts) k = r.choice([2, 2, 3]) forms = [] for i in range(k): forms.append(dumps_call(r, name, args, indent=r.choice([None, 2]), args_first=(i % 2 == 1))) style = r.choice(["json array", "newline separated", "comma separated"]) if style == "json array": text = "[\n" + ",\n".join(forms) + "\n]" elif style == "newline separated": text = "\n".join(forms) else: text = ",\n".join(forms) tools = declare(r, name) return dict(variant="%d identical calls, %s" % (k, style), tools=tools, input=text, expected_kind="value", expected={"name": name, "arguments": args}, rationale=("The model emitted the same call %d times (key order and " "whitespace differ, the canonical form does not) where one call " "was allowed. Because every copy canonicalises to the same " "value, collapsing them picks nothing. Two calls that differ " "are ambiguous and live in `unrecoverable`." % k)) def g_hallucinated_tool(r): called = r.choice(HALLUCINATED) donor = r.choice(TOOL_NAMES) args = args_for(r, donor, []) decl_names = [n for n in TOOL_NAMES] r.shuffle(decl_names) tools = [BY_NAME[n] for n in decl_names[:r.choice([2, 3])]] text = dumps_call(r, called, args) return dict(variant="call to a tool that was never declared", tools=tools, input=text, expected_kind="unrecoverable", rationale=("%r is not among the declared tools %s and does not resolve to " "one under the identity rule. Choosing a 'nearest' tool would be " "the repairer inventing an intent the model never expressed, so " "the only honest output is a refusal." % (called, [t["name"] for t in tools]))) def g_truncated(r): cands = [(n, p) for n in TOOL_NAMES for p in TRUNCATABLE[n]] name, opt = r.choice(cands) req = BY_NAME[name]["parameters"]["required"] full = args_for(r, name, OPTIONAL_ARGS[name] + ["fields_note"]) kept = {k: v for k, v in full.items() if k in req} if opt not in full: raise AssertionError("optional property %r was not generated for %s" % (opt, name)) ordered = dict(kept) ordered[opt] = full[opt] text = json.dumps({"name": name, "arguments": ordered}, indent=2, ensure_ascii=False) tok = '"%s"' % opt start = text.rindex(tok) end = len(text) - len("\n }\n}") cut = r.randint(start + 1, end - 1) return dict(variant="cut inside the optional property %r" % opt, tools=declare(r, name), input=text[:cut], expected_kind="value", expected={"name": name, "arguments": kept}, rationale=("The stream stopped inside %r, which is optional. Every required " "property (%s) was completely written before the cut, so the " "rule is: drop the incomplete tail, close the open containers, " "invent nothing. The result validates. A cut that lands inside a " "REQUIRED property leaves a hole no default can fill and lives " "in `unrecoverable`." % (opt, ", ".join(req)))) def _u_missing_no_default(r): cands = [(n, p) for n in TOOL_NAMES for p in BY_NAME[n]["parameters"]["required"] if "default" not in BY_NAME[n]["parameters"]["properties"][p]] name, prop = r.choice(cands) args = args_for(r, name, []) sent = {k: v for k, v in args.items() if k != prop} return (name, dumps_call(r, name, sent), "required property omitted, no declared default", "%r is required and has no default in the schema. Nothing in the declaration or " "in the model's output determines it. Supplying a value here would be the " "repairer answering a question only the caller can answer." % prop) def _u_enum_unmappable(r): cands = [(n, p) for n in TOOL_NAMES for p, s in BY_NAME[n]["parameters"]["properties"].items() if "enum" in s] name, prop = r.choice(cands) args = args_for(r, name, []) if prop not in args: args[prop] = BY_NAME[name]["parameters"]["properties"][prop]["enum"][0] bad = r.choice(["asap", "whatever_is_cheapest", "auto", "default", "same as last time", "tbd", "unknown", "highest"]) sent = dict(args) sent[prop] = bad return (name, dumps_call(r, name, sent), "enum value that matches no member", "%r must be one of %s. %s matches none of them under case-folding, trimming or " "separator normalisation. Picking the closest-looking member would be a guess " "dressed as a repair." % (prop, BY_NAME[name]["parameters"]["properties"][prop]["enum"], json.dumps(bad))) def _u_uncoercible(r): cands = [(n, p) for n in TOOL_NAMES for p in BY_NAME[n]["parameters"]["required"] if BY_NAME[n]["parameters"]["properties"][p].get("type") in ("integer", "number") and "default" not in BY_NAME[n]["parameters"]["properties"][p]] name, prop = r.choice(cands) args = args_for(r, name, []) bad = r.choice(["soon", "a few", "several", "as many as needed", "", "N/A", "many"]) sent = dict(args) sent[prop] = bad return (name, dumps_call(r, name, sent), "value of the wrong type that cannot be coerced", "%r is declared %s and arrived as %s, which is not the JSON literal of any " "number. There is no lossless conversion, and there is no default to fall back " "on." % (prop, BY_NAME[name]["parameters"]["properties"][prop]["type"], json.dumps(bad))) def _u_ambiguous_multi(r): name = r.choice(TOOL_NAMES) a1 = args_for(r, name, []) a2 = args_for(r, name, []) if canon(a1) == canon(a2): a2 = args_for(r, name, OPTIONAL_ARGS[name]) other = r.choice([n for n in TOOL_NAMES if n != name]) two_names = r.random() < 0.4 c1 = dumps_call(r, name, a1) c2 = dumps_call(r, other, args_for(r, other, [])) if two_names else dumps_call(r, name, a2) text = "[\n" + c1 + ",\n" + c2 + "\n]" return (name, text, "two different calls where one was allowed", "Two calls arrived and they do not canonicalise to the same value, so collapsing " "them means choosing one. The schema does not say which, and neither does the " "model. Order of appearance is not evidence of intent.") def _u_no_call(r): name = r.choice(TOOL_NAMES) text = r.choice([r.choice(POOL["prose"]), "", " ", "\n\n", r.choice(POOL["prose"]) + "\n" + r.choice(POOL["prose"])]) return (name, text, "no call in the output at all", "There is no tool call in this output -- only prose or nothing. Returning an " "empty object here is the failure this suite exists to measure: it turns " "'the model did not call a tool' into 'the model called a tool with no " "arguments', which a server will happily execute.") def _u_truncated_required(r): cands = [(n, p) for n in TOOL_NAMES for p in BY_NAME[n]["parameters"]["required"] if "default" not in BY_NAME[n]["parameters"]["properties"][p]] name, last = r.choice(cands) req = BY_NAME[name]["parameters"]["required"] args = args_for(r, name, []) ordered = {k: v for k, v in args.items() if k != last} ordered[last] = args[last] text = json.dumps({"name": name, "arguments": ordered}, indent=2, ensure_ascii=False) tok = '"%s"' % last start = text.rindex(tok) end = len(text) - len("\n }\n}") cut = r.randint(start + 1, end - 1) return (name, text[:cut], "cut inside a required property", "The stream stopped inside %r, which is required and has no default. Dropping " "the incomplete tail -- the correct rule -- leaves a required property missing, " "and no rule fills it. Completing the half-written value would be invention." % last) def _u_args_string_broken(r): """The cut lands inside a REQUIRED property that has no default, so the case is a refusal under either reading -- 'the string does not decode' (rule 9) and 'drop the incomplete tail' (rule 11) both end with a hole nothing can fill.""" style = r.choice(["truncated", "prose"]) if style == "prose": name = r.choice(TOOL_NAMES) broken = r.choice(["see above", "the arguments from my previous message", "as discussed", "same as before", "unchanged"]) why = ("`arguments` arrived as a string, which is normal, but the string is prose. " "There is no object to decode and no call to validate.") else: cands = [(n, p) for n in TOOL_NAMES for p in BY_NAME[n]["parameters"]["required"] if "default" not in BY_NAME[n]["parameters"]["properties"][p]] name, last = r.choice(cands) args = args_for(r, name, []) ordered = {k: v for k, v in args.items() if k != last} ordered[last] = args[last] s = json.dumps(ordered, ensure_ascii=False) tok = '"%s"' % last start = s.rindex(tok) broken = s[:r.randint(start + 1, len(s) - 2)] why = ("`arguments` arrived as a string and the string is cut inside %r, which is " "required and has no default. Decoding fails; and closing the open " "containers -- the truncation rule -- still leaves %r missing. Both " "readings end in a refusal." % (last, last)) text = json.dumps({"name": name, "arguments": broken}, ensure_ascii=False) return (name, text, "arguments string that does not decode", why) U_CAUSES = [_u_missing_no_default, _u_enum_unmappable, _u_uncoercible, _u_ambiguous_multi, _u_no_call, _u_truncated_required, _u_args_string_broken] def g_unrecoverable(r, which=None): f = which if which is not None else r.choice(U_CAUSES) name, text, variant, why = f(r) return dict(variant=variant, tools=declare(r, name), input=text, expected_kind="unrecoverable", rationale=why) GENERATORS = { "wrong_tool_name": g_wrong_tool_name, "missing_required_arg": g_missing_required_arg, "extra_undeclared_arg": g_extra_undeclared_arg, "type_coercion": g_type_coercion, "enum_violation": g_enum_violation, "nested_flattened": g_nested_flattened, "array_vs_scalar": g_array_vs_scalar, "args_as_string": g_args_as_string, "multiple_calls": g_multiple_calls, "hallucinated_tool": g_hallucinated_tool, "truncated": g_truncated, "unrecoverable": g_unrecoverable, } def build(seed): r = random.Random(seed) cases, seen, n = [], set(), 0 for cat in CATEGORIES: made = 0 tries = 0 while made < N_PER_CATEGORY: tries += 1 if tries > 4000: raise AssertionError("could not make %d unique inputs for %s" % (N_PER_CATEGORY, cat)) if cat == "unrecoverable": cause = U_CAUSES[made % len(U_CAUSES)] c = g_unrecoverable(r, cause) else: c = GENERATORS[cat](r) if c["input"] in seen: continue seen.add(c["input"]) n += 1 case = {"id": "tc300-%04d" % n, "category": cat, "variant": c["variant"], "tools": c["tools"], "input": c["input"], "expected_kind": c["expected_kind"], "rationale": c["rationale"]} if c["expected_kind"] == "value": case["expected"] = c["expected"] cases.append(case) made += 1 return cases def stratified_sample(cases): """2 or 3 per category, deterministic, 30 in total.""" out = [] for i, cat in enumerate(CATEGORIES): pool = [c for c in cases if c["category"] == cat] k = 3 if i < 6 else 2 idxs = [0, len(pool) // 2, len(pool) - 1][:k] out += [pool[j] for j in idxs] return out # ---------------------------------------------------------------------- checks CHECK_LOG = [] def check(name, cond, detail=""): CHECK_LOG.append(name) if not cond: raise AssertionError("INTEGRITY CHECK FAILED -- %s %s" % (name, detail)) return True CHECKS = """ 1 corpus holds exactly 300 cases 2 every category holds exactly 25 cases 3 all 300 ids are unique and sequential 4 all 300 input texts are unique 5 every case declares at least one tool, and every declared tool schema is well formed (object type, properties present, required is a subset of properties) 6 every expected call names a tool that IS declared in that case 7 every expected argument object validates against that tool's declared schema 8 no recoverable case is already schema-valid as sent -- no freebies 9 no unrecoverable case is schema-valid as sent either 10 every unrecoverable case carries no `expected` key, every value case carries one 11 every mangled tool name in wrong_tool_name resolves to exactly one declared tool 12 no hallucinated name collides with any tool in the catalogue, under the identity rule 13 every enum_violation value maps to exactly one declared member, and every unrecoverable enum value maps to none 14 every truncated input genuinely fails to parse as JSON 15 every missing_required_arg case drops a property that HAS a declared default, and every unrecoverable missing-property case drops one that does NOT 16 sample30 covers all 12 categories, is 30 cases, and is a subset of the corpus 17 sample30 and paid270 are disjoint and together are exactly the corpus 18 the build is deterministic: the same seed rebuilds byte-identical JSONL 19 no case text contains a NUL or an unpaired surrogate 20 every multiple_calls case's copies canonicalise to one value; every ambiguous unrecoverable multi-call's copies do not 21 every broken `arguments` string stays unrecoverable even under the most generous truncation reading (close the containers, drop the tail) 22 no undeclared key in extra_undeclared_arg is also the name of a nested property -- otherwise "drop it" and "re-nest it" would both be defensible 23 every flattened key belongs to exactly one nested property and to no top-level one""" def run_checks(cases, sample, paid, seed): check("1 exactly 300 cases", len(cases) == 300, len(cases)) for cat in CATEGORIES: k = sum(1 for c in cases if c["category"] == cat) check("2 category %s == 25" % cat, k == N_PER_CATEGORY, k) ids = [c["id"] for c in cases] check("3 ids unique and sequential", len(set(ids)) == 300 and ids == ["tc300-%04d" % i for i in range(1, 301)]) inputs = [c["input"] for c in cases] check("4 inputs unique", len(set(inputs)) == 300, len(set(inputs))) for c in cases: check("5 tools well formed %s" % c["id"], len(c["tools"]) >= 1) for t in c["tools"]: p = t["parameters"] check("5 schema shape %s/%s" % (c["id"], t["name"]), p.get("type") == "object" and isinstance(p.get("properties"), dict) and set(p.get("required", [])) <= set(p["properties"])) for c in cases: if c["expected_kind"] != "value": continue names = [t["name"] for t in c["tools"]] check("6 expected name declared %s" % c["id"], c["expected"]["name"] in names, c["expected"]["name"]) tool = next(t for t in c["tools"] if t["name"] == c["expected"]["name"]) errs = validate(c["expected"]["arguments"], tool["parameters"]) check("7 expected validates %s" % c["id"], not errs, errs) for c in cases: av = already_valid(c["input"], c["tools"]) if c["expected_kind"] == "value": check("8 no freebie %s" % c["id"], not av) else: check("9 no valid unrecoverable %s" % c["id"], not av) for c in cases: check("10 expected key presence %s" % c["id"], ("expected" in c) == (c["expected_kind"] == "value")) for c in cases: if c["category"] != "wrong_tool_name": continue called = json.loads(c["input"])["name"] hits = [t["name"] for t in c["tools"] if norm_name(t["name"]) == norm_name(called)] check("11 name resolves uniquely %s" % c["id"], len(hits) == 1 and hits[0] == c["expected"]["name"], (called, hits)) for c in cases: if c["category"] != "hallucinated_tool": continue called = json.loads(c["input"])["name"] check("12 hallucinated name unknown %s" % c["id"], norm_name(called) not in {norm_name(n) for n in TOOL_NAMES}, called) for c in cases: if c["category"] == "enum_violation": sent = json.loads(c["input"])["arguments"] tool = next(t for t in c["tools"] if t["name"] == c["expected"]["name"]) props = tool["parameters"]["properties"] hit = 0 for k, v in sent.items(): s = props.get(k, {}) if "enum" in s and v not in s["enum"]: ms = [m for m in s["enum"] if norm_enum(m) == norm_enum(v)] check("13 enum maps to one %s" % c["id"], len(ms) == 1 and ms[0] == c["expected"]["arguments"][k], (v, ms)) hit += 1 check("13 exactly one enum broken %s" % c["id"], hit == 1, hit) elif c["category"] == "unrecoverable" and "enum value that matches no" in c["variant"]: sent = json.loads(c["input"])["arguments"] tool = next(t for t in c["tools"] if t["name"] == json.loads(c["input"])["name"]) props = tool["parameters"]["properties"] bad = 0 for k, v in sent.items(): s = props.get(k, {}) if "enum" in s and v not in s["enum"]: check("13 enum maps to none %s" % c["id"], not [m for m in s["enum"] if norm_enum(m) == norm_enum(v)], v) bad += 1 check("13 one broken enum %s" % c["id"], bad == 1, bad) for c in cases: if c["category"] == "truncated" or "cut inside" in c["variant"]: bad = False try: json.loads(c["input"]) except Exception: bad = True check("14 truncated does not parse %s" % c["id"], bad) for c in cases: if c["category"] == "missing_required_arg": tool = next(t for t in c["tools"] if t["name"] == c["expected"]["name"]) sent = json.loads(c["input"])["arguments"] miss = [k for k in tool["parameters"]["required"] if k not in sent] check("15 one required property missing %s" % c["id"], len(miss) == 1, miss) check("15 missing property has a default %s" % c["id"], "default" in tool["parameters"]["properties"][miss[0]]) check("15 default is the expected value %s" % c["id"], c["expected"]["arguments"][miss[0]] == tool["parameters"]["properties"][miss[0]]["default"]) elif c["category"] == "unrecoverable" and "no declared default" in c["variant"]: d = json.loads(c["input"]) tool = next(t for t in c["tools"] if t["name"] == d["name"]) miss = [k for k in tool["parameters"]["required"] if k not in d["arguments"]] check("15 missing without default %s" % c["id"], len(miss) == 1, miss) check("15 no default to fill %s" % c["id"], "default" not in tool["parameters"]["properties"][miss[0]]) check("16 sample is 30", len(sample) == 30, len(sample)) check("16 sample covers 12 categories", {c["category"] for c in sample} == set(CATEGORIES)) sids, cids = {c["id"] for c in sample}, {c["id"] for c in cases} check("16 sample is a subset", sids <= cids) pids = {c["id"] for c in paid} check("17 sample and paid are disjoint", not (sids & pids)) check("17 sample plus paid is the corpus", sids | pids == cids and len(paid) == 270) again = build(seed) check("18 build is deterministic", jsonl(again) == jsonl(cases)) for c in cases: s = c["input"] check("19 clean text %s" % c["id"], "\x00" not in s and s.encode("utf-8", "strict") is not None) for c in cases: if c["category"] == "unrecoverable" and "does not decode" in c["variant"]: d = json.loads(c["input"]) tool = next(t for t in c["tools"] if t["name"] == d["name"]) got = close_repair(d["arguments"]) if isinstance(got, dict): holes = [k for k in tool["parameters"]["required"] if k not in got and "default" not in tool["parameters"]["properties"][k]] check("21 broken args string stays unrecoverable %s" % c["id"], len(holes) >= 1, got) else: check("21 broken args string does not close %s" % c["id"], got is None) if c["category"] == "extra_undeclared_arg": tool = next(t for t in c["tools"] if t["name"] == c["expected"]["name"]) props = tool["parameters"]["properties"] nested_keys = set() for k, sc in props.items(): if sc.get("type") == "object": nested_keys |= set(sc.get("properties", {})) sent = json.loads(c["input"])["arguments"] extra = [k for k in sent if k not in props] check("22 undeclared keys are droppable, not re-nestable %s" % c["id"], not (set(extra) & nested_keys), extra) if c["category"] == "nested_flattened": tool = next(t for t in c["tools"] if t["name"] == c["expected"]["name"]) props = tool["parameters"]["properties"] sent = json.loads(c["input"])["arguments"] for k in sent: if k in props: continue leaf = k.split(".")[-1] owners = [p for p, sc in props.items() if sc.get("type") == "object" and leaf in sc.get("properties", {})] check("23 flattened key has exactly one owner %s" % c["id"], len(owners) == 1, (k, owners)) check("23 flattened key is not a top-level property %s" % c["id"], leaf not in props, k) if c["category"] == "multiple_calls": calls = split_calls(c["input"]) check("20 copies are identical %s" % c["id"], len(calls) >= 2 and len({canon(x) for x in calls}) == 1, len(calls)) elif c["category"] == "unrecoverable" and "two different calls" in c["variant"]: calls = split_calls(c["input"]) check("20 copies differ %s" % c["id"], len(calls) == 2 and len({canon(x) for x in calls}) == 2) def close_repair(s): """The most generous reading of a truncated fragment: drop the incomplete tail and close what is open. Used ONLY by the integrity checks, to prove that a case labelled unrecoverable stays unrecoverable even under that reading.""" t = s.rstrip() for cut in range(len(t), 0, -1): frag = t[:cut] stack, instr, esc = [], False, False for ch in frag: if instr: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': instr = False continue if ch == '"': instr = True elif ch in "{[": stack.append(ch) elif ch in "}]": if stack: stack.pop() if instr: continue cand = frag.rstrip() while cand and cand[-1] in ",:": cand = cand[:-1].rstrip() closers = "".join("}" if c == "{" else "]" for c in reversed(stack)) try: return json.loads(cand + closers) except Exception: continue return None def split_calls(text): """Split a text holding several JSON objects into those objects. Used by the integrity checks only; the scorer never needs it.""" t = text.strip() try: v = json.loads(t) if isinstance(v, list): return v return [v] except Exception: pass out, depth, start, instr, esc = [], 0, None, False, False for i, ch in enumerate(t): if instr: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': instr = False continue if ch == '"': instr = True elif ch == "{": if depth == 0: start = i depth += 1 elif ch == "}": depth -= 1 if depth == 0 and start is not None: try: out.append(json.loads(t[start:i + 1])) except Exception: pass start = None return out # ----------------------------------------------------------------------- output def jsonl(cases, drop=()): buf = io.StringIO() for c in cases: d = {k: v for k, v in c.items() if k not in drop} buf.write(json.dumps(d, ensure_ascii=False) + "\n") return buf.getvalue() def write(path, text): with io.open(path, "w", encoding="utf-8", newline="\n") as f: f.write(text) return os.path.getsize(path) def sha(path): h = hashlib.sha256() with open(path, "rb") as f: h.update(f.read()) return h.hexdigest() def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", default=".") ap.add_argument("--seed", type=int, default=SEED_DEFAULT) ap.add_argument("--check-only", action="store_true") ap.add_argument("--list-checks", action="store_true") a = ap.parse_args() if a.list_checks: print(CHECKS) return 0 cases = build(a.seed) sample = stratified_sample(cases) sids = {c["id"] for c in sample} paid = [c for c in cases if c["id"] not in sids] run_checks(cases, sample, paid, a.seed) print("%d integrity checks ran, %d failed" % (len(CHECK_LOG), 0)) if a.check_only: full = os.path.join(a.out, "toolcall300.jsonl") if os.path.exists(full): on_disk = io.open(full, encoding="utf-8").read() same = on_disk == jsonl(cases) print("on-disk corpus matches a fresh build: %s" % same) print("sha256 %s %s" % (sha(full), full)) return 0 if same else 2 return 0 n1 = write(os.path.join(a.out, "toolcall300.jsonl"), jsonl(cases)) n2 = write(os.path.join(a.out, "sample30.jsonl"), jsonl(sample, drop=("rationale",))) n3 = write(os.path.join(a.out, "paid270.jsonl"), jsonl(paid)) for nm, n in (("toolcall300.jsonl", n1), ("sample30.jsonl", n2), ("paid270.jsonl", n3)): p = os.path.join(a.out, nm) print("%-22s %8d B sha256 %s" % (nm, n, sha(p))) counts = {} for c in cases: counts[c["category"]] = counts.get(c["category"], 0) + 1 print("\nper category: " + " ยท ".join("%s %d" % (k, counts[k]) for k in CATEGORIES)) print("refusal-expected cases: %d of %d" % (sum(1 for c in cases if c["expected_kind"] == "unrecoverable"), len(cases))) return 0 if __name__ == "__main__": sys.exit(main())