306 lines
18 KiB
Python
306 lines
18 KiB
Python
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
import asyncio, copy, json
|
|
import pytest
|
|
from dataclasses import replace
|
|
from viano_bug_analysis.analysis import DiagnosisBuilder, DiagnosisConfig, DiagnosisLimits, DiagnosisError, Investigator
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.lifecycle import State, CheckpointStore, Workspace
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_case_interpreter import gateway
|
|
from test_investigator import LoopAdapter, CASE
|
|
|
|
class FinalAdapter:
|
|
def __init__(self, confirmed=False, change=None):
|
|
self.confirmed=confirmed; self.change=change; self.calls=0; self.requests=[]
|
|
async def generate(self,config,request,*,max_response_bytes):
|
|
self.calls+=1; self.requests.append(copy.deepcopy(request))
|
|
c=restore(json.loads(request['messages'][1]['content'])['input'])
|
|
eid=next(iter(c['evidence']))
|
|
result={'status':'confirmed' if self.confirmed else 'inconclusive','selected_hypothesis':'H1' if self.confirmed else None,
|
|
'root_cause':{'summary':'Activation uses the wrong state','technical_description':'Packaged activation flow conflicts with observation; inspect the stated transition.','evidence_refs':[eid]} if self.confirmed else None,
|
|
'confidence':'supported' if self.confirmed else 'plausible',
|
|
'evaluations':[{'hypothesis_id':h,'verdict':'supported' if h=='H1' and self.confirmed else 'unresolved','reason':'Scoped investigation results preserve uncertainty','evidence_refs':[eid],
|
|
'countercheck':{'adequate':True,'reason':'Alternative was compared against the scoped finding','evidence_refs':[next(ref for ref,entry in c['evidence'].items() if entry['record']['hypothesis_id']==h)]}} for h in c['investigation']['hypotheses']],
|
|
'remaining_uncertainty':['Snapshot does not establish current live behavior.'],'reason':'Evidence supports a cause' if self.confirmed else 'No causal explanation established',
|
|
'recommended_human_actions':[] if self.confirmed else ['Observe the activation under the specified build.']}
|
|
if 'decision_basis' in request['output_schema']['required']:
|
|
result['decision_basis']={'causal_explanation':'Scoped findings explain the observed activation failure.',
|
|
'scope_limits':['Only the supplied snapshot was examined.'],
|
|
'sufficient_for_change':self.confirmed,'contradictions':[]}
|
|
if self.change:self.change(result,c)
|
|
return ProviderResponse(result,'fixture-final',100,50)
|
|
|
|
def final_phase_loop(store):
|
|
# Reach the final phase via the bounded production protocol. Legacy full
|
|
# investigation prompts exceed the hard input cap before these tests start.
|
|
from test_investigation_startup import Adapter
|
|
from viano_bug_analysis.analysis import InvestigationConfig
|
|
class Bounded(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body=json.loads(request['messages'][1]['content'])
|
|
task=body['task']
|
|
if task=='startup_choose_query':
|
|
return ProviderResponse({'assessment':None,'observation':'Read packaged implementation.',
|
|
'next':{'kind':'tool','operation':'source.read_context',
|
|
'arguments':{'source_ref':'tool/code.cpp','start_line':1,'end_line':2}}},'fixture',100,50)
|
|
if task=='investigate_one_step':
|
|
return ProviderResponse({'assessment':None,'observation':'Preserve the scoped result.',
|
|
'next':{'kind':'checkpoint','intent':'completion'}},'fixture',100,50)
|
|
if task=='consolidate_investigation_session':
|
|
return ProviderResponse({'assessment':None,'confidence':'plausible','findings':[],
|
|
'uncertainties':['Runtime remains untested.'],'next_question':'No further question.',
|
|
'retire':[],'next':{'kind':'finish','reason':'Scoped fixture investigation complete.'}},'fixture',100,50)
|
|
return await super().generate(config,request,**kwargs)
|
|
from test_working_brief import service_for
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
router=InvestigationToolRouter(project_research=service_for((store,store.latest(*CASE))))
|
|
return Investigator(store,gateway(Bounded()),InvestigationConfig(lean_loop=True,
|
|
lean_protocol=2,startup_batch_review=True,lean_checkpoint_steps=0,prompt_window_version=2,
|
|
max_iterations_per_hypothesis=1),router=router)
|
|
|
|
|
|
@pytest.fixture
|
|
def investigated(tool_case):
|
|
result=asyncio.run(final_phase_loop(tool_case[0]).run(*CASE,max_steps=20))
|
|
return tool_case[0],result.package
|
|
|
|
def build(builder,**kwargs):return asyncio.run(builder.build(*CASE,**kwargs))
|
|
|
|
def test_confirmed_and_replay_preserve_package_budget_and_boundaries(investigated):
|
|
store,original=investigated; adapter=FinalAdapter(True)
|
|
before=BudgetManager(store).snapshot(*CASE).state
|
|
builder=DiagnosisBuilder(store,gateway(adapter)); result=build(builder)
|
|
assert result.status=='COMPLETED' and adapter.calls==1, result
|
|
with PackageReader(original) as old,PackageReader(result.package) as new:
|
|
assert new.manifest['state']==State.BUILDING_DIAGNOSIS
|
|
assert new.manifest['artifacts']['change_request'] is None
|
|
for name in old.files:
|
|
if name!='manifest.json':assert old.read_bytes(name)==new.read_bytes(name)
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
assert PackageValidator().validate(result.package).valid
|
|
assert build(builder).reused and adapter.calls==1
|
|
|
|
def test_inconclusive_review_and_portable_replay(investigated,tmp_path):
|
|
store,_=investigated; adapter=FinalAdapter(); result=build(DiagnosisBuilder(store,gateway(adapter)))
|
|
assert result.status=='INCONCLUSIVE' and result.failure_ref
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['state']==State.FAILED_TO_ANALYZE and r.manifest['routing']['destination']=='review'
|
|
assert r.manifest['artifacts']['change_request'] is None
|
|
failure=json.loads(r.read_bytes(result.failure_ref))
|
|
assert set(failure['hypothesis_end_states'])=={'H1','H2','H3'}
|
|
assert failure['remaining_questions']
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(result.package)
|
|
assert build(DiagnosisBuilder(portable,gateway(adapter))).reused and adapter.calls==1
|
|
|
|
@pytest.mark.parametrize('change',[
|
|
lambda p,c:p['root_cause'].update(evidence_refs=['E999999']),
|
|
lambda p,c:p['evaluations'].pop(),
|
|
lambda p,c:p['evaluations'][1].update(hypothesis_id='H1'),
|
|
lambda p,c:p.update(selected_hypothesis='H99'),
|
|
lambda p,c:p.update(confidence='plausible'),
|
|
lambda p,c:[e['countercheck'].update(adequate=False) for e in p['evaluations']],
|
|
lambda p,c:p['root_cause'].update(evidence_refs=[]),
|
|
lambda p,c:p.update(budget=99999),
|
|
])
|
|
def test_invalid_reply_no_partial_diagnosis_or_budget_change(investigated,change):
|
|
store,_=investigated; before=BudgetManager(store).snapshot(*CASE).state
|
|
result=build(DiagnosisBuilder(store,gateway(FinalAdapter(True,change))))
|
|
assert result.status=='REJECTED'
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['artifacts']['diagnosis'] is None
|
|
assert r.manifest['state']==State.BUILDING_DIAGNOSIS
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
def test_rejection_limit_deterministic_review(investigated):
|
|
store,_=investigated; adapter=FinalAdapter(True,lambda p,c:p.update(selected_hypothesis='H999'))
|
|
builder=DiagnosisBuilder(store,gateway(adapter),DiagnosisConfig(2))
|
|
assert build(builder).status=='REJECTED'
|
|
result=build(builder);assert result.status=='INCONCLUSIVE' and adapter.calls==2
|
|
assert build(DiagnosisBuilder(store,gateway(adapter))).reused and adapter.calls==2
|
|
|
|
def test_frozen_policy_and_context_limits(investigated):
|
|
store,_=investigated; adapter=FinalAdapter(True,lambda p,c:p.update(selected_hypothesis='H999'))
|
|
assert build(DiagnosisBuilder(store,gateway(adapter))).status=='REJECTED'
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_POLICY_CHANGED'):
|
|
build(DiagnosisBuilder(store,gateway(adapter),DiagnosisConfig(4)))
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_CONTEXT_LIMIT'):
|
|
build(DiagnosisBuilder(store,gateway(adapter),limits=DiagnosisLimits(max_context_bytes=10)))
|
|
assert adapter.calls==1
|
|
|
|
def test_unfinished_investigation_rejected_without_model(tool_case):
|
|
adapter=FinalAdapter()
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_PHASE'):
|
|
build(DiagnosisBuilder(tool_case[0],gateway(adapter)))
|
|
assert adapter.calls==0
|
|
|
|
def test_provider_unavailable_does_not_spend_budget_or_create_review(investigated):
|
|
from viano_bug_analysis.llm import ProviderUnavailable
|
|
class Offline:
|
|
async def generate(self,*args,**kwargs):raise ProviderUnavailable('offline')
|
|
store,_=investigated; before=BudgetManager(store).snapshot(*CASE).state
|
|
g=gateway(Offline()); g.config=replace(g.config,max_attempts=1)
|
|
result=build(DiagnosisBuilder(store,g))
|
|
assert result.status=='BLOCKED'
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['artifacts']['diagnosis'] is None and r.manifest['artifacts'].get('failure_report') is None
|
|
|
|
def test_provider_configuration_error_is_blocked_not_review(investigated):
|
|
from viano_bug_analysis.llm import LLMError
|
|
class BadConfig:
|
|
async def generate(self,*args,**kwargs):raise LLMError('LLM_CREDENTIALS')
|
|
store,_=investigated;result=build(DiagnosisBuilder(store,gateway(BadConfig())))
|
|
assert result.status=='BLOCKED'
|
|
with PackageReader(result.package) as r:
|
|
assert r.manifest['routing']['reason']=='LLM_CONFIGURATION'
|
|
assert r.manifest['artifacts']['diagnosis'] is None
|
|
|
|
def test_resume_after_model_publication_without_second_call(investigated,monkeypatch):
|
|
store,_=investigated;adapter=FinalAdapter(True);builder=DiagnosisBuilder(store,gateway(adapter))
|
|
publish=builder._publish
|
|
def crash(*args):raise RuntimeError('simulated process interruption')
|
|
monkeypatch.setattr(builder,'_publish',crash)
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_INPUT'):build(builder)
|
|
monkeypatch.setattr(builder,'_publish',publish)
|
|
assert build(builder).status=='COMPLETED' and adapter.calls==1
|
|
|
|
def test_missing_pointer_is_not_rebuilt(investigated,tmp_path):
|
|
from viano_bug_analysis.package import PackageWriter
|
|
store,_=investigated;adapter=FinalAdapter(True);builder=DiagnosisBuilder(store,gateway(adapter))
|
|
result=build(builder);incoming=tmp_path/'pointer.zip'
|
|
PackageWriter().write(result.package,incoming,manifest_updates={'artifacts':{'diagnosis':None}})
|
|
store.intake(incoming)
|
|
with pytest.raises(DiagnosisError,match='DIAGNOSIS_POINTER_CHANGED'):build(builder)
|
|
assert adapter.calls==1
|
|
|
|
def test_config_and_schemas(tmp_path):
|
|
from importlib.resources import files
|
|
from jsonschema import Draft202012Validator
|
|
path=tmp_path/'config.toml';path.write_text('[diagnosis]\nmax_invalid_responses = 5\n')
|
|
assert DiagnosisConfig.load(path).max_invalid_responses==5
|
|
for invalid in (True,0,-1):
|
|
with pytest.raises(DiagnosisError):DiagnosisConfig(invalid)
|
|
path.write_text('[diagnosis]\nunknown = 5\n')
|
|
with pytest.raises(DiagnosisError):DiagnosisConfig.load(path)
|
|
for name in ('diagnosis-output','diagnosis-result','diagnosis-context','failure-report'):
|
|
Draft202012Validator.check_schema(json.loads(files('viano_bug_analysis.schemas').joinpath(name+'-v1.schema.json').read_text()))
|
|
|
|
def test_explicit_provider_resume_preserves_context_and_budget(investigated):
|
|
from datetime import datetime,timedelta,timezone
|
|
from viano_bug_analysis.llm import ProviderUnavailable
|
|
class Toggle(FinalAdapter):
|
|
offline=True
|
|
async def generate(self,*args,**kwargs):
|
|
if self.offline:raise ProviderUnavailable('offline')
|
|
return await super().generate(*args,**kwargs)
|
|
store,_=investigated;before=BudgetManager(store).snapshot(*CASE).state;adapter=Toggle(True)
|
|
g=gateway(adapter);g.config=replace(g.config,max_attempts=1)
|
|
builder=DiagnosisBuilder(store,g,clock=lambda:datetime.now(timezone.utc)-timedelta(days=1))
|
|
assert build(builder).status=='BLOCKED'
|
|
store.advance(*CASE,State.BUILDING_DIAGNOSIS,step_id='manual-diagnosis-provider-resume')
|
|
adapter.offline=False
|
|
result=build(builder,retry_id='provider-restored')
|
|
assert result.status=='COMPLETED' and adapter.calls==1
|
|
assert BudgetManager(store).snapshot(*CASE).state==before
|
|
|
|
|
|
def test_retry_receives_specific_feedback_and_replays(investigated):
|
|
store,_=investigated
|
|
adapter=FinalAdapter(True)
|
|
adapter.change=lambda p,c: p.update(selected_hypothesis='H999') if adapter.calls==1 else None
|
|
builder=DiagnosisBuilder(store,gateway(adapter))
|
|
assert build(builder).status=='REJECTED'
|
|
assert build(builder).status=='COMPLETED'
|
|
first=restore(json.loads(adapter.requests[0]['messages'][1]['content'])['input'])
|
|
second=restore(json.loads(adapter.requests[1]['messages'][1]['content'])['input'])
|
|
assert 'validation_feedback' not in first
|
|
assert second['validation_feedback'][0]['code']=='DIAGNOSIS_SELECTION'
|
|
assert second['validation_feedback'][0]['llm_result_ref']
|
|
assert build(builder).reused and adapter.calls==2
|
|
|
|
|
|
def test_legacy_rejection_upgrades_next_attempt_without_rewriting_context(investigated):
|
|
from viano_bug_analysis.evidence.store import digest
|
|
from viano_bug_analysis.lifecycle.storage import json_bytes
|
|
store,_=investigated
|
|
adapter=FinalAdapter(True)
|
|
adapter.change=lambda p,c: p.update(selected_hypothesis='H999') if adapter.calls==1 else None
|
|
builder=DiagnosisBuilder(store,gateway(adapter))
|
|
budget=BudgetManager(store)
|
|
snapshot=budget.snapshot(*CASE)
|
|
builder.loop.router=final_phase_loop(store).router
|
|
with PackageReader(store.latest(*CASE)) as reader:
|
|
saved=builder.loop._load(reader)
|
|
working=builder.loop._context(saved,'H1',budget)
|
|
investigation_ref=reader.manifest['artifacts']['investigation_state']
|
|
legacy={'format':'viano-diagnosis-context','format_version':1,'case':working['case'],
|
|
'interpretation':working['interpretation'],'investigation':saved,
|
|
'investigation_ref':investigation_ref,'budget':snapshot.state,'evidence':working['evidence'],
|
|
'max_invalid_responses':3}
|
|
ref='analysis/diagnosis/0001/context.json'
|
|
raw=json_bytes(legacy)
|
|
store.advance(*CASE,State.BUILDING_DIAGNOSIS,step_id='diagnosis:v1:1:start',
|
|
details={'operation':'diagnosis-start','context_ref':ref,'context_sha256':digest(legacy)},
|
|
additions={ref:raw},artifacts={'diagnosis_context':ref})
|
|
assert build(builder).status=='REJECTED'
|
|
assert adapter.requests[0]['prompt_version']=='diagnosis_v1'
|
|
assert build(builder).status=='COMPLETED'
|
|
assert adapter.requests[1]['prompt_version']=='diagnosis_v8'
|
|
received=restore(json.loads(adapter.requests[1]['messages'][1]['content'])['input'])
|
|
assert received['validation_feedback'][0]['code']=='DIAGNOSIS_SELECTION'
|
|
assert build(builder).reused and adapter.calls==2
|
|
with PackageReader(store.latest(*CASE)) as reader:assert reader.read_bytes(ref)==raw
|
|
assert budget.snapshot(*CASE).state==snapshot.state
|
|
|
|
|
|
def test_lean_reply_derivation_and_portable_replay(investigated,tmp_path):
|
|
store,_ = investigated
|
|
def change(payload,context):
|
|
payload.pop('reason')
|
|
payload['evaluations'][0]['evidence_refs'] = []
|
|
adapter = FinalAdapter(True,change)
|
|
builder = DiagnosisBuilder(store,gateway(adapter))
|
|
result = build(builder)
|
|
assert result.status == 'COMPLETED'
|
|
assert adapter.requests[0]['prompt_version'] == 'diagnosis_v8'
|
|
with PackageReader(result.package) as reader:
|
|
doc=json.loads(reader.read_bytes(result.diagnosis_ref))
|
|
call=json.loads(reader.read_bytes(doc['llm_result_ref']))
|
|
assert 'reason' not in call['outcome']['document']
|
|
assert call['outcome']['document']['evaluations'][0]['evidence_refs'] == []
|
|
assert doc['diagnosis']['evaluations'][0]['evidence_refs'] == doc['diagnosis']['root_cause']['evidence_refs']
|
|
portable=CheckpointStore(Workspace(tmp_path/'lean-portable'));portable.intake(result.package)
|
|
assert build(DiagnosisBuilder(portable,gateway(adapter))).reused
|
|
assert adapter.calls == 1
|
|
|
|
|
|
def test_targeted_feedback_after_unknown_evidence(investigated):
|
|
store,_=investigated
|
|
def change(payload,context):
|
|
payload['root_cause']['evidence_refs']=['E999999']
|
|
adapter=FinalAdapter(True,change);builder=DiagnosisBuilder(store,gateway(adapter))
|
|
assert build(builder).status=='REJECTED'
|
|
adapter.change=None
|
|
assert build(builder).status=='COMPLETED'
|
|
visible=restore(json.loads(adapter.requests[-1]['messages'][1]['content'])['input'])
|
|
assert visible['validation_feedback'][-1]['details']['requirement']=='cite_only_available_evidence_ids'
|
|
|
|
|
|
def test_frozen_v5_request_replays_without_lean_derivation(investigated):
|
|
store,_=investigated
|
|
adapter=FinalAdapter(True)
|
|
builder=DiagnosisBuilder(store,gateway(adapter))
|
|
original=builder._request
|
|
def frozen(context,reader,attempt,**kwargs):
|
|
kwargs.update(compact=True,consolidated=True,selected=True,lean=False)
|
|
return original(context,reader,attempt,**kwargs)
|
|
builder._request=frozen
|
|
result=build(builder)
|
|
assert result.status=='COMPLETED'
|
|
assert adapter.requests[0]['prompt_version']=='diagnosis_v5'
|
|
assert build(DiagnosisBuilder(store,gateway(adapter))).reused
|
|
assert adapter.calls==1
|