160 lines
8.7 KiB
Python
160 lines
8.7 KiB
Python
"""Offline regressions for phase-specific review and actionable safe rejection."""
|
|
import asyncio
|
|
import copy
|
|
import json
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from test_compact_investigation import context, entry
|
|
from test_lean_investigation_decision import reply
|
|
from test_llm_caching import config, native, scripted
|
|
from test_llm_gateway import request
|
|
from test_investigation_tools import tool_case
|
|
from test_case_builder import case_setup
|
|
from viano_bug_analysis.analysis.investigation_compact import model_view, review_view, validate_tool_shape
|
|
from viano_bug_analysis.analysis.investigation_config import InvestigationError
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema
|
|
from viano_bug_analysis.analysis.investigation_decision import normalize
|
|
from viano_bug_analysis.llm import LLMError
|
|
from viano_bug_analysis.llm.diagnostics import parse_response_object, sanitize
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
from viano_bug_analysis.llm.stable_context import split_document, restore_document
|
|
|
|
|
|
def review_context():
|
|
c=context();c['prompt_version']='investigation_project_v16'
|
|
c['early_review']={'attempts':1,'limit':3,'request':{'status':'preparing','hypothesis_id':'H2',
|
|
'reason':'Other mechanism has scoped support, independent diagnosis pending.'}}
|
|
return c
|
|
|
|
|
|
def test_review_is_narrower_but_keeps_pending_results_and_counterfindings():
|
|
c=review_context();before=copy.deepcopy(c)
|
|
r=build_prompt(c);old=build_prompt(dict(c,prompt_version='investigation_project_v15'))
|
|
view=json.loads(r['messages'][1]['content'])['input']
|
|
assert 'project.query' not in r['messages'][0]['content']
|
|
assert all(k not in view for k in ('known_schemas','tool_schema','evidence_catalog','budget'))
|
|
assert view['evidence']==model_view(c)['evidence']
|
|
assert view['working_brief']['findings'][0]['direction']=='against'
|
|
assert view['other_working_briefs']==model_view(c)['other_working_briefs']
|
|
assert c==before
|
|
assert sum(len(m['content']) for m in r['messages']) < sum(len(m['content']) for m in old['messages'])
|
|
stable,live=split_document(json.loads(r['messages'][1]['content']),r['prompt_version'])
|
|
assert restore_document(stable,live)==json.loads(r['messages'][1]['content'])
|
|
|
|
|
|
@pytest.mark.parametrize('kind',['tool','recall','test','finish','request_diagnosis'])
|
|
def test_review_contract_rejects_new_research_or_diagnosis_actions(kind):
|
|
c=review_context()
|
|
for strict in (False,True):
|
|
schema=output_schema(c,strict=strict)['properties']['next']
|
|
assert not Draft202012Validator(schema).is_valid({'kind':kind,'reason':'New work'})
|
|
assert Draft202012Validator(schema).is_valid({'kind':'defer','reason':'Pending result assessed'})
|
|
assert Draft202012Validator(schema).is_valid({'kind':'reject','reason':'Direct counterevidence'})
|
|
|
|
|
|
def test_review_strict_validation_and_normalization_preserve_assessment():
|
|
c=review_context();r=reply();r['next']={'kind':'defer','reason':'Ready for independent review'}
|
|
validate_tool_shape(c,r)
|
|
normalized=normalize(c,r)
|
|
assert normalized['assessment']['evidence_refs']==['E000020']
|
|
assert normalized['update']['working_brief']['findings']==c['working_brief']['brief']['findings']
|
|
assert normalized['update']['causal_review']['effect']=='contradicts'
|
|
r['next']={'kind':'tool','operation':'project.query','arguments':{'kind':'database.schema','table':'creature'}}
|
|
with pytest.raises(InvestigationError,match='INVESTIGATION_OUTPUT'):validate_tool_shape(c,r)
|
|
|
|
|
|
def test_reference_namespace_error_is_actionable_without_echoing_values():
|
|
c=review_context();r=reply()
|
|
r['update']['working_brief']['findings'][0]['evidence_refs']=['F0001']
|
|
with pytest.raises(InvestigationError) as error:validate_tool_shape(c,r)
|
|
assert 'research_E_ids_only_not_case_F_ids' in error.value.feedback['requirement']
|
|
assert 'F0001' not in json.dumps(error.value.feedback)
|
|
assert '/evidence_refs/0' in error.value.feedback['validation_path']
|
|
|
|
|
|
def test_origin_consolidation_is_lossless_and_does_not_touch_db_cells():
|
|
c=context();c['prompt_version']='investigation_project_v16';c['recalled_evidence']=['E000019']
|
|
origin={'snapshot_id':'snapshot-'+('a'*150),'database':'world','revision':'fixed'}
|
|
for eid in ('E000019','E000020'):
|
|
capture=c['evidence'][eid]['content']['items'][0]
|
|
capture['origin']=copy.deepcopy(origin)
|
|
capture['result']['items']=[{'row':{'origin':copy.deepcopy(origin),'metadata_ref':'literal'}}]
|
|
original=copy.deepcopy(c);v=model_view(c)
|
|
assert len(v['shared_metadata'])==1
|
|
for e in v['evidence'].values():
|
|
capture=e['content']['items'][0]
|
|
assert v['shared_metadata'][capture['origin']['metadata_ref']]==origin
|
|
assert capture['result']['items'][0]['row']=={'origin':origin,'metadata_ref':'literal'}
|
|
assert original==c
|
|
|
|
|
|
@pytest.mark.parametrize('raw,category',[(b'{"private":', 'invalid_json'),
|
|
(b'{"private":1,"private":2}','json_policy'),(b'[]','root_not_object'),
|
|
(b'{"private":NaN}','json_policy')])
|
|
def test_json_diagnostics_are_precise_but_never_echo_content(raw,category):
|
|
with pytest.raises(LLMError) as caught:parse_response_object(raw,stage='output_json')
|
|
info=caught.value.diagnostics
|
|
assert info['response_error']==category and info['response_stage']=='output_json'
|
|
assert 'private' not in json.dumps(info)
|
|
assert sanitize(info)==info
|
|
|
|
|
|
@pytest.mark.parametrize('broken,stage,category',[
|
|
({'finishReason':'STOP','content':{'parts':[{'text':'{"secret":'}]}},'output_json','invalid_json'),
|
|
({'finishReason':'STOP'},'candidate','invalid_structure')])
|
|
def test_gemini_failed_output_keeps_usage_and_safe_category(monkeypatch,broken,stage,category):
|
|
adapter=GeminiProvider();envelope=native();envelope['candidates']=[broken]
|
|
scripted(adapter,monkeypatch,[envelope])
|
|
with pytest.raises(LLMError) as caught:
|
|
asyncio.run(adapter.generate(replace(config(),context_cache=False),request(),max_response_bytes=10000))
|
|
info=caught.value.diagnostics
|
|
assert info['response_stage']==stage and info['response_error']==category
|
|
assert info['input_tokens']==100 and info['reasoning_tokens']==5
|
|
assert 'secret' not in json.dumps(info)
|
|
|
|
|
|
def test_unrecognized_diagnostic_values_are_removed():
|
|
assert sanitize({'response_stage':'secret','response_error':'secret','parse_line':-1})=={'parse_line':None}
|
|
|
|
|
|
def test_v15_uses_frozen_general_prompt_even_when_review_pending():
|
|
c=review_context();c['prompt_version']='investigation_project_v15'
|
|
prompt=build_prompt(c)
|
|
assert json.loads(prompt['messages'][1]['content'])['task']=='investigate_hypothesis'
|
|
assert Draft202012Validator(output_schema(c)['properties']['next']).is_valid({'kind':'recall','evidence_refs':['E000001']})
|
|
|
|
|
|
def test_provider_parse_details_survive_checkpoint_and_repair_without_extra_queries(tool_case):
|
|
from test_investigator import LoopAdapter, run, CASE
|
|
from test_working_brief import service_for
|
|
from test_project_research import rows
|
|
from test_case_interpreter import gateway
|
|
from viano_bug_analysis.analysis import Investigator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
class RepairAdapter(LoopAdapter):
|
|
async def generate(self, config, req, *, max_response_bytes):
|
|
if not hasattr(self, 'failed'):
|
|
self.failed=True
|
|
raise LLMError('LLM_RESPONSE',diagnostics={'response_stage':'output_json',
|
|
'response_error':'invalid_json','parse_line':3,'parse_column':8})
|
|
return await super().generate(config,req,max_response_bytes=max_response_bytes)
|
|
def change(reply,c):
|
|
reply['next']={'kind':'tool','operation':'project.query','arguments':rows()}
|
|
store=tool_case[0];adapter=RepairAdapter(change)
|
|
result=run(Investigator(store,gateway(adapter),router=InvestigationToolRouter(
|
|
project_research=service_for(tool_case))),steps=2)
|
|
info=adapter.contexts[0]['previous_rejection_details']
|
|
assert info['response_error']=='invalid_json' and info['parse_line']==3
|
|
assert info['requirement']=='return_one_valid_JSON_object_no_fences_or_duplicate_keys'
|
|
assert len(BudgetManager(store).snapshot(*CASE).state['accounts']['H1']['actions'])==1
|
|
assert PackageValidator().validate(result.package).valid
|
|
with PackageReader(result.package) as reader:
|
|
failed=[json.loads(reader.read_bytes(n)) for n in reader.files
|
|
if n.startswith('llm/operations/') and '/round-' in n]
|
|
assert next(r for r in failed if r.get('code')=='LLM_RESPONSE')['outcome']['diagnostics']['parse_column']==8
|