230 lines
14 KiB
Python
230 lines
14 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_case_interpreter import gateway
|
|
from test_investigator import LoopAdapter,CASE,state
|
|
from test_working_brief import service_for
|
|
from test_project_research import rows
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
from viano_bug_analysis.llm import ProviderResponse,LLMError
|
|
from viano_bug_analysis.llm.response_recovery import normalize,parse_text,MARKER,REPAIR_PROMPT,apply_repair,sha
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
|
|
|
|
def loop(tool_case,adapter):
|
|
return Investigator(tool_case[0],gateway(adapter),InvestigationConfig(response_recovery_version=1),
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
|
|
|
|
def test_local_normalization_is_bounded_explicit_and_preserves_values():
|
|
raw={'update':json.dumps({'confidence':'plausible'}),'next':{'kind':'tool','operation':'project.query',
|
|
'arguments':rows(filters=['id','eq',1])}}
|
|
before=copy.deepcopy(raw);fixed,audit=normalize(raw)
|
|
assert raw==before and audit['original_document']==raw
|
|
assert fixed['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
|
|
assert fixed['update']=={'confidence':'plausible'}
|
|
assert len(audit['rules'])==2
|
|
assert normalize(fixed)==(fixed,None)
|
|
|
|
|
|
@pytest.mark.parametrize('filters',[[],['id','eq'],['id','unknown',1],['id','eq',1,'extra']])
|
|
def test_ambiguous_filters_are_not_guessed(filters):
|
|
raw={'next':{'kind':'tool','operation':'project.query','arguments':rows(filters=filters)}}
|
|
assert normalize(raw)==(raw,None)
|
|
|
|
|
|
def test_fence_is_shared_and_duplicate_keys_remain_invalid():
|
|
req={'prompt_version':'investigation_bootstrap_v22','output_schema':{MARKER:1}}
|
|
original='```json\n{"next":{}}\n```'
|
|
doc,audit=parse_text(original,req)
|
|
assert doc=={'next':{}} and audit['original_text']==original
|
|
assert parse_text(original,{'output_schema':{}})[0]=={'next':{}}
|
|
with pytest.raises(LLMError):parse_text('```json\n{"a":1,"a":2}\n```',req)
|
|
doc,audit=parse_text(json.dumps('{"next":{}}'),req)
|
|
assert doc=={'next':{}} and audit['rules']==['decode_root_object:/']
|
|
|
|
|
|
def test_flat_key_value_filter_pairs_are_unambiguous():
|
|
raw={'next':{'kind':'tool','operation':'project.query',
|
|
'arguments':rows(filters=['column','id','operator','eq','value',1])}}
|
|
fixed,audit=normalize(raw)
|
|
assert fixed['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
|
|
assert audit['rules']==['flat_filter_pairs:/next/arguments/filters']
|
|
|
|
|
|
def test_flat_filter_executes_once_with_audit_and_deferred_notice(tool_case):
|
|
def change(reply,c):reply['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}
|
|
adapter=LoopAdapter(change);worker=loop(tool_case,adapter)
|
|
result=asyncio.run(worker.step(*CASE));assert result.status=='PROGRESSED'
|
|
assert len(adapter.contexts)==1
|
|
with PackageReader(result.package) as r:
|
|
s=state(result);call=json.loads(r.read_bytes(s['hypotheses']['H1']['notes'][-1]['llm_ref']))
|
|
assert call['outcome']['normalization']['original_document']['next']['arguments']['filters']==['id','eq',1]
|
|
assert call['outcome']['document']['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
|
|
assert len(BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|
|
second=asyncio.run(worker.step(*CASE));assert second.status=='PROGRESSED'
|
|
with PackageReader(second.package) as r:
|
|
requests=[json.loads(r.read_bytes(n)) for n in r.files if n.startswith('llm/') and n.endswith('/request.json')]
|
|
assert any('format_notice' in req['messages'][1]['content'] for req in requests)
|
|
assert PackageValidator().validate(second.package).valid
|
|
|
|
|
|
class RepairAdapter(LoopAdapter):
|
|
def __init__(self,tamper=False):super().__init__();self.requests=[];self.tamper=tamper
|
|
async def generate(self,config,request,**kwargs):
|
|
self.requests.append(copy.deepcopy(request))
|
|
if request['prompt_version']==REPAIR_PROMPT:
|
|
replacement={'column':'id','operator':'eq','value':999 if self.tamper else 1}
|
|
return ProviderResponse({'replacement':replacement},'fixture',100,20)
|
|
response=await super().generate(config,request,**kwargs)
|
|
response.document['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=[['id','eq',1]])}
|
|
return response
|
|
|
|
|
|
def test_small_repair_resumes_portably_and_preserves_whole_response(tool_case,tmp_path):
|
|
adapter=RepairAdapter();worker=loop(tool_case,adapter)
|
|
first=asyncio.run(worker.step(*CASE));assert first.status=='REJECTED'
|
|
assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
|
|
with PackageReader(first.package) as r:
|
|
c=json.loads(r.read_bytes(state(first)['active']['context_ref']))
|
|
assert c['response_repair']['path']==['next','arguments','filters',0]
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
resumed=Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
second=asyncio.run(resumed.step(*CASE));assert second.status=='PROGRESSED'
|
|
assert adapter.requests[-1]['prompt_version']==REPAIR_PROMPT
|
|
sizes=[sum(len(m['content']) for m in req['messages']) for req in adapter.requests]
|
|
assert sizes[-1]<sizes[0]/2
|
|
assert 'case_interpretation' not in adapter.requests[-1]['messages'][1]['content']
|
|
assert len(BudgetManager(portable).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|
|
with PackageReader(second.package) as r:
|
|
s=state(second);call=json.loads(r.read_bytes(s['hypotheses']['H1']['notes'][-1]['llm_ref']))
|
|
before=c['response_repair']['document'];after=call['outcome']['document']
|
|
assert before['update']==after['update'] and before['assessment']==after['assessment']
|
|
assert after['next']['arguments']['filters']==[{'column':'id','operator':'eq','value':1}]
|
|
assert call['outcome']['repair']['patch']=={'replacement':{'column':'id','operator':'eq','value':1}}
|
|
assert PackageValidator().validate(second.package).valid
|
|
# Re-reading/replaying the completed tool action must not execute it twice.
|
|
assert second.package==portable.latest(*CASE)
|
|
|
|
|
|
def test_repair_cannot_change_a_query_value(tool_case):
|
|
adapter=RepairAdapter(tamper=True);worker=loop(tool_case,adapter)
|
|
first=asyncio.run(worker.step(*CASE));assert first.status=='REJECTED'
|
|
second=asyncio.run(worker.step(*CASE));assert second.status=='REJECTED'
|
|
assert second.code=='LLM_REPAIR_CONTENT_CHANGED'
|
|
assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
|
|
|
|
|
|
@pytest.mark.parametrize('field,value',[('kind','invented_action'),('evidence_refs',['E999999'])])
|
|
def test_unknown_action_or_invented_evidence_is_not_normalized(field,value,tool_case):
|
|
def change(r,c):
|
|
if field=='kind':r['next']['kind']=value
|
|
else:r['update']['working_brief']={'format_version':2,'findings':[{
|
|
'id':'B0001','direction':'support','text':'A claim.','scope':'Snapshot.',
|
|
'evidence_refs':value}],'uncertainties':[],'next_question':'Next?','retired':[]}
|
|
worker=loop(tool_case,LoopAdapter(change));result=asyncio.run(worker.step(*CASE))
|
|
assert result.status=='REJECTED'
|
|
assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
|
|
with PackageReader(result.package) as r:
|
|
c=json.loads(r.read_bytes(state(result)['active']['context_ref']))
|
|
assert 'response_repair' not in c
|
|
|
|
|
|
def test_repaired_lean_turn_remains_bound_after_portable_resume(tool_case,tmp_path):
|
|
from test_investigation_hybrid import HybridAdapter
|
|
class Adapter(HybridAdapter):
|
|
def __init__(self):super().__init__(continuous=True);self.bad_sent=False;self.requests=[]
|
|
async def generate(self,config,request,**kwargs):
|
|
self.requests.append(copy.deepcopy(request))
|
|
if request['prompt_version']==REPAIR_PROMPT:
|
|
return ProviderResponse({'replacement':{'column':'id','operator':'eq','value':1}},'fixture',100,20)
|
|
response=await super().generate(config,request,**kwargs)
|
|
if request['prompt_version']=='investigation_project_v20' and not self.bad_sent:
|
|
self.bad_sent=True
|
|
response.document['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=[['id','eq',1]])}
|
|
return response
|
|
adapter=Adapter();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
worker=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,
|
|
lean_checkpoint_steps=0,response_recovery_version=1),router=router)
|
|
for _ in range(3):assert asyncio.run(worker.step(*CASE)).status=='PROGRESSED'
|
|
failed=asyncio.run(worker.step(*CASE));assert failed.status=='REJECTED'
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(failed.package)
|
|
resumed=Investigator(portable,gateway(adapter),router=router)
|
|
fixed=asyncio.run(resumed.step(*CASE));assert fixed.status=='PROGRESSED'
|
|
assert adapter.requests[-1]['prompt_version']==REPAIR_PROMPT
|
|
# Loading the next step verifies the repair call's previous-result/read-ID binding.
|
|
following=asyncio.run(resumed.step(*CASE));assert following.status=='PROGRESSED'
|
|
assert PackageValidator().validate(following.package).valid
|
|
|
|
|
|
def test_patch_never_changes_scalar_bindings_or_source_hash():
|
|
original={'next':{'arguments':{'filters':[['id','eq',1]]}}}
|
|
data={'document':original,'document_sha256':sha(original),'path':['next','arguments','filters',0]}
|
|
request={'messages':[{}, {'content':json.dumps({'repair':data})}]}
|
|
assert apply_repair(request,{'replacement':{'column':'id','operator':'eq','value':1}})['next']['arguments']['filters'][0]['value']==1
|
|
with pytest.raises(LLMError,match='CONTENT_CHANGED'):
|
|
apply_repair(request,{'replacement':{'column':'id','operator':'eq','value':2}})
|
|
data['document_sha256']='0'*64;request['messages'][1]['content']=json.dumps({'repair':data})
|
|
with pytest.raises(LLMError,match='BINDING'):apply_repair(request,{'replacement':{}})
|
|
original={'next':{'arguments':{'filters':{'column':'id','operator':'eq','value':1}}}}
|
|
data.update(document=original,document_sha256=sha(original),path=['next','arguments','filters'])
|
|
request['messages'][1]['content']=json.dumps({'repair':data})
|
|
with pytest.raises(LLMError,match='CONTENT_CHANGED'):
|
|
apply_repair(request,{'replacement':[{'column':'eq','operator':'id','value':1}]})
|
|
original={'next':{'arguments':{'filters':[['id','eq','name']]}}}
|
|
data.update(document=original,document_sha256=sha(original),path=['next','arguments','filters',0])
|
|
request['messages'][1]['content']=json.dumps({'repair':data})
|
|
with pytest.raises(LLMError,match='CONTENT_CHANGED'):
|
|
apply_repair(request,{'replacement':{'column':'name','operator':'eq','value':'id'}})
|
|
|
|
|
|
def test_native_adapter_retains_fenced_original_only_when_enabled(monkeypatch):
|
|
from dataclasses import replace
|
|
from test_parameter_and_bootstrap_guard import bootstrap
|
|
from test_lean_investigation_decision import reply
|
|
from test_llm_caching import config,native,scripted
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
request=bootstrap();request['output_schema'][MARKER]=1
|
|
document=reply();document['update']['next_step']=None
|
|
text='```json\n'+json.dumps(document)+'\n```'
|
|
envelope=native(cached=0);envelope['candidates'][0]['content']['parts'][0]['text']=text
|
|
adapter=GeminiProvider();scripted(adapter,monkeypatch,[envelope])
|
|
result=adapter._generate(replace(config(),context_cache=False,response_format='json_object'),request,1000000)
|
|
assert result.document==document and result.normalization['original_text']==text
|
|
|
|
|
|
def test_crash_after_normalized_plan_publication_does_not_repeat_llm_or_charge(tool_case,tmp_path,monkeypatch):
|
|
def change(reply,c):reply['next']={'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}
|
|
adapter=LoopAdapter(change);worker=loop(tool_case,adapter)
|
|
class SimulatedCrash(Exception):pass
|
|
def crash(*args):raise SimulatedCrash('simulated interruption')
|
|
monkeypatch.setattr(worker,'_execute_plan',crash)
|
|
with pytest.raises(SimulatedCrash):asyncio.run(worker.step(*CASE))
|
|
assert len(adapter.contexts)==1
|
|
assert not BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions']
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(tool_case[0].latest(*CASE))
|
|
resumed=Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
result=asyncio.run(resumed.step(*CASE))
|
|
assert result.status=='PROGRESSED' and len(adapter.contexts)==1
|
|
assert len(BudgetManager(portable).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|
|
|
|
|
|
def test_nonexistent_or_semantically_incomplete_repair_path_is_not_used():
|
|
from viano_bug_analysis.llm.response_recovery import make_repair
|
|
context={'previous_rejection_details':{'validation_keyword':'type','validation_path':'/update/working_brief/findings/0'}}
|
|
assert make_repair(context,{'outcome':{'document':{'other':'data'}}},'unused') is None
|
|
|
|
|
|
def test_oversized_document_is_not_duplicated_for_normalization():
|
|
raw={'text':'x'*65536,'next':{'kind':'tool','operation':'project.query','arguments':rows(filters=['id','eq',1])}}
|
|
assert normalize(raw)==(raw,None)
|