158 lines
8.5 KiB
Python
158 lines
8.5 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from test_compact_investigation import context,entry
|
|
from test_lean_investigation_decision import reply
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from viano_bug_analysis.analysis.investigation_plan import framed_view,observed_keys
|
|
from viano_bug_analysis.analysis.investigation_compact import model_view,validate_tool_shape
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt,output_schema
|
|
from viano_bug_analysis.analysis.investigation_decision import normalize
|
|
from viano_bug_analysis.analysis.investigation_config import InvestigationError
|
|
from viano_bug_analysis.case.contracts import validate
|
|
|
|
|
|
def plan():
|
|
return {'goal':'Determine whether these observed objects are event-bound.',
|
|
'supports_if':'No objects exist in the affected scope.',
|
|
'weakens_if':'Verified objects already exist there.',
|
|
'investigate_if':'The result is incomplete or another creation path is untested.'}
|
|
|
|
|
|
def current():
|
|
c=context();c['prompt_version']='investigation_project_v18';return c
|
|
|
|
|
|
def test_model_goal_and_criteria_persist_unchanged_without_mutating_reply():
|
|
c=current();r=reply();r['update']['next_step']=plan();before=copy.deepcopy(r)
|
|
d=normalize(c,r)
|
|
assert r==before and d['update']['working_brief']['next_step']==plan()
|
|
validate(d['update']['working_brief'],'investigation-working-brief')
|
|
c['working_brief']['brief']=d['update']['working_brief']
|
|
view=framed_view(c,model_view(c))
|
|
assert view['investigation_step']['decision']==plan()
|
|
assert view['investigation_step']['decision_source']=='previous_model_plan'
|
|
assert 'next_step' not in view['working_brief']
|
|
assert c['working_brief']['brief']['next_step']==plan()
|
|
|
|
|
|
def test_action_needs_a_plan_but_completion_can_use_null():
|
|
c=current();r=reply();r['update']['next_step']=None
|
|
r['next']={'kind':'recall','evidence_refs':['E000002']}
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_PLAN'):normalize(c,r)
|
|
r['next']={'kind':'finish','reason':'Enough scoped evidence.'}
|
|
assert normalize(c,r)['update']['working_brief']['next_step'] is None
|
|
r['update'].pop('next_step')
|
|
assert not Draft202012Validator(output_schema(c)).is_valid(r)
|
|
|
|
|
|
def test_missing_plan_criteria_produce_precise_repair_fields():
|
|
c=current();r=reply();r['update']['next_step']={'goal':'PRIVATE_PLAN'}
|
|
with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r)
|
|
assert caught.value.feedback['validation_path']=='/update/next_step'
|
|
assert caught.value.feedback['validation_fields']==['supports_if','weakens_if','investigate_if']
|
|
assert 'PRIVATE_PLAN' not in json.dumps(caught.value.feedback)
|
|
|
|
|
|
def key_context(count=18,origin=None):
|
|
c=current();c['evidence']={};origin=origin or {'snapshot_id':'verified'}
|
|
schema=entry(1,'database.schema');capture=schema['content']['items'][0]
|
|
capture['origin']=copy.deepcopy(origin)
|
|
capture['result']['items'][0]['columns']=[{'COLUMN_NAME':'object_key','COLUMN_TYPE':'int','COLUMN_KEY':'PRI'}]
|
|
rows=entry(20);rows['content']['items']=[{'query':{'kind':'database.rows','table':'creature','offset':0},
|
|
'origin':copy.deepcopy(origin),'result':{'coverage':'complete_filtered_rows','truncated':False,'next_offset':None,
|
|
'items':[{'row':{'object_key':i,'irrelevant_payload':'large raw text'}} for i in range(count)]}}]
|
|
c['evidence']={'E000001':schema,'E000020':rows};c['previous_result']=rows
|
|
return c
|
|
|
|
|
|
def test_known_inputs_are_observed_key_tuples_with_evidence_and_scope():
|
|
c=key_context();before=copy.deepcopy(c);items=observed_keys(c,model_view(c))
|
|
assert items==[{'evidence_id':'E000020','capture_index':0,'schema_id':'E000001',
|
|
'table':'creature','key_columns':['object_key'],'rows':[[i] for i in range(18)],
|
|
'all_returned_keys_included':True,'query_complete':True}]
|
|
assert c==before
|
|
|
|
|
|
def test_key_list_limit_and_query_paging_are_distinct_and_explicit():
|
|
c=key_context(40);items=observed_keys(c,model_view(c))
|
|
assert len(items[0]['rows'])==32 and items[0]['all_returned_keys_included'] is False
|
|
assert items[0]['query_complete'] is True
|
|
c['evidence']['E000020']['content']['items'][0]['result']['truncated']=True
|
|
assert observed_keys(c,model_view(c))[0]['query_complete'] is False
|
|
|
|
|
|
def test_composite_keys_preserve_tuple_relationships_and_scalar_types():
|
|
c=key_context(2);schema=c['evidence']['E000001']['content']['items'][0]
|
|
schema['result']['items'][0]['columns'].append({'COLUMN_NAME':'group_key','COLUMN_TYPE':'varchar','COLUMN_KEY':'PRI'})
|
|
rows=c['evidence']['E000020']['content']['items'][0]['result']['items']
|
|
rows[0]['row']['group_key']='alpha';rows[1]['row']['group_key']='beta'
|
|
item=observed_keys(c,model_view(c))[0]
|
|
assert item['key_columns']==['object_key','group_key'] and item['rows']==[[0,'alpha'],[1,'beta']]
|
|
|
|
|
|
@pytest.mark.parametrize('change',['origin','no_primary','unknown_shape','bool_value','missing_column'])
|
|
def test_key_hints_do_not_guess_keys_values_or_snapshot_relationships(change):
|
|
c=key_context();s=c['evidence']['E000001']['content']['items'][0];r=c['evidence']['E000020']['content']['items'][0]
|
|
if change=='origin':s['origin']={'snapshot_id':'different'}
|
|
elif change=='no_primary':s['result']['items'][0]['columns'][0].pop('COLUMN_KEY')
|
|
elif change=='unknown_shape':s['result']['items']=['unknown provider format']
|
|
elif change=='bool_value':r['result']['items'][0]['row']['object_key']=True
|
|
else:r['result']['items'][0]['row'].pop('object_key')
|
|
assert observed_keys(c,model_view(c))==[]
|
|
|
|
|
|
def test_unread_unrelated_archived_rows_are_not_silently_added_as_hints():
|
|
c=key_context();c['evidence']['E000020']['record']['hypothesis_id']='H2';c['previous_result']=None
|
|
assert observed_keys(c,model_view(c))==[]
|
|
|
|
|
|
def test_v18_prompt_is_shorter_and_all_existing_actions_still_available():
|
|
c=current();old=build_prompt(dict(c,prompt_version='investigation_project_v17'));new=build_prompt(c)
|
|
assert len(new['messages'][0]['content'])<len(old['messages'][0]['content'])
|
|
view=json.loads(new['messages'][1]['content'])
|
|
assert view['task']=='advance_investigation_goal'
|
|
assert set(new['output_schema']['properties']['next']['properties']['kind']['enum'])=={
|
|
'tool','test','finish','reject','request_diagnosis','defer','recall'}
|
|
|
|
|
|
def test_narrow_review_keeps_existing_contract_and_low_effort_routing():
|
|
from test_investigation_review import review_context
|
|
from viano_bug_analysis.llm.gateway import is_investigation_review
|
|
c=review_context();old=build_prompt(c);c['prompt_version']='investigation_project_v18'
|
|
new=build_prompt(c)
|
|
assert new['messages']==old['messages'] and new['output_schema']==old['output_schema']
|
|
assert is_investigation_review(new)
|
|
|
|
|
|
def test_plan_survives_real_checkpoint_portable_resume_and_result_assessment(tool_case,tmp_path):
|
|
from test_investigator import LoopAdapter,CASE,run,state
|
|
from test_working_brief import service_for
|
|
from test_project_research import rows
|
|
from test_case_interpreter import gateway
|
|
from viano_bug_analysis.analysis import Investigator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
store=tool_case[0];service=service_for(tool_case);seen=[]
|
|
def change(reply,c):
|
|
if c['previous_result']:
|
|
assert c['investigation_step']['decision']==plan();seen.append(1)
|
|
reply['next']={'kind':'finish','reason':'Scoped result assessed.'};reply['update']['next_step']=None
|
|
else:
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
reply['update']['next_step']=plan()
|
|
adapter=LoopAdapter(change);loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
first=run(loop,steps=1)
|
|
with PackageReader(first.package) as reader:
|
|
record=json.loads(reader.read_bytes(state(first)['hypotheses']['H1']['working_brief_ref']))
|
|
assert record['brief']['next_step']==plan()
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
final=run(Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service)))
|
|
assert final.status=='COMPLETED' and len(seen)==3 and len(adapter.contexts)==6
|
|
assert PackageValidator().validate(final.package).valid
|