49 lines
2.4 KiB
Python
49 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute only manually reviewed first_success AST nodes in a local test process.
|
|
|
|
No model output is a shell command. All completions must be inspected before
|
|
using this helper; this is a functional checker, not a security sandbox.
|
|
"""
|
|
import ast,asyncio,json,pathlib,re,sys
|
|
|
|
async def check(source, pass_tasks):
|
|
tree=ast.parse(source)
|
|
node=next(n for n in tree.body if isinstance(n,ast.AsyncFunctionDef) and n.name=='first_success')
|
|
scope={'asyncio':asyncio}
|
|
exec(compile(ast.Module(body=[node],type_ignores=[]),'<reviewed-answer>','exec'),scope)
|
|
async def fetch(delay, value=None, fail=False):
|
|
await asyncio.sleep(delay)
|
|
if fail: raise ValueError('synthetic failure')
|
|
return value
|
|
coros=[fetch(.001,fail=True),fetch(.02,7),fetch(.1,9)]
|
|
inputs=[asyncio.create_task(c) for c in coros] if pass_tasks else coros
|
|
try:
|
|
value=await asyncio.wait_for(scope['first_success'](inputs),timeout=1)
|
|
return {'fast_failure_then_success':value==7,'returned':value}
|
|
except Exception as e:
|
|
return {'fast_failure_then_success':False,'error':type(e).__name__+': '+str(e)}
|
|
finally:
|
|
for x in inputs:
|
|
if isinstance(x,asyncio.Task):
|
|
if not x.done():x.cancel()
|
|
elif asyncio.iscoroutine(x): x.close()
|
|
tasks=[x for x in inputs if isinstance(x,asyncio.Task)]
|
|
if tasks:await asyncio.gather(*tasks,return_exceptions=True)
|
|
|
|
results=[]
|
|
for p in pathlib.Path(sys.argv[1]).glob('*/result.json'):
|
|
data=json.loads(p.read_text())
|
|
for phase in ['quality','quality_followup']:
|
|
for item in data.get(phase,[]):
|
|
if item['id']!='i3_code_debugging':continue
|
|
content=item['response']['choices'][0]['message'].get('content','')
|
|
blocks=re.findall(r'```python\s*\n(.*?)```',content,re.S)
|
|
candidates=[b for b in blocks if 'async def first_success' in b and ('create_task' in b or 'ensure_future' in b or 'asyncio.wait' in b)]
|
|
if not candidates:continue
|
|
source=candidates[-1]
|
|
# Contract used by the answer's own main(): tasks versus bare coroutines.
|
|
main=source.split('async def main',1)[-1]
|
|
pass_tasks='asyncio.create_task(fetch(' in main
|
|
results.append({'case':data['case']['label'],'phase':phase,'input_contract':'tasks' if pass_tasks else 'coroutines',**asyncio.run(check(source,pass_tasks))})
|
|
print(json.dumps(results,indent=2))
|