import copy import json from pathlib import Path import pytest from jsonschema import Draft202012Validator from test_bounded_contract_v2 import context, loop_for from test_case_builder import case_setup from test_investigation_tools import tool_case from test_investigation_startup import Adapter from test_investigator import CASE, run, state from test_project_research import rows from viano_bug_analysis.analysis.investigation_prompt import build_prompt, output_schema from viano_bug_analysis.analysis.investigation_contract import feedback from viano_bug_analysis.budget import BudgetManager from viano_bug_analysis.llm import ProviderResponse from viano_bug_analysis.llm.response_recovery import normalize from viano_bug_analysis.package import PackageReader, PackageValidator CORPUS = Path(__file__).parent/'fixtures/llm_rejections/20261020/bug17-recall-assessment' @pytest.mark.parametrize('pending',[False,True]) def test_versioned_schema_only_allows_applicable_assessment(pending): c = context() if not pending: c['previous_result'] = None legacy_schema = output_schema(c,strict=True) assert 'anyOf' in legacy_schema['properties']['assessment'] c['assessment_scope_version'] = 1 schema = output_schema(c,strict=True)['properties']['assessment'] assert Draft202012Validator(schema).is_valid(None) == (not pending) assert Draft202012Validator(schema).is_valid({'semantic_score':50,'reason':'New result.'}) == pending @pytest.mark.parametrize('name',['8f58f733','6ccfd3ed','4115b3b2']) def test_actual_rejections_keep_query_and_observation_without_budget_score(name): folder = CORPUS/name request = json.loads((folder/'request.json').read_text(encoding='utf-8')) answer = json.loads((folder/'parsed-answer.json').read_text(encoding='utf-8')) original = copy.deepcopy(answer) result,audit = normalize(answer,request=request) assert result['assessment'] is None assert result['next'] == answer['next'] and result['observation'] == answer['observation'] assert answer == original and audit['original_document'] == original assert audit['rules'] == ['non_applicable_budget_assessment:/assessment'] c = context();c['previous_result'] = None;c['assessment_scope_version'] = 1 assert Draft202012Validator(output_schema(c,strict=True)).is_valid(result) assert normalize(result,request=request) == (result,None) @pytest.mark.parametrize('change', ['pending','missing_previous','unknown_evidence','unknown_action', 'unknown_query','extra_evidence_refs','missing_score','bad_score','float_score','bool_score','empty_reason', 'causal_findings','checkpoint','legacy_window','disabled','missing_assessment']) def test_ambiguous_or_out_of_scope_scores_are_not_neutralized(change): folder = CORPUS/'8f58f733' request = json.loads((folder/'request.json').read_text(encoding='utf-8')) answer = json.loads((folder/'parsed-answer.json').read_text(encoding='utf-8')) body = json.loads(request['messages'][1]['content']) if change == 'pending': body['input']['previous_result'] = {'evidence_id':'E000005'} if change == 'missing_previous': body['input'].pop('previous_result') if change == 'unknown_evidence': answer['assessment']['reason'] = 'E999999 proves it.' if change == 'unknown_action': answer['next']['operation'] = 'shell.exec' if change == 'unknown_query': answer['next']['arguments']['kind'] = 'invented.query' if change == 'extra_evidence_refs': answer['assessment']['evidence_refs'] = ['E999999'] if change == 'missing_score': answer['assessment'].pop('semantic_score') if change == 'bad_score': answer['assessment']['semantic_score'] = 101 if change == 'float_score': answer['assessment']['semantic_score'] = 1.0 if change == 'bool_score': answer['assessment']['semantic_score'] = True if change == 'empty_reason': answer['assessment']['reason'] = ' ' if change == 'causal_findings': answer['findings'] = [{'text':'Unverified.'}] if change == 'checkpoint': body['task'] = 'consolidate_investigation_session' if change == 'legacy_window': request['output_schema']['x-viano-investigation-window'] = 1 if change == 'disabled': request['output_schema'].pop('x-viano-response-recovery') if change == 'missing_assessment': answer.pop('assessment') request['messages'][1]['content'] = json.dumps(body) result,audit = normalize(answer,request=request) assert result == answer and audit is None def test_null_assessment_error_has_actionable_feedback(): c = context();c['assessment_scope_version'] = 1;c['previous_result'] = None rule = output_schema(c,strict=True)['properties']['assessment'] validator = Draft202012Validator({'properties':{'assessment':rule}}) error = next(validator.iter_errors({'assessment':{'semantic_score':50,'reason':'Old evidence.'}})) info = feedback(error,{'assessment':{'semantic_score':50,'reason':'Old evidence.'}}) assert info['validation_path'] == '/assessment' assert 'must be null' in info['requirement'] and 'refund' in info['requirement'] def test_valid_query_after_recall_is_recovered_without_refund_or_extra_llm(tool_case): class Read(Adapter): async def generate(self,config,request,**kwargs): body = json.loads(request['messages'][1]['content']) if body['task'] != 'investigate_one_step': return await super().generate(config,request,**kwargs) self.calls.append(body) assert request['output_schema']['properties']['assessment']['type'] == 'null' if len(self.calls) == 5: out = {'assessment':None,'observation':'Read existing evidence.', 'next':{'kind':'recall','evidence_refs':['E000001'],'page':0}} else: out = {'assessment':{'semantic_score':90,'reason':'E000001 is useful old evidence.'}, 'observation':'Query a bounded table.', 'next':{'kind':'tool','operation':'project.query','arguments':rows()}} return ProviderResponse(out,'fixture',100,50) adapter = Read();loop = loop_for(tool_case,adapter) run(loop,steps=5) before = BudgetManager(tool_case[0]).snapshot(*CASE).state result = run(loop,steps=1) saved = state(result) assert len(adapter.calls) == 6 and saved['hypotheses']['H1']['invalid_responses'] == 0 after = BudgetManager(tool_case[0]).snapshot(*CASE).state assert after['accounts']['H1']['rewarded'] == before['accounts']['H1']['rewarded'] assert after['accounts']['H1']['spent'] == before['accounts']['H1']['spent'] + 10 with PackageReader(result.package) as reader: ref = saved['hypotheses']['H1']['notes'][-1]['llm_ref'] call = json.loads(reader.read_bytes(ref)) assert call['outcome']['document']['assessment'] is None assert call['outcome']['normalization']['original_document']['assessment']['semantic_score'] == 90 assert PackageValidator().validate(result.package).valid def test_pending_result_feedback_requires_a_real_assessment(): c = context();c['assessment_scope_version'] = 1 rule = output_schema(c,strict=True)['properties']['assessment'] error = next(Draft202012Validator({'properties':{'assessment':rule}}).iter_errors({'assessment':None})) info = feedback(error,{'assessment':None}) assert 'semantic_score' in info['requirement'] and 'pending new' in info['requirement']