241 lines
14 KiB
Python
241 lines
14 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from test_compact_investigation import context
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_investigator import CASE,run,state
|
|
from test_working_brief import service_for
|
|
from test_case_interpreter import gateway
|
|
from test_project_research import rows
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig,InvestigationError
|
|
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.llm import ProviderResponse
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
from viano_bug_analysis.package import PackageReader,PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
|
|
|
|
def lean_context():
|
|
c=context();c['prompt_version']='investigation_project_v18'
|
|
c['lean_session']={'turns':[],'checkpoint':False}
|
|
return c
|
|
|
|
|
|
def test_short_contract_does_not_rewrite_brief_or_confidence():
|
|
c=lean_context();before=copy.deepcopy(c)
|
|
r={'assessment':{'semantic_score':40,'reason':'Useful observation.'} if c['previous_result'] else None,
|
|
'observation':'A bounded result, not a diagnosis.','next':{'kind':'checkpoint'}}
|
|
assert Draft202012Validator(output_schema(c)).is_valid(r)
|
|
d=normalize(c,r)
|
|
assert 'working_brief' not in d['update'] and c==before
|
|
assert d['next']['kind']=='defer'
|
|
assert d['update']['confidence']==c['hypotheses'][c['selected_hypothesis']]['confidence']
|
|
r['update']={'confidence':'strongly_supported'}
|
|
assert not Draft202012Validator(output_schema(c)).is_valid(r)
|
|
|
|
|
|
def test_stable_cache_roundtrip_and_no_hard_result_truncation():
|
|
from viano_bug_analysis.llm.stable_context import split_document,restore_document
|
|
c=lean_context();r=build_prompt(c);document=json.loads(r['messages'][1]['content'])
|
|
split=split_document(document,r['prompt_version'])
|
|
assert split and restore_document(*split)==document
|
|
assert 'request_contract' in split[0]['input']['tool_schema']
|
|
assert 'recent_observations' in split[1]['input']
|
|
assert 'next_step' not in r['output_schema']['properties']
|
|
|
|
|
|
class LeanAdapter:
|
|
def __init__(self):self.calls=[]
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
doc=json.loads(request['messages'][1]['content']);c=doc['input'];self.calls.append(doc)
|
|
if doc['task']=='investigate_one_step':
|
|
result={'assessment':({'semantic_score':50,'reason':'Bounded result.'} if c['previous_result'] else None),'observation':'Inspect bounded project records.',
|
|
'next':({'kind':'checkpoint'} if c['previous_result'] else {'kind':'tool','operation':'project.query','arguments':rows()})}
|
|
else:
|
|
previous=c['previous_result']
|
|
result={'assessment':{'semantic_score':50,'reason':'Scoped result assessed.'} if previous else None,
|
|
'update':{'confidence':'plausible','add_questions':[],'resolve_questions':[],
|
|
'working_brief':{'format_version':2,'findings':[],'retired':[],
|
|
'uncertainties':['Cause remains unconfirmed.'],'next_question':'No further bounded checks.'}},
|
|
'next':{'kind':'finish','reason':'Bounded investigation complete.'}}
|
|
return ProviderResponse(result,'lean-fixture',100,50)
|
|
|
|
|
|
def test_switch_flush_resume_budget_and_portability(tool_case,tmp_path):
|
|
adapter=LeanAdapter();service=service_for(tool_case);router=InvestigationToolRouter(project_research=service)
|
|
config=InvestigationConfig(lean_loop=True)
|
|
first=run(Investigator(tool_case[0],gateway(adapter),config,router=router),steps=1)
|
|
h=state(first)['hypotheses']['H1']
|
|
assert len(h['lean_session']['turns'])==1 and not h.get('working_brief_ref')
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
final=run(Investigator(portable,gateway(adapter),router=router),steps=20)
|
|
assert final.status=='COMPLETED'
|
|
assert sum(d['task']=='consolidate_investigation_session' for d in adapter.calls)==3
|
|
assert sum(d['task']=='investigate_one_step' for d in adapter.calls)==4
|
|
assert all(not h['lean_session']['turns'] and h.get('working_brief_ref') for h in state(final)['hypotheses'].values())
|
|
assert all(all(a['status']=='settled' for a in h['actions'].values()) for h in BudgetManager(portable).snapshot(*CASE).state['accounts'].values())
|
|
assert PackageValidator().validate(final.package).valid
|
|
|
|
|
|
@pytest.mark.parametrize('limit', [1,2])
|
|
def test_iteration_boundary_consolidates_before_leaving_hypothesis(tool_case,limit):
|
|
adapter=LeanAdapter();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
final=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,max_iterations_per_hypothesis=limit),router=router))
|
|
assert final.status=='COMPLETED'
|
|
assert all(not h.get('lean_session',{}).get('turns') for h in state(final)['hypotheses'].values())
|
|
|
|
|
|
def test_invalid_short_tool_reply_has_no_budget_or_evidence_side_effect(tool_case):
|
|
class Invalid(LeanAdapter):
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
return ProviderResponse({'assessment':None,'observation':'Attempt forbidden execution.',
|
|
'next':{'kind':'tool','operation':'shell.exec','arguments':{}}},'fixture',100,50)
|
|
loop=Investigator(tool_case[0],gateway(Invalid()),InvestigationConfig(lean_loop=True),
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='REJECTED'
|
|
assert all(not h['actions'] and h['remaining']==100 for h in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
assert not state(result)['hypotheses']['H1'].get('lean_session')
|
|
|
|
|
|
def test_checkpoint_has_no_next_step_and_retains_review_low_routing():
|
|
from test_investigation_review import review_context
|
|
from viano_bug_analysis.llm.gateway import is_investigation_review
|
|
c=review_context();c['prompt_version']='investigation_project_v18'
|
|
c['lean_session']={'turns':[],'checkpoint':True}
|
|
request=build_prompt(c)
|
|
assert is_investigation_review(request)
|
|
assert 'next_step' not in request['output_schema']['properties']['update']['properties']
|
|
|
|
|
|
def test_session_consolidation_includes_all_prior_payloads():
|
|
c=lean_context();ids=list(c['evidence'])
|
|
c['lean_session']={'checkpoint':True,'turns':[{'llm_ref':'stored/ref.json','evidence_id':eid,
|
|
'observation':'Provisional observation.'} for eid in ids]}
|
|
doc=json.loads(build_prompt(c)['messages'][1]['content'])
|
|
assert set(ids)<=set(doc['input']['evidence'])
|
|
assert doc['input']['session_log']==c['lean_session']['turns']
|
|
|
|
|
|
|
|
def test_periodic_checkpoint_and_defer_continue_without_confidence_update(tool_case):
|
|
class Continue(LeanAdapter):
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
doc=json.loads(request['messages'][1]['content']);self.calls.append(doc);c=doc['input']
|
|
if doc['task']=='investigate_one_step':
|
|
answer={'assessment':{'semantic_score':60,'reason':'Read scoped result.'} if c['previous_result'] else None,
|
|
'observation':'No established cause yet.','next':{'kind':'tool','operation':'project.query','arguments':rows()}}
|
|
else:
|
|
answer={'assessment':{'semantic_score':60,'reason':'Read scoped result.'} if c['previous_result'] else None,
|
|
'update':{'confidence':'plausible','add_questions':[],'resolve_questions':[],
|
|
'working_brief':{'format_version':2,'findings':[],'retired':[],
|
|
'uncertainties':['Unknown causal mechanism.'],'next_question':'Investigate another relationship.'}},
|
|
'next':{'kind':'defer','reason':'Continue bounded research.'}}
|
|
return ProviderResponse(answer,'fixture',100,50)
|
|
adapter=Continue();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_checkpoint_steps=2),router=router)
|
|
result=run(loop,steps=10)
|
|
assert result.status=='YIELDED'
|
|
checkpoints=[d for d in adapter.calls if d['task']=='consolidate_investigation_session']
|
|
assert any(len(d['input']['session_log'])==2 for d in checkpoints)
|
|
assert all(len(h.get('lean_session',{}).get('turns',[]))<=2 for h in state(result)['hypotheses'].values())
|
|
|
|
|
|
def test_unknown_evidence_cannot_enter_consolidated_findings():
|
|
from viano_bug_analysis.analysis.investigation_compact import validate_tool_shape
|
|
from viano_bug_analysis.analysis.investigation_config import InvestigationError
|
|
from test_lean_investigation_decision import reply
|
|
c=lean_context();c['lean_session']['checkpoint']=True;r=reply()
|
|
r['next']={'kind':'finish','reason':'Done.'}
|
|
r['update']['working_brief']['findings']=[{'id':'B9999','direction':'support','text':'Unseen.',
|
|
'scope':'Unknown scope.','evidence_refs':['E999999']}]
|
|
with pytest.raises(InvestigationError):validate_tool_shape(c,r)
|
|
|
|
|
|
|
|
def test_budget_exhaustion_flushes_session_and_settles_pending_result(tool_case):
|
|
from viano_bug_analysis.budget import BudgetConfig
|
|
adapter=LeanAdapter();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
result=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),router=router,
|
|
budget_config=BudgetConfig(default_action_cost=100,max_reward_bps=0)))
|
|
assert result.status=='COMPLETED'
|
|
assert all(not h.get('lean_session',{}).get('turns') for h in state(result)['hypotheses'].values())
|
|
assert all(all(a['status']=='settled' for a in h['actions'].values()) for h in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
|
|
|
|
def test_replay_after_tool_publication_does_not_repeat_model_or_budget(tool_case,monkeypatch):
|
|
adapter=LeanAdapter();loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
original=loop.evidence.store_tool_results
|
|
def interrupted(*args,**kwargs):
|
|
original(*args,**kwargs)
|
|
raise RuntimeError('Injected interruption after evidence publication')
|
|
monkeypatch.setattr(loop.evidence,'store_tool_results',interrupted)
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_INPUT'):asyncio.run(loop.step(*CASE))
|
|
monkeypatch.setattr(loop.evidence,'store_tool_results',original)
|
|
result=asyncio.run(loop.step(*CASE))
|
|
assert result.status=='PROGRESSED' and len(adapter.calls)==1
|
|
assert len(state(result)['hypotheses']['H1']['lean_session']['turns'])==1
|
|
assert len(BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|
|
|
|
|
|
|
|
def test_bad_consolidation_finding_id_repairs_without_corrupting_context(tool_case,tmp_path):
|
|
from viano_bug_analysis.case.contracts import validate
|
|
class BadIdOnce(LeanAdapter):
|
|
failed=False
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
response=await super().generate(config,request,max_response_bytes=max_response_bytes)
|
|
c=self.calls[-1]['input']
|
|
if self.calls[-1]['task']=='consolidate_investigation_session' and not self.failed:
|
|
self.failed=True
|
|
eid=next(iter(c['evidence']))
|
|
response.document['update']['working_brief']['findings']=[{'id':'F01','direction':'neutral',
|
|
'text':'Schema available.','scope':'Captured schema.','evidence_refs':[eid]}]
|
|
return response
|
|
adapter=BadIdOnce();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),router=router)
|
|
first=run(loop,steps=1)
|
|
before=BudgetManager(tool_case[0]).snapshot(*CASE).state
|
|
rejected=asyncio.run(loop.step(*CASE))
|
|
assert rejected.status=='REJECTED'
|
|
assert BudgetManager(tool_case[0]).snapshot(*CASE).state==before
|
|
with PackageReader(rejected.package) as reader:
|
|
c=json.loads(reader.read_bytes(state(rejected)['active']['context_ref']))
|
|
validate(c,'investigation-context')
|
|
detail=c['previous_rejection_details']
|
|
assert 'validation_keyword' not in detail
|
|
assert 'B0001' in detail['requirement']
|
|
# The original released version persisted this extra keyword. Its
|
|
# immutable context must remain loadable without a package rewrite.
|
|
c['previous_rejection_details']={'validation_path':'/update/working_brief/findings/0/id',
|
|
'validation_fields':[],'validation_keyword':'pattern'}
|
|
validate(c,'investigation-context')
|
|
prompt=build_prompt(c)
|
|
assert 'B0001' in json.loads(prompt['messages'][1]['content'])['input']['previous_rejection_details']['requirement']
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(rejected.package)
|
|
final=run(Investigator(portable,gateway(adapter),router=router))
|
|
assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid
|
|
|
|
|
|
|
|
def test_invalid_context_is_not_published(tool_case):
|
|
from viano_bug_analysis.case.errors import CaseBuildError
|
|
adapter=LeanAdapter();loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True),
|
|
router=InvestigationToolRouter(project_research=service_for(tool_case)))
|
|
result=run(loop,steps=1);saved=state(result)
|
|
# A new prompt context is validated before a checkpoint can reference it.
|
|
saved['active']={'hypothesis_id':'H1','iteration':2,'attempt':0,'phase':'prompt',
|
|
'context_ref':'analysis/investigation/0001/contexts/invalid.json','context_sha256':'0'*64,
|
|
'decision':None,'llm_ref':None,'test_request':None,'test_deadline':None,'retry_token':'initial'}
|
|
before=tool_case[0].latest(*CASE)
|
|
with pytest.raises(CaseBuildError) as caught:loop._save(saved,context={'unexpected':'invalid'})
|
|
assert caught.value.code=='INPUT_SCHEMA'
|
|
assert tool_case[0].latest(*CASE)==before
|