247 lines
13 KiB
Python
247 lines
13 KiB
Python
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','investigation_project_v13','investigation_project_v14'])
|
|
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
|
|
if version in {'investigation_project_v13','investigation_project_v14'}:
|
|
for h in c['hypotheses'].values():h['disposition']='active'
|
|
c['budget']['accounts']['H2']=copy.deepcopy(c['budget']['accounts']['H1'])
|
|
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','investigation_project_v13','investigation_project_v14'])
|
|
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
|
|
if version in {'investigation_project_v13','investigation_project_v14'}:
|
|
for h in c['hypotheses'].values():h['disposition']='active'
|
|
c['budget']['accounts']['H2']=copy.deepcopy(c['budget']['accounts']['H1'])
|
|
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']==(version if version in {'investigation_project_v12','investigation_project_v13','investigation_project_v14'} 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
|