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']