123 lines
6.9 KiB
Python
123 lines
6.9 KiB
Python
import copy
|
|
import json
|
|
from dataclasses import replace
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
from viano_bug_analysis.analysis.database_value_grounding import unsupported_assignments
|
|
from viano_bug_analysis.analysis.change_request_contracts import validate_request,ChangeRequestError
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
from viano_bug_analysis.llm import LLMError
|
|
from test_investigation_lean import lean_context
|
|
from test_case_builder import case_setup
|
|
from test_investigation_tools import tool_case
|
|
from test_llm_caching import config,native,scripted
|
|
from test_lean_investigation_decision import reply
|
|
|
|
|
|
def rows_context():
|
|
return {'evidence':{'E000001':{'content':{'items':[{'query':{'kind':'database.rows','table':'objects'},
|
|
'result':{'items':[{'row':{'phaseMask':1,'zoneId':14,'map':1}},{'row':{'map':0,'zoneId':1}}]}}]}}},'case':{}}
|
|
|
|
|
|
@pytest.mark.parametrize('text', ['Insert phaseMask = 14.','Set phaseMask=14, map=0.', 'Use phaseMask 14.'])
|
|
def test_value_from_another_column_cannot_become_phase_mask(text):
|
|
assert unsupported_assignments({'implementation_guidance':[text]},rows_context())==['phasemask']
|
|
|
|
|
|
def test_known_value_placeholder_and_source_only_are_not_rejected():
|
|
c=rows_context()
|
|
assert unsupported_assignments({'implementation_guidance':['Use phaseMask=1, map=0; zoneId=1.']},c)==[]
|
|
assert unsupported_assignments({'implementation_guidance':['Use phaseMask={{PHASE_MASK}} after human verification.']},c)==[]
|
|
assert unsupported_assignments({'implementation_guidance':['Repair source activation.']},{'evidence':{},'case':{}})==[]
|
|
c['evidence']['E000001']['content']['items'][0]['result']['items'][0]['row']['phaseMask']=True
|
|
assert unsupported_assignments({'implementation_guidance':['Use phaseMask=14.']},c)==[] # no observed numeric column; guard makes no broader claim
|
|
|
|
|
|
def bootstrap():
|
|
c=lean_context();c.pop('lean_session');c['bootstrap_json_framing']=True
|
|
return build_prompt(c)
|
|
|
|
|
|
def test_new_bootstrap_native_guard_does_not_change_full_local_contract(monkeypatch):
|
|
request=bootstrap();assert request['prompt_version']=='investigation_bootstrap_v21'
|
|
r=reply();r['update']['next_step']=None
|
|
assert Draft202012Validator(request['output_schema']).is_valid(r)
|
|
env=native(cached=0);env['candidates'][0]['content']['parts'][0]['text']=json.dumps(r)
|
|
adapter=GeminiProvider();seen=scripted(adapter,monkeypatch,[env])
|
|
response=adapter._generate(replace(config(),context_cache=False,response_format='json_object'),request,1000000)
|
|
assert response.document==r
|
|
frame=seen[0][1]['generationConfig']['responseJsonSchema']
|
|
assert frame['required']==['assessment','update','next'] and len(json.dumps(frame))<1000
|
|
assert Draft202012Validator(frame).is_valid(r)
|
|
bad=copy.deepcopy(r);bad['next']={'kind':'shell','command':'bad'}
|
|
assert not Draft202012Validator(request['output_schema']).is_valid(bad)
|
|
assert 'oneOf' not in json.dumps(frame)
|
|
|
|
|
|
def test_old_bootstrap_transport_stays_json_mode(monkeypatch):
|
|
request=bootstrap();request['prompt_version']='investigation_project_v18'
|
|
adapter=GeminiProvider();seen=scripted(adapter,monkeypatch,[native(cached=0)])
|
|
adapter._generate(replace(config(),context_cache=False,response_format='json_object'),request,1000000)
|
|
assert 'responseJsonSchema' not in seen[0][1]['generationConfig']
|
|
|
|
|
|
|
|
def test_unknown_numeric_schema_field_requires_placeholder():
|
|
c={'evidence':{'E000001':{'content':{'items':[{'query':{'kind':'database.schema'},
|
|
'result':{'items':[{'columns':[{'COLUMN_NAME':'custom_flag','COLUMN_TYPE':'int(11)'}]}]}}]}}}}
|
|
assert unsupported_assignments({'implementation_guidance':['Set custom_flag=14.']},c)==['custom_flag']
|
|
|
|
|
|
def test_bad_json_diagnostics_are_specific_without_echoing_content():
|
|
from viano_bug_analysis.llm.diagnostics import parse_response_object
|
|
with pytest.raises(LLMError) as caught:parse_response_object(b'{"private_value":"secret"',stage='output_json')
|
|
d=caught.value.diagnostics
|
|
assert d['json_error_kind']=='missing_delimiter' and d['json_error_at_end']
|
|
assert 'secret' not in json.dumps(d) and 'private_value' not in json.dumps(d)
|
|
|
|
|
|
|
|
def test_coding_contract_rejects_wrong_column_value_but_accepts_placeholder():
|
|
from test_handoff_outcomes import minimal
|
|
payload,c=minimal();c['parameter_grounding']=1
|
|
c['evidence']['E000001']['content']=rows_context()['evidence']['E000001']['content']
|
|
# Keep the original exact scope/source reference required by this fixture.
|
|
for item in c['evidence']['E000001']['content']['items']:item['source_ref']=payload['affected_scope'][0]['source_ref']
|
|
payload['implementation_guidance']=['Insert phaseMask = 14.']
|
|
with pytest.raises(ChangeRequestError,match='UNGROUNDED_DATABASE_VALUE'):validate_request(payload,c)
|
|
payload['implementation_guidance']=['Use a human-verified mask; do not guess it.']
|
|
validate_request(payload,c)
|
|
|
|
|
|
@pytest.mark.parametrize('framing_version',[1,2,3])
|
|
def test_bootstrap_marker_is_persisted_and_replayable(tool_case,tmp_path,framing_version):
|
|
import asyncio
|
|
from test_investigation_hybrid import HybridAdapter
|
|
from test_case_interpreter import gateway
|
|
from test_working_brief import service_for
|
|
from test_investigator import CASE,run,state
|
|
from viano_bug_analysis.analysis import Investigator,InvestigationConfig
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.lifecycle import CheckpointStore,Workspace
|
|
class Framed(HybridAdapter):
|
|
versions=[]
|
|
async def generate(self,config,request,**kwargs):
|
|
self.versions.append(request['prompt_version'])
|
|
if request['prompt_version'] in {'investigation_bootstrap_v21','investigation_bootstrap_v22','investigation_bootstrap_v23'}:
|
|
request=dict(request,prompt_version='investigation_project_v18')
|
|
return await super().generate(config,request,**kwargs)
|
|
adapter=Framed();router=InvestigationToolRouter(project_research=service_for(tool_case))
|
|
loop=Investigator(tool_case[0],gateway(adapter),InvestigationConfig(lean_loop=True,lean_protocol=2,
|
|
lean_checkpoint_steps=0,bootstrap_json_framing=True,bootstrap_framing_version=framing_version),router=router)
|
|
first=run(loop,steps=3)
|
|
expected=f'investigation_bootstrap_v{20+framing_version}'
|
|
assert adapter.versions==[expected]*3
|
|
portable=CheckpointStore(Workspace(tmp_path/'portable'));portable.intake(first.package)
|
|
result=run(Investigator(portable,gateway(adapter),router=router))
|
|
assert result.status=='COMPLETED' and adapter.versions[3]=='investigation_project_v20'
|
|
from viano_bug_analysis.llm.cache_measurement import measure_package
|
|
measurement=measure_package(result.package)
|
|
assert measurement['all_exactly_reconstructed'] and measurement['unmatched_requests']==0
|
|
assert sum(s['historical_prompt']==expected for s in measurement['samples'])==3
|