137 lines
7.4 KiB
Python
137 lines
7.4 KiB
Python
import asyncio
|
|
import copy
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from viano_bug_analysis.analysis.confidence_review import effective_confidence, validate_review
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema
|
|
from viano_bug_analysis.analysis import Investigator, InvestigationError
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.tools.project_query import ProjectResearch
|
|
from test_investigation_tools import tool_case
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway
|
|
from test_investigator import LoopAdapter, CASE, run, state
|
|
from test_project_research import FakeDB, rows
|
|
|
|
|
|
def review(effect='contradicts',support=None,against=None):
|
|
return {'effect':effect,'support':support or [],'against':against or [],
|
|
'reason':'The required dispatch is implemented; locating it is counterevidence to its absence.'}
|
|
|
|
|
|
def update(effect='contradicts',support=None,against=None):
|
|
return {'confidence':'strongly_supported','evidence_refs':['E000001'],
|
|
'causal_review':review(effect,support,against)}
|
|
|
|
|
|
@pytest.mark.parametrize('effect,expected',[
|
|
('contradicts','weakened'),('mixed','plausible'),('neutral','plausible'),('supports','strongly_supported')])
|
|
def test_direction_caps_confidence_independently_of_information_score(effect,expected):
|
|
u=update(effect)
|
|
assert effective_confidence(u)==expected
|
|
assert u['confidence']=='strongly_supported' # Never rewrite the saved LLM answer.
|
|
assert effective_confidence({'confidence':'supported'})=='supported'
|
|
|
|
|
|
def test_real_scoped_empty_result_can_support_missing_data():
|
|
u=update('supports',['E000001'])
|
|
evidence={'E000001':{'content':{'coverage':'complete_filtered_rows','items':[
|
|
{'query':{'table':'objects'},'result':{'items':[],'truncated':False,'coverage':'complete_filtered_rows'}}]}}}
|
|
validate_review(u,evidence)
|
|
assert effective_confidence(u)=='strongly_supported'
|
|
|
|
|
|
@pytest.mark.parametrize('change,code',[
|
|
(lambda u:u['causal_review'].update(against=['E999999']),'INVESTIGATION_CAUSAL_REFERENCE'),
|
|
(lambda u:u['causal_review'].update(effect='supports',support=[]),'INVESTIGATION_CAUSAL_EFFECT'),
|
|
(lambda u:u['causal_review'].update(effect='mixed',support=[]),'INVESTIGATION_CAUSAL_EFFECT'),
|
|
(lambda u:u['causal_review'].update(effect='neutral'),'INVESTIGATION_CAUSAL_EFFECT')])
|
|
def test_causal_refs_and_direction_are_validated(change,code):
|
|
u=update(against=['E000001']);change(u)
|
|
with pytest.raises(InvestigationError,match=code):
|
|
validate_review(u,{'E000001':{'content':{'coverage':'selected_lines'}}})
|
|
|
|
|
|
def test_schema_evidence_can_test_a_schema_defect_without_proving_runtime_rows():
|
|
validate_review(update('supports',['E000001']),{'E000001':{'content':{'coverage':'schema_only'}}})
|
|
|
|
|
|
def test_new_schema_requires_review_but_saved_v11_contract_is_unchanged():
|
|
from test_investigation_consolidation import populated
|
|
c=populated();c['prompt_version']='investigation_project_v12'
|
|
new=output_schema(c)['properties']['update']
|
|
old=output_schema(dict(c,prompt_version='investigation_project_v11'))['properties']['update']
|
|
assert 'causal_review' in new['required'] and 'causal_review' not in old['properties']
|
|
assert not Draft202012Validator(new).is_valid({'confidence':'unassessed','evidence_refs':[],
|
|
'resolve_questions':[],'add_questions':[],'note':'Initial check'})
|
|
assert 'normal code' in build_prompt(c)['messages'][0]['content']
|
|
|
|
|
|
def test_counterevidence_changes_scheduler_without_losing_reward_budget_or_history(tool_case):
|
|
store=tool_case[0]
|
|
with PackageReader(tool_case[1]) as reader:project=reader.manifest['project']
|
|
service=ProjectResearch(project,database=FakeDB([{'row':{'id':1}}]))
|
|
def change(reply,context):
|
|
if context['selected_hypothesis']=='H1' and context['previous_result']:
|
|
eid=context['previous_result']['evidence_id']
|
|
reply['assessment']['semantic_score']=100
|
|
reply['update'].update(confidence='strongly_supported',causal_review=review(against=[eid]))
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
adapter=LoopAdapter(change)
|
|
loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
result=run(loop,steps=5)
|
|
saved=state(result)
|
|
assert result.status=='YIELDED'
|
|
assert [c['selected_hypothesis'] for c in adapter.contexts][:5]==['H1','H2','H3','H1','H2']
|
|
h=saved['hypotheses']['H1'];assert h['confidence']=='weakened' and h['invalid_responses']==0
|
|
note=h['notes'][-1]
|
|
assert note['requested_confidence']=='strongly_supported' and note['effective_confidence']=='weakened'
|
|
assert note['causal_review']['against']
|
|
account=BudgetManager(store).snapshot(*CASE).state['accounts']['H1']
|
|
assert account['rewarded']>0 and account['remaining']>0 and len(account['actions'])==2
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
def test_invalid_review_is_atomic_before_assessment_or_tool_spending(tool_case):
|
|
store=tool_case[0]
|
|
with PackageReader(tool_case[1]) as reader:project=reader.manifest['project']
|
|
service=ProjectResearch(project,database=FakeDB([{'row':{'id':1}}]))
|
|
def change(reply,context):
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
if context['previous_result']:
|
|
reply['update']['causal_review']=review(against=['E999999'])
|
|
loop=Investigator(store,gateway(LoopAdapter(change)),router=InvestigationToolRouter(project_research=service))
|
|
original=loop._context
|
|
loop._context=lambda *args:dict(original(*args),prompt_version='investigation_project_v13')
|
|
run(loop,steps=3)
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='REJECTED' and result.code=='INVESTIGATION_CAUSAL_REFERENCE'
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
|
|
def test_early_diagnosis_cannot_bypass_calibrated_counterevidence(tool_case):
|
|
store=tool_case[0]
|
|
with PackageReader(tool_case[1]) as reader:project=reader.manifest['project']
|
|
service=ProjectResearch(project,database=FakeDB([{'row':{'id':1}}]))
|
|
def change(reply,context):
|
|
if context['selected_hypothesis']=='H1' and context['previous_result']:
|
|
eid=context['previous_result']['evidence_id']
|
|
other=next(e for e,v in context['evidence'].items() if v['record']['hypothesis_id']=='H2')
|
|
reply['update'].update(confidence='strongly_supported',causal_review=review(against=[eid]))
|
|
reply['next']={'kind':'request_diagnosis','reason':'Incorrect strong claim','countercheck_refs':[other]}
|
|
else:reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
loop=Investigator(store,gateway(LoopAdapter(change)),router=InvestigationToolRouter(project_research=service))
|
|
original=loop._context
|
|
loop._context=lambda *args:dict(original(*args),prompt_version='investigation_project_v14')
|
|
run(loop,steps=3)
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='REJECTED' and result.code=='EARLY_STOP_STRONG_EVIDENCE_REQUIRED'
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
assert 'early_stop' not in state(result)
|