232 lines
12 KiB
Python
232 lines
12 KiB
Python
import copy
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from test_case_builder import case_setup
|
|
from test_case_interpreter import gateway
|
|
from test_investigation_tools import tool_case
|
|
from test_investigation_startup import Adapter
|
|
from test_investigation_window import context as old_context
|
|
from test_investigator import CASE, run, state
|
|
from test_project_research import rows
|
|
from test_working_brief import service_for
|
|
from test_source_context import source
|
|
from viano_bug_analysis.analysis import Investigator, InvestigationConfig, InvestigationError
|
|
from viano_bug_analysis.analysis.investigation_contract import entity_catalog, feedback
|
|
from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema
|
|
from viano_bug_analysis.budget import BudgetManager
|
|
from viano_bug_analysis.llm import ProviderResponse
|
|
from viano_bug_analysis.llm.gemini import GeminiProvider
|
|
from viano_bug_analysis.llm.investigation_size import check_wire, LIMIT
|
|
from viano_bug_analysis.llm.response_recovery import normalize
|
|
from viano_bug_analysis.package import PackageReader, PackageValidator
|
|
from viano_bug_analysis.tools import InvestigationToolRouter
|
|
|
|
CORPUS = Path(__file__).parent/'fixtures/llm_rejections/20261020/bounded-startup-contract'
|
|
|
|
|
|
def context():
|
|
c = old_context()
|
|
c['policy']['prompt_window_version'] = 2
|
|
c['case_sources'] = {
|
|
'N0012': {'kind':'entity','data':{'type':'creature','id':'77','roles':['observed'],
|
|
'name_bindings':[{'name':'Observed giver','provenance':{'large':'unused'}}]}},
|
|
'N0003': {'kind':'entity','data':{'type':'item','id':'77','name':'Different typed identity','roles':[]}},
|
|
'N0042': {'kind':'entity','data':{'type':'creature','id':'88','name':'Affected creature','roles':['affected_subject']}},
|
|
'F0001': {'kind':'fact','data':{'type':'creature','id':'999'}},
|
|
}
|
|
return c
|
|
|
|
|
|
def test_signatures_are_text_and_literal_example_matches_strict_schema():
|
|
c = context()
|
|
before = copy.deepcopy(c)
|
|
request = build_prompt(c)
|
|
body = json.loads(request['messages'][1]['content'])
|
|
text = body['response_contract']
|
|
assert isinstance(text,str) and '"const"' not in text and '"choices"' not in text
|
|
example = json.loads(text.rsplit('\n',1)[1])
|
|
assert Draft202012Validator(output_schema(c,strict=True)).is_valid(example)
|
|
assert example['next']['kind'] == 'tool'
|
|
assert 'arguments' in example['next']
|
|
assert c == before
|
|
system,prefix,live = GeminiProvider._messages(request,stable_case_context=True)
|
|
assert check_wire(system,prefix+live) <= LIMIT
|
|
|
|
|
|
def test_catalog_uses_exact_canonical_refs_and_keeps_types_distinct():
|
|
c = context()
|
|
catalog = entity_catalog(c)
|
|
assert set(catalog) == {'N0012','N0003','N0042'}
|
|
assert catalog['N0012']['names'] == ['Observed giver']
|
|
assert catalog['N0012']['type'] == 'creature' and catalog['N0003']['type'] == 'item'
|
|
assert catalog['N0012']['id'] == catalog['N0003']['id'] == '77'
|
|
assert 'provenance' not in json.dumps(catalog)
|
|
body = json.loads(build_prompt(c)['messages'][1]['content'])
|
|
assert body['input']['entities'] == catalog
|
|
|
|
|
|
@pytest.mark.parametrize('name,recovered,path', [
|
|
('query-choice-wrapper',True,None),
|
|
('flattened-tool-arguments',False,'/next/arguments'),
|
|
('literal-schema-values',False,'/next/kind'),
|
|
('unavailable-entity-reference-catalog',False,'/next/arguments/entity_refs'),
|
|
])
|
|
def test_real_rejected_answers(name,recovered,path):
|
|
folder = CORPUS/name
|
|
document = json.loads((folder/'parsed-answer.json').read_text(encoding='utf-8'))
|
|
request = json.loads((folder/'request.json').read_text(encoding='utf-8'))
|
|
schema = json.loads((folder/'validation-context.json').read_text(encoding='utf-8'))['strict_output_schema']
|
|
request['output_schema']['x-viano-investigation-window'] = 2
|
|
before = copy.deepcopy(document)
|
|
candidate,audit = normalize(document,request=request)
|
|
errors = list(Draft202012Validator(schema).iter_errors(candidate))
|
|
assert document == before
|
|
if recovered:
|
|
assert not errors
|
|
assert audit['original_document'] == before
|
|
assert audit['rules'] == ['unwrap_single_query_choice:/next/arguments']
|
|
assert candidate['next']['arguments'] == before['next']['arguments']['choices'][0]
|
|
assert candidate['observation'] == before['observation']
|
|
else:
|
|
assert errors and not audit and candidate == before
|
|
info = feedback(errors[0],candidate)
|
|
assert info['validation_path'].startswith(path)
|
|
assert len(json.dumps(info)) < 600
|
|
if 'entity_refs' in path:
|
|
assert 'N references' in info['requirement']
|
|
|
|
|
|
@pytest.mark.parametrize('change', ['many','sibling','unknown_operation','unknown_kind','no_query_kind','legacy'])
|
|
def test_ambiguous_or_unversioned_wrappers_are_not_repaired(change):
|
|
document = {'assessment':None,'observation':'Read.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':{'choices':[rows()]}}}
|
|
request = {'prompt_version':'investigation_startup_v1','output_schema':{
|
|
'x-viano-response-recovery':1,'x-viano-investigation-window':2}}
|
|
if change == 'many': document['next']['arguments']['choices'].append(rows())
|
|
if change == 'sibling': document['next']['arguments']['other'] = 9
|
|
if change == 'unknown_operation': document['next']['operation'] = 'shell.exec'
|
|
if change == 'unknown_kind': document['next']['kind'] = 'execute'
|
|
if change == 'no_query_kind': document['next']['arguments']['choices'][0].pop('kind')
|
|
if change == 'legacy': request['output_schema']['x-viano-investigation-window'] = 1
|
|
candidate,audit = normalize(document,request=request)
|
|
assert candidate == document and audit is None
|
|
|
|
|
|
def loop_for(tool_case,adapter,source=None):
|
|
service = service_for(tool_case)
|
|
if source is not None:
|
|
from viano_bug_analysis.tools.source_context import SourceContextAnalysis
|
|
service.source = source
|
|
service.source_context = SourceContextAnalysis(source,{'protocol':2})
|
|
return Investigator(tool_case[0],gateway(adapter),InvestigationConfig(
|
|
lean_loop=True,lean_protocol=2,startup_batch_review=True,
|
|
lean_checkpoint_steps=0,prompt_window_version=2,response_recovery_version=1),
|
|
router=InvestigationToolRouter(project_research=service))
|
|
|
|
|
|
class Wrapped(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
body = json.loads(request['messages'][1]['content'])
|
|
if body['task'] != 'startup_choose_query':
|
|
return await super().generate(config,request,**kwargs)
|
|
self.calls.append(body)
|
|
return ProviderResponse({'assessment':None,'observation':'Read scoped records.',
|
|
'next':{'kind':'tool','operation':'project.query','arguments':{'choices':[rows()]}}},'fixture',100,50)
|
|
|
|
|
|
def test_recovered_start_queries_execute_once_then_joint_review(tool_case):
|
|
adapter = Wrapped()
|
|
result = run(loop_for(tool_case,adapter),steps=4)
|
|
saved = state(result)
|
|
assert saved['startup']['done'] and len(adapter.calls) == 4
|
|
assert all(h['invalid_responses'] == 0 for h in saved['hypotheses'].values())
|
|
with PackageReader(result.package) as reader:
|
|
ref = saved['hypotheses']['H1']['notes'][0]['llm_ref']
|
|
outcome = json.loads(reader.read_bytes(ref))['outcome']
|
|
assert outcome['normalization']['original_document']['next']['arguments']['choices']
|
|
assert outcome['document']['next']['arguments'] == rows()
|
|
assert PackageValidator().validate(result.package).valid
|
|
|
|
|
|
@pytest.mark.parametrize('error_kind', ['flattened','raw_entity','unknown_entity','missing_offset','unknown_action'])
|
|
def test_invalid_content_gets_precise_feedback_and_no_budget_side_effects(tool_case,error_kind,source):
|
|
class Invalid(Adapter):
|
|
async def generate(self,config,request,**kwargs):
|
|
query = rows()
|
|
action = {'kind':'tool','operation':'project.query','arguments':query}
|
|
if error_kind == 'flattened': action = {'kind':'tool','operation':'project.query','profile':'p','value':77}
|
|
if error_kind in {'raw_entity','unknown_entity'}:
|
|
action['arguments'] = {'kind':'source.context','text':'Observed','path_prefix':'',
|
|
'entity_refs':['77' if error_kind == 'raw_entity' else 'N9999']}
|
|
if error_kind == 'missing_offset':
|
|
query.pop('offset');action['arguments'] = {'choices':[query]}
|
|
if error_kind == 'unknown_action': action['operation'] = 'shell.exec'
|
|
return ProviderResponse({'assessment':None,'observation':'Read.','next':action},'fixture',100,50)
|
|
loop = loop_for(tool_case,Invalid(),source=source)
|
|
result = run(loop,steps=1)
|
|
saved = state(result)
|
|
assert not saved['startup']['explored']
|
|
assert all(not a['actions'] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
with PackageReader(result.package) as reader:
|
|
c = json.loads(reader.read_bytes(saved['active']['context_ref']))
|
|
info = c['previous_rejection_details']
|
|
assert info['validation_path'].startswith('/next')
|
|
if error_kind == 'flattened': assert 'next.arguments must contain' in info['requirement']
|
|
if error_kind in {'raw_entity','unknown_entity'}: assert 'N references' in info['requirement']
|
|
if error_kind == 'missing_offset': assert 'offset' in info['validation_fields']
|
|
|
|
|
|
def test_recovered_action_survives_restart_before_execution(tool_case,monkeypatch):
|
|
adapter = Wrapped();loop = loop_for(tool_case,adapter)
|
|
execute = loop._execute_plan
|
|
def interrupt(*args): raise RuntimeError('interrupt before tool execution')
|
|
monkeypatch.setattr(loop,'_execute_plan',interrupt)
|
|
with pytest.raises(InvestigationError): run(loop,steps=1)
|
|
assert len(adapter.calls) == 1
|
|
assert all(not a['actions'] for a in BudgetManager(tool_case[0]).snapshot(*CASE).state['accounts'].values())
|
|
monkeypatch.setattr(loop,'_execute_plan',execute)
|
|
result = run(loop,steps=1)
|
|
assert state(result)['startup']['explored'] == ['H1'] and len(adapter.calls) == 1
|
|
|
|
|
|
def test_v2_cache_partition_and_early_review():
|
|
from viano_bug_analysis.llm.stable_context import split_document, restore_document
|
|
from viano_bug_analysis.llm.gateway import is_investigation_review
|
|
c = context()
|
|
request = build_prompt(c)
|
|
body = json.loads(request['messages'][1]['content'])
|
|
assert restore_document(*split_document(body, request['prompt_version'])) == body
|
|
system, prefix, live = GeminiProvider._messages(request, stable_case_context=True)
|
|
assert '"output_schema"' not in json.dumps(prefix)
|
|
assert check_wire(system, prefix + live) <= LIMIT
|
|
c['early_review'] = {'request': {'status': 'preparing'}}
|
|
request = build_prompt(c)
|
|
assert is_investigation_review(request)
|
|
body = json.loads(request['messages'][1]['content'])
|
|
assert '"observation":"' not in body['response_contract']
|
|
|
|
|
|
def test_v2_completion_after_checkpoints(tool_case):
|
|
class Finish(Adapter):
|
|
async def generate(self, config, request, **kwargs):
|
|
body = json.loads(request['messages'][1]['content'])
|
|
system, prefix, live = GeminiProvider._messages(request, stable_case_context=True)
|
|
assert check_wire(system, prefix + live) <= LIMIT
|
|
if body['task'] == 'investigate_one_step':
|
|
out = {'assessment': None, 'observation': 'Ready to consolidate.',
|
|
'next': {'kind': 'checkpoint', 'intent': 'completion'}}
|
|
elif body['task'] == 'consolidate_investigation_session':
|
|
out = {'assessment': None, 'confidence': 'plausible', 'findings': [],
|
|
'uncertainties': ['No proven cause.'], 'next_question': None,
|
|
'retire': [], 'next': {'kind': 'finish', 'reason': 'Investigation complete.'}}
|
|
else:
|
|
return await super().generate(config, request, **kwargs)
|
|
return ProviderResponse(out, 'fixture', 100, 50)
|
|
result = run(loop_for(tool_case, Finish()), steps=20)
|
|
assert result.status == 'COMPLETED'
|
|
assert PackageValidator().validate(result.package).valid
|