191 lines
11 KiB
Python
191 lines
11 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_investigator import LoopAdapter,CASE,run,state
|
|
from test_investigation_lean import lean_context
|
|
from test_working_brief import service_for
|
|
from test_case_interpreter import gateway
|
|
from test_project_research import rows
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt,output_schema
|
|
from viano_bug_analysis.analysis.investigation_decision import normalize
|
|
from viano_bug_analysis.analysis.investigation_lean_v20 import wire
|
|
from viano_bug_analysis.case.contracts import validate
|
|
from viano_bug_analysis.package import PackageValidator
|
|
from viano_bug_analysis.budget import BudgetConfig
|
|
|
|
|
|
def consolidation(c):
|
|
return {'assessment':{'semantic_score':50,'reason':'Scoped result.'} if c['previous_result'] else None,
|
|
'confidence':'plausible','findings':[],'uncertainties':['Cause not established.'],
|
|
'next_question':None,'retire':[],'next':{'kind':'finish','reason':'Bounded question examined.'}}
|
|
|
|
|
|
class HybridAdapter:
|
|
def __init__(self, continuous=False,switch=False):
|
|
self.calls=[];self.continuous=continuous;self.switch=switch
|
|
def bootstrap(r,c):r['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
self.bootstrap=LoopAdapter(bootstrap)
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
doc=json.loads(request['messages'][1]['content']);self.calls.append(doc)
|
|
if request['prompt_version']=='investigation_project_v18':
|
|
return await self.bootstrap.generate(config,request,max_response_bytes=max_response_bytes)
|
|
c=doc['input']
|
|
if doc['task']=='investigate_one_step':
|
|
reply={'assessment':{'semantic_score':50,'reason':'Scoped result.'} if c['previous_result'] else None,
|
|
'observation':'A bounded observation.',
|
|
'next':({'kind':'tool','operation':'project.query','arguments':rows()} if self.continuous else
|
|
{'kind':'checkpoint','intent':'switch' if self.switch else 'completion'})}
|
|
else:
|
|
reply=consolidation(c)
|
|
if self.switch and c['selected_hypothesis']=='H1':
|
|
eid=next(iter(c['evidence']))
|
|
reply.update(confidence='supported',findings=[{'direction':'support','text':'Scoped support for H1.','evidence_refs':[eid]}])
|
|
reply['next']={'kind':'switch','hypothesis_id':'H2','reason':'Check the alternative before concluding.'}
|
|
return ProviderResponse(reply,'hybrid-fixture',100,50)
|
|
|
|
|
|
def loop_for(tool_case,adapter,**kwargs):
|
|
return Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case)),**kwargs)
|
|
|
|
|
|
def test_bootstrap_full_once_each_without_extra_consolidation(tool_case):
|
|
adapter=HybridAdapter();loop=loop_for(tool_case,adapter)
|
|
first=run(loop,steps=3)
|
|
assert [d['task'] for d in adapter.calls]==['advance_investigation_goal']*3
|
|
assert [d['input']['selected_hypothesis'] for d in adapter.calls]==['H1','H2','H3']
|
|
assert all(not h.get('lean_session') and h.get('working_brief_ref') for h in state(first)['hypotheses'].values())
|
|
final=run(loop)
|
|
assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
|
|
assert sum(d['task']=='consolidate_investigation_session' for d in adapter.calls)==3
|
|
from viano_bug_analysis.llm.cache_measurement import measure_package
|
|
measurement=measure_package(final.package)
|
|
assert measurement['unmatched_requests']==0 and measurement['all_exactly_reconstructed']
|
|
assert any(s['rebuilt_prompt']=='investigation_project_v20' for s in measurement['samples'])
|
|
|
|
|
|
def test_periodic_checkpoint_disabled_past_four_steps(tool_case):
|
|
adapter=HybridAdapter(continuous=True)
|
|
loop=loop_for(tool_case,adapter,budget_config=BudgetConfig(default_action_cost=1,max_low_information_steps=100,max_reward_bps=0))
|
|
result=run(loop,steps=9)
|
|
assert result.status=='YIELDED'
|
|
assert [d['task'] for d in adapter.calls[3:]]==['investigate_one_step']*6
|
|
assert len(state(result)['hypotheses']['H1']['lean_session']['turns'])==6
|
|
|
|
|
|
def test_explicit_switch_honored_even_with_higher_current_confidence(tool_case):
|
|
adapter=HybridAdapter(switch=True);loop=loop_for(tool_case,adapter)
|
|
result=run(loop,steps=6)
|
|
assert result.status=='YIELDED'
|
|
assert adapter.calls[-1]['input']['hypothesis']['id']=='H2'
|
|
assert state(result)['hypotheses']['H1']['confidence']=='supported'
|
|
|
|
|
|
def test_controller_ids_and_lossless_long_text_with_existing_findings():
|
|
c=lean_context();c['lean_protocol']=2;c['lean_session']['checkpoint']=True
|
|
old=copy.deepcopy(c.get('working_brief',{}).get('brief',{}).get('findings',[]))
|
|
r={'assessment':None,'confidence':'plausible','findings':[{'direction':'neutral','text':'X'*650,
|
|
'evidence_refs':[next(iter(c['evidence']))]}],'uncertainties':[],
|
|
'next_question':'Q'*450,'retire':[],'next':{'kind':'finish','reason':'Done.'}}
|
|
assert Draft202012Validator(output_schema(c,strict=True)).is_valid(r)
|
|
d=normalize(c,r);brief=d['update']['working_brief'];validate(brief,'investigation-working-brief')
|
|
new=[f for f in brief['findings'] if f not in old]
|
|
assert ''.join(f['text'] for f in new)=='X'*650
|
|
assert brief['next_question_details']=='Q'*450 and brief['uncertainties']==[]
|
|
assert all(f in brief['findings'] for f in old)
|
|
assert [f['id'] for f in new]==[f['id'] for f in normalize(c,r)['update']['working_brief']['findings'] if f not in old]
|
|
assert r['findings'][0]['text']=='X'*650 and 'id' not in r['findings'][0]
|
|
|
|
|
|
def test_unknown_switch_or_retirement_rejected():
|
|
c=lean_context();c['lean_protocol']=2;c['lean_session']['checkpoint']=True
|
|
r={'assessment':None,'confidence':'plausible','findings':[],'uncertainties':[],
|
|
'next_question':None,'retire':[],'next':{'kind':'switch','hypothesis_id':'H999','reason':'Unknown.'}}
|
|
from viano_bug_analysis.analysis import InvestigationError
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_SWITCH_TARGET'):wire(c,r)
|
|
r['next']={'kind':'finish','reason':'Done.'};r['retire']=[{'finding_id':'B9999','reason':'Unknown.'}]
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_BRIEF_ID'):wire(c,r)
|
|
|
|
|
|
|
|
def test_retirement_is_explicit_and_counterevidence_is_preserved_by_default():
|
|
c=lean_context();c['lean_protocol']=2;c['lean_session']['checkpoint']=True
|
|
c['working_brief']={'brief':{'findings':[{'id':'B0001','direction':'against','text':'Contrary observation.',
|
|
'scope':'Exact query scope.','evidence_refs':[next(iter(c['evidence']))]}],
|
|
'uncertainties':[],'next_question':'Check uncertainty.','changes':[],'format_version':1}}
|
|
r={'assessment':None,'confidence':'plausible','findings':[],'uncertainties':[],
|
|
'next_question':None,'retire':[],'next':{'kind':'finish','reason':'Done.'}}
|
|
d=normalize(c,r)
|
|
assert d['update']['working_brief']['findings']==c['working_brief']['brief']['findings']
|
|
r['retire']=[{'finding_id':'B0001','reason':'Additional evidence established the earlier observation was from another deployment.'}]
|
|
d=normalize(c,r)
|
|
assert d['update']['working_brief']['findings']==[]
|
|
assert d['update']['working_brief']['changes'][0]['id']=='B0001'
|
|
assert r['retire'][0]['reason'].startswith('Additional evidence')
|
|
|
|
|
|
def test_v20_review_profile_and_stable_context_transport():
|
|
from test_investigation_review import review_context
|
|
from viano_bug_analysis.llm.gateway import is_investigation_review
|
|
from viano_bug_analysis.llm.stable_context import split_document,restore_document
|
|
c=review_context();c.update(prompt_version='investigation_project_v18',lean_protocol=2,
|
|
lean_session={'turns':[],'checkpoint':True})
|
|
request=build_prompt(c)
|
|
assert request['prompt_version']=='investigation_project_v20' and is_investigation_review(request)
|
|
document=json.loads(request['messages'][1]['content'])
|
|
parts=split_document(document,request['prompt_version'])
|
|
assert parts and restore_document(*parts)==document
|
|
assert 'id' not in request['output_schema']['properties']['findings']['items']['properties']
|
|
assert 'update' not in request['output_schema']['properties']
|
|
|
|
|
|
def test_config_disables_periodic_checkpoint_without_disabling_mode():
|
|
cfg=InvestigationConfig.load('config/investigation.lean.toml')
|
|
assert cfg.lean_loop and cfg.lean_protocol==2 and cfg.lean_checkpoint_steps==0
|
|
for value in (-1,False,1.5):
|
|
from viano_bug_analysis.analysis import InvestigationError
|
|
with pytest.raises(InvestigationError):InvestigationConfig(lean_checkpoint_steps=value)
|
|
|
|
|
|
|
|
def test_explicit_switch_survives_portable_resume(tool_case,tmp_path):
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
adapter=HybridAdapter(switch=True);loop=loop_for(tool_case,adapter)
|
|
first=run(loop,steps=5)
|
|
assert state(first)['lean_preferred_hypothesis']=='H2'
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
resumed=Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
result=run(resumed,steps=1)
|
|
assert result.status=='YIELDED' and adapter.calls[-1]['input']['hypothesis']['id']=='H2'
|
|
assert 'lean_preferred_hypothesis' not in state(result)
|
|
|
|
|
|
def test_hybrid_budget_end_settles_all_bootstrap_results(tool_case):
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
adapter=HybridAdapter();loop=loop_for(tool_case,adapter,budget_config=BudgetConfig(default_action_cost=100,max_reward_bps=0))
|
|
result=run(loop)
|
|
assert result.status=='COMPLETED'
|
|
assert all(all(action['status']=='settled' for action in h['actions'].values())
|
|
for h in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
assert all(not h.get('lean_session',{}).get('turns') for h in state(result)['hypotheses'].values())
|
|
|
|
|
|
|
|
def test_long_next_question_stays_in_correct_field_and_next_prompt():
|
|
c=lean_context();c['lean_protocol']=2;c['lean_session']['checkpoint']=True
|
|
r={'assessment':None,'confidence':'plausible','findings':[],'uncertainties':[],
|
|
'next_question':'Check relationship. '*30,'retire':[],'next':{'kind':'defer','reason':'Continue.'}}
|
|
d=normalize(c,r);validate(d['update']['working_brief'],'investigation-working-brief')
|
|
c['working_brief']={'brief':d['update']['working_brief']}
|
|
c['lean_session']['checkpoint']=False
|
|
prompt=json.loads(build_prompt(c)['messages'][1]['content'])
|
|
assert prompt['input']['checkpoint_brief']['next_question']==r['next_question']
|
|
assert prompt['input']['checkpoint_brief']['uncertainties']==[]
|