#!/usr/bin/env python3 """Does the compiler keep an LLM honest? Generate -> static check -> feed errors back -> run tests. One local model writes the same HumanEval tasks in six languages (MultiPL-E translations). Each arm has a static gate (the language's own compiler or checker). When the gate rejects a program, only the sanitized gate diagnostics are fed back, up to MAX_REPAIRS times. Hidden tests are never shown to the model; they only score the final program. """ import argparse, gzip, hashlib, json, os, random, re, resource, secrets, shutil, subprocess, sys, tempfile, time, urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[3] BASE = ROOT / 'paper/compiler-keeps-it-honest' ASSETS = ROOT / 'assets' TOOLS = ASSETS / 'tools' HUMANEVAL = Path(os.environ.get('CKH_HUMANEVAL', str(ASSETS / 'free-tokens-coding/HumanEval.jsonl.gz'))) HUMANEVAL_SHA = 'b796127e635a67f93fb35c04f4cb03cf06f38c8072ee7cee8833d7bee06979ef' MODEL = os.environ.get('CKH_MODEL', 'qwen2.5-coder:1.5b') SEED = 60060 MAX_REPAIRS = 2 NUM_PREDICT = 384 POLICY = '(version 1) (allow default) (deny network*) (deny file-write*) (allow file-write-data (literal "/dev/null")) (deny process-fork)' PY_STOPS = ['\ndef ', '\n#', '\nif __name__', '\nclass ', '\nprint('] GO = TOOLS / 'go/bin/go' RUSTC = TOOLS / 'rust/bin/rustc' TSC = TOOLS / 'node_modules/.bin/tsc' MYPY = TOOLS / 'venv/bin/mypy' JAR = TOOLS / 'javatuples-1.2.jar' # arm -> (generation language, gate description) ARMS = { 'python': ('py', 'python -m py_compile (syntax only)'), 'python+mypy': ('py', 'py_compile + mypy --check-untyped-defs'), 'javascript': ('js', 'node --check (syntax only)'), 'typescript': ('ts', 'tsc --strict --noEmit'), 'java': ('java', 'javac'), 'go': ('go', 'go vet + go test -c'), 'rust': ('rs', 'rustc --edition 2021'), # Sensitivity arm: MultiPL-E's TypeScript stop list lacks the '\nconsole.log' stop its JavaScript list has. 'typescript-jsstops': ('ts', 'tsc --strict --noEmit'), } EXTRA_STOPS = {'typescript-jsstops': ['\nconsole.log']} COMMENT = {'py': '#', 'js': '//', 'ts': '//', 'java': '//', 'go': '//', 'rs': '//'} LANG_NAME = {'py': 'Python', 'js': 'JavaScript', 'ts': 'TypeScript', 'java': 'Java', 'go': 'Go', 'rs': 'Rust'} def sha(p): return hashlib.sha256(p.read_bytes()).hexdigest() def load_tasks(): import pandas as pd if sha(HUMANEVAL) != HUMANEVAL_SHA: raise RuntimeError('ASSET_HASH_MISMATCH: HumanEval') pins = json.loads((BASE / 'evidence/dataset-pins.json').read_text())['sha256'] py = {int(x['task_id'].split('/')[1]): x for x in map(json.loads, gzip.open(HUMANEVAL, 'rt'))} per = {} for lang in ['js', 'ts', 'java', 'go', 'rs']: f = ASSETS / f'multipl-e/humaneval-{lang}.parquet' if sha(f) != pins['humaneval-' + lang]: raise RuntimeError('ASSET_HASH_MISMATCH: ' + lang) rows = pd.read_parquet(f).to_dict('records') per[lang] = {int(r['name'].split('_')[1]): {'prompt': r['prompt'], 'tests': r['tests'], 'stops': list(r['stop_tokens'])} for r in rows} ids = sorted(set(py).intersection(*[set(v) for v in per.values()])) tasks = [] for i in ids: t = {'id': i, 'py': {'prompt': py[i]['prompt'], 'tests': '\n' + py[i]['test'] + f"\ncheck({py[i]['entry_point']})\n", 'stops': PY_STOPS, 'entry': py[i]['entry_point']}} for lang in per: t[lang] = per[lang][i] sig = [ln for ln in t[lang]['prompt'].rstrip().splitlines() if '(' in ln][-1] t[lang]['entry'] = re.search(r'(\w+)\s*\(', sig.split('fn ')[-1].split('func ')[-1].split('function ')[-1]).group(1) tasks.append(t) return tasks _cache = {} def join(prompt, completion, tests): # A completion can end in a line comment with no newline; gluing the tests onto that line would # comment out their first statement. Always start the tests on a fresh line. sep = '' if completion.endswith('\n') else '\n' return prompt + completion + sep + tests def generate(prompt, stops): key = hashlib.sha256((MODEL + '\0' + prompt + '\0' + json.dumps(stops)).encode()).hexdigest() if key in _cache: return dict(_cache[key], cached=True) payload = {'model': MODEL, 'prompt': prompt, 'raw': True, 'stream': False, 'keep_alive': '30m', 'options': {'temperature': 0, 'seed': SEED, 'num_predict': NUM_PREDICT, 'num_ctx': 4096, 'stop': stops}} t = time.monotonic() req = urllib.request.Request('http://127.0.0.1:11434/api/generate', data=json.dumps(payload).encode(), headers={'Content-Type': 'application/json'}) with urllib.request.urlopen(req, timeout=600) as r: out = json.load(r) g = {'completion': out['response'], 'eval_count': out.get('eval_count', 0), 'eval_ns': out.get('eval_duration', 0), 'prompt_eval_count': out.get('prompt_eval_count', 0), 'wall_s': round(time.monotonic() - t, 3), 'hit_cap': out.get('done_reason') == 'length'} _cache[key] = g return dict(g, cached=False) def run(cmd, cwd, timeout=90, env=None): e = {'PATH': f'{GO.parent}:{RUSTC.parent}:/usr/local/bin:/usr/bin:/bin', 'HOME': str(TOOLS), 'LANG': 'C.UTF-8', 'GOCACHE': str(TOOLS / 'gocache'), 'GOPATH': str(TOOLS / 'gopath'), 'GOFLAGS': '-mod=mod', 'GO111MODULE': 'off', 'CGO_ENABLED': '0', 'GOTOOLCHAIN': 'local', 'MYPY_CACHE_DIR': '/dev/null'} e.update(env or {}) try: cp = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, env=e) return cp.returncode, cp.stdout + cp.stderr except subprocess.TimeoutExpired: return 124, 'gate timeout' DIAG = { 'py': re.compile(r'(Error|error)'), 'mypy': re.compile(r':\d+: error:'), 'js': re.compile(r'^(\S+:\d+|\w*Error:)'), 'ts': re.compile(r'\(\d+,\d+\): error TS\d+'), 'java': re.compile(r'(:\d+: error:|^\s+symbol:)'), 'go': re.compile(r'\.go:\d+:\d+:'), 'rs': re.compile(r':\d+:\d+: error'), } def sanitize(kind, text, workdir): """Keep diagnostic headline lines only: no echoed source lines, so test inputs are not leaked.""" lines = [] for ln in re.sub(r'(/[^\s:()]+/)+', '', text.replace(str(workdir) + '/', '')).splitlines(): if DIAG[kind].search(ln) and 'aborting due to' not in ln and 'could not compile' not in ln: lines.append(ln.strip()[:200]) return lines[:6] or [text.strip().splitlines()[-1][:200] if text.strip() else 'unknown error'] def gate(arm, program, workdir, solution=None): """Static check of the full assembled program (solution + test call sites). Returns (ok, diagnostics).""" w = Path(workdir) if arm in ('python', 'python+mypy'): (w / 'prog.py').write_text(program) rc, out = run([sys.executable, '-c', 'import sys;compile(open(sys.argv[1]).read(),"prog.py","exec")', 'prog.py'], w) if rc: return False, sanitize('py', out, w) if arm == 'python+mypy': # HumanEval's test harness is untyped and carries an unannotated METADATA dict, so mypy checks the solution only. (w / 'prog.py').write_text(solution if solution is not None else program) rc, out = run([str(MYPY), '--check-untyped-defs', '--no-incremental', '--no-error-summary', '--hide-error-context', '--no-pretty', 'prog.py'], w) if rc: return False, sanitize('mypy', out, w) return True, [] if arm == 'javascript': (w / 'prog.js').write_text(program) rc, out = run(['node', '--check', 'prog.js'], w) return (rc == 0), ([] if rc == 0 else sanitize('js', out, w)) if arm in ('typescript', 'typescript-jsstops'): (w / 'prog.ts').write_text(program) rc, out = run([str(TSC), '--strict', '--noEmit', '--pretty', 'false', '--target', 'es2020', '--lib', 'es2020,dom', 'prog.ts'], w) return (rc == 0), ([] if rc == 0 else sanitize('ts', out, w)) if arm == 'java': (w / 'Problem.java').write_text(program) rc, out = run(['javac', '-nowarn', '-cp', str(JAR), '-d', 'cls', 'Problem.java'], w) return (rc == 0), ([] if rc == 0 else sanitize('java', out, w)) if arm == 'go': (w / 'prog_test.go').write_text(program) rc, out = run([str(GO), 'vet', 'prog_test.go'], w) if rc: return False, sanitize('go', out, w) rc, out = run([str(GO), 'test', '-c', '-vet=off', '-o', 'gotest.bin', 'prog_test.go'], w) return (rc == 0), ([] if rc == 0 else sanitize('go', out, w)) if arm == 'rust': (w / 'prog.rs').write_text(program) rc, out = run([str(RUSTC), '--edition', '2021', '--error-format=short', '-A', 'warnings', '-o', 'rs.bin', 'prog.rs'], w) return (rc == 0), ([] if rc == 0 else sanitize('rs', out, w)) raise ValueError(arm) def limits(): resource.setrlimit(resource.RLIMIT_CPU, (10, 10)) resource.setrlimit(resource.RLIMIT_FSIZE, (0, 0)) def sandboxed(cmd, cwd): try: cp = subprocess.run(['/usr/bin/sandbox-exec', '-p', POLICY] + cmd, cwd=cwd, capture_output=True, text=True, timeout=15, preexec_fn=limits, env={'PATH': '/usr/local/bin:/usr/bin:/bin', 'LANG': 'C.UTF-8', 'PYTHONDONTWRITEBYTECODE': '1'}) return cp.returncode, cp.stdout, cp.stderr except subprocess.TimeoutExpired: return None, '', 'timeout' def test(arm, program, workdir): """Run hidden tests. Untrusted generated code runs under macOS sandbox-exec (no network, writes or forks).""" w = Path(workdir) marker = 'CKH_DONE_' + secrets.token_hex(8) if arm in ('python', 'python+mypy'): (w / 'run.py').write_text(program + f'\nprint({marker!r})\n') rc, out, err = sandboxed([sys.executable, 'run.py'], w) elif arm == 'javascript': (w / 'run.js').write_text(program + f'\nconsole.log({marker!r});\n') rc, out, err = sandboxed(['node', 'run.js'], w) elif arm in ('typescript', 'typescript-jsstops'): # tsc emits JavaScript even when type checking fails, so an un-gated TS program can still be run. (w / 'run.ts').write_text(program + f'\nconsole.log({marker!r});\n') run([str(TSC), '--target', 'es2020', '--module', 'commonjs', '--lib', 'es2020,dom', '--outDir', 'emit', 'run.ts'], w) if not (w / 'emit/run.js').exists(): return {'status': 'no_build', 'passed': False} rc, out, err = sandboxed(['node', 'emit/run.js'], w) else: ok, _ = gate(arm, program, w) if not ok: return {'status': 'no_build', 'passed': False} cmd = {'java': ['java', '-ea', '-XX:-UsePerfData', '-Xshare:off', '-cp', f'cls:{JAR}', 'Problem'], 'go': ['./gotest.bin'], 'rust': ['./rs.bin']}[arm] rc, out, err = sandboxed(cmd, w) marker = None if rc is None: return {'status': 'timeout', 'passed': False} passed = rc == 0 and (marker is None or out.splitlines()[-1:] == [marker]) return {'status': 'pass' if passed else 'fail', 'passed': passed, 'stderr_tail': (err or '')[-200:]} def chat(msg): key = hashlib.sha256((MODEL + '\0chat\0' + msg).encode()).hexdigest() if key in _cache: return dict(_cache[key], cached=True) payload = {'model': MODEL, 'stream': False, 'keep_alive': '30m', 'messages': [{'role': 'user', 'content': msg}], 'options': {'temperature': 0, 'seed': SEED, 'num_predict': 2 * NUM_PREDICT, 'num_ctx': 4096}} t = time.monotonic() req = urllib.request.Request('http://127.0.0.1:11434/api/chat', data=json.dumps(payload).encode(), headers={'Content-Type': 'application/json'}) with urllib.request.urlopen(req, timeout=600) as r: out = json.load(r) g = {'reply': out['message']['content'], 'eval_count': out.get('eval_count', 0), 'eval_ns': out.get('eval_duration', 0), 'prompt_eval_count': out.get('prompt_eval_count', 0), 'wall_s': round(time.monotonic() - t, 3), 'hit_cap': out.get('done_reason') == 'length'} _cache[key] = g return dict(g, cached=False) def match_brace(text, open_idx): depth = 0 for i in range(open_idx, len(text)): if text[i] == '{': depth += 1 elif text[i] == '}': depth -= 1 if depth == 0: return i return None DEF = {'go': r'func\s+{n}\s*\(', 'rs': r'fn\s+{n}\s*(<[^>]*>)?\s*\(', 'ts': r'function\s+{n}\s*(<[^>]*>)?\s*\(', 'js': r'function\s+{n}\s*\(', 'java': r'\b(?:public|private|protected|static)[\w\s<>\[\],.?]*\s{n}\s*\('} def brace_function(code, lang, name): m = re.search(DEF[lang].format(n=re.escape(name)), code) if not m: return None open_idx = code.find('{', m.end()) close_idx = match_brace(code, open_idx) if open_idx >= 0 else None if close_idx is None: return None line_start = code.rfind('\n', 0, m.start()) + 1 return line_start, open_idx, close_idx def extract_completion(lang, task, reply): """Turn a chat reply into a completion for the original, unchanged prompt. The signature and docstring stay fixed; the function body is taken from the reply. Imports and helper functions in the reply are kept, placed where each language allows them.""" t = task[lang] blocks = re.findall(r'```[\w+-]*\s*\n(.*?)```', reply, re.S) or [reply] code = max(blocks, key=len) name = t['entry'] if lang == 'py': lines = code.splitlines() defs = [i for i, ln in enumerate(lines) if re.match(r'def \w+\s*\(', ln)] idx = next((i for i in defs if re.match(rf'def {name}\s*\(', lines[i])), None) if idx is None: return None def block(i): j = i + 1 while j < len(lines) and not lines[j - 1].rstrip().endswith(':'): j += 1 out = [] for ln in lines[j:]: if ln.strip() and not ln[0].isspace(): break out.append(ln) return lines[i:j], out _, body = block(idx) helpers = [] for i in defs: if i != idx: head, hb = block(i) helpers += [' ' + ln if ln.strip() else '' for ln in head + hb] imports = [' ' + ln.strip() for ln in lines if re.match(r'(import |from \S+ import )', ln)] return '\n'.join(imports + helpers + body).rstrip() + '\n' code = re.sub(r'^\s*//.*$', '', code, flags=re.M) found = brace_function(code, lang, name) if not found: return None _, open_idx, close_idx = found body = code[open_idx + 1:close_idx].strip('\n') rest = code[:found[0]] + code[close_idx + 1:] helpers = [] names = re.findall({'go': r'func\s+(\w+)\s*\(', 'rs': r'fn\s+(\w+)', 'ts': r'function\s+(\w+)', 'js': r'function\s+(\w+)', 'java': r'\bstatic[\w\s<>\[\],.?]*\s(\w+)\s*\('}[lang], rest) for h in dict.fromkeys(names): if h in (name, 'main') or h.startswith('Test'): continue f = brace_function(rest, lang, h) if f: helpers.append(rest[f[0]:f[2] + 1]) if lang == 'rs': uses = [ln.strip() for ln in code.splitlines() if ln.strip().startswith('use ')] return '\n'.join([' ' + u for u in uses] + helpers + [body]) + '\n' if lang == 'java': # The hidden tests begin by closing the method, so helpers go after an extra closer and a pad method. if not helpers: return body + '\n' return body + '\n }\n' + '\n'.join(helpers) + '\n private static void ckhPad() {\n' return body + '\n}\n' + ('\n' + '\n\n'.join(helpers) + '\n' if helpers else '') def go_imports(prompt, reply): """Merge standard-library imports named in the reply into the prompt's import block.""" pkgs = set(re.findall(r'^\s*"([\w/]+)"', reply, re.M)) | set(re.findall(r'^import\s+"([\w/]+)"', reply, re.M)) have = set(re.findall(r'^\s*"([\w/]+)"', prompt.split(')')[0], re.M)) add = sorted(p for p in pkgs - have if '.' not in p) if not add: return prompt return prompt.replace('import (\n', 'import (\n' + ''.join(f' "{p}"\n' for p in add), 1) def repair(lang, task, completion, diags, tool): t = task[lang] msg = (f"This {LANG_NAME[lang]} code was rejected by {tool}.\n\n```{lang}\n{t['prompt'] + completion}\n```\n\n" f"Diagnostics:\n" + '\n'.join(diags) + f"\n\nFix the errors. Keep the function name and signature unchanged. Reply with only the complete corrected " f"function (and any imports it needs) in one ```{lang} code block.") g = chat(msg) comp = extract_completion(lang, task, g['reply']) prompt = go_imports(t['prompt'], g['reply']) if lang == 'go' else t['prompt'] return dict(g, completion=comp if comp is not None else '', prompt=prompt, extract_failed=comp is None) def run_arm(arm, task): lang, tool = ARMS[arm] t = dict(task[lang], stops=list(task[lang]['stops']) + EXTRA_STOPS.get(arm, [])) attempts = [] with tempfile.TemporaryDirectory(prefix='ckh-') as d: g = dict(generate(t['prompt'], t['stops']), prompt=t['prompt'], extract_failed=False) for k in range(MAX_REPAIRS + 1): prog = join(g['prompt'], g['completion'], t['tests']) for f in Path(d).iterdir(): shutil.rmtree(f) if f.is_dir() else f.unlink() ok, diags = gate(arm, prog, d, solution=g['prompt'] + g['completion']) res = test(arm, prog, d) attempts.append({'k': k, 'completion': g['completion'], 'gate_ok': ok, 'diagnostics': diags, 'test': res, 'eval_count': g['eval_count'], 'eval_ns': g['eval_ns'], 'prompt_eval_count': g['prompt_eval_count'], 'cached': g['cached'], 'hit_cap': g['hit_cap'], 'extract_failed': g['extract_failed'], 'reply': g.get('reply')}) if ok or k == MAX_REPAIRS: break g = repair(lang, task, g['completion'], diags, tool.split(' (')[0]) first, last = attempts[0], attempts[-1] return {'task': task['id'], 'arm': arm, 'lang': lang, 'model': MODEL, 'first_gate_ok': first['gate_ok'], 'first_pass': first['test']['passed'], 'final_gate_ok': last['gate_ok'], 'final_pass': last['test']['passed'], 'repairs': len(attempts) - 1, 'tokens': sum(a['eval_count'] for a in attempts if not a['cached'] or a['k'] == 0), 'attempts': attempts} def measure(out, arms, limit=None): tasks = load_tasks() if limit: tasks = sorted(random.Random(SEED).sample(tasks, limit), key=lambda t: t['id']) out = Path(out) done = set() if out.exists(): done = {(r['task'], r['arm']) for r in map(json.loads, out.read_text().splitlines())} print(f'{len(tasks)} tasks x {len(arms)} arms; model {MODEL}; {len(done)} already done', flush=True) with out.open('a') as f: for t in tasks: for arm in arms: if (t['id'], arm) in done: continue t0 = time.monotonic() r = run_arm(arm, t) r['elapsed_s'] = round(time.monotonic() - t0, 2) f.write(json.dumps(r, sort_keys=True) + '\n'); f.flush() print(f"HumanEval/{t['id']:<3} {arm:<12} gate0={'Y' if r['first_gate_ok'] else 'n'} pass0={'Y' if r['first_pass'] else 'n'} " f"repairs={r['repairs']} final={'PASS' if r['final_pass'] else 'fail'} {r['elapsed_s']}s", flush=True) if __name__ == '__main__': a = argparse.ArgumentParser() a.add_argument('action', choices=['measure', 'tasks']) a.add_argument('--out', default=str(BASE / 'results/runs.jsonl')) a.add_argument('--arms', default=','.join(a for a in ARMS if a not in EXTRA_STOPS)) a.add_argument('--limit', type=int) x = a.parse_args() if x.action == 'tasks': ts = load_tasks(); print(len(ts), 'tasks in all six languages:', [t['id'] for t in ts]) else: measure(x.out, x.arms.split(','), x.limit)