import asyncio import copy import json from dataclasses import asdict,replace import pytest from jsonschema import Draft202012Validator from viano_bug_analysis.analysis.investigation_compact import model_view,source_aliases,_content,validate_tool_shape from viano_bug_analysis.analysis.investigation_prompt import build_prompt,output_schema from viano_bug_analysis.analysis.investigation_config import InvestigationConfig,InvestigationError from viano_bug_analysis.analysis.investigation_decision import normalize from viano_bug_analysis.analysis import Investigator,DiagnosisBuilder from viano_bug_analysis.budget import BudgetManager from viano_bug_analysis.package import PackageReader,PackageValidator from viano_bug_analysis.lifecycle import CheckpointStore,Workspace from viano_bug_analysis.llm.gemini import GeminiProvider from viano_bug_analysis.llm.stable_context import split_document,restore_document from viano_bug_analysis.tools import InvestigationToolRouter from test_investigation_tools import tool_case from test_case_builder import case_setup from test_case_interpreter import gateway from test_investigator import LoopAdapter,CASE,run,state from test_working_brief import service_for,brief from test_project_research import rows def entry(number,kind='source.read'): query={'kind':kind,'path':'code.cpp','start_line':1,'end_line':2} items=[{'path':'code.cpp','line':1,'text':'important-'+str(number)+'x'*1000}, {'path':'code.cpp','line':2,'text':'next line'}] if kind=='database.schema': query={'kind':kind,'table':'creature'} items=[{'table':'creature','columns':[{'COLUMN_NAME':'id','COLUMN_TYPE':'int'}, {'COLUMN_NAME':'guid','COLUMN_TYPE':'int'}]}] return {'record':{'id':f'E{number:06d}','hypothesis_id':'H1','action_id':str(number)}, 'content':{'coverage':'schema_only' if kind=='database.schema' else 'selected_lines', 'items':[{'query':query,'origin':{'commit':'a'*40},'result':{'coverage':'schema_only' if kind=='database.schema' else 'selected_lines', 'truncated':False,'items':items}}]}} def context(): h={'proposal':{'title':'Missing hook','mechanism':'Required hook is absent.', 'expected_evidence':['Missing registration'],'falsifying_evidence':['Hook exists']}, 'confidence':'plausible','disposition':'active','questions':[], 'notes':[]} evidence={f'E{i:06d}':entry(i) for i in range(1,21)} c={'prompt_version':'investigation_project_v15','case':{'case':{'case_id':'c','package_id':'p','analysis_revision':1}, 'source_context':{'refs':[{'path':'original/code.cpp'}]},'database_context':{'refs':[]}}, 'interpretation':{'observed_behavior':[{'text':'Interaction fails','source_ids':['O0001']}], 'confirmed_technical_facts':[{'text':'Captured quest giver differs from affected subject.','source_ids':['F0001']}]}, 'hypotheses':{'H1':h,'H2':copy.deepcopy(h)},'selected_hypothesis':'H1','evidence':evidence, 'previous_result':evidence['E000020'],'working_brief':{'brief':brief()}, 'other_working_briefs':{'H2':{'brief':brief()}}, 'budget':{'accounts':{'H1':{'remaining':90,'status':'investigating','actions':{}},'H2':{'remaining':90,'status':'investigating','actions':{}}}}, 'budget_policy':{'default_action_cost':10,'action_costs':{}},'policy':asdict(InvestigationConfig()), 'tool_schema':{'x-project-bindings':{'source':{'commit':'a'*40},'database':{'snapshot_id':'snap'}, 'max_rows':50,'max_lines':80}},'recalled_evidence':[]} return c def test_working_memory_replaces_payload_history_without_hiding_counterfindings(): c=context();original=copy.deepcopy(c);v=model_view(c) assert c==original assert set(v['evidence'])=={'E000020'} assert v['working_brief']['findings'][0]['direction']=='against' assert v['other_working_briefs']['H2']['findings'][0]['evidence_refs']==['E000001'] assert 'E000001' in {r['id'] for r in v['evidence_catalog']} assert next(r for r in v['evidence_catalog'] if r['id']=='E000001')['queries'][0]['origin']=={'commit':'a'*40} assert 'E000002' in v['archive_ids'] assert len(json.dumps(v)) < len(json.dumps(c))/2 def test_fresh_pending_recalled_and_schema_results_stay_available(): c=context();c['recalled_evidence']=['E000002'] c['budget']['accounts']['H1']['actions']['3']={'status':'recorded'} c['evidence']['E000004']=entry(4,'database.schema') v=model_view(c) assert {'E000002','E000003','E000020'}<=set(v['evidence']) assert 'E000004' in v['known_schemas'] assert v['known_schemas']['E000004']['captures'][0]['coverage']=='schema_only' assert v['known_schemas']['E000004']['captures'][0]['tables'][0]['columns']==[['id','int'],['guid','int']] c['policy']['prompt_target_characters']=1 c['evidence']['E000020']['content']['items'][0]['result']['items'][0]['text']='CRITICAL'*20000 v=model_view(c) assert 'CRITICAL'*20000 in json.dumps(v) assert v['evidence_policy']['character_target']==1 def test_memoryless_migration_bootstraps_all_evidence_once(): c=context();c.pop('working_brief');v=model_view(c) assert set(v['evidence'])==set(c['evidence']) assert v['working_brief_status']=='bootstrap_complete_history' def test_row_table_preserves_literal_scalar_types_and_duplicates(): rows=[{'id':1,'map':0,'null':None,'literal':{'encoding':'result_records_v1','rows':[]}}, {'id':2,'map':0,'null':None,'literal':{'encoding':'result_records_v1','rows':[]}}]*10 raw={'items':[{'query':{'kind':'database.rows'},'result':{'items':[{'table':'creature','row':r} for r in rows], 'truncated':True,'next_offset':20,'coverage':'page'}}]} encoded=_content(raw)['items'][0]['result'];table=encoded['items']['records'] restored=[{**table['common'],**dict(zip(table['fields'],r))} for r in table['rows']] assert restored==rows assert encoded['truncated'] is True and encoded['next_offset']==20 def test_stable_cache_partition_survives_new_findings_and_hypothesis_switch(): c=context();first=build_prompt(c);doc=json.loads(first['messages'][1]['content']) stable,live=split_document(doc,'investigation_project_v15') assert restore_document(stable,live)==doc c['selected_hypothesis']='H2';c['previous_result']=None;c['evidence']['E000021']=entry(21) updated=json.loads(build_prompt(c)['messages'][1]['content']);stable2,_=split_document(updated,'investigation_project_v15') assert stable2==stable def test_wire_contract_keeps_finish_reject_early_review_test_and_recall(): c=context();wire=output_schema(c) for action in [{'kind':'finish','reason':'Done'},{'kind':'reject','reason':'Refuted'}, {'kind':'defer','reason':'Await review'}, {'kind':'request_diagnosis','reason':'Cause supported','countercheck_refs':['E000001']}, {'kind':'recall','evidence_refs':['E000002']}]: assert Draft202012Validator(wire['properties']['next']).is_valid(action) assert 'working_brief' not in output_schema(dict(c,prompt_version='investigation_project_v12'))['properties']['update']['properties'] def test_loose_wire_does_not_relax_controller_argument_or_brief_validation(): from test_lean_investigation_decision import reply c=context();r=reply();r['next']={'kind':'tool','operation':'project.query','arguments':{'kind':'database.rows','table':'creature'}} assert Draft202012Validator(output_schema(c)).is_valid(r) with pytest.raises(InvestigationError,match='INVESTIGATION_OUTPUT'):validate_tool_shape(c,r) r['next']={'kind':'finish','reason':'Done'};r['update']['working_brief']['findings'][0]['text']='x'*301 with pytest.raises(InvestigationError):validate_tool_shape(c,r) def test_packaged_source_alias_roundtrip_and_no_mutation(): c=context();before=copy.deepcopy(c);sources,aliases=source_aliases(c) alias=next(iter(sources));assert alias.startswith('PS') and aliases[alias]=='original/code.cpp' assert sources[alias]=='code.cpp' and c==before def test_recalled_evidence_stays_pinned_without_extra_queries_and_portable_replay(tool_case,tmp_path): store=tool_case[0];service=service_for(tool_case);calls={};seen=[] def change(reply,c): hid=c['selected_hypothesis'];calls[hid]=calls.get(hid,0)+1;n=calls[hid] if hid=='H1' and n in (1,2):reply['next']={'kind':'tool','operation':'project.query','arguments':rows()} elif hid=='H1' and n==3:reply['next']={'kind':'recall','evidence_refs':['E000001']} elif hid=='H1' and n==4: assert 'E000001' in c['evidence'];seen.append(c) reply['next']={'kind':'finish','reason':'Scoped investigation complete'} adapter=LoopAdapter(change) loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) result=run(loop) assert result.status=='COMPLETED' and seen saved=state(result);assert saved['recalled_evidence']['H1']==['E000001'] assert len(BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['actions'])==2 portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(result.package) resumed=run(Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service))) assert resumed.status=='COMPLETED' assert PackageValidator().validate(result.package).valid def test_v15_small_prefix_attempts_cache_instead_of_failing_local_character_heuristic(monkeypatch): from test_llm_caching import config,scripted,CACHE,native from test_llm_gateway import request req=request();req['prompt_version']='investigation_project_v15' adapter=GeminiProvider();cfg=replace(config(),cache_min_prefix_chars=100000,cache_required=True) seen=scripted(adapter,monkeypatch,[CACHE,native()]) result=asyncio.run(adapter.generate(cfg,req,max_response_bytes=10000)) assert result.diagnostics['cache_status']=='created' and len(seen)==2 def test_unseen_archive_ids_are_not_newly_citable_before_local_read(): from test_lean_investigation_decision import reply c=context();r=reply();r['update']['working_brief']['findings'][0]['evidence_refs']=['E000002'] with pytest.raises(InvestigationError,match='EVIDENCE_NOT_READ'):validate_tool_shape(c,r) c['recalled_evidence']=['E000002'] validate_tool_shape(c,r) def test_other_pending_payload_has_inventory_and_is_full_on_its_own_turn(): c=context();e=c['evidence']['E000003'];e['record']['hypothesis_id']='H2' c['budget']['accounts']['H2']['actions']['3']={'status':'recorded'} v=model_view(c) assert 'E000003' not in v['evidence'] and 'E000003' in v['evidence_policy']['pending_other_ids'] assert any(r['id']=='E000003' for r in v['evidence_catalog']) c['selected_hypothesis']='H2';c['previous_result']=e assert 'E000003' in model_view(c)['evidence'] @pytest.mark.parametrize('action',[{'kind':'finish'},{'kind':'tool','operation':'project.query','arguments':{'kind':'database.rows','table':'creature'}}]) def test_invalid_compact_wire_has_no_tool_or_budget_side_effect(tool_case,action): store=tool_case[0];service=service_for(tool_case) adapter=LoopAdapter(lambda r,c:r.update(next=action)) result=asyncio.run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)).step(*CASE)) assert result.status=='REJECTED' and result.code=='INVESTIGATION_OUTPUT' assert not service._captures assert all(not a['actions'] for a in BudgetManager(store).snapshot(*CASE).state['accounts'].values()) def test_already_visible_local_read_is_rejected_before_external_work(tool_case): store=tool_case[0];service=service_for(tool_case) def change(r,c): if c['previous_result']:r['next']={'kind':'recall','evidence_refs':[c['previous_result']['evidence_id']]} else:r['next']={'kind':'tool','operation':'project.query','arguments':rows()} adapter=LoopAdapter(change) loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) for _ in range(4):result=asyncio.run(loop.step(*CASE)) assert result.status=='REJECTED' and result.code=='INVESTIGATION_RECALL_VISIBLE' snapshot=BudgetManager(store).snapshot(*CASE).state assert len(snapshot['accounts']['H1']['actions'])==1 def test_frozen_v14_active_request_remains_replayable(tool_case): store=tool_case[0];service=service_for(tool_case);adapter=LoopAdapter() loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) original=loop._context def old_context(*args): value=original(*args);value['prompt_version']='investigation_project_v14' value.pop('other_working_briefs',None);value.pop('recalled_evidence',None) return value loop._context=old_context execute=loop._execute_plan def stop_before_execution(state,budget):raise RuntimeError('simulated crash before execution') loop._execute_plan=stop_before_execution with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):run(loop,steps=1) resumed=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=1) assert resumed.status=='YIELDED' and len(adapter.contexts)==1 def test_v15_early_completion_reads_local_countercheck_then_independent_review(tool_case): from test_diagnosis_builder import FinalAdapter,build store=tool_case[0];service=service_for(tool_case);local_reads=[] def change(reply,c): early=c['early_review']['request'] if early and early['status']=='preparing': reply['next']={'kind':'defer','reason':'Settle pending evidence.'};return own=[e for e,v in c['evidence'].items() if v['record']['hypothesis_id']=='H1'] if c['selected_hypothesis']=='H1' and own: eid=own[0] reply['update'].update(confidence='strongly_supported',causal_review={ 'effect':'supports','support':[eid],'against':[],'reason':'Fixture scoped causal defect.'}) other=[e for e,v in c['evidence'].items() if v['record']['hypothesis_id']=='H2'] if not other: local_reads.append(1) ref=next(r['id'] for r in c['evidence_catalog'] if r['collector']=='H2') reply['next']={'kind':'recall','evidence_refs':[ref]} else:reply['next']={'kind':'request_diagnosis','reason':'Scoped cause and observed alternative check.','countercheck_refs':[other[0]]} else:reply['next']={'kind':'tool','operation':'project.query','arguments':rows()} adapter=LoopAdapter(change) done=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=12) assert done.status=='COMPLETED' and state(done)['early_stop']['status']=='requested' assert len(local_reads)==1 before=BudgetManager(store).snapshot(*CASE).state assert all(a['remaining']>0 for a in before['accounts'].values()) final=build(DiagnosisBuilder(store,gateway(FinalAdapter(True,lambda p,c:p.update(confidence='strongly_supported'))))) assert final.status=='COMPLETED' and BudgetManager(store).snapshot(*CASE).state==before def test_v15_counterevidence_still_weakens_a_strong_model_claim(): from test_lean_investigation_decision import reply from viano_bug_analysis.analysis.confidence_review import effective_confidence c=context();decision=normalize(c,reply()) assert decision['update']['causal_review']['effect']=='contradicts' assert effective_confidence(decision['update'])=='weakened' def test_recall_turn_cannot_publish_unread_finding_before_reading_it(): from test_lean_investigation_decision import reply c=context();r=reply() r['update']['working_brief']['findings'][0]['evidence_refs']=['E000002'] r['next']={'kind':'recall','evidence_refs':['E000002']} with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r) assert caught.value.feedback['requirement']=='preserve_existing_brief_omit_new_unread_citations_and_use_recall_first' r['update']['working_brief']['findings'][0]['evidence_refs']=['E000001'] validate_tool_shape(c,r) def test_lossless_table_never_collapses_boolean_integer_or_float_cells(): from viano_bug_analysis.analysis.investigation_compact import _table values=[True,1,1.0,False,0,0.0,None,'1']*4 rows=[{'value':v,'long_common':'unchanged metadata '*30} for v in values] table=_table(rows) assert isinstance(table,dict) and 'value' not in table['common'] decoded=[{**table['common'],**dict(zip(table['fields'],r))} for r in table['rows']] assert [type(r['value']) for r in decoded]==[type(v) for v in values] assert decoded==rows