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