import asyncio import copy import json import pytest from jsonschema import Draft202012Validator from viano_bug_analysis.analysis.investigation_decision import normalize from viano_bug_analysis.analysis.confidence_review import effective_confidence from viano_bug_analysis.analysis.investigation_prompt import output_schema from viano_bug_analysis.analysis.working_brief import model_view from viano_bug_analysis.analysis import Investigator, InvestigationError, DiagnosisBuilder from viano_bug_analysis.budget import BudgetManager from viano_bug_analysis.case.contracts import validate from viano_bug_analysis.llm import LLMError, ProviderResponse from viano_bug_analysis.llm.diagnostics import sanitize, schema_feedback from viano_bug_analysis.package import PackageReader from viano_bug_analysis.tools import InvestigationToolRouter from test_investigation_consolidation import populated from test_working_brief import brief, service_for from test_case_builder import case_setup from test_case_interpreter import gateway from test_investigation_tools import tool_case from test_investigator import LoopAdapter, CASE, run, state from test_project_research import rows from test_diagnosis_builder import FinalAdapter, build def context(): return {'prompt_version':'investigation_project_v14', 'evidence':{'E000001':{},'E000002':{}}, 'previous_result':{'record':{'id':'E000002'}},'working_brief':{'brief':brief()}} def reply(): b=brief();b['format_version']=2;b.pop('changes');b['retired']=[] return {'assessment':{'semantic_score':90,'reason':'Useful counterevidence'}, 'update':{'confidence':'strongly_supported','resolve_questions':[], 'add_questions':[],'working_brief':b}, 'next':{'kind':'finish','reason':'Mechanism contradicted'}} def test_one_findings_set_derives_refs_direction_and_assessment_binding_without_mutation(): c=context();raw=reply();before=copy.deepcopy(raw) d=normalize(c,raw) assert raw==before assert d['assessment']['evidence_refs']==['E000002'] assert d['update']['evidence_refs']==['E000001','E000002'] assert d['update']['causal_review']['effect']=='contradicts' assert effective_confidence(d['update'])=='weakened' assert d['assessment']['semantic_score']==90 assert d['update']['working_brief']['format_version']==1 validate(d['update']['working_brief'],'investigation-working-brief') def test_derived_union_keeps_all_128_refs_without_a_second_model_list_limit(): c={'prompt_version':'investigation_project_v14','evidence':{f'E{i:06d}':{} for i in range(1,129)},'previous_result':None} r=reply();r['assessment']=None r['update']['working_brief']['findings']=[{ 'id':f'B{i+1:04d}','direction':'support','text':'Scoped finding.', 'scope':'Snapshot A', 'evidence_refs':[f'E{j:06d}' for j in range(i*8+1,i*8+9)]} for i in range(16)] d=normalize(c,r) assert len(d['update']['causal_review']['support'])==128 assert len(d['update']['causal_review']['reason'])<=600 from importlib.resources import files schema=json.loads(files('viano_bug_analysis.schemas').joinpath('investigation-state-v1.schema.json').read_text()) note={'kind':'assessment','text':'Controller-derived brief','llm_ref':'call.json', 'causal_review':d['update']['causal_review']} assert Draft202012Validator(schema['properties']['hypotheses']['additionalProperties']['properties']['notes']['items']).is_valid(note) def test_text_scope_precision_and_added_refs_automatically_audited(): r=reply();f=r['update']['working_brief']['findings'][0] f.update(text='The required hook exists in the inspected source.',scope='Commit A, not deployment.',evidence_refs=['E000001','E000002']) result=normalize(context(),r) audit=result['brief_change_audit'][0] assert not audit['requires_reason'] assert set(audit['fields'])=={'text','scope','evidence_refs'} assert result['update']['working_brief']['changes'][0]['reason'].startswith('Controller') @pytest.mark.parametrize('change',[ lambda f:f.update(direction='neutral'),lambda f:f.update(evidence_refs=['E000002'])]) def test_direction_change_or_lost_counterevidence_ref_requires_explicit_reason(change): r=reply();f=r['update']['working_brief']['findings'][0];change(f) with pytest.raises(InvestigationError,match='INVESTIGATION_BRIEF_CHANGE') as caught: normalize(context(),r) assert caught.value.feedback['ids']==['B0001'] f['revision_reason']='A newer observation changes the applicable scope.' assert normalize(context(),r)['brief_change_audit'][0]['requires_reason'] def test_removed_counterevidence_requires_retirement_and_keeps_old_references(): r=reply();b=r['update']['working_brief'];b['findings']=[] with pytest.raises(InvestigationError,match='INVESTIGATION_BRIEF_CHANGE'): normalize(context(),r) b['retired']=[{'id':'B0001','reason':'Observation belongs to an obsolete deployment.'}] d=normalize(context(),r) assert d['update']['working_brief']['changes'][0]['evidence_refs']==['E000001'] assert d['update']['causal_review']['effect']=='neutral' assert effective_confidence(d['update'])=='plausible' def test_unknown_reference_has_concrete_safe_feedback(): r=reply();r['update']['working_brief']['findings'][0]['evidence_refs']=['E999999'] with pytest.raises(InvestigationError) as caught:normalize(context(),r) assert caught.value.feedback=={'field':'update.working_brief.findings','ids':['E999999'], 'requirement':'known_evidence_ids'} def test_both_sides_produce_mixed_effect_and_cannot_force_high_confidence(): r=reply();f=copy.deepcopy(r['update']['working_brief']['findings'][0]);f.update(id='B0002',direction='support') r['update']['working_brief']['findings'].append(f) d=normalize(context(),r) assert d['update']['causal_review']['effect']=='mixed' assert effective_confidence(d['update'])=='plausible' assert d['update']['evidence_refs']==['E000001','E000002'] def test_wire_does_not_accept_redundant_fields_and_old_contract_is_frozen(): c=populated();c['prompt_version']='investigation_project_v14' schema=output_schema(c);r=reply() assert Draft202012Validator(schema).is_valid(r) r['update']['evidence_refs']=['E000001'] assert not Draft202012Validator(schema).is_valid(r) old=output_schema(dict(c,prompt_version='investigation_project_v13'))['properties']['update'] assert 'evidence_refs' in old['required'] and 'causal_review' in old['required'] def test_v14_model_view_keeps_all_old_foreign_and_fresh_evidence_payloads(): c=populated() for h in c['hypotheses'].values():h['disposition']='active' c['prompt_version']='investigation_project_v13';old=model_view(c) c['prompt_version']='investigation_project_v14';new=model_view(c) for eid in old['evidence']: assert new['evidence'][eid]['content']==old['evidence'][eid]['content'] for key in ('shared_sources','shared_results','sources','interpretation'): assert new.get(key)==old.get(key) assert new['previous_result']==old['previous_result'] def test_schema_feedback_never_echoes_response_values_or_unknown_property_names(): schema={'type':'object','properties':{'update':{'type':'object', 'properties':{'confidence':{'type':'integer'}},'required':['confidence']}}} err=next(Draft202012Validator(schema).iter_errors({'update':{}})) assert schema_feedback(err)=={'validation_path':'/update','validation_fields':['confidence']} assert sanitize({'validation_path':'/update/secret_key_content','validation_fields':['secret_value','confidence']})=={'validation_fields':['confidence']} def test_schema_feedback_selects_matching_action_instead_of_wrong_oneof_branch(): schema={'type':'object','properties':{'next':{'oneOf':[ {'type':'object','properties':{'kind':{'const':'finish'},'reason':{'type':'string'}},'required':['kind','reason']}, {'type':'object','properties':{'kind':{'const':'tool'},'arguments':{'type':'object', 'properties':{'start_line':{'type':'integer'}},'required':['start_line']}},'required':['kind','arguments']}]}}} err=next(Draft202012Validator(schema).iter_errors({'next':{'kind':'tool','arguments':{}}})) info=schema_feedback(err) assert info=={'validation_path':'/next/arguments','validation_fields':['start_line']} def test_controller_rejection_details_reach_retry_before_any_budget_effect(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['format_version']=2;b.pop('changes');b['retired']=[] 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 BudgetManager(tool_case[0]).snapshot(*CASE).state==before saved=state(result) with PackageReader(result.package) as r: c=json.loads(r.read_bytes(saved['active']['context_ref'])) assert c['previous_rejection_details']['ids']==['E999999'] def test_gateway_schema_location_persists_in_retry_without_leaking_values(tool_case): service=service_for(tool_case) class Bad: async def generate(self,config,request,*,max_response_bytes): r=reply();r['update']['working_brief'].pop('retired') if request['prompt_version']=='investigation_project_v18':r['update']['next_step']=None return ProviderResponse(r,'fixture',10,10) loop=Investigator(tool_case[0],gateway(Bad()),router=InvestigationToolRouter(project_research=service)) result=asyncio.run(loop.step(*CASE));assert result.status=='REJECTED' with PackageReader(result.package) as reader: c=json.loads(reader.read_bytes(state(result)['active']['context_ref'])) info=c['previous_rejection_details'] assert info['validation_path']=='/update/working_brief' and info['validation_fields']==['retired'] def test_oversized_legacy_v14_request_is_blocked_without_spending_budget(tool_case): # The old verbose v14 contract exceeds the new universal character limit. # Compatibility does not authorize sending an oversized request. loop=Investigator(tool_case[0],gateway(LoopAdapter()), router=InvestigationToolRouter(project_research=service_for(tool_case))) original=loop._context loop._context=lambda *args:dict(original(*args),prompt_version='investigation_project_v14') done=run(loop,steps=10) assert done.status=='BLOCKED' and done.code=='INVESTIGATION_PROMPT_LIMIT' budget=BudgetManager(tool_case[0]).snapshot(*CASE).state assert all(a['remaining']==a['initial'] and not a['actions'] for a in budget['accounts'].values()) def test_v14_can_finish_inconclusively_without_invented_cause(tool_case): service=service_for(tool_case) loop=Investigator(tool_case[0],gateway(LoopAdapter()),router=InvestigationToolRouter(project_research=service)) done=run(loop,steps=10);assert done.status=='COMPLETED' result=build(DiagnosisBuilder(tool_case[0],gateway(FinalAdapter(False)))) assert result.status=='INCONCLUSIVE' with PackageReader(result.package) as reader: assert reader.manifest['state']=='FAILED_TO_ANALYZE' assert reader.manifest['artifacts']['change_request'] is None