| """Run the actual RSI controller/paired gate/checkpoint/audit with offline evaluator doubles.""" |
| import copy |
| from pathlib import Path |
| from common import digest,read_json,read_jsonl,write_json,paired_episode_seed |
| from helpers import state |
| from test_release_integration import command,ROOT |
|
|
|
|
| def test_controller_accept_reject_resume_and_audit_isolation(monkeypatch,tmp_path,config): |
| import rsi_loop |
| from harness import manager |
| from controller import seed_service |
| from contextlib import contextmanager |
| cfg=copy.deepcopy(config);cfg.update(iterations=3,operators=['harness']) |
| config_path=tmp_path/'input.json';write_json(config_path,cfg) |
| campaign=tmp_path/'campaign' |
| prepared=command(ROOT/'prepare.py','--config',config_path,'--output',campaign) |
| assert prepared.returncode==0,prepared.stdout+prepared.stderr |
| cfg=read_json(campaign/'config.json') |
| selection=read_jsonl(campaign/'partition/selection.jsonl') |
| by_category={c:[r for r in selection if r['category']==c] for c in cfg['bfcl_only']['categories']} |
| base_ids={r['task_id'] for rows in by_category.values() for r in rows[:3]} |
| gain0=(base_ids-{by_category['multiple'][0]['task_id']})|{ |
| by_category['simple_python'][3]['task_id'],by_category['simple_python'][4]['task_id']} |
| gain2=gain0|{by_category['simple_python'][5]['task_id']} |
| paths=[] |
| for i in range(3): |
| artifact=manager.initial_artifact();artifact['instructions_extra']=f'Check declared argument structure; revision {i}.' |
| paths.append(str(manager.save_version(tmp_path/f'candidate-{i}',artifact))) |
| proposals=[];analyzed_roles=[];evaluation_roles=[] |
| def evaluate(rows,current,effective,directory): |
| roles={r['split_role'] for r in rows};assert len(roles)==1 |
| role=next(iter(roles));evaluation_roles.append(role) |
| assert role in ('improve','selection','audit') |
| if role=='selection': |
| ids=base_ids if current['harness'] not in paths else [gain0,set(),gain2][paths.index(current['harness'])] |
| else:ids=set() |
| results=[{'domain':'bfcl','task_id':r['task_id'],'resolved':r['task_id'] in ids, |
| 'det_reward':float(r['task_id'] in ids),'failure_labels':[] if r['task_id'] in ids else ['bfcl_argument_value'], |
| 'split_role':role,'seed':paired_episode_seed(r,effective),'output_tokens':0, |
| 'episode_dir':str(Path(directory)/r['task_id'])} for r in rows] |
| score=sum(r['resolved'] for r in results)/len(results) |
| return {'complete':True,'results':results,'score':score,'output_tokens':0, |
| 'by_domain':{'bfcl':{'score':score,'resolved':sum(r['resolved'] for r in results),'total':len(results)}}} |
| def analyze(results,output,**kwargs): |
| analyzed_roles.extend(r['split_role'] for r in results) |
| value={'counts':{'bfcl_argument_value':len(results)},'summary':'Observed improve argument errors.', |
| 'evidence':[{'domain':'bfcl','task_id':r['task_id'],'split_role':'improve', |
| 'failure_labels':['bfcl_argument_value'],'benchmark_resolved':False} for r in results[:12]]} |
| write_json(output,value);return value |
| def search(operator,current,diagnosis,rows,effective,directory,iteration,*args,**kwargs): |
| assert operator=='harness' and all(r['split_role']=='improve' for r in rows) |
| assert 'audit' not in diagnosis |
| proposals.append(copy.deepcopy(current)) |
| return {**current,'harness':paths[iteration]},{'suboperator':'prompt_update','proposal':{}, |
| 'nested':{'status':'candidate_ready','trials':[],'cost':{'output_tokens':0}}} |
| @contextmanager |
| def replicas(*args,**kwargs):yield [],['Offline controller integration fixture'] |
| monkeypatch.setattr(seed_service,'replicas',replicas) |
| monkeypatch.setattr(rsi_loop.runner,'evaluate',evaluate) |
| monkeypatch.setattr(rsi_loop.failure_analyzer,'analyze',analyze) |
| monkeypatch.setattr(rsi_loop.nested_rsi,'search',search) |
| monkeypatch.setattr(rsi_loop.model_client,'verify_serving_profile',lambda req:{'fixture':True}) |
| monkeypatch.setattr('modeling.service_probe.check_generation',lambda *args:{'fixture':True}) |
| output=campaign/'run' |
| rsi_loop._run(cfg,output) |
| history=read_json(output/'history.json') |
| assert [r['accepted'] for r in history]==[True,False,True] |
| assert all(r['selection_comparison_status']=='completed' for r in history) |
| assert proposals[1]['harness']==paths[0] and proposals[2]['harness']==paths[0] |
| assert history[0]['gate']['robust']['by_source']['bfcl/multiple']<0 |
| assert history[0]['gate']['acceptance_rule']=='source_balanced_positive_mean' |
| assert read_json(output/'controller-state.json')['state']['harness']==paths[2] |
| assert read_json(output/'status.json')['status']=='pilot_iterations_complete' |
| curve=read_json(output/'audit-curve.json') |
| assert curve['used_for_any_decision'] is False and len(curve['iterations'])==3 |
| assert set(analyzed_roles)=={'improve'} and set(evaluation_roles)=={'improve','selection','audit'} |
| count=len(proposals);rsi_loop._run(cfg,output) |
| assert len(proposals)==count and read_json(output/'history.json')==history |
|
|
