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