"""Kimi K3.1 Pro reasoning-effort test (AY Automate, 2026-10-25). 8 tasks with automatic checks, run on moonshotai/kimi-k3.1-pro at reasoning_effort low / high / max and on anthropic/claude-sonnet-5.5 at its default, via OpenRouter. Records correctness, completion + reasoning tokens, wall-clock latency and the cost OpenRouter bills (usage.cost). Usage: OPENROUTER_API_KEY=... python3 run.py out.jsonl """ import json, os, re, subprocess, sys, tempfile, time, urllib.request KEY = os.environ["OPENROUTER_API_KEY"] OUT = sys.argv[1] if len(sys.argv) > 1 else "results.jsonl" def fib_mod(n, m=1_000_000_007): def fd(k): if k == 0: return (0, 1) a, b = fd(k >> 1) c = a * ((2 * b - a) % m) % m d = (a * a + b * b) % m return (d, (c + d) % m) if k & 1 else (c, d) return fd(n)[0] def run_code(code, test): with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "t.py") open(p, "w").write(code + "\n\n" + test) try: r = subprocess.run([sys.executable, "-I", p], capture_output=True, timeout=20, cwd=d) return r.returncode == 0 except subprocess.TimeoutExpired: return False def code_block(text): m = re.findall(r"```(?:python)?\n(.*?)```", text, re.S) return m[-1] if m else text def last_line(text): lines = [l.strip() for l in text.strip().splitlines() if l.strip()] return lines[-1] if lines else "" def num(text): m = re.findall(r"-?\d[\d,]*\.?\d*", last_line(text)) return m[-1].replace(",", "") if m else None TASKS = [ {"id": "primes", "prompt": "What is the sum of all prime numbers below 100? End with a final line containing only the number.", "check": lambda t: num(t) == "1060"}, {"id": "letters", "prompt": "How many times does the letter 'r' appear in the phrase 'strawberry raspberry rhubarb'? End with a final line containing only the number.", "check": lambda t: num(t) == "8"}, {"id": "fake-date", "prompt": "What day of the week was February 29, 2026? End with a final line containing only the answer.", "check": lambda t: bool(re.search(r"not|no such|doesn|did not|didn|isn|invalid|leap", last_line(t), re.I))}, {"id": "pens", "prompt": "Pens cost $4 for a pack of 3 or $1.50 each. What is the minimum cost, in dollars, to buy exactly 17 pens? End with a final line containing only the number.", "check": lambda t: num(t) in ("23", "23.0", "23.00")}, {"id": "extract", "prompt": "Extract JSON with keys invoice_number (string), total (number), currency (ISO code), due_date (YYYY-MM-DD) from: 'Inv. no. A-1043 issued 3 Oct 2026. Amount due: EUR 1.250,40, payable within 30 days of issue.' Reply with only the JSON object.", "check": lambda t: (lambda j: j and j.get("invoice_number") == "A-1043" and abs(float(j.get("total", 0)) - 1250.40) < 0.01 and j.get("currency") == "EUR" and j.get("due_date") == "2026-11-02")( (lambda s: json.loads(s) if s else None)((re.search(r"\{.*\}", t, re.S) or [None])[0] if re.search(r"\{.*\}", t, re.S) else None))}, {"id": "roman", "prompt": "Write a Python function roman_to_int(s: str) -> int that converts a Roman numeral (1-3999) to an integer and raises ValueError for invalid numerals such as 'IIII', 'VX', 'IC' or ''. Reply with only the code in one ```python block.", "check": lambda t: run_code(code_block(t), "assert roman_to_int('MCMXCIV')==1994\nassert roman_to_int('MMMCMXCIX')==3999\nassert roman_to_int('XLII')==42\nfor b in ['IIII','VX','IC','','MMMM','IL']:\n try:\n roman_to_int(b)\n raise SystemExit(1)\n except ValueError:\n pass")}, {"id": "fib", "prompt": "Write a Python function fib_mod(n: int) -> int returning the n-th Fibonacci number (fib_mod(0)=0, fib_mod(1)=1) modulo 1_000_000_007. It must handle n up to 10**18 in well under a second. Reply with only the code in one ```python block.", "check": lambda t: run_code(code_block(t), f"assert fib_mod(0)==0 and fib_mod(1)==1 and fib_mod(10)==55\nassert fib_mod(10**18)=={fib_mod(10**18)}")}, {"id": "intervals", "prompt": "Write a Python function merge(intervals: list[list[int]]) -> list[list[int]] that merges overlapping or touching closed intervals and returns them sorted. merge([[1,3],[2,6],[8,10],[10,12]]) == [[1,6],[8,12]]. Reply with only the code in one ```python block.", "check": lambda t: run_code(code_block(t), "assert merge([[1,3],[2,6],[8,10],[10,12]])==[[1,6],[8,12]]\nassert merge([])==[]\nassert merge([[5,7],[1,2]])==[[1,2],[5,7]]\nassert merge([[1,10],[2,3]])==[[1,10]]")}, ] CONFIGS = [ ("moonshotai/kimi-k3.1-pro", "low"), ("moonshotai/kimi-k3.1-pro", "high"), ("moonshotai/kimi-k3.1-pro", "max"), ("anthropic/claude-sonnet-5.5", None), ] def call(model, effort, prompt): body = {"model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": 32000, "usage": {"include": True}} if effort: body["reasoning_effort"] = effort req = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=json.dumps(body).encode(), headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}) t0 = time.time() with urllib.request.urlopen(req, timeout=900) as r: d = json.load(r) return d, time.time() - t0 with open(OUT, "a") as f: for model, effort in CONFIGS: for task in TASKS: try: d, secs = call(model, effort, task["prompt"]) text = d["choices"][0]["message"].get("content") or "" u = d.get("usage", {}) try: ok = bool(task["check"](text)) except Exception: ok = False rec = {"model": model, "effort": effort or "default", "task": task["id"], "correct": ok, "seconds": round(secs, 1), "prompt_tokens": u.get("prompt_tokens"), "completion_tokens": u.get("completion_tokens"), "reasoning_tokens": (u.get("completion_tokens_details") or {}).get("reasoning_tokens"), "cost_usd": u.get("cost"), "provider": d.get("provider"), "finish": d["choices"][0].get("finish_reason"), "answer_tail": text[-300:]} except Exception as e: rec = {"model": model, "effort": effort or "default", "task": task["id"], "error": str(e)[:300]} f.write(json.dumps(rec) + "\n"); f.flush() print(json.dumps({k: rec.get(k) for k in ("model", "effort", "task", "correct", "seconds", "reasoning_tokens", "cost_usd", "error")}), flush=True)