284 lines
16 KiB
Python
284 lines
16 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
from dataclasses import asdict,replace
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis.investigation_compact import model_view,source_aliases,_content,validate_tool_shape
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt,output_schema
|
|
from viano_bug_analysis.analysis.investigation_config import InvestigationConfig,InvestigationError
|
|
from viano_bug_analysis.analysis.investigation_decision import normalize
|
|
from viano_bug_analysis.analysis import Investigator,DiagnosisBuilder
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
from viano_bug_analysis.llm.stable_context import split_document,restore_document
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
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_working_brief import service_for,brief
|
|
from test_project_research import rows
|
|
|
|
|
|
def entry(number,kind='source.read'):
|
|
query={'kind':kind,'path':'code.cpp','start_line':1,'end_line':2}
|
|
items=[{'path':'code.cpp','line':1,'text':'important-'+str(number)+'x'*1000},
|
|
{'path':'code.cpp','line':2,'text':'next line'}]
|
|
if kind=='database.schema':
|
|
query={'kind':kind,'table':'creature'}
|
|
items=[{'table':'creature','columns':[{'COLUMN_NAME':'id','COLUMN_TYPE':'int'},
|
|
{'COLUMN_NAME':'guid','COLUMN_TYPE':'int'}]}]
|
|
return {'record':{'id':f'E{number:06d}','hypothesis_id':'H1','action_id':str(number)},
|
|
'content':{'coverage':'schema_only' if kind=='database.schema' else 'selected_lines',
|
|
'items':[{'query':query,'origin':{'commit':'a'*40},'result':{'coverage':'schema_only' if kind=='database.schema' else 'selected_lines',
|
|
'truncated':False,'items':items}}]}}
|
|
|
|
|
|
def context():
|
|
h={'proposal':{'title':'Missing hook','mechanism':'Required hook is absent.',
|
|
'expected_evidence':['Missing registration'],'falsifying_evidence':['Hook exists']},
|
|
'confidence':'plausible','disposition':'active','questions':[], 'notes':[]}
|
|
evidence={f'E{i:06d}':entry(i) for i in range(1,21)}
|
|
c={'prompt_version':'investigation_project_v15','case':{'case':{'case_id':'c','package_id':'p','analysis_revision':1},
|
|
'source_context':{'refs':[{'path':'original/code.cpp'}]},'database_context':{'refs':[]}},
|
|
'interpretation':{'observed_behavior':[{'text':'Interaction fails','source_ids':['O0001']}],
|
|
'confirmed_technical_facts':[{'text':'Captured quest giver differs from affected subject.','source_ids':['F0001']}]},
|
|
'hypotheses':{'H1':h,'H2':copy.deepcopy(h)},'selected_hypothesis':'H1','evidence':evidence,
|
|
'previous_result':evidence['E000020'],'working_brief':{'brief':brief()},
|
|
'other_working_briefs':{'H2':{'brief':brief()}},
|
|
'budget':{'accounts':{'H1':{'remaining':90,'status':'investigating','actions':{}},'H2':{'remaining':90,'status':'investigating','actions':{}}}},
|
|
'budget_policy':{'default_action_cost':10,'action_costs':{}},'policy':asdict(InvestigationConfig()),
|
|
'tool_schema':{'x-project-bindings':{'source':{'commit':'a'*40},'database':{'snapshot_id':'snap'},
|
|
'max_rows':50,'max_lines':80}},'recalled_evidence':[]}
|
|
return c
|
|
|
|
|
|
def test_working_memory_replaces_payload_history_without_hiding_counterfindings():
|
|
c=context();original=copy.deepcopy(c);v=model_view(c)
|
|
assert c==original
|
|
assert set(v['evidence'])=={'E000020'}
|
|
assert v['working_brief']['findings'][0]['direction']=='against'
|
|
assert v['other_working_briefs']['H2']['findings'][0]['evidence_refs']==['E000001']
|
|
assert 'E000001' in {r['id'] for r in v['evidence_catalog']}
|
|
assert next(r for r in v['evidence_catalog'] if r['id']=='E000001')['queries'][0]['origin']=={'commit':'a'*40}
|
|
assert 'E000002' in v['archive_ids']
|
|
assert len(json.dumps(v)) < len(json.dumps(c))/2
|
|
|
|
|
|
def test_fresh_pending_recalled_and_schema_results_stay_available():
|
|
c=context();c['recalled_evidence']=['E000002']
|
|
c['budget']['accounts']['H1']['actions']['3']={'status':'recorded'}
|
|
c['evidence']['E000004']=entry(4,'database.schema')
|
|
v=model_view(c)
|
|
assert {'E000002','E000003','E000020'}<=set(v['evidence'])
|
|
assert 'E000004' in v['known_schemas']
|
|
assert v['known_schemas']['E000004']['captures'][0]['coverage']=='schema_only'
|
|
assert v['known_schemas']['E000004']['captures'][0]['tables'][0]['columns']==[['id','int'],['guid','int']]
|
|
c['policy']['prompt_target_characters']=1
|
|
c['evidence']['E000020']['content']['items'][0]['result']['items'][0]['text']='CRITICAL'*20000
|
|
v=model_view(c)
|
|
assert 'CRITICAL'*20000 in json.dumps(v)
|
|
assert v['evidence_policy']['character_target']==1
|
|
|
|
|
|
def test_memoryless_migration_bootstraps_all_evidence_once():
|
|
c=context();c.pop('working_brief');v=model_view(c)
|
|
assert set(v['evidence'])==set(c['evidence'])
|
|
assert v['working_brief_status']=='bootstrap_complete_history'
|
|
|
|
|
|
def test_row_table_preserves_literal_scalar_types_and_duplicates():
|
|
rows=[{'id':1,'map':0,'null':None,'literal':{'encoding':'result_records_v1','rows':[]}},
|
|
{'id':2,'map':0,'null':None,'literal':{'encoding':'result_records_v1','rows':[]}}]*10
|
|
raw={'items':[{'query':{'kind':'database.rows'},'result':{'items':[{'table':'creature','row':r} for r in rows],
|
|
'truncated':True,'next_offset':20,'coverage':'page'}}]}
|
|
encoded=_content(raw)['items'][0]['result'];table=encoded['items']['records']
|
|
restored=[{**table['common'],**dict(zip(table['fields'],r))} for r in table['rows']]
|
|
assert restored==rows
|
|
assert encoded['truncated'] is True and encoded['next_offset']==20
|
|
|
|
|
|
def test_stable_cache_partition_survives_new_findings_and_hypothesis_switch():
|
|
c=context();first=build_prompt(c);doc=json.loads(first['messages'][1]['content'])
|
|
stable,live=split_document(doc,'investigation_project_v15')
|
|
assert restore_document(stable,live)==doc
|
|
c['selected_hypothesis']='H2';c['previous_result']=None;c['evidence']['E000021']=entry(21)
|
|
updated=json.loads(build_prompt(c)['messages'][1]['content']);stable2,_=split_document(updated,'investigation_project_v15')
|
|
assert stable2==stable
|
|
|
|
|
|
def test_wire_contract_keeps_finish_reject_early_review_test_and_recall():
|
|
c=context();wire=output_schema(c)
|
|
for action in [{'kind':'finish','reason':'Done'},{'kind':'reject','reason':'Refuted'},
|
|
{'kind':'defer','reason':'Await review'},
|
|
{'kind':'request_diagnosis','reason':'Cause supported','countercheck_refs':['E000001']},
|
|
{'kind':'recall','evidence_refs':['E000002']}]:
|
|
assert Draft202012Validator(wire['properties']['next']).is_valid(action)
|
|
assert 'working_brief' not in output_schema(dict(c,prompt_version='investigation_project_v12'))['properties']['update']['properties']
|
|
|
|
|
|
def test_loose_wire_does_not_relax_controller_argument_or_brief_validation():
|
|
from test_lean_investigation_decision import reply
|
|
c=context();r=reply();r['next']={'kind':'tool','operation':'project.query','arguments':{'kind':'database.rows','table':'creature'}}
|
|
assert Draft202012Validator(output_schema(c)).is_valid(r)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_OUTPUT'):validate_tool_shape(c,r)
|
|
r['next']={'kind':'finish','reason':'Done'};r['update']['working_brief']['findings'][0]['text']='x'*301
|
|
with pytest.raises(InvestigationError):validate_tool_shape(c,r)
|
|
|
|
|
|
def test_packaged_source_alias_roundtrip_and_no_mutation():
|
|
c=context();before=copy.deepcopy(c);sources,aliases=source_aliases(c)
|
|
alias=next(iter(sources));assert alias.startswith('PS') and aliases[alias]=='original/code.cpp'
|
|
assert sources[alias]=='code.cpp' and c==before
|
|
|
|
|
|
def test_recalled_evidence_stays_pinned_without_extra_queries_and_portable_replay(tool_case,tmp_path):
|
|
store=tool_case[0];service=service_for(tool_case);calls={};seen=[]
|
|
def change(reply,c):
|
|
hid=c['selected_hypothesis'];calls[hid]=calls.get(hid,0)+1;n=calls[hid]
|
|
if hid=='H1' and n in (1,2):reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
elif hid=='H1' and n==3:reply['next']={'kind':'recall','evidence_refs':['E000001']}
|
|
elif hid=='H1' and n==4:
|
|
assert 'E000001' in c['evidence'];seen.append(c)
|
|
reply['next']={'kind':'finish','reason':'Scoped investigation complete'}
|
|
adapter=LoopAdapter(change)
|
|
loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
result=run(loop)
|
|
assert result.status=='COMPLETED' and seen
|
|
saved=state(result);assert saved['recalled_evidence']['H1']==['E000001']
|
|
assert len(BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['actions'])==2
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(result.package)
|
|
resumed=run(Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service)))
|
|
assert resumed.status=='COMPLETED'
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
def test_v15_small_prefix_attempts_cache_instead_of_failing_local_character_heuristic(monkeypatch):
|
|
from test_llm_caching import config,scripted,CACHE,native
|
|
from test_llm_gateway import request
|
|
req=request();req['prompt_version']='investigation_project_v15'
|
|
adapter=GeminiProvider();cfg=replace(config(),cache_min_prefix_chars=100000,cache_required=True)
|
|
seen=scripted(adapter,monkeypatch,[CACHE,native()])
|
|
result=asyncio.run(adapter.generate(cfg,req,max_response_bytes=10000))
|
|
assert result.diagnostics['cache_status']=='created' and len(seen)==2
|
|
|
|
|
|
def test_unseen_archive_ids_are_not_newly_citable_before_local_read():
|
|
from test_lean_investigation_decision import reply
|
|
c=context();r=reply();r['update']['working_brief']['findings'][0]['evidence_refs']=['E000002']
|
|
with pytest.raises(InvestigationError,match='EVIDENCE_NOT_READ'):validate_tool_shape(c,r)
|
|
c['recalled_evidence']=['E000002']
|
|
validate_tool_shape(c,r)
|
|
|
|
|
|
def test_other_pending_payload_has_inventory_and_is_full_on_its_own_turn():
|
|
c=context();e=c['evidence']['E000003'];e['record']['hypothesis_id']='H2'
|
|
c['budget']['accounts']['H2']['actions']['3']={'status':'recorded'}
|
|
v=model_view(c)
|
|
assert 'E000003' not in v['evidence'] and 'E000003' in v['evidence_policy']['pending_other_ids']
|
|
assert any(r['id']=='E000003' for r in v['evidence_catalog'])
|
|
c['selected_hypothesis']='H2';c['previous_result']=e
|
|
assert 'E000003' in model_view(c)['evidence']
|
|
|
|
|
|
@pytest.mark.parametrize('action',[{'kind':'finish'},{'kind':'tool','operation':'project.query','arguments':{'kind':'database.rows','table':'creature'}}])
|
|
def test_invalid_compact_wire_has_no_tool_or_budget_side_effect(tool_case,action):
|
|
store=tool_case[0];service=service_for(tool_case)
|
|
adapter=LoopAdapter(lambda r,c:r.update(next=action))
|
|
result=asyncio.run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)).step(*CASE))
|
|
assert result.status=='REJECTED' and result.code=='INVESTIGATION_OUTPUT'
|
|
assert not service._captures
|
|
assert all(not a['actions'] for a in BudgetManager(store).snapshot(*CASE).state['accounts'].values())
|
|
|
|
|
|
def test_already_visible_local_read_is_rejected_before_external_work(tool_case):
|
|
store=tool_case[0];service=service_for(tool_case)
|
|
def change(r,c):
|
|
if c['previous_result']:r['next']={'kind':'recall','evidence_refs':[c['previous_result']['evidence_id']]}
|
|
else:r['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
adapter=LoopAdapter(change)
|
|
loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
for _ in range(4):result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='REJECTED' and result.code=='INVESTIGATION_RECALL_VISIBLE'
|
|
snapshot=BudgetManager(store).snapshot(*CASE).state
|
|
assert len(snapshot['accounts']['H1']['actions'])==1
|
|
|
|
|
|
def test_frozen_v14_active_request_remains_replayable(tool_case):
|
|
store=tool_case[0];service=service_for(tool_case);adapter=LoopAdapter()
|
|
loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
original=loop._context
|
|
def old_context(*args):
|
|
value=original(*args);value['prompt_version']='investigation_project_v14'
|
|
value.pop('other_working_briefs',None);value.pop('recalled_evidence',None)
|
|
return value
|
|
loop._context=old_context
|
|
execute=loop._execute_plan
|
|
def stop_before_execution(state,budget):raise RuntimeError('simulated crash before execution')
|
|
loop._execute_plan=stop_before_execution
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):run(loop,steps=1)
|
|
resumed=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=1)
|
|
assert resumed.status=='YIELDED' and len(adapter.contexts)==1
|
|
|
|
|
|
def test_v15_early_completion_reads_local_countercheck_then_independent_review(tool_case):
|
|
from test_diagnosis_builder import FinalAdapter,build
|
|
store=tool_case[0];service=service_for(tool_case);local_reads=[]
|
|
def change(reply,c):
|
|
early=c['early_review']['request']
|
|
if early and early['status']=='preparing':
|
|
reply['next']={'kind':'defer','reason':'Settle pending evidence.'};return
|
|
own=[e for e,v in c['evidence'].items() if v['record']['hypothesis_id']=='H1']
|
|
if c['selected_hypothesis']=='H1' and own:
|
|
eid=own[0]
|
|
reply['update'].update(confidence='strongly_supported',causal_review={
|
|
'effect':'supports','support':[eid],'against':[],'reason':'Fixture scoped causal defect.'})
|
|
other=[e for e,v in c['evidence'].items() if v['record']['hypothesis_id']=='H2']
|
|
if not other:
|
|
local_reads.append(1)
|
|
ref=next(r['id'] for r in c['evidence_catalog'] if r['collector']=='H2')
|
|
reply['next']={'kind':'recall','evidence_refs':[ref]}
|
|
else:reply['next']={'kind':'request_diagnosis','reason':'Scoped cause and observed alternative check.','countercheck_refs':[other[0]]}
|
|
else:reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
adapter=LoopAdapter(change)
|
|
done=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=12)
|
|
assert done.status=='COMPLETED' and state(done)['early_stop']['status']=='requested'
|
|
assert len(local_reads)==1
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
assert all(a['remaining']>0 for a in before['accounts'].values())
|
|
final=build(DiagnosisBuilder(store,gateway(FinalAdapter(True,lambda p,c:p.update(confidence='strongly_supported')))))
|
|
assert final.status=='COMPLETED' and BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
|
|
def test_v15_counterevidence_still_weakens_a_strong_model_claim():
|
|
from test_lean_investigation_decision import reply
|
|
from viano_bug_analysis.analysis.confidence_review import effective_confidence
|
|
c=context();decision=normalize(c,reply())
|
|
assert decision['update']['causal_review']['effect']=='contradicts'
|
|
assert effective_confidence(decision['update'])=='weakened'
|
|
|
|
|
|
def test_recall_turn_cannot_publish_unread_finding_before_reading_it():
|
|
from test_lean_investigation_decision import reply
|
|
c=context();r=reply()
|
|
r['update']['working_brief']['findings'][0]['evidence_refs']=['E000002']
|
|
r['next']={'kind':'recall','evidence_refs':['E000002']}
|
|
with pytest.raises(InvestigationError) as caught:validate_tool_shape(c,r)
|
|
assert caught.value.feedback['requirement']=='preserve_existing_brief_omit_new_unread_citations_and_use_recall_first'
|
|
r['update']['working_brief']['findings'][0]['evidence_refs']=['E000001']
|
|
validate_tool_shape(c,r)
|
|
|
|
|
|
def test_lossless_table_never_collapses_boolean_integer_or_float_cells():
|
|
from viano_bug_analysis.analysis.investigation_compact import _table
|
|
values=[True,1,1.0,False,0,0.0,None,'1']*4
|
|
rows=[{'value':v,'long_common':'unchanged metadata '*30} for v in values]
|
|
table=_table(rows)
|
|
assert isinstance(table,dict) and 'value' not in table['common']
|
|
decoded=[{**table['common'],**dict(zip(table['fields'],r))} for r in table['rows']]
|
|
assert [type(r['value']) for r in decoded]==[type(v) for v in values]
|
|
assert decoded==rows
|