337 lines
19 KiB
Python
337 lines
19 KiB
Python
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_v15"} if recall == "upgrade" else {"investigation_project_v9" if recall else "investigation_project_v15"})
|
|
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_v15':
|
|
reply['update']['causal_review']={'effect':'neutral','support':[],'against':[],
|
|
'reason':'Scoped observations are not independent evidence of a defect.'}
|
|
if request['prompt_version']=='investigation_project_v15':
|
|
reply['update']['working_brief']={'format_version':1,'findings':[],
|
|
'uncertainties':['Scoped observations only.'],'next_question':'Check remaining uncertainty.','changes':[]}
|
|
if request['prompt_version']=='investigation_project_v15':
|
|
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 not recall:
|
|
original_context=loop._context
|
|
def frozen_v15_context(*args):
|
|
context=original_context(*args)
|
|
context['prompt_version']='investigation_project_v15'
|
|
return context
|
|
loop._context=frozen_v15_context
|
|
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
|
|
else:
|
|
def frozen_v15_resume(*args):
|
|
context=original_context(*args)
|
|
context['prompt_version']='investigation_project_v15'
|
|
return context
|
|
loop._context=frozen_v15_resume
|
|
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
|