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)