157 lines
8.1 KiB
Python
157 lines
8.1 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from test_bounded_contract_v2 import context, loop_for
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_investigation_startup import Adapter
|
|
from test_investigator import CASE, run, state
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from viano_bug_analysis.analysis.investigation_contract import feedback
|
|
from viano_bug_analysis.analysis.research_guidance import source_access, namespace_feedback
|
|
from viano_bug_analysis.llm.response_recovery import normalize
|
|
from viano_bug_analysis.llm.stable_context import split_document, restore_document
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
from viano_bug_analysis.llm.investigation_size import check_wire
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
|
|
CORPUS = Path(__file__).parent/'fixtures/llm_rejections/20261020/bug17-source-search-contract'
|
|
|
|
|
|
def captured(name):
|
|
folder = CORPUS/name
|
|
return tuple(json.loads((folder/f).read_text(encoding='utf-8')) for f in ['request.json','parsed-answer.json'])
|
|
|
|
|
|
def test_nested_recall_is_recovered_without_changing_reference_page_or_assessment():
|
|
request,answer = captured('nested-recall')
|
|
before = copy.deepcopy(answer)
|
|
result,audit = normalize(answer,request=request)
|
|
assert result['next'] == {'kind':'recall',**answer['next']['arguments']}
|
|
assert result['assessment'] == answer['assessment']
|
|
assert result['observation'] == answer['observation'] and answer == before
|
|
assert audit['original_document'] == before
|
|
assert audit['rules'] == ['unwrap_recall_arguments:/next']
|
|
assert normalize(result,request=request) == (result,None)
|
|
|
|
|
|
@pytest.mark.parametrize('change',['extra','sibling','many','unknown','negative','boolean','string','operation','legacy','disabled'])
|
|
def test_ambiguous_recall_is_not_repaired(change):
|
|
request,answer = captured('nested-recall')
|
|
if change == 'extra': answer['next']['arguments']['other'] = 1
|
|
if change == 'sibling': answer['next']['page'] = 4
|
|
if change == 'many': answer['next']['arguments']['evidence_refs'].append('E000001')
|
|
if change == 'unknown': answer['next']['arguments']['evidence_refs'] = ['E999999']
|
|
if change == 'negative': answer['next']['arguments']['page'] = -1
|
|
if change == 'boolean': answer['next']['arguments']['page'] = True
|
|
if change == 'string': answer['next']['arguments']['page'] = '1'
|
|
if change == 'operation': answer['next']['operation'] = 'invented'
|
|
if change == 'legacy': request['output_schema']['x-viano-investigation-window'] = 1
|
|
if change == 'disabled': request['output_schema'].pop('x-viano-response-recovery')
|
|
assert normalize(answer,request=request) == (answer,None)
|
|
|
|
|
|
@pytest.mark.parametrize('name,fields',[('eb01f48a',{'offset','limit'}),('a99adf87',{'kind'})])
|
|
def test_actual_search_errors_get_precise_feedback_without_guessing(name,fields):
|
|
request,answer = captured(name)
|
|
query_schema = ProjectResearch.schema()
|
|
schema = {'properties':{'next':{'properties':{'arguments':query_schema}}}}
|
|
error = next(Draft202012Validator(schema).iter_errors(answer))
|
|
detail = feedback(error,answer)
|
|
assert fields <= set(detail['validation_fields'])
|
|
assert 'source.search' in detail['requirement']
|
|
assert len(detail['requirement']) <= 160
|
|
assert normalize(answer,request=request) == (answer,None)
|
|
|
|
|
|
def test_package_reference_feedback_does_not_rewrite_or_echo_path():
|
|
request,answer = captured('60d98274')
|
|
detail = namespace_feedback(answer['next'])
|
|
assert detail['validation_path'] == '/next/arguments/source_refs'
|
|
assert 'project.query' in detail['requirement'] and 'package' in detail['requirement']
|
|
assert 'arathi' not in json.dumps(detail)
|
|
assert len(detail['requirement']) <= 160
|
|
assert normalize(answer,request=request) == (answer,None)
|
|
|
|
|
|
def test_versioned_guidance_is_cached_bounded_and_generic():
|
|
c = context();old = build_prompt(c)
|
|
c['case']['source_context'] = {'refs':[{'path':f'original/source-{i}.cpp'} for i in range(9)]}
|
|
before = copy.deepcopy(c)
|
|
c['research_guidance_version'] = 1
|
|
request = build_prompt(c)
|
|
body = json.loads(request['messages'][1]['content'])
|
|
assert len(body['input']['source_access']['package_source_refs']) == 3
|
|
assert body['input']['source_access']['omitted_package_source_refs'] == 6
|
|
assert 'overrides' in request['messages'][0]['content']
|
|
assert 'QuestAccept' not in request['messages'][0]['content']
|
|
parts = split_document(body,request['prompt_version'])
|
|
assert 'source_access' in parts[0]['input']
|
|
assert restore_document(*parts) == body
|
|
system,prefix,live = GeminiProvider._messages(request,stable_case_context=True)
|
|
assert check_wire(system,prefix+live) <= 20000
|
|
c.pop('research_guidance_version')
|
|
assert c == before
|
|
assert 'source_access' not in json.loads(old['messages'][1]['content'])['input']
|
|
|
|
|
|
def test_long_package_paths_do_not_expand_catalog():
|
|
c = context();c['case']['source_context'] = {'refs':[{'path':'x'*701}]}
|
|
result = source_access(c)
|
|
assert result['package_source_refs'] == [] and result['omitted_package_source_refs'] == 1
|
|
|
|
|
|
def test_bad_namespace_gets_controller_feedback_without_budget_side_effect(tool_case):
|
|
class Wrong(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body = json.loads(request['messages'][1]['content'])
|
|
if body['task'] != 'investigate_one_step':
|
|
return await super().generate(config,request,**kwargs)
|
|
return ProviderResponse({'assessment':None,'observation':'Search the existing source.',
|
|
'next':{'kind':'tool','operation':'source.find_text','arguments':{
|
|
'source_refs':['src/not-a-package-reference.cpp'],'text':'Start',
|
|
'case_sensitive':True,'offset':0,'limit':10}}},'fixture',100,50)
|
|
loop = loop_for(tool_case,Wrong());run(loop,steps=4)
|
|
before = BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
result = run(loop,steps=1)
|
|
saved = state(result)
|
|
assert saved['hypotheses']['H1']['notes'][-1]['code'] == 'TOOL_SOURCE_NOT_DECLARED'
|
|
with PackageReader(result.package) as reader:
|
|
c = json.loads(reader.read_bytes(saved['active']['context_ref']))
|
|
assert 'package_source_refs' in c['previous_rejection_details']['requirement']
|
|
assert c['previous_rejection_details']['validation_path'] == '/next/arguments/source_refs'
|
|
assert BudgetManager(tool_case[0]).snapshot(*CASE).state == before
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
def test_nested_recall_is_executed_without_new_research_charge(tool_case):
|
|
class Recall(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body = json.loads(request['messages'][1]['content'])
|
|
if body['task'] != 'investigate_one_step':
|
|
return await super().generate(config,request,**kwargs)
|
|
self.calls.append(body)
|
|
return ProviderResponse({'assessment':None,'observation':'Read recorded evidence.',
|
|
'next':{'kind':'recall','arguments':{'evidence_refs':['E000001'],'page':0}}},
|
|
'fixture',100,50)
|
|
adapter = Recall();loop = loop_for(tool_case,adapter)
|
|
run(loop,steps=4)
|
|
before = BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
result = run(loop,steps=1)
|
|
saved = state(result)
|
|
assert saved['hypotheses']['H1']['invalid_responses'] == 0
|
|
assert BudgetManager(tool_case[0]).snapshot(*CASE).state == before
|
|
with PackageReader(result.package) as reader:
|
|
ref = saved['hypotheses']['H1']['notes'][-1]['llm_ref']
|
|
call = json.loads(reader.read_bytes(ref))
|
|
assert call['outcome']['document']['next'] == {'kind':'recall','evidence_refs':['E000001'],'page':0}
|
|
assert call['outcome']['normalization']['rules'] == ['unwrap_recall_arguments:/next']
|
|
assert PackageValidator().validate(result.package).valid
|