import asyncio import copy import json from dataclasses import replace import pytest from viano_bug_analysis.llm import GatewayConfig from viano_bug_analysis.llm.gemini import GeminiProvider from viano_bug_analysis.llm.stable_context import split_document, restore_document from test_llm_caching import config, scripted, native, CACHE from test_llm_gateway import request def document(): return {'task':'investigate_hypothesis','output_schema':request()['output_schema'], 'input':{ 'case':{'case':{'case_id':'example','package_id':'stable-package','analysis_revision':1}}, 'format':'investigation','format_version':1,'prompt_version':'investigation_project_v10', 'sources':{'O0001':{'text':'Tester sees no movement','origin':{'metadata_ref':'M0001'}}, 'F0001':{'type':'creature','id':2768,'name':'Verified name'}}, 'interpretation':{'summary':'Quest accepted, no observed movement','sources':['O0001','F0001']}, 'shared_metadata':{'M0001':{'snapshot':'original','parent':{'metadata_ref':'M0002'}}, 'M0002':{'commit':'abcdef'},'M0003':{'snapshot':'research'}}, 'policy':{'max_iterations':20},'budget_policy':{'cost':10},'tool_schema':{'project':'example'}, 'selected_hypothesis':'H1','hypotheses':{'H1':{'confidence':'weak','notes':['Retain this assessment']}}, 'budget':{'remaining':80},'previous_result':{'evidence_id':'E000002'}, 'evidence':{'E000001':{'content':{'origin':{'metadata_ref':'M0001'},'text':'First full result'}}, 'E000002':{'content':{'origin':{'metadata_ref':'M0003'},'row':{'metadata_ref':'literal database cell', 'null':None,'number':123,'bool':False},'text':'Second full result'}}}, 'unknown_future_field':{'preserve':'everything'}}} def case_request(value=None): req=request() req['prompt_version']='investigation_project_v10' req['messages'][1]['content']=json.dumps(value or document()) return req def test_exact_reconstruction_preserves_full_history_provenance_and_json_types(): original=document(); before=copy.deepcopy(original) stable,live=split_document(original,'investigation_project_v10') assert restore_document(stable,live)==original==before assert set(stable['input']['shared_metadata'])=={'M0001','M0002'} assert set(live['input']['shared_metadata'])=={'M0003'} assert live['input']['evidence']==original['input']['evidence'] assert live['input']['hypotheses']==original['input']['hypotheses'] assert 'unknown_future_field' in live['input'] assert 'evidence' not in stable['input'] and 'budget' not in stable['input'] def test_selected_hypothesis_new_evidence_and_budget_do_not_invalidate_stable_prefix(): original=document(); first,_=split_document(original,'investigation_project_v10') original['input']['selected_hypothesis']='H2' original['input']['budget']['remaining']=60 original['input']['hypotheses']['H2']={'confidence':'supported','notes':['New note']} original['input']['evidence']['E000003']={'text':'Entire new research payload'} original['input']['shared_metadata']['M0004']={'snapshot':'new research'} second,live=split_document(original,'investigation_project_v10') assert first==second assert restore_document(second,live)==original @pytest.mark.parametrize('target',['case_id','package_id','analysis_revision','observation','identity','interpretation','metadata']) def test_changed_stable_identity_or_content_invalidates_digest(target): original=document(); first,_=split_document(original,'investigation_project_v10') view=original['input'] if target in {'case_id','package_id','analysis_revision'}:view['case']['case'][target]='changed' elif target=='observation':view['sources']['O0001']['text']='Changed observation' elif target=='identity':view['sources']['F0001']['id']=999 elif target=='interpretation':view['interpretation']['summary']='Changed interpretation' else:view['shared_metadata']['M0001']['snapshot']='different' second,_=split_document(original,'investigation_project_v10') assert first['id']!=second['id'] def test_metadata_alias_rebinding_cannot_reuse_stale_prefix(): value=document(); first,_=split_document(value,'investigation_project_v10') value['input']['shared_metadata']['M0001']={'snapshot':'another source'} second,live=split_document(value,'investigation_project_v10') assert first['id']!=second['id'] and restore_document(second,live)==value @pytest.mark.parametrize('change',['binding','tamper','duplicate','metadata']) def test_ambiguous_or_mismatched_reconstruction_rejected(change): stable,live=split_document(document(),'investigation_project_v10') if change=='binding':live['stable_context_ref']='wrong' elif change=='tamper':stable['input']['sources']['O0001']['text']='tampered' elif change=='duplicate':live['input']['sources']={} else:live['input']['shared_metadata']['M0001']={'snapshot':'override'} with pytest.raises(ValueError):restore_document(stable,live) def test_missing_metadata_definition_keeps_full_old_view(): value=document(); del value['input']['shared_metadata']['M0002'] assert split_document(value,'investigation_project_v10') is None @pytest.mark.parametrize('version',['investigation_project_v8','investigation_project_v9','diagnosis_v5','unknown']) def test_focused_legacy_and_independent_diagnosis_are_not_reinterpreted(version): assert split_document(document(),version) is None def test_native_wire_reconstructs_canonical_request_without_mutation(): req=case_request(); original=copy.deepcopy(req) system,prefix,contents=GeminiProvider._messages(req,stable_case_context=True) cached=json.loads(prefix[0]['parts'][0]['text']) current=json.loads(contents[0]['parts'][0]['text']) current['output_schema']=cached['output_schema'] assert restore_document(cached['stable_case_context'],current)==json.loads(req['messages'][1]['content']) assert req==original and 'stable_case_v1' in system['parts'][-1]['text'] def test_stable_context_works_without_duplicated_user_schema(): value=document(); del value['output_schema'] req=case_request(value) _,prefix,contents=GeminiProvider._messages(req,stable_case_context=True) cached=json.loads(prefix[0]['parts'][0]['text']) assert restore_document(cached['stable_case_context'],json.loads(contents[0]['parts'][0]['text']))==value def test_native_cache_reuse_keeps_all_fresh_research_and_assessments(monkeypatch): adapter=GeminiProvider(); seen=scripted(adapter,monkeypatch,[CACHE,native(),native()]) cfg=config() asyncio.run(adapter.generate(cfg,case_request(),max_response_bytes=100000)) value=document(); value['input']['evidence']['E000003']={'text':'Fresh finding'} asyncio.run(adapter.generate(cfg,case_request(value),max_response_bytes=100000)) assert len(seen)==3 and seen[0][0].endswith('/cachedContents') assert 'Tester sees no movement' in json.dumps(seen[0][1]) assert 'Entire new research payload' not in json.dumps(seen[0][1]) for _,payload,_ in seen[1:]: assert payload['cachedContent']==CACHE['name'] text=json.dumps(payload) assert 'First full result' in text and 'Second full result' in text and 'Retain this assessment' in text assert 'Fresh finding' in json.dumps(seen[2][1]) def test_instruction_only_compatibility_mode_preserves_previous_snapshot_and_wire(): cfg=replace(config(),cache_context_version='instructions_v1') snapshot=GatewayConfig((cfg,)).snapshot()['providers'][0] assert 'cache_context_version' not in snapshot assert GatewayConfig((config(),)).snapshot()['providers'][0]['cache_context_version']=='stable_case_v1' _,prefix,_=GeminiProvider._messages(case_request()) assert 'stable_case_context' not in json.loads(prefix[0]['parts'][0]['text']) @pytest.mark.parametrize('version',['investigation_project_v11','investigation_project_v12']) def test_v11_aliases_and_cached_case_do_not_shift_with_new_research(version): from test_investigation_consolidation import populated, capture from viano_bug_analysis.analysis.investigation_prompt import build_prompt c=populated();c['format']='viano-investigation-context';c['prompt_version']=version c['interpretation']={'summary':'Preserve the interpretation'} before=copy.deepcopy(c) first=build_prompt(c) _,prefix,_=GeminiProvider._messages(first,stable_case_context=True) c['selected_hypothesis']='H2' new=copy.deepcopy(c['evidence']['E000001']) new['record']['id']='E000003' new['content']['items']=[capture(range(121,160),sha='c'*64,commit='d'*40)] c['evidence']['E000003']=new second=build_prompt(c) _,later,_=GeminiProvider._messages(second,stable_case_context=True) assert prefix==later for eid in c['evidence']: assert eid in json.loads(second['messages'][1]['content'])['input']['evidence'] assert build_prompt(before)==first def test_v11_short_identifier_aliases_are_stable_and_reversible(): from viano_bug_analysis.analysis.model_context import compact_context first={'sources':{'path':'original/'+'a'*64+'/file.cpp'}} aliases={} projected=compact_context(first,stable_aliases=True,aliases_out=aliases) expanded={'evidence':{'path':'original/'+'b'*64+'/other.cpp'},**first} more=compact_context(expanded,stable_aliases=True) alias=projected['sources']['path'] assert more['sources']['path']==alias and aliases[alias]==first['sources']['path'] assert compact_context(first)['sources']['path'].startswith('P0001:') @pytest.mark.parametrize('version',['investigation_project_v10','investigation_project_v12']) def test_offline_measurement_matches_saved_context_without_provider_calls(make_package,monkeypatch,version): from test_investigation_consolidation import populated from viano_bug_analysis.analysis.investigation_prompt import build_prompt from viano_bug_analysis.llm.cache_measurement import measure_package c=populated();c['format']='viano-investigation-context';c['prompt_version']=version c['interpretation']={'summary':'Original interpretation'} entries={ 'analysis/investigation/0001/contexts/H1-0001.json':json.dumps(c), 'llm/operations/example/request.json':json.dumps(build_prompt(c))} path=make_package(entries=entries) def forbidden(*args,**kwargs):raise AssertionError('No provider calls in offline measurement') monkeypatch.setattr(GeminiProvider,'_post',forbidden) report=measure_package(path) assert report['requests']==1 and report['unmatched_requests']==0 assert report['unique_prefixes']==1 and report['all_exactly_reconstructed'] assert report['samples'][0]['rebuilt_prompt']==('investigation_project_v12' if version.endswith('v12') else 'investigation_project_v11') @pytest.mark.parametrize('failure',[None,400,404]) def test_stable_context_full_fallback_keeps_same_information(monkeypatch,failure): from viano_bug_analysis.llm import LLMError adapter=GeminiProvider() if failure is None: actions=[LLMError('LLM_HTTP_CONFIGURATION',http_status=400),native(0)] else: actions=[CACHE,LLMError('LLM_HTTP_CONFIGURATION',http_status=failure),native(0)] seen=scripted(adapter,monkeypatch,actions) asyncio.run(adapter.generate(config(),case_request(),max_response_bytes=100000)) payload=seen[-1][1] assert 'cachedContent' not in payload and 'systemInstruction' in payload prefix=json.loads(payload['contents'][0]['parts'][0]['text']) live=json.loads(payload['contents'][1]['parts'][0]['text']) live['output_schema']=prefix['output_schema'] assert restore_document(prefix['stable_case_context'],live)==document() def test_restart_rebuilds_identical_case_prefix_from_portable_request(monkeypatch): prefixes=[] for adapter in (GeminiProvider(),GeminiProvider()): seen=scripted(adapter,monkeypatch,[CACHE,native()]) asyncio.run(adapter.generate(config(),case_request(),max_response_bytes=100000)) prefixes.append(seen[0][1]) assert prefixes[0]==prefixes[1] def test_provider_failover_receives_complete_canonical_request(monkeypatch): from viano_bug_analysis.llm import LLMGateway, ProviderConfig, ProviderResponse, ProviderUnavailable req=case_request();original=copy.deepcopy(req) adapter=GeminiProvider() scripted(adapter,monkeypatch,[CACHE,ProviderUnavailable('LLM_CONNECTION')]) class Fallback: async def generate(self,config,request,**kwargs): assert request['messages']==original['messages'] assert 'First full result' in request['messages'][1]['content'] return ProviderResponse({'answer':'ok'},'fallback') fallback=ProviderConfig('fallback','fallback','http://localhost',adapter='fixture') result=asyncio.run(LLMGateway(GatewayConfig((config(),fallback)), {'gemini':adapter,'fixture':Fallback()}).complete_round(req)) assert result['metadata']['provider']=='fallback' and req==original