295 lines
16 KiB
Python
295 lines
16 KiB
Python
import asyncio
|
|
import copy
|
|
import json
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis import CaseInterpreter,HypothesisConfig
|
|
from viano_bug_analysis.analysis.joint_v2 import VERSION,REPAIR
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.package import PackageReader
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import built_case,payload,gateway
|
|
from test_joint_planning import JointAdapter,hypotheses
|
|
|
|
|
|
class Adapter(JointAdapter):
|
|
def __init__(self, change=None, review=None):
|
|
super().__init__(change);self.review=review
|
|
async def generate(self,config,request,**kwargs):
|
|
if request['prompt_version']!=REPAIR:return await super().generate(config,request,**kwargs)
|
|
self.requests.append(copy.deepcopy(request))
|
|
wire=json.loads(request['messages'][1]['content'])
|
|
from viano_bug_analysis.analysis.json_consolidation import restore
|
|
body=restore(wire['repair'])
|
|
patches=[]
|
|
for issue in body['issues']:
|
|
path=issue['path'];parts=path.strip('/').split('/')
|
|
value=payload()[parts[0]]
|
|
if len(parts)>1:value=value[int(parts[1])]
|
|
patches.append({'path':path,'replacement':copy.deepcopy(value)})
|
|
reviews=[{'position':i,'decision':'retain','reason':'Neither direct inputs nor implicit causal prerequisites rely on the corrected statement.','replacement':None} for i,_ in enumerate((body['hypotheses'] or {}).get('hypotheses',[]))]
|
|
out={'patches':patches,'reviews':reviews}
|
|
if self.review:self.review(out,body)
|
|
return ProviderResponse(out,'fixture',100,50)
|
|
|
|
|
|
def interpret(store,gw):
|
|
from viano_bug_analysis.analysis.correction import CorrectionConfig
|
|
return asyncio.run(CaseInterpreter(store,gw,joint_config=HypothesisConfig(),correction_config=CorrectionConfig()).interpret('case-A','stable-package'))
|
|
|
|
|
|
def test_source_enums_and_explicit_assumptions(built_case):
|
|
adapter=Adapter();store=built_case[0]
|
|
result=interpret(store,gateway(adapter));assert result.status=='COMPLETED'
|
|
req=adapter.requests[0];assert req['prompt_version']==VERSION
|
|
body=json.loads(req['messages'][1]['content']);schema=body['block_contracts']['interpretation']
|
|
obs=schema['properties']['observed_behavior']['items']['properties']['source_ids']['items']['enum']
|
|
assert 'O0001' in obs and 'F0001' not in obs
|
|
assert 'premises' in body['block_contracts']['hypotheses']['$defs']['proposal']['required']
|
|
assert hypotheses(store,gateway(adapter)).status=='COMPLETED'
|
|
assert len(adapter.requests)==1
|
|
|
|
|
|
def test_statement_repair_retains_independent_hypotheses_and_replays(built_case):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
fixed=interpret(store,gw);assert fixed.status=='COMPLETED'
|
|
assert adapter.requests[-1]['prompt_version']==REPAIR
|
|
assert 'input' not in json.loads(adapter.requests[-1]['messages'][1]['content'])
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).reused
|
|
assert interpret(store,gw).reused
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
def test_implicit_uncertainty_locks_other_candidates_in_followup(built_case):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
def review(d,_):d['reviews'][1]['decision']='uncertain'
|
|
store=built_case[0];adapter=Adapter(bad,review);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
result=hypotheses(store,gw);assert result.status=='REJECTED'
|
|
with PackageReader(result.package) as r:
|
|
c=json.loads(r.read_bytes(r.manifest['artifacts']['hypothesis_context']))
|
|
assert set(c['joint_retained']['proposals'])=={'0','2'}
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert adapter.requests[-1]['output_schema']['properties']['hypotheses']['prefixItems'][0].get('const')
|
|
assert len(adapter.requests)==3
|
|
|
|
|
|
def test_changed_direct_input_cannot_be_retained(built_case):
|
|
def bad(d):
|
|
d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
d['hypotheses']['hypotheses'][1]['interpretation_paths']=['/interpretation/observed_behavior/0']
|
|
d['hypotheses']['hypotheses'][1]['premises'][0]['interpretation_paths']=['/interpretation/observed_behavior/0']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).code=='HYPOTHESIS_DEPENDENCY_CHANGED'
|
|
|
|
|
|
def test_unlisted_patch_is_rejected(built_case):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
def review(d,_):d['patches'][0]['path']='/confirmed_technical_facts/0'
|
|
store=built_case[0];adapter=Adapter(bad,review);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
with PackageReader(store.latest('case-A','stable-package')) as r:assert not r.manifest['artifacts'].get('case_interpretation')
|
|
|
|
|
|
def test_rejected_corpus_produces_local_issues_without_modifying_original():
|
|
from pathlib import Path
|
|
from viano_bug_analysis.analysis.joint_v2 import issues
|
|
root=Path('tests/fixtures/llm_rejections/20261019/joint-interpretation-source-kinds')
|
|
context=json.loads((root/'validation-context.json').read_text(encoding='utf-8'))
|
|
context['joint_generation']['version']=2
|
|
data=json.loads((root/'parsed-answer.json').read_text(encoding='utf-8'))
|
|
before=copy.deepcopy(data)
|
|
found=issues(data['interpretation'],context)
|
|
assert {'/observed_behavior/2','/tester_assumptions'}<={i['path'] for i in found}
|
|
assert data==before
|
|
|
|
|
|
def test_direct_dependency_can_be_explicitly_replaced(built_case):
|
|
def bad(d):
|
|
d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
h=d['hypotheses']['hypotheses'][1]
|
|
h['interpretation_paths']=['/interpretation/observed_behavior/0']
|
|
h['premises'][0]['interpretation_paths']=h['interpretation_paths']
|
|
def review(d,body):
|
|
d['reviews'][1].update(decision='replace',replacement={'hypotheses':[copy.deepcopy(body['hypotheses']['hypotheses'][1])]})
|
|
store=built_case[0];adapter=Adapter(bad,review);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
def test_premise_cannot_invent_an_input_reference(built_case):
|
|
def bad(d):d['hypotheses']['hypotheses'][0]['premises'][0]['source_ids']=['F9999']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).code=='HYPOTHESIS_PREMISE_SOURCE'
|
|
with PackageReader(store.latest('case-A','stable-package')) as r:
|
|
assert r.manifest['artifacts']['case_interpretation']
|
|
assert not r.manifest['artifacts'].get('hypotheses')
|
|
|
|
|
|
def test_followup_cannot_rewrite_retained_candidates(built_case):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
def review(d,_):d['reviews'][1]['decision']='uncertain'
|
|
store=built_case[0];adapter=Adapter(bad,review);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED';assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='REJECTED'
|
|
from test_hypothesis_builder import ProposalAdapter
|
|
wrong=ProposalAdapter(change=lambda d,c:d['hypotheses'][0].update(title='Changed retained candidate'))
|
|
result=hypotheses(store,gateway(wrong));assert result.status=='REJECTED'
|
|
with PackageReader(result.package) as r:assert not r.manifest['artifacts'].get('hypotheses')
|
|
|
|
|
|
def test_patch_call_reused_after_interrupted_publication(built_case,monkeypatch):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
original=store.advance
|
|
def fail(*args,**kwargs):
|
|
if kwargs.get('details',{}).get('operation')=='case-interpreted':raise RuntimeError('simulated crash')
|
|
return original(*args,**kwargs)
|
|
monkeypatch.setattr(store,'advance',fail)
|
|
from viano_bug_analysis.analysis.errors import InterpretationError
|
|
with pytest.raises(InterpretationError,match='INTERPRETATION_INPUT'):interpret(store,gw)
|
|
assert len(adapter.requests)==2
|
|
monkeypatch.setattr(store,'advance',original)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
@pytest.mark.parametrize('broken',[None,17,'broken',{'text':'bad','source_ids':17}])
|
|
def test_issue_localization_handles_malformed_model_fields(built_case,broken):
|
|
def bad(d):d['interpretation']['unknowns']=broken
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
result=interpret(store,gw);assert result.status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
|
|
|
|
def test_repaired_plan_reimports_without_new_calls(built_case,tmp_path):
|
|
from viano_bug_analysis.lifecycle import Workspace,CheckpointStore
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED';assert interpret(store,gw).status=='COMPLETED'
|
|
final=hypotheses(store,gw)
|
|
imported=CheckpointStore(Workspace(tmp_path/'recipient'))
|
|
imported.intake(final.package)
|
|
assert interpret(imported,gw).reused
|
|
assert hypotheses(imported,gw).reused
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
def test_budget_origin_validates_the_reviewed_source_chain(built_case):
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED';assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
result=BudgetManager(store).initialize('case-A','stable-package')
|
|
assert len(result.state['accounts'])==3
|
|
assert {v['remaining'] for v in result.state['accounts'].values()}=={100}
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
@pytest.mark.parametrize('malformed',[None,17,[{'position':0}],'invalid'])
|
|
def test_bad_review_does_not_discard_valid_interpretation_patch(built_case,malformed):
|
|
def bad(d):d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
def review(d,_):d['reviews']=malformed
|
|
store=built_case[0];adapter=Adapter(bad,review);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='REJECTED'
|
|
with PackageReader(store.latest('case-A','stable-package')) as r:assert r.manifest['artifacts']['case_interpretation']
|
|
|
|
|
|
def test_valid_premise_dependencies_are_unioned_without_mutating_answer(built_case):
|
|
def missing(d):
|
|
h=d['hypotheses']['hypotheses'][0]
|
|
h['premises'][0]['interpretation_paths'].append('/interpretation/observed_behavior/0')
|
|
h['interpretation_paths']=['/interpretation/unknowns/0']
|
|
store=built_case[0];adapter=Adapter(missing);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert len(adapter.requests)==1
|
|
|
|
|
|
def test_single_invalid_candidate_locks_valid_siblings(built_case):
|
|
def bad(d):d['hypotheses']['hypotheses'][1]['premises'][0]['source_ids']=['F9999']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
result=hypotheses(store,gw);assert result.code=='HYPOTHESIS_PREMISE_SOURCE'
|
|
with PackageReader(result.package) as r:
|
|
c=json.loads(r.read_bytes(r.manifest['artifacts']['hypothesis_context']))
|
|
assert set(c['joint_retained']['proposals'])=={'0','2'}
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
assert len(adapter.requests)==2
|
|
|
|
|
|
def test_real_dependency_corpus_normalization_is_lossless_and_bounded():
|
|
from pathlib import Path
|
|
import hashlib
|
|
from viano_bug_analysis.analysis.joint_v2 import complete_dependencies,validate_premises
|
|
root=Path('tests/fixtures/llm_rejections/20261020/joint-v2-dependencies')
|
|
for name,digest in json.loads((root/'hashes.json').read_text(encoding='utf-8')).items():
|
|
assert hashlib.sha256((root/name).read_bytes()).hexdigest()==digest
|
|
context=json.loads((root/'context.json').read_text(encoding='utf-8'))
|
|
accepted=json.loads((root/'case_interpretation.json').read_text(encoding='utf-8'))['interpretation']
|
|
doc=json.loads((root/'case_plan_v2.json').read_text(encoding='utf-8'))['outcome']['document']
|
|
raw=doc['hypotheses'];before=copy.deepcopy(raw)
|
|
context['joint_generation']['recovery_version']=1
|
|
fixed=complete_dependencies(raw,accepted,context)
|
|
validate_premises(fixed,accepted,context)
|
|
from viano_bug_analysis.analysis.hypothesis_contracts import build_context,validate_proposals
|
|
from viano_bug_analysis.analysis.hypothesis_config import HypothesisLimits
|
|
accepted_doc=json.loads((root/'case_interpretation.json').read_text(encoding='utf-8'))
|
|
hc=build_context(context,accepted_doc,'accepted.json',context['joint_generation']['policy'],[],HypothesisLimits())
|
|
mapping={v['interpretation_pointer']:k for k,v in hc['interpretation_sources'].items()}
|
|
canonical=copy.deepcopy(fixed)
|
|
for proposal in canonical['hypotheses']:
|
|
proposal.pop('premises')
|
|
proposal['interpretation_ids']=[mapping[p] for p in proposal.pop('interpretation_paths')]
|
|
validate_proposals(canonical,hc)
|
|
assert raw==before
|
|
for old,new in zip(raw['hypotheses'],fixed['hypotheses']):
|
|
assert {k:v for k,v in old.items() if k not in {'source_ids','interpretation_paths'}}=={k:v for k,v in new.items() if k not in {'source_ids','interpretation_paths'}}
|
|
assert set(old['interpretation_paths'])<=set(new['interpretation_paths'])
|
|
del context['joint_generation']['recovery_version']
|
|
assert complete_dependencies(raw,accepted,context)==raw
|
|
|
|
|
|
def test_invalid_premise_path_is_not_promoted(built_case):
|
|
def bad(d):d['hypotheses']['hypotheses'][0]['premises'][0]['interpretation_paths']=['/interpretation/unknowns/9999']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).code=='HYPOTHESIS_PREMISE_SOURCE'
|
|
|
|
|
|
def test_malformed_candidate_does_not_discard_siblings(built_case):
|
|
def bad(d):d['hypotheses']['hypotheses'][1].pop('mechanism')
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
result=hypotheses(store,gw);assert result.code=='HYPOTHESIS_SCHEMA'
|
|
with PackageReader(result.package) as r:
|
|
context=json.loads(r.read_bytes(r.manifest['artifacts']['hypothesis_context']))
|
|
assert set(context['joint_retained']['proposals'])=={'0','2'}
|
|
assert hypotheses(store,gw).status=='COMPLETED'
|
|
|
|
|
|
def test_added_dependency_prevents_false_independence_after_patch(built_case):
|
|
def bad(d):
|
|
d['interpretation']['observed_behavior'][0]['source_ids']=['F0001']
|
|
# Explicit premise points at changed statement, missing only from parent list.
|
|
d['hypotheses']['hypotheses'][1]['premises'][0]['interpretation_paths']=['/interpretation/observed_behavior/0']
|
|
store=built_case[0];adapter=Adapter(bad);gw=gateway(adapter)
|
|
assert interpret(store,gw).status=='REJECTED'
|
|
assert interpret(store,gw).status=='COMPLETED'
|
|
assert hypotheses(store,gw).code=='HYPOTHESIS_DEPENDENCY_CHANGED'
|