import copy import json from pathlib import Path import subprocess from types import SimpleNamespace import pytest from viano_bug_analysis.tools import ToolError from viano_bug_analysis.tools.project_query import ProjectResearch from viano_bug_analysis.tools.repository import GitSourceProvider from viano_bug_analysis.tools.source_relevance import case_seeds from test_case_builder import case_setup from test_investigation_tools import tool_case @pytest.fixture def source(tmp_path): root=tmp_path/'repo' for d in ['src','sql/base','sql/old']: (root/d).mkdir(parents=True,exist_ok=True) (root/'src/a_noise.cpp').write_text('\n'.join('SPELL_NOISE_'+str(i)+' = 77;' for i in range(20))+'\nvoid AddSC_noise() {}\n',encoding='utf-8') (root/'src/target.cpp').write_text('enum { NPC_TARGET = 77 };\nvoid npc_target() { summon(NPC_TARGET); }\nvoid AddSC_target() { npc_target(); }\n',encoding='utf-8') (root/'src/loader.cpp').write_text('void AddSC_target();\nvoid load() { AddSC_target(); }\nvoid AddSC_unrelated();\n',encoding='utf-8') (root/'src/coordinates.cpp').write_text('float x = 77.123;\nint larger = 177;\nint random = 88;\n',encoding='utf-8') (root/'src/other_area.cpp').write_text('void npc_cross_zone() {}\n',encoding='utf-8') (root/'src/zone.cpp').write_text('void npc_local() {}\n',encoding='utf-8') (root/'sql/base/world.sql').write_text('CREATE TABLE target (id int);\n',encoding='utf-8') (root/'sql/old/noise.sql').write_text('INSERT INTO spell VALUES (77);\n',encoding='utf-8') def git(*args):return subprocess.run(['git','-C',str(root),*args],check=True,capture_output=True).stdout.decode().strip() git('init');git('add','.');git('-c','user.name=Test','-c','user.email=test@example.invalid','commit','-m','fixture') return GitSourceProvider(root,git('rev-parse','HEAD'),'https://example.invalid/repo') def service(source,**settings): return ProjectResearch({},source=source,source_analysis={'protocol':2,'max_searches':8, 'code_prefixes':['src/'],'schema_prefixes':['sql/base/'],'historical_prefixes':['sql/old/'], 'registration_prefixes':['AddSC_'],'type_markers':{'actor':['NPC'],'effect':['SPELL']},**settings}) def query(text='77',**extra):return {'kind':'source.context','text':text,'path_prefix':'',**extra} def seeds(*anchors):return {'anchors':list(anchors),'omitted_anchors':0} def anchor(term,kind='id',group='N0001:id',**extra): return {'term':term,'kind':kind,'group':group,'source_ref':'case/case.json', 'json_pointer':'/entities/0','entity_type':'actor',**extra} def test_typed_collision_not_frequency_selects_implementation(source): r=json.loads(service(source).collect(query(),source_context=seeds(anchor('77'))))['result'] items={i['path']:i for i in r['items']} assert items['src/target.cpp']['tier']=='primary_candidate' assert items['src/a_noise.cpp']['tier']=='weak_numeric' and items['src/a_noise.cpp']['blocks']==[] assert 'src/coordinates.cpp' not in items assert all('different_or_ambiguous_identifier_namespace' in [s['reason'] for s in h['signals']] for h in items['src/a_noise.cpp']['hits']) assert 'AddSC_noise' not in [s['query']['text'] for s in r['searches']] assert {'NPC_TARGET','AddSC_target'} <= {s['query']['text'] for s in r['searches']} def test_untyped_numbers_do_not_promote_or_expand_even_when_only_hits(source): r=json.loads(service(source).collect(query()))['result'] assert r['selection']['primary_files']==[] assert all(i['tier']=='weak_numeric' and not i['blocks'] for i in r['items']) assert len(r['searches'])==1 assert len([h for i in r['items'] for h in i['hits']])<=8 def test_names_variants_are_not_independent_corroboration(source): r=json.loads(service(source).collect(query('77'),source_context=seeds( anchor('npc_target','name','same_name'),anchor('NPC_TARGET','name','same_name'))))['result'] item=next(i for i in r['items'] if i['path']=='src/target.cpp') assert item['tier']=='supporting_context' and item['ranking_score']==25 r=json.loads(service(source).collect(query('npc_target'),source_context=seeds( anchor('npc_target','name','same_name'))))['result'] assert r['selection']['primary_files']==[] def test_known_other_area_remains_usable_context(source): r=json.loads(service(source,location_paths={'Home':['src/zone.cpp'],'Elsewhere':['src/other_area.cpp']}).collect( query('npc_local'),source_context=seeds(anchor('npc_cross_zone','script','script'), anchor('Home','context','place'))))['result'] items={i['path']:i for i in r['items']} assert items['src/other_area.cpp']['tier']=='supporting_other_location' assert items['src/other_area.cpp']['blocks'] assert 'src/other_area.cpp' in r['selection']['supporting_files'] def test_scopes_do_not_silently_mix_code_schema_and_history(source): s=service(source) assert not any(i['path'].endswith('.sql') for i in json.loads(s.collect(query()))['result']['items']) r=json.loads(s.collect(query('CREATE TABLE',scope='schema')))['result'] assert [i['path'] for i in r['items']]==['sql/base/world.sql'] r=json.loads(s.collect(query(scope='historical_sql')))['result'] assert [i['path'] for i in r['items']]==['sql/old/noise.sql'] with pytest.raises(ToolError,match='SCOPE_CONFLICT'):s.collect(query(path_prefix='sql/old/')) with pytest.raises(ToolError):s.collect(query(scope='all')) def test_new_args_require_a_case_and_do_not_accept_model_fabricated_anchors(source): s=service(source) with pytest.raises(ToolError,match='CONTEXT_REQUIRED'):s.collect(query(entity_refs=['N0001'])) with pytest.raises(ToolError):s.collect({**query(),'anchors':[anchor('npc_fake','script')]}) assert ProjectResearch({},source=source,source_analysis={}).source_context.catalog()['version']==1 with pytest.raises(ToolError):ProjectResearch({},source=source,source_analysis={'code_prefixes':['src/']}) def test_selection_limits_are_explicit_and_source_worktree_is_ignored(source): (source.path/'src/target.cpp').write_text('wrong mutable worktree\n',encoding='utf-8') r=json.loads(service(source,max_total_lines=1).collect(query(),source_context=seeds(anchor('77'))))['result'] assert r['truncated'] and 'line_limit' in r['limits_reached'] assert any(i.get('omitted_context_lines',0)>0 for i in r['items']) assert 'wrong mutable' not in json.dumps(r) @pytest.mark.parametrize('settings',[{'type_markers':{'actor':['.*']}},{'protocol':True}, {'code_prefixes':['../escape']},{'max_seed_terms':0},{'location_paths':{'Home':['C:/bad']}}]) def test_invalid_adapter_rules_fail_controlled(source,settings): with pytest.raises(ToolError):service(source,**settings) def test_case_collector_names_roles_scripts_and_provenance(source,monkeypatch): import viano_bug_analysis.case.contracts as contracts sources={'N0001':{'kind':'entity','data':{'type':'actor','id':'77','name':'Target actor','roles':['affected_subject']}}, 'N0002':{'kind':'entity','data':{'type':'effect','id':'88','name':'Related quest','roles':['selected_focus']}}} captured={'query':{'kind':'database.rows','table':'actors'},'result':{'items':[{'row':{'entry':77,'ScriptName':'npc_target'}}]}} fact={'id':'F0001','name':'project_query','value':{'capture_ref':'research/capture.json',**captured}} case={'runtime_facts':{'facts':[fact,{'id':'F0002','name':'location_name','value':'Home'}]}} ctx=SimpleNamespace(case=case,case_ref='case/case.json',reader=SimpleNamespace( manifest={'artifacts':{'interpretation_context':'case/context.json'}},files=['research/capture.json'])) monkeypatch.setattr(contracts,'read_json',lambda r,ref,*args: {'sources':sources} if ref=='case/context.json' else captured) s=service(source,focus_roles=['selected_focus'],context_fact_names=['location_name'],entity_tables=[ {'table':'actors','entity_type':'actor','id_column':'entry','script_columns':['ScriptName']}]) c=case_seeds(ctx,query(),s.source_context.config) assert c['entity_refs']==['N0001','N0002'] assert any(a['term']=='npc_target' and a['source_ref']=='research/capture.json' for a in c['anchors']) assert any(a['term']=='Home' for a in c['anchors']) assert all(a['source_ref'] and a['json_pointer'] for a in c['anchors']) with pytest.raises(ToolError,match='ENTITY_REF'):case_seeds(ctx,query(entity_refs=['N9999']),s.source_context.config) def test_ambiguous_id_does_not_inherit_a_namespace(source,monkeypatch): import viano_bug_analysis.case.contracts as contracts sources={sid:{'kind':'entity','data':{'type':typ,'id':'77','roles':[]}} for sid,typ in [('N0001','actor'),('N0002','effect')]} monkeypatch.setattr(contracts,'read_json',lambda *args:{'sources':sources}) ctx=SimpleNamespace(case={},case_ref='case/case.json',reader=SimpleNamespace( manifest={'artifacts':{'interpretation_context':'case/context.json'}})) s=service(source);c=case_seeds(ctx,query(),s.source_context.config) assert c['ambiguous_identity'] and all(a.get('entity_type') is None for a in c['anchors']) r=json.loads(s.collect(query(),source_context=c))['result'] assert r['selection']['primary_files']==[] def test_v2_fixture_loop_captures_case_seeds_and_resumes(source,tool_case,tmp_path): from test_case_interpreter import gateway from test_investigation_hybrid import HybridAdapter from test_investigator import run from viano_bug_analysis.analysis import Investigator,InvestigationConfig from viano_bug_analysis.tools import InvestigationToolRouter from viano_bug_analysis.package import PackageReader,PackageValidator from viano_bug_analysis.lifecycle import Workspace,CheckpointStore s=service(source) with PackageReader(tool_case[1]) as reader:s.project=reader.manifest['project'] adapter=HybridAdapter();adapter.bootstrap.change=lambda r,c:r.update(next={'kind':'tool','operation':'project.query','arguments':query('npc_target')}) router=InvestigationToolRouter(project_research=s) first=run(Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,lean_checkpoint_steps=0),router=router),steps=3) portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package) final=run(Investigator(portable,gateway(adapter),router=router)) assert final.status=='COMPLETED' and PackageValidator().validate(final.package).valid from viano_bug_analysis.llm.cache_measurement import measure_package assert measure_package(final.package)['all_exactly_reconstructed']