153 lines
7.8 KiB
Python
153 lines
7.8 KiB
Python
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from test_compact_investigation import context, entry
|
|
from viano_bug_analysis.analysis.investigation_step import step_view, _result_status
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from viano_bug_analysis.llm.stable_context import split_document, restore_document
|
|
from viano_bug_analysis.llm.gateway import is_investigation_review
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
|
|
|
|
def capture(kind='database.rows', **result):
|
|
return {'query':{'kind':kind,'table':'creature','offset':0,'filters':[]},
|
|
'origin':{'snapshot_id':'fixed'},'result':{'items':[],
|
|
'coverage':'complete_filtered_rows','truncated':False,'next_offset':None,**result}}
|
|
|
|
|
|
def test_empty_complete_query_is_scoped_metadata_not_a_causal_finding():
|
|
c=capture();s=_result_status(c)
|
|
assert s['returned_records']==0 and s['completeness']=='complete_for_exact_query'
|
|
assert 'confidence' not in s and 'absence' not in s
|
|
|
|
|
|
@pytest.mark.parametrize('change',[
|
|
{'truncated':True},{'truncated':None},{'next_offset':50},{'coverage':'partial_filtered_rows'},
|
|
{'coverage':'unknown'},{'items':None}])
|
|
def test_partial_or_unknown_metadata_never_asserts_complete_query(change):
|
|
assert _result_status(capture(**change))['completeness']=='partial_or_unknown'
|
|
|
|
|
|
@pytest.mark.parametrize('offset',[20,True,-1])
|
|
def test_offset_result_is_not_whole_query(offset):
|
|
c=capture();c['query']['offset']=offset
|
|
assert _result_status(c)['completeness']=='partial_or_unknown'
|
|
|
|
|
|
def test_schema_and_count_are_not_data_rows_or_empty_searches():
|
|
schema=capture('database.schema',coverage='schema_only',items=[{'table':'creature','columns':[]}])
|
|
assert _result_status(schema)['completeness']=='schema_only'
|
|
count=capture('database.count',coverage='complete_filtered_count',items=[{'matched_rows':0}])
|
|
status=_result_status(count)
|
|
assert status['returned_records']==1 and status['matched_rows']==0 and status['completeness']=='count_only'
|
|
count['result']['items'][0]['matched_rows']=False
|
|
assert 'matched_rows' not in _result_status(count)
|
|
assert _result_status(capture('source.search'))['completeness']=='partial_or_unknown'
|
|
|
|
|
|
def test_lexical_search_is_complete_only_with_explicit_no_skipped_files():
|
|
c=capture('source.search',coverage='complete_selected_text_search',skipped_files=[])
|
|
assert _result_status(c)['completeness']=='complete_for_exact_query'
|
|
c['result']['skipped_files']=['binary-or-oversized.cpp']
|
|
assert _result_status(c)['completeness']=='partial_or_unknown'
|
|
c['result'].pop('skipped_files')
|
|
assert _result_status(c)['completeness']=='partial_or_unknown'
|
|
|
|
|
|
def test_question_selection_uses_existing_memory_then_unresolved_question_then_mechanism():
|
|
c=context();assert step_view(c)['question']==c['working_brief']['brief']['next_question']
|
|
c['working_brief']=None
|
|
c['hypotheses']['H1']['questions']=[{'id':'Q0001','text':'Already answered','resolved':True},
|
|
{'id':'Q0002','text':'Which path executes?','resolved':False}]
|
|
assert step_view(c)['question']=='Which path executes?'
|
|
c['hypotheses']['H1']['questions'][1]['resolved']=True
|
|
assert step_view(c)['question_source']=='selected_mechanism'
|
|
|
|
|
|
def test_resolved_question_is_not_reintroduced_from_stale_brief():
|
|
c=context();q=c['working_brief']['brief']['next_question']
|
|
c['hypotheses']['H1']['questions']=[{'id':'Q0001','text':q,'resolved':True}]
|
|
assert step_view(c)['question_source']=='selected_mechanism'
|
|
|
|
|
|
def test_same_query_detection_requires_same_scope_origin_and_json_types():
|
|
c=context();a=capture();a['query']['filters']=[{'column':'id','value':1}]
|
|
e=entry(20);e['content']['items']=[a]
|
|
c['evidence']={'E000020':e};c['previous_result']=e
|
|
for n in range(1,6):c['evidence'][f'E{n:06d}']=copy.deepcopy(e)
|
|
c['evidence']['E000002']['content']['items'][0]['origin']['snapshot_id']='changed'
|
|
c['evidence']['E000003']['content']['items'][0]['query']['offset']=50
|
|
c['evidence']['E000004']['content']['items'][0]['query']['filters'][0]['value']=True
|
|
c['evidence']['E000005']['content']['items'][0]['query']['columns']=['guid']
|
|
before=copy.deepcopy(c)
|
|
s=step_view(c)['preflight']['previous_captures'][0]
|
|
assert s['same_query_and_origin_ids']==['E000001'] and c==before
|
|
e['content']['items'][0]['origin']={}
|
|
assert 'same_query_and_origin_ids' not in step_view(c)['preflight']['previous_captures'][0]
|
|
|
|
|
|
def test_known_schema_ids_do_not_depend_on_research_owner():
|
|
c=context();c['evidence']['E000001']=entry(1,'database.schema')
|
|
c['evidence']['E000001']['record']['hypothesis_id']='H2'
|
|
assert step_view(c)['preflight']['schema_captures']=={'creature':['E000001']}
|
|
|
|
|
|
def test_v17_normal_task_is_focused_without_discarding_context_or_breaking_cache():
|
|
c=context();c['prompt_version']='investigation_project_v17';before=copy.deepcopy(c)
|
|
req=build_prompt(c);doc=json.loads(req['messages'][1]['content'])
|
|
assert doc['task']=='advance_selected_question' and doc['input']['investigation_step']==step_view(c)
|
|
assert doc['input']['working_brief']['findings']==c['working_brief']['brief']['findings']
|
|
assert doc['input']['evidence'] and c==before
|
|
stable,live=split_document(doc,c['prompt_version'])
|
|
assert restore_document(stable,live)==doc
|
|
c['selected_hypothesis']='H2';c['previous_result']=None
|
|
updated=json.loads(build_prompt(c)['messages'][1]['content'])
|
|
assert split_document(updated,c['prompt_version'])[0]==stable
|
|
|
|
|
|
def test_v17_keeps_narrow_early_review_and_effort_routing():
|
|
from test_investigation_review import review_context
|
|
c=review_context();old=build_prompt(c);c['prompt_version']='investigation_project_v17'
|
|
req=build_prompt(c);doc=json.loads(req['messages'][1]['content'])
|
|
assert doc['task']=='assess_pending_result_for_early_review'
|
|
assert 'investigation_step' not in doc['input']
|
|
assert req['messages']==old['messages'] and req['output_schema']==old['output_schema']
|
|
assert is_investigation_review(req)
|
|
|
|
|
|
def test_v16_remains_frozen_without_new_task_fields():
|
|
c=context();c['prompt_version']='investigation_project_v16'
|
|
doc=json.loads(build_prompt(c)['messages'][1]['content'])
|
|
assert doc['task']=='investigate_hypothesis' and 'investigation_step' not in doc['input']
|
|
|
|
|
|
def test_focused_loop_keeps_assessment_and_budget_settlement(tool_case):
|
|
from test_investigator import LoopAdapter,run,CASE
|
|
from test_case_interpreter import gateway
|
|
from test_working_brief import service_for
|
|
from test_project_research import rows
|
|
from viano_bug_analysis.analysis import Investigator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageValidator
|
|
assessed=[]
|
|
def change(reply,c):
|
|
step=c['investigation_step']
|
|
assert step.get('question') or step.get('decision',{}).get('goal')
|
|
if c['previous_result']:
|
|
eid=c['previous_result']['evidence_id']
|
|
assert step['preflight']['previous_id']==eid
|
|
assert step['preflight']['previous_captures']
|
|
assessed.append(eid)
|
|
reply['next']={'kind':'finish','reason':'Scoped question checked; preserve uncertainty.'}
|
|
else:reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
store=tool_case[0];adapter=LoopAdapter(change)
|
|
result=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service_for(tool_case))))
|
|
assert result.status=='COMPLETED' and len(assessed)==3
|
|
accounts=BudgetManager(store).snapshot(*CASE).state['accounts']
|
|
assert all(len(a['actions'])==1 and all(v['status']=='settled' for v in a['actions'].values()) for a in accounts.values())
|
|
assert PackageValidator().validate(result.package).valid
|