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']==[]