import asyncio import copy import json import pytest from jsonschema import Draft202012Validator from test_compact_investigation import context from test_case_builder import case_setup from test_investigation_tools import tool_case from test_investigator import CASE,run,state 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,InvestigationError 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.llm import ProviderResponse from viano_bug_analysis.lifecycle import CheckpointStore,Workspace from viano_bug_analysis.package import PackageReader,PackageValidator from viano_bug_analysis.tools import InvestigationToolRouter from viano_bug_analysis.budget import BudgetManager def lean_context(): c=context();c['prompt_version']='investigation_project_v18' c['lean_session']={'turns':[],'checkpoint':False} return c def test_short_contract_does_not_rewrite_brief_or_confidence(): c=lean_context();before=copy.deepcopy(c) r={'assessment':{'semantic_score':40,'reason':'Useful observation.'} if c['previous_result'] else None, 'observation':'A bounded result, not a diagnosis.','next':{'kind':'checkpoint'}} assert Draft202012Validator(output_schema(c)).is_valid(r) d=normalize(c,r) assert 'working_brief' not in d['update'] and c==before assert d['next']['kind']=='defer' assert d['update']['confidence']==c['hypotheses'][c['selected_hypothesis']]['confidence'] r['update']={'confidence':'strongly_supported'} assert not Draft202012Validator(output_schema(c)).is_valid(r) def test_stable_cache_roundtrip_and_no_hard_result_truncation(): from viano_bug_analysis.llm.stable_context import split_document,restore_document c=lean_context();r=build_prompt(c);document=json.loads(r['messages'][1]['content']) split=split_document(document,r['prompt_version']) assert split and restore_document(*split)==document assert 'request_contract' in split[0]['input']['tool_schema'] assert 'recent_observations' in split[1]['input'] assert 'next_step' not in r['output_schema']['properties'] class LeanAdapter: def __init__(self):self.calls=[] async def generate(self,config,request,*,max_response_bytes): doc=json.loads(request['messages'][1]['content']);c=doc['input'];self.calls.append(doc) if doc['task']=='investigate_one_step': result={'assessment':({'semantic_score':50,'reason':'Bounded result.'} if c['previous_result'] else None),'observation':'Inspect bounded project records.', 'next':({'kind':'checkpoint'} if c['previous_result'] else {'kind':'tool','operation':'project.query','arguments':rows()})} else: previous=c['previous_result'] result={'assessment':{'semantic_score':50,'reason':'Scoped result assessed.'} if previous else None, 'update':{'confidence':'plausible','add_questions':[],'resolve_questions':[], 'working_brief':{'format_version':2,'findings':[],'retired':[], 'uncertainties':['Cause remains unconfirmed.'],'next_question':'No further bounded checks.'}}, 'next':{'kind':'finish','reason':'Bounded investigation complete.'}} return ProviderResponse(result,'lean-fixture',100,50) def test_switch_flush_resume_budget_and_portability(tool_case,tmp_path): adapter=LeanAdapter();service=service_for(tool_case);router=InvestigationToolRouter(project_research=service) config=InvestigationConfig(lean_loop=True) first=run(Investigator(tool_case[0],gateway(adapter),config,router=router),steps=1) h=state(first)['hypotheses']['H1'] assert len(h['lean_session']['turns'])==1 and not h.get('working_brief_ref') portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package) final=run(Investigator(portable,gateway(adapter),router=router),steps=20) assert final.status=='COMPLETED' assert sum(d['task']=='consolidate_investigation_session' for d in adapter.calls)==3 assert sum(d['task']=='investigate_one_step' for d in adapter.calls)==4 assert all(not h['lean_session']['turns'] and h.get('working_brief_ref') for h in state(final)['hypotheses'].values()) assert all(all(a['status']=='settled' for a in h['actions'].values()) for h in BudgetManager(portable).snapshot(*CASE).state['accounts'].values()) assert PackageValidator().validate(final.package).valid @pytest.mark.parametrize('limit', [1,2]) def test_iteration_boundary_consolidates_before_leaving_hypothesis(tool_case,limit): adapter=LeanAdapter();router=InvestigationToolRouter(project_research=service_for(tool_case)) final=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,max_iterations_per_hypothesis=limit),router=router)) assert final.status=='COMPLETED' assert all(not h.get('lean_session',{}).get('turns') for h in state(final)['hypotheses'].values()) def test_invalid_short_tool_reply_has_no_budget_or_evidence_side_effect(tool_case): class Invalid(LeanAdapter): async def generate(self,config,request,*,max_response_bytes): return ProviderResponse({'assessment':None,'observation':'Attempt forbidden execution.', 'next':{'kind':'tool','operation':'shell.exec','arguments':{}}},'fixture',100,50) loop=Investigator(tool_case[0],gateway(Invalid()),InvestigationConfig(lean_loop=True), router=InvestigationToolRouter(project_research=service_for(tool_case))) result=asyncio.run(loop.step(*CASE)) assert result.status=='REJECTED' assert all(not h['actions'] and h['remaining']==100 for h in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values()) assert not state(result)['hypotheses']['H1'].get('lean_session') def test_checkpoint_has_no_next_step_and_retains_review_low_routing(): from test_investigation_review import review_context from viano_bug_analysis.llm.gateway import is_investigation_review c=review_context();c['prompt_version']='investigation_project_v18' c['lean_session']={'turns':[],'checkpoint':True} request=build_prompt(c) assert is_investigation_review(request) assert 'next_step' not in request['output_schema']['properties']['update']['properties'] def test_session_consolidation_includes_all_prior_payloads(): c=lean_context();ids=list(c['evidence']) c['lean_session']={'checkpoint':True,'turns':[{'llm_ref':'stored/ref.json','evidence_id':eid, 'observation':'Provisional observation.'} for eid in ids]} doc=json.loads(build_prompt(c)['messages'][1]['content']) assert set(ids)<=set(doc['input']['evidence']) assert doc['input']['session_log']==c['lean_session']['turns'] def test_periodic_checkpoint_and_defer_continue_without_confidence_update(tool_case): class Continue(LeanAdapter): async def generate(self,config,request,*,max_response_bytes): doc=json.loads(request['messages'][1]['content']);self.calls.append(doc);c=doc['input'] if doc['task']=='investigate_one_step': answer={'assessment':{'semantic_score':60,'reason':'Read scoped result.'} if c['previous_result'] else None, 'observation':'No established cause yet.','next':{'kind':'tool','operation':'project.query','arguments':rows()}} else: answer={'assessment':{'semantic_score':60,'reason':'Read scoped result.'} if c['previous_result'] else None, 'update':{'confidence':'plausible','add_questions':[],'resolve_questions':[], 'working_brief':{'format_version':2,'findings':[],'retired':[], 'uncertainties':['Unknown causal mechanism.'],'next_question':'Investigate another relationship.'}}, 'next':{'kind':'defer','reason':'Continue bounded research.'}} return ProviderResponse(answer,'fixture',100,50) adapter=Continue();router=InvestigationToolRouter(project_research=service_for(tool_case)) loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_checkpoint_steps=2),router=router) result=run(loop,steps=10) assert result.status=='YIELDED' checkpoints=[d for d in adapter.calls if d['task']=='consolidate_investigation_session'] assert any(len(d['input']['session_log'])==2 for d in checkpoints) assert all(len(h.get('lean_session',{}).get('turns',[]))<=2 for h in state(result)['hypotheses'].values()) def test_unknown_evidence_cannot_enter_consolidated_findings(): from viano_bug_analysis.analysis.investigation_compact import validate_tool_shape from viano_bug_analysis.analysis.investigation_config import InvestigationError from test_lean_investigation_decision import reply c=lean_context();c['lean_session']['checkpoint']=True;r=reply() r['next']={'kind':'finish','reason':'Done.'} r['update']['working_brief']['findings']=[{'id':'B9999','direction':'support','text':'Unseen.', 'scope':'Unknown scope.','evidence_refs':['E999999']}] with pytest.raises(InvestigationError):validate_tool_shape(c,r) def test_budget_exhaustion_flushes_session_and_settles_pending_result(tool_case): from viano_bug_analysis.budget import BudgetConfig adapter=LeanAdapter();router=InvestigationToolRouter(project_research=service_for(tool_case)) result=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),router=router, budget_config=BudgetConfig(default_action_cost=100,max_reward_bps=0))) assert result.status=='COMPLETED' assert all(not h.get('lean_session',{}).get('turns') for h in state(result)['hypotheses'].values()) assert all(all(a['status']=='settled' for a in h['actions'].values()) for h in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values()) def test_replay_after_tool_publication_does_not_repeat_model_or_budget(tool_case,monkeypatch): adapter=LeanAdapter();loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True), router=InvestigationToolRouter(project_research=service_for(tool_case))) original=loop.evidence.store_tool_results def interrupted(*args,**kwargs): original(*args,**kwargs) raise RuntimeError('Injected interruption after evidence publication') monkeypatch.setattr(loop.evidence,'store_tool_results',interrupted) with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):asyncio.run(loop.step(*CASE)) monkeypatch.setattr(loop.evidence,'store_tool_results',original) result=asyncio.run(loop.step(*CASE)) assert result.status=='PROGRESSED' and len(adapter.calls)==1 assert len(state(result)['hypotheses']['H1']['lean_session']['turns'])==1 assert len(BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions'])==1 def test_bad_consolidation_finding_id_repairs_without_corrupting_context(tool_case,tmp_path): from viano_bug_analysis.case.contracts import validate class BadIdOnce(LeanAdapter): failed=False async def generate(self,config,request,*,max_response_bytes): response=await super().generate(config,request,max_response_bytes=max_response_bytes) c=self.calls[-1]['input'] if self.calls[-1]['task']=='consolidate_investigation_session' and not self.failed: self.failed=True eid=next(iter(c['evidence'])) response.document['update']['working_brief']['findings']=[{'id':'F01','direction':'neutral', 'text':'Schema available.','scope':'Captured schema.','evidence_refs':[eid]}] return response adapter=BadIdOnce();router=InvestigationToolRouter(project_research=service_for(tool_case)) loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),router=router) first=run(loop,steps=1) before=BudgetManager(tool_case[0]).snapshot(*CASE).state rejected=asyncio.run(loop.step(*CASE)) assert rejected.status=='REJECTED' assert BudgetManager(tool_case[0]).snapshot(*CASE).state==before with PackageReader(rejected.package) as reader: c=json.loads(reader.read_bytes(state(rejected)['active']['context_ref'])) validate(c,'investigation-context') detail=c['previous_rejection_details'] assert 'validation_keyword' not in detail assert 'B0001' in detail['requirement'] # The original released version persisted this extra keyword. Its # immutable context must remain loadable without a package rewrite. c['previous_rejection_details']={'validation_path':'/update/working_brief/findings/0/id', 'validation_fields':[],'validation_keyword':'pattern'} validate(c,'investigation-context') prompt=build_prompt(c) assert 'B0001' in json.loads(prompt['messages'][1]['content'])['input']['previous_rejection_details']['requirement'] portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(rejected.package) final=run(Investigator(portable,gateway(adapter),router=router)) assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid def test_invalid_context_is_not_published(tool_case): from viano_bug_analysis.case.errors import CaseBuildError adapter=LeanAdapter();loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True), router=InvestigationToolRouter(project_research=service_for(tool_case))) result=run(loop,steps=1);saved=state(result) # A new prompt context is validated before a checkpoint can reference it. saved['active']={'hypothesis_id':'H1','iteration':2,'attempt':0,'phase':'prompt', 'context_ref':'analysis/investigation/0001/contexts/invalid.json','context_sha256':'0'*64, 'decision':None,'llm_ref':None,'test_request':None,'test_deadline':None,'retry_token':'initial'} before=tool_case[0].latest(*CASE) with pytest.raises(CaseBuildError) as caught:loop._save(saved,context={'unexpected':'invalid'}) assert caught.value.code=='INPUT_SCHEMA' assert tool_case[0].latest(*CASE)==before