126 lines
6.2 KiB
Python
126 lines
6.2 KiB
Python
import asyncio
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from test_investigation_review import review_context
|
|
from test_lean_investigation_decision import reply
|
|
from test_llm_gateway import provider, http_endpoint, active_case
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from viano_bug_analysis.llm import (GatewayConfig, ProviderConfig, ProviderResponse,
|
|
ProviderUnavailable, LLMGateway, LLMError, CaseLLMRunner)
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
|
|
|
|
def review_request():
|
|
return build_prompt(review_context())
|
|
|
|
|
|
def review_reply():
|
|
result=reply();result['next']={'kind':'defer','reason':'Pending evidence assessed'}
|
|
return result
|
|
|
|
|
|
class RecordingAdapter:
|
|
def __init__(self, fail_first=False, error=False):
|
|
self.calls=[];self.fail_first=fail_first;self.error=error
|
|
|
|
async def generate(self, config, request, *, max_response_bytes):
|
|
self.calls.append((config.id,config.reasoning_effort))
|
|
if self.error:raise LLMError('LLM_RESPONSE')
|
|
if self.fail_first and len(self.calls)==1:raise ProviderUnavailable('LLM_CONNECTION')
|
|
return ProviderResponse(review_reply(),'fixture-model',100,20)
|
|
|
|
|
|
@pytest.mark.parametrize('mode',['review','normal','diagnosis','old_version','wrong_task','wrong_actions'])
|
|
def test_effort_override_applies_only_to_narrow_controller_review(mode):
|
|
req=review_request();c=review_context()
|
|
if mode=='normal':c['early_review']['request']=None;req=build_prompt(c)
|
|
elif mode=='old_version':c['prompt_version']='investigation_project_v15';req=build_prompt(c)
|
|
elif mode=='diagnosis':req['prompt_version']='diagnosis_early_v6'
|
|
elif mode=='wrong_task':
|
|
doc=json.loads(req['messages'][1]['content']);doc['task']='investigate_hypothesis'
|
|
req['messages'][1]['content']=json.dumps(doc)
|
|
elif mode=='wrong_actions':req['output_schema']['properties']['next']['properties']['kind']['enum'].append('tool')
|
|
cfg=GatewayConfig((provider(reasoning_effort='medium',investigation_review_reasoning_effort='low'),))
|
|
adapter=RecordingAdapter();out=asyncio.run(LLMGateway(cfg,{'fixture':adapter}).complete_round(req))
|
|
assert adapter.calls==[('first','low' if mode=='review' else 'medium')]
|
|
assert cfg.providers[0].reasoning_effort=='medium'
|
|
assert out['metadata'].get('reasoning_profile')==('investigation_review' if mode=='review' else None)
|
|
|
|
|
|
def test_provider_failover_uses_each_providers_review_setting():
|
|
cfg=GatewayConfig((provider(reasoning_effort='high',investigation_review_reasoning_effort='low'),
|
|
provider('second',reasoning_effort='medium',investigation_review_reasoning_effort='minimal')))
|
|
adapter=RecordingAdapter(fail_first=True)
|
|
out=asyncio.run(LLMGateway(cfg,{'fixture':adapter}).complete_round(review_request()))
|
|
assert adapter.calls==[('first','low'),('second','minimal')]
|
|
assert out['metadata']['reasoning_effort']=='minimal'
|
|
|
|
|
|
@pytest.mark.parametrize('bad',[True,0,'unknown',[],{}])
|
|
def test_invalid_override_is_rejected(bad):
|
|
with pytest.raises(LLMError,match='LLM_CONFIG'):
|
|
provider(investigation_review_reasoning_effort=bad)
|
|
|
|
|
|
def test_omitted_override_preserves_snapshot_and_inherits_general_effort():
|
|
cfg=GatewayConfig((provider(reasoning_effort='medium'),))
|
|
assert 'investigation_review_reasoning_effort' not in cfg.snapshot()['providers'][0]
|
|
adapter=RecordingAdapter();asyncio.run(LLMGateway(cfg,{'fixture':adapter}).complete_round(review_request()))
|
|
assert adapter.calls==[('first','medium')]
|
|
changed=GatewayConfig((replace(cfg.providers[0],investigation_review_reasoning_effort='low'),))
|
|
assert changed.snapshot()['providers'][0]['investigation_review_reasoning_effort']=='low'
|
|
|
|
|
|
def test_toml_loads_override(tmp_path):
|
|
p=tmp_path/'llm.toml';p.write_text('''[llm]
|
|
[[llm.providers]]
|
|
id="fixture"
|
|
model="fixture"
|
|
base_url="http://localhost/v1"
|
|
reasoning_effort="medium"
|
|
investigation_review_reasoning_effort="low"
|
|
''')
|
|
assert GatewayConfig.load(p).providers[0].investigation_review_reasoning_effort=='low'
|
|
|
|
|
|
def test_native_gemini_receives_low_thinking_level(monkeypatch):
|
|
from test_llm_caching import config, native, scripted
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
adapter=GeminiProvider();envelope=native()
|
|
envelope['candidates'][0]['content']['parts'][0]['text']=json.dumps(review_reply())
|
|
seen=scripted(adapter,monkeypatch,[envelope])
|
|
cfg=GatewayConfig((replace(config(),context_cache=False,reasoning_effort='medium',
|
|
investigation_review_reasoning_effort='low'),))
|
|
out=asyncio.run(LLMGateway(cfg,{'gemini':adapter}).complete_round(review_request()))
|
|
assert seen[0][1]['generationConfig']['thinkingConfig']=={'thinkingLevel':'low'}
|
|
assert out['metadata']['reasoning_effort']=='low'
|
|
with pytest.raises(LLMError,match='LLM_CONFIG'):
|
|
replace(cfg.providers[0],investigation_review_reasoning_effort='none')
|
|
|
|
|
|
def test_compatible_transport_receives_low_reasoning_effort(http_endpoint):
|
|
url,seen,response=http_endpoint
|
|
response['body']['choices'][0]['message']['content']=json.dumps(review_reply())
|
|
cfg=GatewayConfig((ProviderConfig(id='test',model='configured',base_url=url,
|
|
reasoning_effort='medium',investigation_review_reasoning_effort='low'),))
|
|
asyncio.run(LLMGateway(cfg).complete_round(review_request()))
|
|
assert seen[0]['body']['reasoning_effort']=='low'
|
|
|
|
|
|
@pytest.mark.parametrize('failed',[False,True])
|
|
def test_effective_setting_survives_checkpoint_and_replay(active_case,failed):
|
|
store,_,_=active_case;adapter=RecordingAdapter(error=failed)
|
|
cfg=GatewayConfig((provider(reasoning_effort='medium',investigation_review_reasoning_effort='low'),))
|
|
runner=CaseLLMRunner(store,LLMGateway(cfg,{'fixture':adapter}))
|
|
req=review_request()
|
|
result=asyncio.run(runner.run('case-A','stable-package',operation_id='review-effort',request=req))
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
call=json.loads(reader.read_bytes(result.result_ref))
|
|
assert call['outcome']['metadata']['reasoning_effort']=='low'
|
|
assert asyncio.run(runner.run('case-A','stable-package',operation_id='review-effort',request=req))==result
|
|
assert len(adapter.calls)==1
|