import copy import json from pathlib import Path import subprocess import pytest from viano_bug_analysis.tools.project_query import ProjectResearch, ProjectLimits from viano_bug_analysis.tools.repository import GitSourceProvider from viano_bug_analysis.tools.sql_database import SQLDatabaseProvider from viano_bug_analysis.tools import InvestigationToolRouter, ToolError from viano_bug_analysis.case import CaseBuilder from viano_bug_analysis.package import PackageReader, PackageValidator from viano_bug_analysis.budget import BudgetManager from viano_bug_analysis.evidence import EvidenceStore from viano_bug_analysis.lifecycle import CheckpointStore, Workspace from test_case_builder import case_setup from test_investigation_tools import tool_case, batch from test_case_interpreter import gateway from test_investigator import LoopAdapter from viano_bug_analysis.analysis import Investigator import asyncio @pytest.fixture def source(tmp_path): root = tmp_path / "repo" root.mkdir() def git(*args): return subprocess.run(["git","-C",str(root),*args],check=True,capture_output=True).stdout.decode().strip() git("init") (root/"module.cpp").write_text("void activate() {}\nactivate();\nactivate_more();\n",encoding="utf-8") (root/"schema.sql").write_text("CREATE TABLE `switches` (\n `id` int NOT NULL\n);\n",encoding="utf-8") git("add",".") git("-c","user.name=Test","-c","user.email=test@example.invalid","commit","-m","fixture") commit = git("rev-parse","HEAD") (root/"module.cpp").write_text("changed worktree\n",encoding="utf-8") return GitSourceProvider(root,commit,"https://example.invalid/repo") def search(kind="source.search",text="activate",**kw): return {"kind":kind,"text":text,"path_prefix":"","case_sensitive":True,"offset":0,"limit":10,**kw} def test_git_commit_not_worktree_and_word_search(source): result=source.query(search("source.symbol")) assert [m["line"] for m in result["items"]]==[1,2] assert result["coverage"]=="complete_selected_text_search" assert "changed" not in str(result) assert source.query(search(text="activate_more"))["items"][0]["line"]==3 def test_git_schema_and_range(source): result=source.query(search("source.schema","CREATE TABLE")) assert result["items"][0]["path"]=="schema.sql" assert source.query({"kind":"source.read","path":"schema.sql","start_line":1,"end_line":2})["items"][1]["line"]==2 def test_git_pagination_and_empty_scope(source): result=source.query(search(limit=1)) assert result["truncated"] and result["next_offset"]==1 assert result["coverage"]=="partial_search" assert source.query(search(text="absent"))["items"]==[] assert source.query(search(text="absent",path_prefix="unknown/"))["coverage"]=="complete_selected_text_search" @pytest.mark.parametrize("path",["../escape","C:/escape","/escape",".git/config","a\\b","a/../b","x\x00y"]) def test_git_paths_rejected(source,path): with pytest.raises(ToolError): source.query({"kind":"source.read","path":path,"start_line":1,"end_line":1}) def test_git_no_model_regex_or_options(source): assert source.query(search(text=".*"))["items"]==[] assert source.query(search(text="--help"))["items"]==[] with pytest.raises(ToolError):source.query(search("source.symbol","a.*")) with pytest.raises(ToolError):source.query(search(text="activate\nother")) class FakeDB: def __init__(self,items=None):self.calls=0;self.items=items or [] def binding(self):return {"kind":"test", "snapshot_id":"snapshot-A"} def validate(self,args):pass def query(self,args): self.calls+=1 return {"items":copy.deepcopy(self.items), "coverage":"complete_filtered_rows", "truncated":False} def rows(**kw): return {"kind":"database.rows","table":"switches","columns":["id"], "filters":[{"column":"id","operator":"eq","value":1}],"offset":0,"limit":5,**kw} @pytest.mark.parametrize("change",[ {"sql":"DROP TABLE switches"},{"table":"switches;DROP TABLE x"},{"limit":True},{"limit":1.0}, {"filters":[]},{"limit":999},{"offset":-1},{"kind":"shell.exec"}, ]) def test_project_invalid_queries_before_access(change): db=FakeDB();s=ProjectResearch({"type":"generic","profile":"test"},database=db) with pytest.raises(ToolError):s.collect({**rows(),**change}) assert db.calls==0 def test_capture_is_frozen_and_limits(): db=FakeDB([{"row":{"id":1}}]);s=ProjectResearch({},database=db) one=s.collect(rows());db.items=[{"row":{"id":2}}] assert s.collect(rows())==one and db.calls==1 assert json.loads(one)["query"]==rows() with pytest.raises(ToolError):ProjectResearch({},database=db,limits=ProjectLimits(max_result_bytes=10)).collect(rows()) def test_capture_cache_bounded_and_invalid_provider_contract(): db=FakeDB();s=ProjectResearch({},database=db,limits=ProjectLimits(max_cached_queries=1)) s.collect(rows());s.collect(rows(offset=1)) assert len(s._captures)==1 class Invalid(FakeDB): def query(self,args):return {"items":[],"coverage":"invented_complete","truncated":False} from viano_bug_analysis.case.errors import CaseBuildError with pytest.raises(CaseBuildError):ProjectResearch({},database=Invalid()).collect(rows()) def test_sql_compiler_parameters_and_wildcards(): sql,values=SQLDatabaseProvider.compile_filters([ {"column":"name","operator":"contains","value":"x%' OR 1=1 --_="}, {"column":"id","operator":"in","value":[1,2]}, {"column":"gone","operator":"eq","value":None}]) assert "OR 1=1" not in sql and "IS NULL" in sql and "IN (%s,%s)" in sql assert values==["%x=%' OR 1==1 --=_==%",1,2] with pytest.raises(ToolError):SQLDatabaseProvider.compile_filters([{"column":"x; DROP TABLE t","operator":"eq","value":1}]) def test_database_allowlist_and_missing_secret(monkeypatch): p=SQLDatabaseProvider("127.0.0.1",33079,"test","reader","snapshot",["switches"],password_env="VBA_TEST_SECRET") with pytest.raises(ToolError):p.query(rows(table="accounts")) monkeypatch.delenv("VBA_TEST_SECRET",raising=False) with pytest.raises(ToolError,match="PROJECT_DATABASE_CREDENTIALS"):p.query(rows()) assert "reader" not in str(p.binding()) and "33079" not in str(p.binding()) def test_external_evidence_is_portable_and_budgeted(tool_case,tmp_path): store=tool_case[0] with PackageReader(tool_case[1]) as reader:project=reader.manifest["project"] db=FakeDB([]);service=ProjectResearch(project,database=db) router=InvestigationToolRouter(project_research=service) budget=BudgetManager(store);budget.initialize("case-A","stable-package") reserved=budget.reserve("case-A","stable-package","H1","A1","project.query") request=batch("project.query",rows());reply=router.query(reserved.package,request) saved=EvidenceStore(store,router=router).store_tool_results("case-A","stable-package","project-query-1",request,reply) assert db.calls==1 assert PackageValidator().validate(saved.package).valid assert CaseBuilder(store).build("case-A","stable-package").reused with PackageReader(saved.package) as reader: capture=reply["results"][0]["data"]["checked_sources"][0]["source_ref"] assert json.loads(reader.read_bytes(capture))["result"]["items"]==[] record=json.loads(reader.read_bytes(saved.record_refs[0])) content=json.loads(reader.read_bytes(record["content_ref"])) assert content["items"][0]["query"]==rows() assert content["items"][0]["result"]["coverage"]=="complete_filtered_rows" budget.record_evidence("case-A","stable-package","H1","A1",saved.record_refs[0]) portable=CheckpointStore(Workspace(tmp_path/"portable"));portable.intake(store.latest("case-A","stable-package")) assert EvidenceStore(portable).snapshot("case-A","stable-package").record_refs==saved.record_refs def test_case_builder_shared_provider_resolves_names(case_setup,source): store=case_setup[0] with PackageReader(store.latest("case-A","stable-package")) as reader:project=reader.manifest["project"] db=FakeDB([{"table":"switches","row":{"id":1,"name":"Correct entity name"}}]) s=ProjectResearch(project,database=db,source=source) s.enrichment={"queries":[rows(),search("source.schema","CREATE TABLE")]} result=CaseBuilder(store,project_research=s).build("case-A","stable-package") with PackageReader(result.package) as reader: case=json.loads(reader.read_bytes(result.case_ref)) assert "Correct entity name" in str(case["runtime_facts"]) assert any(ref["path"].startswith("research/project/") for ref in case["database_context"]["refs"]) assert "schema.sql" in str(case["runtime_facts"]) assert CaseBuilder(store,project_research=s).build("case-A","stable-package").reused assert db.calls==1 def test_case_builder_provider_outage_blocks_without_sealing(case_setup): store=case_setup[0] with PackageReader(store.latest("case-A","stable-package")) as reader:project=reader.manifest["project"] class Offline(FakeDB): def query(self,args):raise ToolError("PROJECT_DATABASE_UNAVAILABLE") s=ProjectResearch(project,database=Offline());s.enrichment={"queries":[rows()]} result=CaseBuilder(store,project_research=s).build("case-A","stable-package") assert result.status=="BLOCKED" and result.code=="PROJECT_DATABASE_UNAVAILABLE" and result.case_ref is None with PackageReader(result.package) as reader: assert reader.manifest["state"]=="BLOCKED" and reader.manifest["artifacts"].get("case") is None @pytest.mark.parametrize("recall", [False, True, "unknown", "upgrade"]) def test_model_loop_uses_project_tools_and_compact_previous_result(tool_case, recall): store=tool_case[0] with PackageReader(tool_case[1]) as reader:project=reader.manifest["project"] service=ProjectResearch(project,database=FakeDB([{"row":{"id":1}}])) class Adapter(LoopAdapter): async def generate(self,config,request,*,max_response_bytes): from viano_bug_analysis.llm import ProviderResponse context=json.loads(request["messages"][1]["content"])["input"] self.contexts.append(context) previous=context["previous_result"] eid=previous["evidence_id"] if previous else None if previous: assert context["evidence"][eid]["content"]["items"] assert "inputs" not in context["case"] assert request["prompt_version"] in ({"investigation_project_v9","investigation_project_v14"} if recall == "upgrade" else {"investigation_project_v9" if recall else "investigation_project_v14"}) reply={"assessment":{"semantic_score":50,"reason":"Scoped fact","evidence_refs":[eid]} if previous else None, "update":{"confidence":"plausible","evidence_refs":[eid] if previous else [],"resolve_questions":[],"add_questions":[],"note":"Query typed identity"}, "next":{"kind":"finish","reason":"Check complete"} if previous else {"kind":"tool","operation":"project.query","arguments":rows()}} if recall: if previous and request['prompt_version']=='investigation_project_v9': reply['next'] = {'kind':'recall','evidence_refs':['E999999' if recall == 'unknown' else eid]} elif not previous and any(e['record']['hypothesis_id'] == context['selected_hypothesis'] for e in context.get('evidence', {}).values()): recalled = next(eid for eid,e in context['evidence'].items() if e['record']['hypothesis_id'] == context['selected_hypothesis']) assert context['evidence'][recalled]['content']['items'] reply['update']['evidence_refs'] = [recalled] reply['next'] = {'kind':'finish','reason':'Recalled exact evidence; uncertainty remains'} if request['prompt_version']=='investigation_project_v14': reply['update']['causal_review']={'effect':'neutral','support':[],'against':[], 'reason':'Scoped observations are not independent evidence of a defect.'} if request['prompt_version']=='investigation_project_v14': reply['update']['working_brief']={'format_version':1,'findings':[], 'uncertainties':['Scoped observations only.'],'next_question':'Check remaining uncertainty.','changes':[]} if request['prompt_version']=='investigation_project_v14': b=reply['update']['working_brief'];b['format_version']=2;b.pop('changes');b['retired']=[] for k in ('evidence_refs','causal_review','note'):reply['update'].pop(k,None) if reply['assessment']:reply['assessment'].pop('evidence_refs',None) return ProviderResponse(reply,"fixture",100,10) adapter=Adapter() loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) if recall: original_context=loop._context def frozen_context(*args): context=original_context(*args) context['prompt_version']='investigation_project_v9' context['recalled_evidence']=args[0].get('recalled_evidence',{}).get(args[1],[]) return context loop._context=frozen_context if recall is True or recall == "upgrade": partial=asyncio.run(loop.run("case-A","stable-package",max_steps=4)) assert partial.status == 'YIELDED' loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) original_context=loop._context def frozen_resume_context(*args): context=original_context(*args) context['prompt_version']='investigation_project_v9' context['recalled_evidence']=args[0].get('recalled_evidence',{}).get(args[1],[]) return context if recall is True:loop._context=frozen_resume_context result=asyncio.run(loop.run("case-A","stable-package",max_steps=20)) if recall == 'unknown': assert result.status == 'BLOCKED' and result.code == 'INVESTIGATION_EVIDENCE_UNKNOWN' else: assert result.status=="COMPLETED" and len(adapter.contexts)==(7 if recall == "upgrade" else 9 if recall else 6) assert len(EvidenceStore(store).snapshot("case-A","stable-package").record_refs)==3 accounts=BudgetManager(store).snapshot("case-A","stable-package").state['accounts'] assert sum(len(a['actions']) for a in accounts.values()) == 3 if recall == 'unknown': assert any(a['status'] != 'settled' for a in accounts['H1']['actions'].values()) else: assert all(a['status'] == 'settled' for account in accounts.values() for a in account['actions'].values()) def test_provider_outage_blocks_without_ending_hypothesis(tool_case): store=tool_case[0] with PackageReader(tool_case[1]) as reader:project=reader.manifest["project"] class Offline(FakeDB): def query(self,args):raise ToolError("PROJECT_DATABASE_UNAVAILABLE") service=ProjectResearch(project,database=Offline()) def change(reply,context):reply["next"]={"kind":"tool","operation":"project.query","arguments":rows()} from datetime import datetime, timezone adapter=LoopAdapter(change) loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service), clock=lambda:datetime(2000,1,1,tzinfo=timezone.utc)) result=asyncio.run(loop.step("case-A","stable-package")) assert result.status=="BLOCKED" and result.code=="PROJECT_DATABASE_UNAVAILABLE" with PackageReader(result.package) as reader: state=json.loads(reader.read_bytes(result.state_ref)) assert all(h["disposition"]=="active" for h in state["hypotheses"].values()) assert state["active"]["phase"]=="executing" assert not BudgetManager(store).snapshot("case-A","stable-package").ready_for_evaluation before=BudgetManager(store).snapshot("case-A","stable-package").state store.advance("case-A","stable-package","INVESTIGATING_HYPOTHESES",step_id="operator-resume-1") again=asyncio.run(loop.step("case-A","stable-package")) assert again.status=="BLOCKED" and len(adapter.contexts)==1 assert BudgetManager(store).snapshot("case-A","stable-package").state==before service.database=FakeDB([{"row":{"id":1}}]) store.advance("case-A","stable-package","INVESTIGATING_HYPOTHESES",step_id="operator-resume-2") recovered=asyncio.run(loop.step("case-A","stable-package")) assert recovered.status=="PROGRESSED" and len(adapter.contexts)==1 def test_started_project_loop_keeps_legacy_prompt_protocol(tool_case): from viano_bug_analysis.llm import ProviderResponse store=tool_case[0] with PackageReader(tool_case[1]) as reader: project=reader.manifest['project'] service=ProjectResearch(project,database=FakeDB([{'row':{'id':1}}])) class Adapter: calls=0 async def generate(self,config,request,**kwargs): self.calls+=1 assert request['prompt_version']=='investigation_project_v1' return ProviderResponse({'assessment':None,'update':{'confidence':'unassessed', 'evidence_refs':[],'resolve_questions':[],'add_questions':[],'note':'Read-only query'}, 'next':{'kind':'tool','operation':'project.query','arguments':rows()}},'fixture',100,10) adapter=Adapter() loop=Investigator(store,gateway(adapter),router=InvestigationToolRouter(project_research=service)) loop._budget('case-A','stable-package') with PackageReader(store.latest('case-A','stable-package')) as reader: initial=loop._begin(reader) initial.pop('prompt_protocol') initial.pop('early_stop_protocol') context=loop._context(initial,'H1',loop._budget('case-A','stable-package')) initial['hypotheses']['H1']['iterations']=1 initial['active']={'hypothesis_id':'H1','iteration':1,'attempt':0,'phase':'prompt', 'context_ref':'analysis/investigation/0001/contexts/H1-0001.json', 'context_sha256':'0'*64,'decision':None,'llm_ref':None,'test_request':None, 'test_deadline':None,'retry_token':'initial'} loop._save(initial,context=context) result=asyncio.run(loop.step('case-A','stable-package')) assert result.status=='PROGRESSED' and adapter.calls==1 assert PackageValidator().validate(result.package).valid