1"""Executable task family: small, deterministic coding/data tasks whose correctness is checked by hidden tests.23Family 'exec': an agent receives (a) a source document describing a small dataset + a bug report or analysis request and4(b) a task; it must output a Python function body. Hidden tests (evaluator-only) run in a clean subprocess with a timeout.5The 'document' is the information the sender holds (table schema, bug report, expected behaviour); the receiver must6reconstruct the specification from the message. Ground truth is generated deterministically from a seed.7"""8from __future__ import annotations9import random, json, textwrap, subprocess, sys, tempfile, os10from dataclasses import dataclass, asdict1112COLS = ["region", "product", "units", "unit_price", "discount_pct", "returned", "shipped_days"]13REGIONS = ["Nord", "Sud", "Est", "Ouest", "Centre"]14PRODUCTS = ["alpha", "beta", "gamma", "delta", "epsilon", "zeta"]1516def make_table(rng: random.Random, n: int):17 rows = []18 for _ in range(n):19 rows.append({"region": rng.choice(REGIONS), "product": rng.choice(PRODUCTS), "units": rng.randint(1, 40),20 "unit_price": round(rng.uniform(2.5, 80.0), 2), "discount_pct": rng.choice([0, 0, 5, 10, 15, 25]),21 "returned": rng.random() < 0.12, "shipped_days": rng.randint(1, 14)})22 return rows2324def spec_variants(rng: random.Random, rows):25 """Return (spec_text, reference_fn_source, test_cases). Spec encodes exact business rules that must be transmitted."""26 thr_days = rng.randint(3, 9)27 min_units = rng.randint(2, 6)28 excl_region = rng.choice(REGIONS)29 rounding = rng.choice([0, 1, 2])30 tax = rng.choice([0.0, 5.0, 9.975, 14.975])31 spec = (f"Bug report BR-{rng.randint(1000,9999)}: the function net_revenue(rows) must compute the total net revenue of the "32 f"rows list (each row is a dict with keys {COLS}). Rules agreed with the client on {rng.randint(1,28)} {rng.choice(['March','June','September'])} 2031: "33 f"(1) exclude returned rows; (2) exclude rows whose region is '{excl_region}'; (3) exclude rows with fewer than {min_units} units; "34 f"(4) line revenue = units * unit_price * (1 - discount_pct/100); (5) apply a tax of {tax} % on the sum; "35 f"(6) return the result rounded to {rounding} decimals (Python round). The current implementation forgets rule (2) and applies no tax.")36 ref = textwrap.dedent(f"""37 def net_revenue(rows):38 total = 0.039 for r in rows:40 if r['returned']:41 continue42 if r['region'] == '{excl_region}':43 continue44 if r['units'] < {min_units}:45 continue46 total += r['units'] * r['unit_price'] * (1 - r['discount_pct'] / 100.0)47 total = total * (1 + {tax} / 100.0)48 return round(total, {rounding})49 """)50 ns = {}51 exec(ref, ns)52 tests = []53 for k in range(5):54 sub = [dict(r) for r in rng.sample(rows, k=min(len(rows), rng.randint(4, len(rows))))]55 tests.append({"input": sub, "expected": ns["net_revenue"](sub)})56 return spec, ref, tests, {"thr_days": thr_days, "min_units": min_units, "excl_region": excl_region, "rounding": rounding, "tax": tax}5758def generate_exec_task(seed: int, target_tokens: int = 2000) -> dict:59 rng = random.Random(10_000 + seed)60 rows = make_table(rng, rng.randint(12, 24))61 spec, ref, tests, params = spec_variants(rng, rows)62 sample_rows = json.dumps(rows[:6], indent=None)63 filler = []64 boiler = ["The ticketing system assigns severity on a four-level scale; this ticket is severity 2.",65 "All monetary values are in the client's reporting currency; no FX conversion is applied here.",66 "Unit tests live in the tests/ directory and are executed in CI on every push.",67 "The data engineering team refreshes the sales extract nightly at 02:00 local time.",68 "Code style follows the repository's formatter settings; do not reformat unrelated lines.",69 "Deployment to production requires a reviewer approval and a green pipeline.",70 "Historic tickets BR-1183 and BR-2047 are unrelated and remain closed.",71 "The data dictionary defines shipped_days as the number of calendar days from order to shipment."]72 n_fill = max(0, int(target_tokens * 4 / 95) - 12)73 filler = [rng.choice(boiler) for _ in range(n_fill)]74 doc = ("ENGINEERING TICKET\n\n" + spec + "\n\nSample rows (first 6 of the extract):\n" + sample_rows + "\n\n" +75 "\n".join(filler[: n_fill // 2]) + "\n\nThe existing (buggy) implementation is:\n\n" +76 "def net_revenue(rows):\n total = 0.0\n for r in rows:\n if r['returned']:\n continue\n"77 f" if r['units'] < {params['min_units']}:\n continue\n"78 " total += r['units'] * r['unit_price'] * (1 - r['discount_pct'] / 100.0)\n"79 f" return round(total, {params['rounding']})\n\n" + "\n".join(filler[n_fill // 2:]))80 task = ("Rewrite the Python function net_revenue(rows) so that it implements ALL the agreed rules in the ticket. "81 "Output only Python code defining net_revenue; no explanations, no markdown fences.")82 return {"document_id": f"exec_{seed:05d}_{target_tokens}", "seed": seed, "target_tokens": target_tokens, "text": doc, "char_len": len(doc),83 "questions": [{"question_id": "q00_exec", "qtype": "exec", "question": task, "answer": ref, "answer_type": "exec", "tolerance": 0.0,84 "supporting_fact_ids": ["spec"], "answer_position_frac": 0.02, "aliases": []}],85 "facts": [{"fact_id": "spec", "entity": "net_revenue", "attribute": "rules", "value": json.dumps(params), "kind": "spec", "sentence": spec, "char_start": doc.find(spec), "char_end": doc.find(spec) + len(spec)}],86 "hidden_tests": tests}8788def run_hidden_tests(code: str, tests: list, timeout_s: float = 10.0) -> dict:89 """Execute candidate code in a clean subprocess (no network, no project imports) against hidden tests."""90 code = code.strip()91 if code.startswith("```"):92 code = code.strip("`")93 if code.startswith("python"):94 code = code[6:]95 harness = textwrap.dedent("""96 import json, sys, math97 src = open(sys.argv[1]).read(); tests = json.load(open(sys.argv[2]))98 ns = {}99 try:100 exec(src, ns)101 except Exception as e:102 print(json.dumps({"passed": 0, "total": len(tests), "error": "exec:" + type(e).__name__})); sys.exit(0)103 fn = ns.get('net_revenue')104 if fn is None:105 print(json.dumps({"passed": 0, "total": len(tests), "error": "no_function"})); sys.exit(0)106 p = 0; err = None107 for t in tests:108 try:109 out = fn([dict(r) for r in t['input']])110 if isinstance(out, (int, float)) and math.isclose(float(out), float(t['expected']), rel_tol=1e-9, abs_tol=1e-6):111 p += 1112 except Exception as e:113 err = type(e).__name__114 print(json.dumps({"passed": p, "total": len(tests), "error": err}))115 """)116 with tempfile.TemporaryDirectory() as d:117 cp = os.path.join(d, "cand.py"); tp = os.path.join(d, "tests.json"); hp = os.path.join(d, "h.py")118 open(cp, "w").write(code); json.dump(tests, open(tp, "w")); open(hp, "w").write(harness)119 try:120 r = subprocess.run([sys.executable, "-I", hp, cp, tp], capture_output=True, text=True, timeout=timeout_s, cwd=d)121 res = json.loads(r.stdout.strip().splitlines()[-1]) if r.stdout.strip() else {"passed": 0, "total": len(tests), "error": "no_output"}122 except subprocess.TimeoutExpired:123 res = {"passed": 0, "total": len(tests), "error": "timeout"}124 except Exception as e:125 res = {"passed": 0, "total": len(tests), "error": type(e).__name__}126 res["pass_rate"] = res["passed"] / max(1, res["total"])127 res["correct"] = float(res["passed"] == res["total"])128 return res129