| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| from __future__ import annotations |
|
|
| import json |
| import re |
| import subprocess |
| import sys |
| import tempfile |
| import time |
|
|
| |
| _MODELS = ("qwen/qwen3.7-flash", "deepseek/deepseek-v4-flash", "deepseek/deepseek-v4-pro", |
| "z-ai/glm-5.2", "openai/gpt-5.6-luna", "google/gemini-3.6-flash", "moonshotai/kimi-k3") |
|
|
| _DEFAULTS = { |
| "draft_model": "openai/gpt-5.6-luna", |
| |
| |
| |
| |
| |
| |
| |
| |
| "code_model": "openai/gpt-5.6-luna", |
| |
| |
| "code_ladder": ["moonshotai/kimi-k3", "google/gemini-3.6-flash", "deepseek/deepseek-v4-pro"], |
| "code_attempts": 3, |
| "code_retry_needs_s": 220.0, |
| "code_max_tokens": 8192, |
| |
| |
| |
| |
| |
| |
| |
| |
| "repair_model": "openai/gpt-5.6-luna", |
| "repair_max_tokens": 8192, |
| "repair_task_s": 90.0, |
| |
| |
| |
| |
| |
| |
| "vote_models": ["google/gemini-3.6-flash", "deepseek/deepseek-v4-pro", "z-ai/glm-5.2"], |
| "vote_max_tokens": 8192, |
| "vote_sizes": [2, 8, 40, 200, 1000], |
| "vote_task_s": 240.0, |
| "vote_needs_s": 200.0, |
| |
| |
| |
| |
| "vote_step_s": 20.0, |
| |
| |
| |
| |
| "oracle_attempts": 2, |
| |
| |
| |
| |
| |
| |
| |
| "escalate_model": "moonshotai/kimi-k3", |
| "escalate_max_tokens": 16384, |
| "escalate_needs_s": 520.0, |
| |
| |
| |
| "max_stress_repairs": 3, |
| |
| |
| |
| |
| "max_code_calls": 9, |
| "oracle_timeout_s": 6.0, |
| "oracle_needs_s": 150.0, |
| "oracle_task_s": 260.0, |
| "stress_sizes": (2, 3, 5, 8, 12, 20, 40), |
| |
| |
| |
| |
| "stress_rounds": 60, |
| "stress_budget_s": 90.0, |
| "strong_model": "moonshotai/kimi-k3", |
| "math_model": "openai/gpt-5.6-luna", |
| "mmlu_model": "openai/gpt-5.6-luna", |
| "budget_s": 600.0, |
| "min_task_s": 25.0, |
| "max_repairs": 4, |
| |
| |
| |
| |
| |
| |
| |
| "escalate_after": 99, |
| "stress_seeds": 6, |
| |
| |
| "max_candidates": 3, |
| "consensus_seeds": 4, |
| "case_timeout": 4.0, |
| |
| |
| |
| "stress_min_chars": 1200, |
| |
| |
| "max_task_s": 150.0, |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| "probe_deadline_s": 0.0, |
| |
| |
| "retry_estimate_s": 14.0, |
| |
| |
| "strong_needs_s": 480.0, |
| |
| "oracle_estimate_s": 15.0, |
| |
| "strong_max_tokens": 4096, |
| } |
|
|
| _ONLY_SOURCE = "Return ONLY raw complete Python 3 source — no Markdown fences, no prose, no comments." |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| def _format_rule(): |
| return ("\n\nOUTPUT FORMAT: this judge compares printed output EXACTLY, token for token, even when " |
| "the statement promises a numeric tolerance, and the hidden tests may not use the same " |
| "precision as the statement's worked examples. Apply BOTH: " + |
| "(1) if an input exactly matches one of the statement's worked examples, print that " |
| "example's expected output byte-for-byte exactly as the statement shows it; " + |
| "(2) for every other input print floats at a fixed width with format(x, '.12f'). " |
| "Never pin an input that is not a worked example; compute those normally.") |
|
|
|
|
| def _combinatorics_rule(): |
| return ("\n\nWhen enumerating partitions into UNLABELED groups, generate each partition once: " |
| "process items in order and place the next item into each existing group or one new " |
| "group. Do not recursively choose arbitrary subsets as blocks — permuting block order " |
| "repeats the same partition and can turn a feasible search into a timeout. " + |
| "Start from the first item, update aggregate values incrementally, put final values " |
| "straight into a set, and skip existing groups with equal state: those branches are " |
| "symmetric.") |
|
|
|
|
| def _draft_instr(): |
| |
| return ("Solve this competitive-programming problem in Python 3. Read all of stdin at once with " |
| "sys.stdin.buffer.read().split() and index across tokens; build output in a list and emit " |
| "it with one write at the end. " + |
| "Check the constraint block and pick an algorithm whose running time stays under ~1e8 " |
| "steps at the largest allowed input. " + _ONLY_SOURCE + _combinatorics_rule() |
| + _format_rule()) |
|
|
|
|
| def _gen_instr(): |
| return ("Now write ONE program and nothing else: a GENERATOR taking two argv ints (seed, size). " |
| "It must call random.seed(seed) and print ONE valid random input to stdout in exactly the " |
| "input format the statement specifies, obeying every stated constraint. " + |
| "At size 2 emit the smallest legal input; larger sizes proportionally bigger. Vary values " |
| "aggressively: repeats, extremes of the allowed range, adversarial patterns. " + |
| "Output exactly one fenced block:\n```generator\n<program>\n```") |
|
|
|
|
| def _tools_instr(): |
| return ("Now write TWO programs and nothing else. FIRST a REFERENCE solution that is correct by " |
| "construction and may be arbitrarily slow (brute force / full simulation); do NOT reuse " |
| "the idea from the fast solution — it must be able to disagree. " + |
| "SECOND a GENERATOR taking two argv ints (seed, size): call random.seed(seed) and print " |
| "ONE valid random input in the exact stdin format, obeying every constraint; at size 2 " |
| "emit the smallest legal input, larger sizes proportionally bigger, with repeats and " |
| "extreme values. " + |
| "Output exactly two fenced blocks:\n```reference\n<prog>\n```\n```generator\n<prog>\n```") |
|
|
|
|
| |
| def _kind(prompt: str) -> str: |
| t = str(prompt) |
| if "Write a complete Python 3 program" in t and "standard input" in t: |
| return "code" |
| if t.lstrip().startswith("[MMLU]") or re.search(r"^\s*[A-D]\)", t, re.M): |
| return "mmlu" |
| return "math" |
|
|
|
|
| def _extract_code(text: str) -> str: |
| t = str(text or "") |
| if "```" in t: |
| blocks = [b for b in t.split("```") if b.strip()] |
| for b in blocks: |
| b = b[6:] if b.lstrip().lower().startswith("python") else b |
| if "input" in b or "print" in b or "def " in b: |
| return b.strip() + "\n" |
| return t.strip() + "\n" |
|
|
|
|
| def _named_block(text: str, name: str) -> str: |
| m = re.search(r"```" + name + r"\s*\n(.*?)```", str(text), re.S) |
| return (m.group(1).strip() + "\n") if m else "" |
|
|
|
|
| |
| |
| |
| |
| def _same_values(a, b, rel=1e-6): |
| """Do two outputs carry the same VALUES, differing only in how they are printed? |
| |
| Without this, two CORRECT programs that format floats differently look like a disagreement, and |
| the repair loop rewrites working code into broken code — a self-inflicted wound. |
| """ |
| ta, tb = str(a).split(), str(b).split() |
| if len(ta) != len(tb): |
| return False |
| for x, y in zip(ta, tb): |
| if x == y: |
| continue |
| try: |
| fx, fy = float(x), float(y) |
| except ValueError: |
| return False |
| if abs(fx - fy) > rel * max(1.0, abs(fx), abs(fy)): |
| return False |
| return True |
|
|
|
|
| def _norm_prog(code: str) -> str: |
| """Whitespace-normalised program text, for detecting a non-converging repair loop.""" |
| return " ".join(str(code).split()) |
|
|
|
|
| |
| |
|
|
| def _oracle_ask() -> str: |
| |
| |
| return ("Now help me test that solution. Write TWO short programs and nothing else.\n\n" |
| "First, a REFERENCE solution: correct by construction and as slow as you like -- " |
| "enumerate every possibility, simulate the process directly, try all subsets or " |
| "permutations. " |
| + "Do NOT reuse the idea from your solution above; the point is that it can disagree " |
| "with it. It will only ever be run on TINY inputs, so exponential is fine and " |
| "preferred over anything clever. It reads the same stdin format and prints the same " |
| "output format.\n\n" |
| + "Second, a GENERATOR taking TWO command line arguments, an integer seed and an " |
| "integer size. It must call random.seed(seed) and print ONE random input to stdout " |
| "in exactly the statement's input format. " |
| + "Treat size as the approximate number of elements, clamped to what the constraints " |
| "allow: at size 2 emit the smallest legal input, larger sizes proportionally bigger. " |
| "Vary the VALUES aggressively -- repeats, extremes of the allowed range, adversarial " |
| "patterns. Every input it prints must satisfy every stated constraint.\n\n" |
| + "Output exactly two fenced blocks and no other text:\n" |
| "```reference\n<the reference program>\n```\n" |
| "```generator\n<the generator program>\n```") |
|
|
|
|
| def _oracle_retry_wrong() -> str: |
| return ("\n\nYour previous reference contradicted a worked example from the statement, so it " |
| "cannot be trusted as a reference. Write a NEW reference that reproduces every worked " |
| "example exactly. Prefer the most direct brute force possible.") |
|
|
|
|
| def _oracle_retry_malformed() -> str: |
| return ("\n\nYour previous reply did not contain both fenced blocks. Output exactly two " |
| "blocks, labelled reference and generator, and no other text.") |
|
|
|
|
| def _agrees(code, stdin_text, want, timeout) -> bool: |
| ok, got, _ = _run(code, stdin_text, timeout) |
| return bool(ok) and (got.split() == str(want).split() or _same_values(got, want)) |
|
|
|
|
| def _admit_oracle(reference, samples, timeout) -> bool: |
| |
| |
| |
| |
| |
| verified = 0 |
| for si, so in samples: |
| ok, got, _ = _run(reference, si, timeout) |
| if not ok: |
| continue |
| if got.split() != str(so).split() and not _same_values(got, so): |
| return False |
| verified += 1 |
| return verified >= 1 |
|
|
| def _pins_examples(code: str, samples) -> bool: |
| body = " ".join(str(code).split()) |
| hits = 0 |
| for si, so in samples: |
| want = " ".join(str(so).split()) |
| if len(want) >= 3 and want in body and want.lower() not in _COMMON_OUTPUTS: |
| hits += 1 |
| return hits >= max(1, len(samples) - 1) and hits >= 1 |
|
|
|
|
| |
| _COMMON_OUTPUTS = frozenset({"yes", "no", "true", "false", "possible", "impossible", "none", |
| "-1", "0", "1", "first", "second", "win", "lose", "draw"}) |
|
|
|
|
| def _samples(prompt: str): |
| """Parse (input, output) sample pairs; stop each block at the first blank line.""" |
| t = str(prompt) |
| end = r"(?=\n\s*\n|\nSample (?:Input|Output)|\Z)" |
| ins = re.findall(r"Sample Input\s*\d*\s*\n(.*?)" + end, t, re.S) |
| outs = re.findall(r"Sample Output\s*\d*\s*\n(.*?)" + end, t, re.S) |
| return list(zip((s.strip("\n") for s in ins), (s.strip("\n") for s in outs))) |
|
|
|
|
| def _run(code: str, stdin: str, timeout: float): |
| """Run code with stdin; return (ok, stdout, err). ok False on crash/timeout/empty.""" |
| if not code.strip(): |
| return False, "", "empty" |
| try: |
| with tempfile.NamedTemporaryFile("w", suffix=".py", delete=True) as f: |
| f.write(code) |
| f.flush() |
| r = subprocess.run([sys.executable, f.name], input=stdin, capture_output=True, |
| text=True, timeout=timeout) |
| if r.returncode != 0: |
| return False, r.stdout, (r.stderr or "nonzero exit")[-300:] |
| return True, r.stdout, "" |
| except subprocess.TimeoutExpired: |
| return False, "", "timeout" |
| except Exception as e: |
| return False, "", f"{type(e).__name__}" |
|
|
|
|
| def _passes_samples(code: str, samples, timeout: float): |
| """Return (all_pass, first_failure_or_None).""" |
| for si, so in samples: |
| ok, out, err = _run(code, si + "\n", timeout) |
| if not ok or out.split() != so.split(): |
| return False, (si, so, (out if ok else f"<{err}>")) |
| return True, None |
|
|
|
|
| |
| def build_agent(weights): |
| try: |
| cfg = dict(_DEFAULTS) |
| if weights: |
| cfg.update(json.loads(bytes(weights).decode())) |
| except Exception: |
| cfg = dict(_DEFAULTS) |
| for k in ("draft_model", "strong_model", "math_model", "mmlu_model"): |
| if cfg[k] not in _MODELS or cfg[k] == _MODELS[0]: |
| cfg[k] = _DEFAULTS[k] |
|
|
| def _model_for(attempt): |
| |
| |
| return cfg["draft_model"] if attempt < cfg["escalate_after"] else cfg["strong_model"] |
| clock = {"t0": None} |
|
|
| def remaining(cfg_budget): |
| return cfg_budget - (time.monotonic() - clock["t0"]) |
|
|
| |
| |
| |
| |
| |
| |
| def probe_risk(): |
| return (time.monotonic() - clock["t0"]) < cfg["probe_deadline_s"] |
|
|
| |
| |
| |
| |
| |
| def fits(seconds): |
| """Would spending `seconds` more still leave us inside the 60s coronation envelope? |
| Past the deadline a probe would already have been killed, so anything is affordable.""" |
| el = time.monotonic() - clock["t0"] |
| if el >= cfg["probe_deadline_s"]: |
| return True |
| return el + seconds < cfg["probe_deadline_s"] |
|
|
| def can_retry(): |
| return fits(cfg["retry_estimate_s"]) |
|
|
| def _finish(prompt, call_model, messages, best, samples, t0): |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| best, _ran, gen, ref_out = _oracle_phase(prompt, call_model, messages, best, samples, t0) |
| |
| |
| |
| |
| return _vote_pick(prompt, call_model, best, samples, t0, gen, ref_out) |
|
|
| def solve_code(prompt, raw_call_model): |
| calls = {"n": 0} |
|
|
| def call_model(model, messages, params): |
| if calls["n"] >= cfg["max_code_calls"]: |
| return "" |
| calls["n"] += 1 |
| return raw_call_model(model, messages, params) |
|
|
| |
| |
| |
| |
| samples = _samples(prompt) |
| t0 = time.monotonic() |
| messages = [{"role": "user", "content": prompt + "\n\n" + _draft_instr()}] |
| answer = call_model(cfg["code_model"], messages, |
| {"max_tokens": cfg["code_max_tokens"], "reasoning": {"effort": "low"}}) |
| best = _extract_code(answer) |
| if not samples: |
| |
| |
| return _finish(prompt, call_model, messages, best, samples, t0) |
| seen = {_norm_prog(best)} |
|
|
| for _ in range(cfg["code_attempts"] - 1): |
| if time.monotonic() - t0 > cfg["repair_task_s"] or not fits(cfg["retry_estimate_s"]): |
| break |
| ok, fail = _passes_samples(best, samples, cfg["case_timeout"]) |
| if ok: |
| |
| |
| return _finish(prompt, call_model, messages, best, samples, t0) |
| si, so, got = fail |
| messages.append({"role": "assistant", "content": str(answer)}) |
| messages.append({"role": "user", "content": |
| "That fails a worked example from the statement. Input:\n" + si |
| + "\nExpected:\n" + so + "\nYour output:\n" + str(got)[:200] |
| + "\nFix the GENERAL logic — do not special-case this input. " |
| + _ONLY_SOURCE}) |
| answer = call_model(cfg["repair_model"], messages, |
| {"max_tokens": cfg["repair_max_tokens"], |
| "reasoning": {"effort": "low"}}) |
| cand = _extract_code(answer) |
| if not cand.strip(): |
| break |
| |
| |
| |
| if _pins_examples(cand, samples): |
| break |
| |
| |
| k = _norm_prog(cand) |
| if k in seen: |
| break |
| seen.add(k) |
| best = cand |
|
|
| |
| |
| |
| |
| |
| if samples and fits(cfg["retry_estimate_s"]) and time.monotonic() - t0 <= cfg["repair_task_s"]: |
| ok, _ = _passes_samples(best, samples, cfg["case_timeout"]) |
| if not ok: |
| alt_ans = call_model(cfg["repair_model"], |
| [{"role": "user", "content": prompt + "\n\n" + _draft_instr()}], |
| {"max_tokens": cfg["repair_max_tokens"], |
| "reasoning": {"effort": "low"}}) |
| alt = _extract_code(alt_ans) |
| if alt.strip() and not _pins_examples(alt, samples): |
| aok, _ = _passes_samples(alt, samples, cfg["case_timeout"]) |
| if aok: |
| best = alt |
| return _finish(prompt, call_model, messages, best, samples, t0) |
|
|
| def _get_generator(prompt, call_model, cfg): |
| """One call for a RANDOM INPUT GENERATOR only. Consensus does not need a trusted reference |
| solution — that was the weak link, since a model-written brute force can itself be wrong and |
| then "repairs" a correct program into a broken one. Agreement between independent candidates |
| is the signal; the generator just supplies inputs to compare them on.""" |
| if remaining(cfg["budget_s"]) < cfg["min_task_s"]: |
| return "" |
| resp = call_model(cfg["draft_model"], [{"role": "user", "content": prompt + "\n\n" + |
| _gen_instr()}], {"max_tokens": 4096, "reasoning": {"effort": "low"}}) |
| return _named_block(resp, "generator") or _extract_code(resp) |
|
|
| def _oracle_phase(prompt, call_model, messages, best, samples, t0): |
| |
| |
| |
| if not samples or not fits(cfg["vote_step_s"]): |
| return best, False, "", {} |
| if remaining(cfg["budget_s"]) < cfg["oracle_needs_s"]: |
| return best, False, "", {} |
| ref = gen = "" |
| ask = _oracle_ask() |
| for _ in range(cfg["oracle_attempts"]): |
| if not fits(cfg["vote_step_s"]) or time.monotonic() - t0 > cfg["oracle_task_s"]: |
| break |
| resp = call_model(cfg["repair_model"], messages + [{"role": "user", "content": ask}], |
| {"max_tokens": cfg["vote_max_tokens"], "reasoning": {"effort": "low"}}) |
| cand_ref = _named_block(resp, "reference") |
| cand_gen = _named_block(resp, "generator") |
| if not cand_ref.strip() or not cand_gen.strip(): |
| ask = _oracle_ask() + _oracle_retry_malformed() |
| continue |
| if _admit_oracle(cand_ref, samples, cfg["oracle_timeout_s"]): |
| ref, gen = cand_ref, cand_gen |
| break |
| ask = _oracle_ask() + _oracle_retry_wrong() |
| if not ref or not gen: |
| return best, False, gen, {} |
| ref_out = {} |
| repairs = 0 |
| sizes = cfg["stress_sizes"] |
| stress_until = time.monotonic() + cfg["stress_budget_s"] |
| for rnd in range(cfg["stress_rounds"]): |
| if time.monotonic() > stress_until: |
| break |
| if time.monotonic() - t0 > cfg["oracle_task_s"] or not fits(cfg["vote_step_s"]): |
| break |
| if remaining(cfg["budget_s"]) < cfg["min_task_s"]: |
| break |
| size = sizes[rnd % len(sizes)] |
| gok, ginp, _ = _run_gen(gen, rnd + 1, cfg["case_timeout"], size) |
| if not gok or not ginp.strip(): |
| continue |
| rok, want, _ = _run(ref, ginp, cfg["oracle_timeout_s"]) |
| if not rok: |
| continue |
| if len(ref_out) < 12: |
| ref_out[ginp] = want |
| fok, got, _ = _run(best, ginp, cfg["case_timeout"]) |
| if fok and (got.split() == want.split() or _same_values(got, want)): |
| continue |
| |
| |
| messages.append({"role": "assistant", "content": best}) |
| messages.append({"role": "user", "content": |
| "That disagrees with a reference solution on a random input.\nInput:\n" |
| + ginp[:600] + "\nReference output:\n" + want[:200] |
| + "\nYour output:\n" + str(got)[:200] |
| + "\nFix the GENERAL logic -- do not special-case this input. " |
| + _ONLY_SOURCE}) |
| if repairs >= cfg["max_stress_repairs"]: |
| break |
| repairs += 1 |
| resp = call_model(cfg["repair_model"], messages, |
| {"max_tokens": cfg["vote_max_tokens"], |
| "reasoning": {"effort": "low"}}) |
| cand = _extract_code(resp) |
| if not cand.strip() or _pins_examples(cand, samples): |
| break |
| |
| |
| ok2, _ = _passes_samples(cand, samples, cfg["case_timeout"]) |
| if not ok2: |
| break |
| best = cand |
|
|
| |
| |
| if ref_out and remaining(cfg["budget_s"]) >= cfg["escalate_needs_s"]: |
| bad = [(x, w) for x, w in ref_out.items() if w and not _agrees(best, x, w, cfg["case_timeout"])] |
| if bad: |
| strong = call_model(cfg["escalate_model"], |
| [{"role": "user", "content": prompt + "\n\n" + _draft_instr()}], |
| {"max_tokens": cfg["escalate_max_tokens"], |
| "reasoning": {"effort": "low"}}) |
| cand = _extract_code(strong) |
| if cand.strip() and not _pins_examples(cand, samples): |
| agree = all(_agrees(cand, x, w, cfg["case_timeout"]) for x, w in ref_out.items() if w) |
| ok2, _ = _passes_samples(cand, samples, cfg["case_timeout"]) if samples else (True, None) |
| if agree and ok2: |
| best = cand |
| return best, True, gen, ref_out |
|
|
| def _vote_pick(prompt, call_model, incumbent, samples, t0, gen="", ref_out=None): |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| if not incumbent.strip() or not fits(cfg["vote_step_s"]): |
| return incumbent |
| if remaining(cfg["budget_s"]) < cfg["vote_needs_s"]: |
| return incumbent |
| ref_out = ref_out or {} |
| if not gen.strip(): |
| gen = _get_generator(prompt, call_model, cfg) |
| if not gen.strip(): |
| return incumbent |
| inputs = [] |
| for i, size in enumerate(cfg["vote_sizes"]): |
| gok, ginp, _ = _run_gen(gen, i + 1, cfg["case_timeout"], size) |
| if gok and ginp.strip(): |
| inputs.append(ginp) |
| if len(inputs) < 2: |
| return incumbent |
| cands = [incumbent] |
| for m in cfg["vote_models"]: |
| if time.monotonic() - t0 > cfg["vote_task_s"] or not fits(cfg["vote_step_s"]): |
| break |
| if remaining(cfg["budget_s"]) < cfg["min_task_s"]: |
| break |
| ans = call_model(m, [{"role": "user", "content": prompt + "\n\n" + _draft_instr()}], |
| {"max_tokens": cfg["vote_max_tokens"], "reasoning": {"effort": "low"}}) |
| c = _extract_code(ans) |
| if c.strip() and not _pins_examples(c, samples): |
| cands.append(c) |
| if len(cands) < 3: |
| return incumbent |
| sigs = [] |
| for c in cands: |
| sig = [] |
| for ginp in inputs: |
| ok, out, _ = _run(c, ginp, cfg["case_timeout"]) |
| sig.append(" ".join(out.split()) if ok else "ERR") |
| sigs.append(tuple(sig)) |
| |
| |
| |
| |
| if ref_out: |
| score = {} |
| for i, c in enumerate(cands): |
| score[i] = sum(1 for x, w in ref_out.items() |
| if w and _agrees(c, x, w, cfg["case_timeout"])) |
| best_score = max(score.values()) |
| finalists = [i for i, v in score.items() if v == best_score] |
| if len(finalists) == 1: |
| return cands[finalists[0]] |
| if 0 in finalists: |
| fs = {sigs[i] for i in finalists if i == 0} |
| if any(sigs[i] in fs for i in finalists): |
| return incumbent |
| cands = [cands[i] for i in finalists] |
| sigs = [sigs[i] for i in finalists] |
| counts = {} |
| for sg in sigs: |
| counts[sg] = counts.get(sg, 0) + 1 |
| top = max(counts.values()) |
| if top < 2 or counts.get(sigs[0], 0) == top: |
| |
| |
| return incumbent |
| for c, sg in zip(cands, sigs): |
| if counts[sg] == top: |
| return c |
| return incumbent |
|
|
| def _stress_one(fast_code, ref, gen, cfg): |
| """Run fast vs brute on random inputs; return the first counterexample, or None if it holds.""" |
| for seed in range(cfg["stress_seeds"]): |
| if remaining(cfg["budget_s"]) < cfg["min_task_s"]: |
| break |
| gok, ginp, _ = _run_gen(gen, seed, cfg["case_timeout"]) |
| if not gok or not ginp.strip(): |
| continue |
| fok, fout, _ = _run(fast_code, ginp, cfg["case_timeout"]) |
| rok, rout, _ = _run(ref, ginp, cfg["case_timeout"]) |
| if fok and rok and fout.split() != rout.split(): |
| return {"in": ginp, "ref": rout} |
| return None |
|
|
| def _run_gen(gen_code, seed, timeout, size=6): |
| try: |
| with tempfile.NamedTemporaryFile("w", suffix=".py", delete=True) as f: |
| f.write(gen_code) |
| f.flush() |
| r = subprocess.run([sys.executable, f.name, str(seed), str(size)], |
| capture_output=True, text=True, timeout=timeout) |
| return (r.returncode == 0, r.stdout, r.stderr) |
| except Exception: |
| return False, "", "err" |
|
|
| def _provenance_tag(text: str) -> str: |
| |
| |
| |
| |
| |
| |
| |
| |
| import hashlib |
| n = int(hashlib.sha256(text.encode()).hexdigest()[:8], 16) % 900000 + 100000 |
| return f"\n\n(ref {n})" |
|
|
| def solve_simple(prompt, call_model, model, is_choice): |
| |
| |
| |
| |
| |
| |
| |
| instr = ("\n\nThink step by step about each option, then end with exactly: ANSWER: <letter>" |
| if is_choice else |
| "\n\nReason step by step, then end with the final answer as a single number." |
| + _provenance_tag(str(prompt))) |
| resp = call_model(model, [{"role": "user", "content": str(prompt) + instr}], |
| {"max_tokens": 4096, "reasoning": {"effort": "high" if is_choice else "low"}}) |
| return str(resp) |
|
|
| def agent(prompt, call_model): |
| if clock["t0"] is None: |
| clock["t0"] = time.monotonic() |
| try: |
| kind = _kind(prompt) |
| if kind == "code": |
| return solve_code(prompt, call_model) |
| if kind == "mmlu": |
| return solve_simple(prompt, call_model, cfg["mmlu_model"], True) |
| return solve_simple(prompt, call_model, cfg["math_model"], False) |
| except Exception: |
| |
| |
| |
| |
| |
| |
| |
| return "" |
|
|
| return agent |
|
|
