241 lines
13 KiB
Python
241 lines
13 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
import pytest
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig,InvestigationError
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_working_brief import service_for
|
|
from test_case_interpreter import gateway
|
|
from test_project_research import rows
|
|
from test_investigator import CASE,run,state
|
|
|
|
|
|
class Adapter:
|
|
def __init__(self,bad=None):self.calls=[];self.bad=bad
|
|
async def generate(self,config,request,**kwargs):
|
|
body=json.loads(request['messages'][1]['content']);self.calls.append(body)
|
|
if body['task']=='startup_choose_query':
|
|
out={'assessment':None,'observation':'Read bounded world records.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':rows()}}
|
|
elif body['task']=='review_initial_hypotheses':
|
|
out={'reviews':{}}
|
|
for hid,h in body['input']['hypotheses'].items():
|
|
eid=h['previous_result']
|
|
out['reviews'][hid]={'assessment':{'semantic_score':50,'reason':'Scoped useful data.'} if eid else None,
|
|
'confidence':'plausible','findings':[{'direction':'neutral','text':'Captured rows do not prove the cause.',
|
|
'evidence_refs':[eid]}] if eid else [], 'uncertainties':['Cause remains unknown.'],
|
|
'next_question':'Investigate the causal mechanism.','retire':[]}
|
|
if self.bad:self.bad(out,body)
|
|
else:
|
|
raise AssertionError('Unexpected call '+body['task'])
|
|
return ProviderResponse(out,'fixture',100,50)
|
|
|
|
|
|
def loop_for(tool_case,adapter):
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
return Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,
|
|
startup_batch_review=True,lean_checkpoint_steps=0),router=router)
|
|
|
|
|
|
def test_three_lean_queries_then_one_review(tool_case):
|
|
adapter=Adapter();loop=loop_for(tool_case,adapter)
|
|
first=run(loop,steps=3);s=state(first)
|
|
assert [c['input']['hypothesis']['id'] for c in adapter.calls]==['H1','H2','H3']
|
|
assert len(adapter.calls[0]['input']['evidence'])==0
|
|
assert len(adapter.calls[1]['input']['evidence'])==1
|
|
assert len(adapter.calls[2]['input']['evidence'])==2
|
|
assert all(h['confidence']=='unassessed' and not h.get('working_brief_ref') for h in s['hypotheses'].values())
|
|
before=BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
assert all(a['spent']>0 and a['rewarded']==0 for a in before['accounts'].values())
|
|
assert all(next(iter(a['actions'].values()))['status']=='awaiting_assessment' for a in before['accounts'].values())
|
|
final=run(loop,steps=1);s=state(final)
|
|
assert len(adapter.calls)==4 and adapter.calls[-1]['task']=='review_initial_hypotheses'
|
|
assert s['startup']['done'] and all(h['confidence']=='plausible' and h.get('working_brief_ref') for h in s['hypotheses'].values())
|
|
after=BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
assert all(all(a['status']=='settled' for a in h['actions'].values()) for h in after['accounts'].values())
|
|
assert PackageValidator().validate(final.package).valid
|
|
|
|
|
|
def test_bad_review_retains_valid_siblings(tool_case):
|
|
def bad(out,body):
|
|
if 'H2' in out['reviews'] and not body['input']['already_accepted']:
|
|
out['reviews']['H2']['findings'][0]['evidence_refs']=['E999999']
|
|
adapter=Adapter(bad);loop=loop_for(tool_case,adapter)
|
|
result=run(loop,steps=4);s=state(result)
|
|
assert set(s['startup']['accepted'])=={'H1','H3'}
|
|
assert not s['startup']['done']
|
|
result=run(loop,steps=1)
|
|
assert state(result)['startup']['done']
|
|
assert set(adapter.calls[-1]['input']['hypotheses'])=={'H2'}
|
|
assert len(adapter.calls)==5
|
|
|
|
|
|
def test_bootstrap_round_survives_import(tool_case,tmp_path):
|
|
adapter=Adapter();loop=loop_for(tool_case,adapter)
|
|
result=run(loop,steps=2)
|
|
recipient=CheckpointStore(Workspace(tmp_path/'recipient'));recipient.intake(result.package)
|
|
restored=Investigator(recipient,gateway(adapter),router=loop.router)
|
|
result=run(restored,steps=2)
|
|
assert state(result)['startup']['done']
|
|
assert len(adapter.calls)==4
|
|
|
|
|
|
def test_unknown_tool_has_no_budget_side_effect(tool_case):
|
|
class Bad(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
return ProviderResponse({'assessment':None,'observation':'Bad.',
|
|
'next':{'kind':'tool','operation':'shell.exec','arguments':{}}},'fixture',100,50)
|
|
loop=loop_for(tool_case,Bad());result=run(loop,steps=1)
|
|
assert not state(result)['startup']['explored']
|
|
assert all(not a['actions'] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
|
|
|
|
def test_config_requires_lean_v2():
|
|
with pytest.raises(InvestigationError):InvestigationConfig(startup_batch_review=True)
|
|
|
|
|
|
def test_settlement_and_saved_call_reused_after_interrupted_publication(tool_case,monkeypatch):
|
|
adapter=Adapter();loop=loop_for(tool_case,adapter)
|
|
run(loop,steps=3)
|
|
original=loop._save
|
|
def crash(s,**kw):
|
|
if s.get('startup',{}).get('done'):raise RuntimeError('crash after settlements')
|
|
return original(s,**kw)
|
|
monkeypatch.setattr(loop,'_save',crash)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):run(loop,steps=1)
|
|
before=BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
monkeypatch.setattr(loop,'_save',original)
|
|
result=run(loop,steps=1)
|
|
assert state(result)['startup']['done'] and len(adapter.calls)==4
|
|
assert BudgetManager(tool_case[0]).snapshot(*CASE).state==before
|
|
|
|
|
|
def test_group_review_repeated_failures_block_without_losing_siblings(tool_case):
|
|
def bad(out,body):
|
|
if 'H2' in out['reviews']:out['reviews']['H2']['assessment']['semantic_score']=1000
|
|
adapter=Adapter(bad);loop=loop_for(tool_case,adapter)
|
|
result=run(loop,steps=8)
|
|
assert result.status=='BLOCKED' and result.code=='INVESTIGATION_STARTUP_REVIEW_EXHAUSTED'
|
|
assert set(state(result)['startup']['accepted'])=={'H1','H3'}
|
|
assert len(adapter.calls)==6
|
|
with PackageReader(result.package) as reader:assert reader.manifest['routing']['reason']=='INVALID_LLM_RESPONSES'
|
|
|
|
|
|
def test_terminal_initial_account_is_skipped_without_a_query(tool_case):
|
|
adapter=Adapter();loop=loop_for(tool_case,adapter)
|
|
manager=BudgetManager(tool_case[0]);manager.initialize(*CASE);manager.complete(*CASE,'H2',reason='Fixture terminal account')
|
|
result=run(loop,steps=3)
|
|
assert [c['input']['hypothesis']['id'] for c in adapter.calls if c['task']=='startup_choose_query']==['H1','H3']
|
|
assert state(result)['startup']['done']
|
|
assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H2']['actions']
|
|
|
|
|
|
def test_initial_query_cannot_request_checkpoint_or_early_finish(tool_case):
|
|
class Bad(Adapter):
|
|
async def generate(self,*args,**kwargs):
|
|
return ProviderResponse({'assessment':None,'observation':'Skip research.',
|
|
'next':{'kind':'checkpoint','intent':'completion'}},'fixture',100,50)
|
|
result=run(loop_for(tool_case,Bad()),steps=1)
|
|
assert not state(result)['startup']['explored']
|
|
assert all(not a['actions'] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
|
|
|
|
def test_stable_cache_partition_preserves_initial_query(tool_case):
|
|
from viano_bug_analysis.llm.stable_context import split_document,restore_document
|
|
adapter=Adapter();run(loop_for(tool_case,adapter),steps=1)
|
|
doc=adapter.calls[0]
|
|
split=split_document(doc,'investigation_startup_v1')
|
|
assert split and restore_document(*split)==doc
|
|
assert 'tool_schema' in split[0]['input'] and 'hypothesis' in split[1]['input']
|
|
|
|
|
|
def test_confidence_scheduler_resumes_after_joint_review(tool_case):
|
|
def rank(out,body):
|
|
h=out['reviews']['H2'];h['confidence']='strongly_supported';h['findings'][0]['direction']='support'
|
|
class Ranked(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body=json.loads(request['messages'][1]['content'])
|
|
if body['task']=='investigate_one_step':
|
|
self.calls.append(body)
|
|
return ProviderResponse({'assessment':None,'observation':'Investigate strongest candidate.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':rows()}},'fixture',100,50)
|
|
return await super().generate(config,request,**kwargs)
|
|
adapter=Ranked(rank);loop=loop_for(tool_case,adapter)
|
|
result=run(loop,steps=5)
|
|
assert state(result)['startup']['done']
|
|
assert adapter.calls[-1]['task']=='investigate_one_step'
|
|
assert adapter.calls[-1]['input']['hypothesis']['id']=='H2'
|
|
assert len(adapter.calls)==5
|
|
|
|
|
|
def test_batch_contract_supports_five_and_sparse_numeric_ids():
|
|
from test_investigation_lean import lean_context
|
|
from viano_bug_analysis.analysis.investigation_startup import request_for
|
|
c=lean_context();c['lean_protocol']=2;c['lean_session']['checkpoint']=True
|
|
contexts={}
|
|
for hid in ['H1','H2','H3','H5','H10']:
|
|
item=copy.deepcopy(c);item['selected_hypothesis']=hid
|
|
item['hypotheses'][hid]=copy.deepcopy(c['hypotheses'][c['selected_hypothesis']])
|
|
contexts[hid]=item
|
|
request=request_for(contexts,{},{});body=json.loads(request['messages'][1]['content'])
|
|
assert set(body['input']['hypotheses'])=={'H1','H2','H3','H5','H10'}
|
|
assert 'review_contract' in body and 'review_contract_overrides' not in body
|
|
assert list(request['output_schema']['properties']['reviews']['properties'])==list(contexts)
|
|
|
|
|
|
def test_frozen_review_request_is_checked_before_reuse(tool_case,monkeypatch):
|
|
from viano_bug_analysis.analysis import investigation_startup as startup
|
|
adapter=Adapter();loop=loop_for(tool_case,adapter);run(loop,steps=3)
|
|
original_run=loop.runner.run
|
|
async def crash(*a,**kw):raise RuntimeError('Before reviewer call')
|
|
monkeypatch.setattr(loop.runner,'run',crash)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):run(loop,steps=1)
|
|
monkeypatch.setattr(loop.runner,'run',original_run)
|
|
original_read=startup.raw_request
|
|
def changed(reader,ref,limit):
|
|
request=original_read(reader,ref,limit)
|
|
if '/startup/requests/' in ref:request['messages'][0]['content']='Wrong instructions'
|
|
return request
|
|
monkeypatch.setattr(startup,'raw_request',changed)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_STARTUP_BINDING'):run(loop,steps=1)
|
|
assert len(adapter.calls)==3
|
|
|
|
|
|
def test_format_only_leaf_recovery_does_not_repeat_joint_call(tool_case):
|
|
def encoded(out,body):out['reviews']['H2']=json.dumps(out['reviews']['H2'])
|
|
adapter=Adapter(encoded);result=run(loop_for(tool_case,adapter),steps=4)
|
|
assert state(result)['startup']['done'] and len(adapter.calls)==4
|
|
with PackageReader(result.package) as r:
|
|
refs={entry['llm_ref'] for entry in state(result)['startup']['accepted'].values()}
|
|
assert len(refs)==1
|
|
original=json.loads(r.read_bytes(next(iter(refs))))
|
|
assert isinstance(original['outcome']['document']['reviews']['H2'],str)
|
|
|
|
|
|
def test_existing_lean_completion_still_reaches_diagnosis(tool_case):
|
|
class Complete(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body=json.loads(request['messages'][1]['content'])
|
|
if body['task']=='investigate_one_step':
|
|
self.calls.append(body)
|
|
out={'assessment':None,'observation':'Ready to conclude this hypothesis.',
|
|
'next':{'kind':'checkpoint','intent':'completion'}}
|
|
elif body['task']=='consolidate_investigation_session':
|
|
self.calls.append(body)
|
|
out={'assessment':None,'confidence':'plausible','findings':[],
|
|
'uncertainties':['No proven cause.'],'next_question':None,'retire':[],
|
|
'next':{'kind':'finish','reason':'Available bounded investigation complete.'}}
|
|
else:return await super().generate(config,request,**kwargs)
|
|
return ProviderResponse(out,'fixture',100,50)
|
|
adapter=Complete();result=run(loop_for(tool_case,adapter),steps=20)
|
|
assert result.status=='COMPLETED'
|
|
assert state(result)['startup']['done']
|
|
with PackageReader(result.package) as r:assert r.manifest['state']=='BUILDING_DIAGNOSIS'
|
|
assert sum(c['task']=='review_initial_hypotheses' for c in adapter.calls)==1
|