168 lines
8.5 KiB
Python
168 lines
8.5 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from viano_bug_analysis.analysis import Investigator, InvestigationError
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema
|
|
from viano_bug_analysis.analysis.investigation_focus import full_history_view
|
|
from viano_bug_analysis.analysis.working_brief import model_view, validate_brief
|
|
from viano_bug_analysis.case.contracts import validate
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.lifecycle import CheckpointStore, Workspace
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
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_investigation_consolidation import populated
|
|
from test_case_interpreter import gateway
|
|
from test_case_builder import case_setup
|
|
from test_investigator import LoopAdapter, CASE, run, state
|
|
from test_project_research import FakeDB, rows
|
|
|
|
|
|
def brief():
|
|
return {'format_version':1,'findings':[{'id':'B0001','direction':'against',
|
|
'text':'Required hook exists.','scope':'Pinned source, not runtime proof.',
|
|
'evidence_refs':['E000001']}], 'uncertainties':['Deployment is unverified.'],
|
|
'next_question':'Does the deployed revision match?', 'changes':[]}
|
|
|
|
|
|
def update(b=None):
|
|
return {'working_brief':b or brief(),'evidence_refs':['E000001'],
|
|
'causal_review':{'effect':'contradicts','support':[],'against':['E000001'],
|
|
'reason':'Required hook exists.'}}
|
|
|
|
|
|
@pytest.mark.parametrize('kind',['remove','edit','relabel'])
|
|
def test_counterevidence_cannot_silently_disappear_or_change(kind):
|
|
old=brief();new=copy.deepcopy(old)
|
|
if kind=='remove':new['findings']=[]
|
|
elif kind=='edit':new['findings'][0]['text']='A different claim.'
|
|
else:new['findings'][0]['direction']='support'
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_BRIEF_CHANGE'):
|
|
validate_brief(update(new),old,{'E000001':{}})
|
|
|
|
|
|
def test_explicit_correction_keeps_audit_trail_and_balance():
|
|
old=brief();new=copy.deepcopy(old);new['findings']=[]
|
|
new['changes']=[{'id':'B0001','reason':'New snapshot invalidates the old scope.',
|
|
'evidence_refs':['E000002']}]
|
|
u=update(new);u['causal_review'].update(effect='neutral',against=[])
|
|
validate_brief(u,old,{'E000001':{},'E000002':{}})
|
|
|
|
|
|
@pytest.mark.parametrize('change,code',[
|
|
(lambda u:u['working_brief']['findings'][0].update(evidence_refs=['E999999']),'REFERENCE'),
|
|
(lambda u:u['causal_review'].update(against=[]),'BALANCE'),
|
|
(lambda u:u['working_brief']['findings'].append(copy.deepcopy(u['working_brief']['findings'][0])),'ID'),
|
|
(lambda u:u['working_brief']['changes'].append({'id':'B9999','reason':'Invented old finding.','evidence_refs':['E000001']}),'CHANGE')])
|
|
def test_invalid_brief_refs_ids_changes_and_review_balance(change,code):
|
|
u=update();change(u)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_BRIEF_'+code):
|
|
validate_brief(u,None,{'E000001':{}})
|
|
|
|
|
|
def test_schema_bounds_working_memory_and_legacy_schema_unchanged():
|
|
validate(brief(),'investigation-working-brief')
|
|
c=populated();c['prompt_version']='investigation_project_v13'
|
|
assert 'working_brief' in output_schema(c)['properties']['update']['required']
|
|
for old in ('investigation_project_v11','investigation_project_v12'):
|
|
assert 'working_brief' not in output_schema(dict(c,prompt_version=old))['properties']['update']['properties']
|
|
|
|
|
|
def context():
|
|
c=populated();c['prompt_version']='investigation_project_v13'
|
|
c['interpretation']={'summary':'Observed gap'}
|
|
for h in c['hypotheses'].values():h['disposition']='active'
|
|
c['budget']['accounts']['H2']=copy.deepcopy(c['budget']['accounts']['H1'])
|
|
return c
|
|
|
|
|
|
def test_all_research_and_fresh_scope_remain_and_stable_prefix_survives_brief():
|
|
c=context();before=copy.deepcopy(c)
|
|
old=full_history_view(c);view=model_view(c)
|
|
for eid,entry in old['evidence'].items():
|
|
assert view['evidence'][eid]['content']==entry['content']
|
|
for key in ('shared_sources','shared_results','shared_metadata','sources','interpretation'):
|
|
assert view.get(key)==old.get(key)
|
|
assert view['previous_result']==old['previous_result']
|
|
assert view['hypotheses']['H1']['notes']==old['hypotheses']['H1']['notes']
|
|
assert 'notes' not in view['hypotheses']['H2']
|
|
assert c==before
|
|
_,prefix,_=GeminiProvider._messages(build_prompt(c),stable_case_context=True)
|
|
c['working_brief']={'brief':brief()}
|
|
current=model_view(c)
|
|
assert 'notes' not in current['hypotheses']['H1']
|
|
assert current['working_brief']==brief()
|
|
_,later,_=GeminiProvider._messages(build_prompt(c),stable_case_context=True)
|
|
assert prefix==later
|
|
|
|
|
|
def service_for(tool_case):
|
|
with PackageReader(tool_case[1]) as reader:project=reader.manifest['project']
|
|
return ProjectResearch(project,database=FakeDB([{'row':{'id':1}}]))
|
|
|
|
|
|
def test_portable_brief_checkpoint_resume_and_originals_preserved(tool_case,tmp_path):
|
|
service=service_for(tool_case)
|
|
adapter=LoopAdapter(lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':rows()}))
|
|
first=run(Investigator(tool_case[0],gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=4)
|
|
saved=state(first);h=saved['hypotheses']['H1']
|
|
with PackageReader(first.package) as reader:
|
|
doc=json.loads(reader.read_bytes(h['working_brief_ref']))
|
|
assert doc['sequence']==2 and doc['parent_ref'] and doc['brief']['format_version']==1
|
|
validate(doc,'investigation-working-brief-record')
|
|
for hypothesis in saved['hypotheses'].values():
|
|
assert hypothesis['working_brief_ref'] in reader.files
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'))
|
|
portable.intake(first.package)
|
|
resumed=run(Investigator(portable,gateway(adapter),router=InvestigationToolRouter(project_research=service)),steps=1)
|
|
assert adapter.contexts[-1]['working_brief'] is not None
|
|
assert 'notes' not in adapter.contexts[-1]['hypotheses'][adapter.contexts[-1]['selected_hypothesis']]
|
|
assert PackageValidator().validate(resumed.package).valid
|
|
with PackageReader(first.package) as a,PackageReader(resumed.package) as b:
|
|
for name in a.files:
|
|
if name!='manifest.json':assert a.read_bytes(name)==b.read_bytes(name)
|
|
|
|
|
|
def test_invalid_brief_does_not_settle_previous_or_reserve_new_budget(tool_case):
|
|
service=service_for(tool_case)
|
|
def change(reply,c):
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
if c['previous_result']:
|
|
b=brief();b['findings'][0]['evidence_refs']=['E999999']
|
|
reply['update']['working_brief']=b
|
|
loop=Investigator(tool_case[0],gateway(LoopAdapter(change)),router=InvestigationToolRouter(project_research=service))
|
|
run(loop,steps=3)
|
|
before=BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='REJECTED' and result.code=='INVESTIGATION_BRIEF_REFERENCE'
|
|
assert BudgetManager(tool_case[0]).snapshot(*CASE).state==before
|
|
assert state(result)['hypotheses']['H1']['working_brief_ref'].endswith('H1-0001.json')
|
|
|
|
|
|
def test_interrupted_execution_publishes_brief_once_without_another_model_call(tool_case,monkeypatch):
|
|
service=service_for(tool_case)
|
|
adapter=LoopAdapter(lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':rows()}))
|
|
loop=Investigator(tool_case[0],gateway(adapter),router=InvestigationToolRouter(project_research=service))
|
|
save=loop._save
|
|
def interrupted(*args,**kwargs):
|
|
if kwargs.get('additions'):
|
|
raise RuntimeError('crash before brief publication')
|
|
return save(*args,**kwargs)
|
|
monkeypatch.setattr(loop,'_save',interrupted)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):
|
|
asyncio.run(loop.step(*CASE))
|
|
assert len(adapter.contexts)==1
|
|
monkeypatch.setattr(loop,'_save',save)
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert len(adapter.contexts)==1
|
|
h=state(result)['hypotheses']['H1']
|
|
with PackageReader(result.package) as reader:
|
|
assert len([ref for ref in reader.files if '/briefs/' in ref])==1
|
|
assert json.loads(reader.read_bytes(h['working_brief_ref']))['sequence']==1
|
|
assert len(BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|