237 lines
12 KiB
Python
237 lines
12 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
import pytest
|
|
from viano_bug_analysis.llm import CaseLLMRunner, GatewayConfig, LLMGateway, LLMError, ProviderResponse
|
|
from viano_bug_analysis.llm.answer_recovery import capture, parse_complete, normalize_schema, PATCH_PROMPT
|
|
from viano_bug_analysis.llm.response_recovery import parse_text
|
|
from viano_bug_analysis.case.contracts import read_json
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_llm_gateway import active_case, provider, http_endpoint
|
|
|
|
|
|
@pytest.mark.parametrize('text,expected',[
|
|
('{"a":1 "b":2}', {'a':1,'b':2}),
|
|
('{"a" 1}', {'a':1}),
|
|
('{"a":[1,2,]}', {'a':[1,2]}),
|
|
('{"a":"literal , } [ text",}', {'a':'literal , } [ text'}),
|
|
('```json\n{"a":1}\n```', {'a':1}),
|
|
])
|
|
def test_local_syntax_edits_preserve_literal_values(text,expected):
|
|
document,audit=parse_complete(text)
|
|
assert document==expected and audit['original_text']==text
|
|
assert parse_text(text,{'prompt_version':'hypotheses_v4','output_schema':{}})[0]==expected
|
|
|
|
|
|
@pytest.mark.parametrize('text',[
|
|
'{"a":1', '{"a":"unfinished', '{"a":1,"a":2}',
|
|
'{"a":NaN}', '{"a":undefined}', '{"a":1,,"b":2}',
|
|
'{"a":"text with "broken" quotes"}', 'untrusted prose {"a":1}',
|
|
])
|
|
def test_no_duplicate_key_truncation_or_content_guessing(text):
|
|
assert parse_complete(text)==(None,None)
|
|
with pytest.raises(LLMError):parse_text(text,{'prompt_version':'hypotheses_v4','output_schema':{}})
|
|
|
|
|
|
def test_schema_container_normalization_requires_entire_contract():
|
|
schema={'type':'object','required':['items'],'properties':{'items':{'type':'array','items':{'type':'object','required':['id'],'properties':{'id':{'type':'integer'}}}}}}
|
|
original={'items':{'id':3}}
|
|
result,rules=normalize_schema(original,schema)
|
|
assert result=={'items':[{'id':3}]} and rules and original=={'items':{'id':3}}
|
|
bad={'items':{'wrong':'invented'}}
|
|
assert normalize_schema(bad,schema)==(bad,[])
|
|
|
|
|
|
def test_capture_is_complete_or_omitted_never_a_partial_json():
|
|
retained=capture('SECRET',limit=3)
|
|
assert retained['bytes']==6 and not retained['retained'] and 'text' not in retained
|
|
assert capture('SECRET',complete=False)['complete'] is False
|
|
|
|
|
|
SCHEMA={'type':'object','additionalProperties':False,'required':['title','subject'],
|
|
'properties':{'title':{'type':'string','minLength':1,'maxLength':200},'subject':{'type':'string'}}}
|
|
|
|
|
|
def request(schema=SCHEMA):
|
|
return {'prompt_version':'hypotheses_fixture','privacy':'configured','output_schema':schema,
|
|
'messages':[{'role':'system','content':'Return the supplied contract.'},
|
|
{'role':'user','content':'Subject: known subject.'}]}
|
|
|
|
|
|
class Patches:
|
|
def __init__(self,patch=None):self.requests=[];self.patch=patch or {'replacement':'Known subject'}
|
|
async def generate(self,config,request,**kwargs):
|
|
self.requests.append(copy.deepcopy(request))
|
|
document=self.patch if request['prompt_version']==PATCH_PROMPT else {'title':'','subject':'Known subject'}
|
|
return ProviderResponse(copy.deepcopy(document),'fixture',10,20)
|
|
|
|
|
|
def runner(store,adapter):
|
|
return CaseLLMRunner(store,LLMGateway(GatewayConfig((provider(),)),{'fixture':adapter}))
|
|
|
|
|
|
def run(r):return asyncio.run(r.run('case-A','stable-package',operation_id='bounded-repair',request=request()))
|
|
|
|
|
|
def test_schema_patch_is_durable_atomic_and_not_double_billed(active_case):
|
|
store,_,_=active_case;adapter=Patches();r=runner(store,adapter)
|
|
result=run(r)
|
|
assert result.status=='SUCCESS' and len(adapter.requests)==2 and result.rounds==2
|
|
with PackageReader(result.package) as reader:
|
|
call=read_json(reader,result.result_ref,'llm-call',2**20)
|
|
assert call['derived'] and call['outcome']['document']=={'title':'Known subject','subject':'Known subject'}
|
|
assert call['outcome']['metadata']['input_tokens']==0
|
|
originals=[json.loads(reader.read_bytes(n)) for n in reader.files if n.startswith('llm/operations/') and '/round-' in n]
|
|
assert sum(d['outcome'].get('metadata',{}).get('input_tokens',0) or 0 for d in originals)==20
|
|
assert call['outcome']['response_recovery']['source_ref'] in reader.files
|
|
assert call['outcome']['response_recovery']['patch_ref'] in reader.files
|
|
assert run(r).result_ref==result.result_ref and len(adapter.requests)==2
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
def test_failed_patch_preserves_original_and_is_not_retried_forever(active_case):
|
|
store,_,_=active_case;adapter=Patches({'replacement':'Claim 99999'});r=runner(store,adapter)
|
|
result=run(r)
|
|
assert result.status=='ERROR' and len(adapter.requests)==2
|
|
with PackageReader(result.package) as reader:
|
|
call=read_json(reader,result.result_ref,'llm-call',2**20)
|
|
assert 'document' not in call['outcome']
|
|
assert call['outcome']['response_recovery']['failure_code']=='LLM_REPAIR_CONTENT_CHANGED'
|
|
assert run(r).status=='ERROR' and len(adapter.requests)==2
|
|
|
|
|
|
def test_crash_after_patch_does_not_repeat_paid_call(active_case,monkeypatch):
|
|
store,_,_=active_case;adapter=Patches();r=runner(store,adapter);advance=store.advance
|
|
def crash(*args,**kwargs):
|
|
if any('/round-0002.json' in n for n in kwargs.get('additions',{})):raise RuntimeError('publication interrupted')
|
|
return advance(*args,**kwargs)
|
|
monkeypatch.setattr(store,'advance',crash)
|
|
with pytest.raises(RuntimeError):run(r)
|
|
assert len(adapter.requests)==2
|
|
monkeypatch.setattr(store,'advance',advance)
|
|
assert run(r).status=='SUCCESS' and len(adapter.requests)==2
|
|
|
|
|
|
def test_capture_budget_defaults_preserve_config_snapshot():
|
|
config=GatewayConfig((provider(),))
|
|
assert 'response_capture_bytes' not in config.snapshot()
|
|
small=GatewayConfig((provider(),),response_capture_bytes=4)
|
|
g=LLMGateway(small,{'fixture':Patches()})
|
|
assert g._capture_budget(capture('private'))['retained'] is False
|
|
|
|
|
|
|
|
def test_native_capture_retains_rejected_text_without_logging_it(monkeypatch,caplog):
|
|
from dataclasses import replace
|
|
from test_llm_caching import config,native,scripted
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
envelope=native();raw='{"answer":"PRIVATE_REPLY"'
|
|
envelope['candidates'][0]['content']['parts'][0]['text']=raw
|
|
adapter=GeminiProvider();scripted(adapter,monkeypatch,[envelope])
|
|
with pytest.raises(LLMError) as caught:
|
|
adapter._generate(replace(config(),context_cache=False),request(),100000)
|
|
assert caught.value.response_capture['text']==raw
|
|
assert caught.value.response_capture['complete'] is True
|
|
assert 'PRIVATE_REPLY' not in json.dumps(caught.value.diagnostics)
|
|
assert 'PRIVATE_REPLY' not in caplog.text
|
|
|
|
|
|
def test_incomplete_native_output_is_captured_but_never_repaired(monkeypatch):
|
|
from dataclasses import replace
|
|
from test_llm_caching import config,native,scripted
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
envelope=native();envelope['candidates'][0]['finishReason']='MAX_TOKENS'
|
|
envelope['candidates'][0]['content']['parts'][0]['text']='{"answer":"partial"'
|
|
adapter=GeminiProvider();scripted(adapter,monkeypatch,[envelope])
|
|
with pytest.raises(LLMError,match='INCOMPLETE') as caught:
|
|
adapter._generate(replace(config(),context_cache=False),request(),100000)
|
|
assert caught.value.response_capture['complete'] is False
|
|
assert caught.value.response_capture['text']=='{"answer":"partial"'
|
|
|
|
|
|
def test_full_phase_preflight_rejects_patch_when_other_semantics_are_invalid(active_case):
|
|
from viano_bug_analysis.analysis.hypothesis_config import HypothesisError
|
|
store,_,_=active_case;adapter=Patches();r=runner(store,adapter)
|
|
def invalid(_):raise HypothesisError('HYPOTHESIS_SOURCE_UNKNOWN')
|
|
result=asyncio.run(r.run('case-A','stable-package',operation_id='not-isolated',request=request(),validator=invalid))
|
|
assert result.status=='ERROR' and len(adapter.requests)==1
|
|
with PackageReader(result.package) as reader:
|
|
ref=next(n for n in reader.files if '/recovery-declined-' in n)
|
|
assert json.loads(reader.read_bytes(ref))['reason']=='HYPOTHESIS_SOURCE_UNKNOWN'
|
|
|
|
|
|
def test_derived_document_cannot_change_an_untargeted_field(active_case):
|
|
from viano_bug_analysis.llm.answer_recovery import verify_derived
|
|
store,_,_=active_case;r=runner(store,Patches());result=run(r)
|
|
with PackageReader(result.package) as reader:
|
|
call=read_json(reader,result.result_ref,'llm-call',2**20)
|
|
call['outcome']['document']['subject']='Changed outside target'
|
|
with pytest.raises(LLMError,match='BINDING'):verify_derived(reader,call,2**20)
|
|
|
|
|
|
|
|
def test_field_repairs_can_be_disabled_without_discarding_rejected_document(active_case):
|
|
store,_,_=active_case;adapter=Patches()
|
|
g=LLMGateway(GatewayConfig((provider(),),max_field_repairs=0),{'fixture':adapter})
|
|
result=run(CaseLLMRunner(store,g))
|
|
assert result.status=='ERROR' and len(adapter.requests)==1
|
|
with PackageReader(result.package) as reader:
|
|
call=read_json(reader,result.result_ref,'llm-call',2**20)
|
|
assert call['outcome']['rejected_document']['subject']=='Known subject'
|
|
assert call['outcome']['response_assessment']['decision']=='schema_field_candidate'
|
|
|
|
|
|
|
|
def test_patch_transport_retry_reuses_original_generation(active_case):
|
|
from datetime import datetime, timezone, timedelta
|
|
from viano_bug_analysis.llm import ProviderUnavailable
|
|
store,_,_=active_case
|
|
class Delayed(Patches):
|
|
unavailable=True
|
|
async def generate(self,config,request,**kwargs):
|
|
if request['prompt_version']==PATCH_PROMPT and self.unavailable:
|
|
self.requests.append(copy.deepcopy(request));self.unavailable=False
|
|
raise ProviderUnavailable('LLM_CONNECTION')
|
|
return await super().generate(config,request,**kwargs)
|
|
adapter=Delayed();now=[datetime.now(timezone.utc)]
|
|
g=LLMGateway(GatewayConfig((provider(),),retry_delay_seconds=1),{'fixture':adapter})
|
|
r=CaseLLMRunner(store,g,clock=lambda:now[0])
|
|
assert run(r).status=='RETRY_PENDING' and len(adapter.requests)==2
|
|
assert run(r).status=='DEFERRED' and len(adapter.requests)==2
|
|
now[0]+=timedelta(seconds=2)
|
|
assert run(r).status=='SUCCESS' and len(adapter.requests)==3
|
|
assert sum(q['prompt_version']!=PATCH_PROMPT for q in adapter.requests)==1
|
|
|
|
|
|
|
|
@pytest.mark.parametrize('raw,valid',[('{"answer":"ok",}',True),('{"answer":"PRIVATE_REPLY"',False)])
|
|
def test_openai_candidate_capture_and_common_parser(http_endpoint,raw,valid):
|
|
from viano_bug_analysis.llm import ProviderConfig
|
|
from viano_bug_analysis.llm.openai_compatible import OpenAICompatibleProvider
|
|
url,_,reply=http_endpoint
|
|
reply['body']=json.dumps({'model':'fixture','choices':[{'finish_reason':'stop','message':{'content':raw}}],
|
|
'usage':{'prompt_tokens':10,'completion_tokens':20}}).encode('utf-8')
|
|
config=ProviderConfig(id='local',model='fixture',base_url=url)
|
|
if valid:
|
|
result=OpenAICompatibleProvider._generate(config,request(),100000)
|
|
assert result.document=={'answer':'ok'} and result.response_capture['text']==raw
|
|
assert result.normalization['original_text']==raw
|
|
else:
|
|
with pytest.raises(LLMError) as caught:OpenAICompatibleProvider._generate(config,request(),100000)
|
|
assert caught.value.response_capture['text']==raw
|
|
assert 'PRIVATE_REPLY' not in json.dumps(caught.value.diagnostics)
|
|
|
|
|
|
|
|
def test_normalization_audit_limit_falls_back_to_controlled_rejection(active_case):
|
|
store,_,_=active_case
|
|
class ManyRules:
|
|
async def generate(self,*args,**kwargs):
|
|
return ProviderResponse({'data':'{"id":1}'},'fixture',normalization={
|
|
'version':1,'original_document':{'data':'{"id":1}'},'rules':['prior_rule']*16})
|
|
schema={'type':'object','properties':{'data':{'type':'object','properties':{'id':{'type':'integer'}}}}}
|
|
g=LLMGateway(GatewayConfig((provider(),)),{'fixture':ManyRules()})
|
|
result=asyncio.run(CaseLLMRunner(store,g).run('case-A','stable-package',operation_id='bounded-audit',request=request(schema)))
|
|
assert result.status=='ERROR' and result.code=='LLM_OUTPUT_SCHEMA'
|
|
assert PackageValidator().validate(result.package).valid
|