130 lines
7.3 KiB
Python
130 lines
7.3 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 test_project_research import rows
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema
|
|
from viano_bug_analysis.analysis.investigation_contract import feedback
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.llm.response_recovery import normalize
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
|
|
CORPUS = Path(__file__).parent/'fixtures/llm_rejections/20261020/bug17-recall-assessment'
|
|
|
|
|
|
@pytest.mark.parametrize('pending',[False,True])
|
|
def test_versioned_schema_only_allows_applicable_assessment(pending):
|
|
c = context()
|
|
if not pending: c['previous_result'] = None
|
|
legacy_schema = output_schema(c,strict=True)
|
|
assert 'anyOf' in legacy_schema['properties']['assessment']
|
|
c['assessment_scope_version'] = 1
|
|
schema = output_schema(c,strict=True)['properties']['assessment']
|
|
assert Draft202012Validator(schema).is_valid(None) == (not pending)
|
|
assert Draft202012Validator(schema).is_valid({'semantic_score':50,'reason':'New result.'}) == pending
|
|
|
|
|
|
@pytest.mark.parametrize('name',['8f58f733','6ccfd3ed','4115b3b2'])
|
|
def test_actual_rejections_keep_query_and_observation_without_budget_score(name):
|
|
folder = CORPUS/name
|
|
request = json.loads((folder/'request.json').read_text(encoding='utf-8'))
|
|
answer = json.loads((folder/'parsed-answer.json').read_text(encoding='utf-8'))
|
|
original = copy.deepcopy(answer)
|
|
result,audit = normalize(answer,request=request)
|
|
assert result['assessment'] is None
|
|
assert result['next'] == answer['next'] and result['observation'] == answer['observation']
|
|
assert answer == original and audit['original_document'] == original
|
|
assert audit['rules'] == ['non_applicable_budget_assessment:/assessment']
|
|
c = context();c['previous_result'] = None;c['assessment_scope_version'] = 1
|
|
assert Draft202012Validator(output_schema(c,strict=True)).is_valid(result)
|
|
assert normalize(result,request=request) == (result,None)
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['pending','missing_previous','unknown_evidence','unknown_action',
|
|
'unknown_query','extra_evidence_refs','missing_score','bad_score','float_score','bool_score','empty_reason',
|
|
'causal_findings','checkpoint','legacy_window','disabled','missing_assessment'])
|
|
def test_ambiguous_or_out_of_scope_scores_are_not_neutralized(change):
|
|
folder = CORPUS/'8f58f733'
|
|
request = json.loads((folder/'request.json').read_text(encoding='utf-8'))
|
|
answer = json.loads((folder/'parsed-answer.json').read_text(encoding='utf-8'))
|
|
body = json.loads(request['messages'][1]['content'])
|
|
if change == 'pending': body['input']['previous_result'] = {'evidence_id':'E000005'}
|
|
if change == 'missing_previous': body['input'].pop('previous_result')
|
|
if change == 'unknown_evidence': answer['assessment']['reason'] = 'E999999 proves it.'
|
|
if change == 'unknown_action': answer['next']['operation'] = 'shell.exec'
|
|
if change == 'unknown_query': answer['next']['arguments']['kind'] = 'invented.query'
|
|
if change == 'extra_evidence_refs': answer['assessment']['evidence_refs'] = ['E999999']
|
|
if change == 'missing_score': answer['assessment'].pop('semantic_score')
|
|
if change == 'bad_score': answer['assessment']['semantic_score'] = 101
|
|
if change == 'float_score': answer['assessment']['semantic_score'] = 1.0
|
|
if change == 'bool_score': answer['assessment']['semantic_score'] = True
|
|
if change == 'empty_reason': answer['assessment']['reason'] = ' '
|
|
if change == 'causal_findings': answer['findings'] = [{'text':'Unverified.'}]
|
|
if change == 'checkpoint': body['task'] = 'consolidate_investigation_session'
|
|
if change == 'legacy_window': request['output_schema']['x-viano-investigation-window'] = 1
|
|
if change == 'disabled': request['output_schema'].pop('x-viano-response-recovery')
|
|
if change == 'missing_assessment': answer.pop('assessment')
|
|
request['messages'][1]['content'] = json.dumps(body)
|
|
result,audit = normalize(answer,request=request)
|
|
assert result == answer and audit is None
|
|
|
|
|
|
def test_null_assessment_error_has_actionable_feedback():
|
|
c = context();c['assessment_scope_version'] = 1;c['previous_result'] = None
|
|
rule = output_schema(c,strict=True)['properties']['assessment']
|
|
validator = Draft202012Validator({'properties':{'assessment':rule}})
|
|
error = next(validator.iter_errors({'assessment':{'semantic_score':50,'reason':'Old evidence.'}}))
|
|
info = feedback(error,{'assessment':{'semantic_score':50,'reason':'Old evidence.'}})
|
|
assert info['validation_path'] == '/assessment'
|
|
assert 'must be null' in info['requirement'] and 'refund' in info['requirement']
|
|
|
|
|
|
def test_valid_query_after_recall_is_recovered_without_refund_or_extra_llm(tool_case):
|
|
class Read(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)
|
|
assert request['output_schema']['properties']['assessment']['type'] == 'null'
|
|
if len(self.calls) == 5:
|
|
out = {'assessment':None,'observation':'Read existing evidence.',
|
|
'next':{'kind':'recall','evidence_refs':['E000001'],'page':0}}
|
|
else:
|
|
out = {'assessment':{'semantic_score':90,'reason':'E000001 is useful old evidence.'},
|
|
'observation':'Query a bounded table.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':rows()}}
|
|
return ProviderResponse(out,'fixture',100,50)
|
|
adapter = Read();loop = loop_for(tool_case,adapter)
|
|
run(loop,steps=5)
|
|
before = BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
result = run(loop,steps=1)
|
|
saved = state(result)
|
|
assert len(adapter.calls) == 6 and saved['hypotheses']['H1']['invalid_responses'] == 0
|
|
after = BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
assert after['accounts']['H1']['rewarded'] == before['accounts']['H1']['rewarded']
|
|
assert after['accounts']['H1']['spent'] == before['accounts']['H1']['spent'] + 10
|
|
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']['assessment'] is None
|
|
assert call['outcome']['normalization']['original_document']['assessment']['semantic_score'] == 90
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
def test_pending_result_feedback_requires_a_real_assessment():
|
|
c = context();c['assessment_scope_version'] = 1
|
|
rule = output_schema(c,strict=True)['properties']['assessment']
|
|
error = next(Draft202012Validator({'properties':{'assessment':rule}}).iter_errors({'assessment':None}))
|
|
info = feedback(error,{'assessment':None})
|
|
assert 'semantic_score' in info['requirement'] and 'pending new' in info['requirement']
|