"""Regressions for actual Claude questions and adversarial same-person joins.""" import copy import json from pathlib import Path import pytest from tools.answer_evaluation.checked_generation import generate_checked_answer from tools.answer_evaluation.context_policy import answer_context_issues from tools.answer_evaluation.graph_context import graph_context from tools.answer_evaluation.render_answer import render_graph_answer from tools.answers.answer_delivery import deliver_answer from tools.graphrag import graph_search as gs from tools.graphrag.tests.test_graph_search import FakeSession from tools.retrieval.question_constraints import unsupported_constraint PEOPLE=['Ján Pták','Ján Malinovský','Valerii Kutsenko','Jakub Schwarc'] PROMPT=Path('tools/answer_evaluation/graph_answer_prompt.txt').read_text() def person(index,name): return {'kind':'Person','name':name,'element_id':f'p{index}', 'source_document':f'pages/students/2022/p{index}/README.md'} def source(index): return {'document_path':f'pages/students/2022/p{index}/README.md', 'source_url':f'https://example.test/p{index}'} @pytest.fixture def tagged_projects(): catalog=[person(i,name) for i,name in enumerate(PEOPLE,1)] catalog.append({'kind':'Keyword','name':'RAG','id':'rag','element_id':'krag'}) projects=[{'person':name,'person_element_id':f'p{i}','year':2026, 'work_type':'diploma_project','work_id':f'work:p{i}', 'heading':'Diplomový projekt 2026','title':f'Téma projektu {name}',**source(i)} for i,name in enumerate(PEOPLE,1)] tags=[{'person':name,'person_element_id':f'p{i}','keyword':'RAG', 'keyword_id':'rag','keyword_element_id':'krag',**source(i)} for i,name in enumerate(PEOPLE[:3],1)] # A fifth person has the tag, but no work in the requested year/type. tags.append({'person':'Iná Osoba','person_element_id':'p5','keyword':'RAG', 'keyword_id':'rag','keyword_element_id':'krag',**source(5)}) return FakeSession({gs.CATALOG_QUERY:catalog,gs.WORK_FACTS_QUERY:projects, gs.PERSON_KEYWORD_FACTS_QUERY:tags}) @pytest.mark.parametrize('question',[ 'Ktorí študenti majú tag RAG a zároveň diplomový projekt z roku 2026?', 'Ktorí študenti s tagom RAG majú diplomový projekt v roku 2026?', 'Ktorí študenti majú tag RAG a diplomový projekt z roku 2026?', 'Ktorí študenti majú tag „RAG“ a zároveň diplomový projekt z roku 2026?', 'Ktorí študenti s tagom RAG a zároveň diplomovým projektom z roku 2026?', 'Ktorí študenti s tagom RAG a diplomovým projektom z roku 2026?', ]) def test_tag_and_project_are_witnessed_by_the_same_person(tagged_projects,question): r=gs.search_graph_with_session(tagged_projects,question,limit=50) assert r['status']=='ok' and r['constraints']['keyword_phrase']=='rag' assert {f['person'] for f in r['facts']}==set(PEOPLE[:3]) work=[params for q,params in tagged_projects.calls if q==gs.WORK_FACTS_QUERY][0] tags=[params for q,params in tagged_projects.calls if q==gs.PERSON_KEYWORD_FACTS_QUERY][0] assert work['years']==[2026] and work['work_types']==['diploma_project'] assert tags['keyword_ids']==['krag'] context=graph_context(r) assert {c['person'] for c in context['answer_candidates']}==set(PEOPLE[:3]) assert all(c['work_type']=='diploma_project' for c in context['answer_candidates']) def test_diploma_thesis_cannot_be_replaced_by_a_project(tagged_projects): r=gs.search_graph_with_session(tagged_projects, 'Ktorí študenti majú tag RAG a zároveň diplomovú prácu z roku 2026?',limit=50) assert r['status']=='no_answer' and r['reason']=='no_matching_facts' assert not r['facts'] and not r['constraints']['unresolved_keyword'] def test_unknown_tag_cannot_trigger_a_global_project_listing(tagged_projects): r=gs.search_graph_with_session(tagged_projects, 'Ktorí študenti majú tag XNEZNAMY a zároveň diplomový projekt z roku 2026?') assert r['status']=='no_answer' and r['reason']=='unknown_keyword' and not r['facts'] @pytest.mark.parametrize('question,expected',[ ('Ľudia s tagom strojové učenie a zároveň diplomovou prácou.','strojove ucenie'), ('Ľudia s tagom „výskum a aplikácie“ a zároveň diplomovým projektom.','vyskum a aplikacie'), ('Ľudia s tagom RAG z roku 2026.','rag'), ]) def test_multiword_tag_and_condition_boundary(question,expected): assert gs.explicit_keyword_phrase(question)==expected @pytest.mark.parametrize('question,code',[ ('Ktorí študenti majú aspoň tri práce?','quantity_condition'), ('Ktorí študenti majú aspoň 3 práce?','quantity_condition'), ('Kto má najmenej tri práce?','quantity_condition'), ('Študenti s najviac dvoma projektmi?','quantity_condition'), ('Študenti s minimálne piatimi prácami?','quantity_condition'), ('ktorý študenti týmto tempom ukončia svoje diplomové práce v roku 2027?','completion_prediction'), ('Ktorí študenti ukončia diplomové práce v roku 2027?','completion_prediction'), ('Odhadni termín dokončenia diplomovej práce Jána Holpa.','completion_prediction'), ('Kedy Ján Holp ukončil diplomovú prácu?','work_event_date'), ('Aké metódy predikcie počasia používa diplomová práca?',None), ('Aký je názov práce „Kto dokončí prácu v roku 2027“?',None), ]) def test_unsupported_condition_wording(question,code): assert unsupported_constraint(question)==code @pytest.mark.parametrize('engine',['rag','graphrag']) @pytest.mark.parametrize('question,word',[ ('Ktorí študenti majú aspoň tri práce?','počet'), ('Ktorí študenti týmto tempom ukončia diplomové práce v roku 2027?','predpovedať'), ]) def test_unsupported_condition_never_calls_a_model(engine,question,word): r=deliver_answer(question,engine=engine,complete=lambda _:pytest.fail('No model for unsupported condition')) assert r['status']=='ambiguous' and r['accepted_for_delivery'] and word in r['answer'] assert not r['claims'] and not r['source_urls'] def selected(context): fields=['person','record_id','title','year','work_type','work_label','keyword','category','document_author'] rows=context.get('required_academic_records') or context['facts'] claims=[{k:r[k] for k in fields if r.get(k) is not None} for r in rows] return {'status':'ok','answer':'','claims':claims, 'source_urls':[s['source_url'] for s in context['sources']]} def generated(answer,context): return generate_checked_answer(context.get('question',''),context,PROMPT, lambda _: {'choices':[{'message':{'content':json.dumps(answer)}}]}) def test_titleless_records_survive_and_are_delivered_without_invented_titles(): catalog=[person(1,'Ján Holp')] records=[{'person':'Ján Holp','person_element_id':'p1','work_id':f'project-{i}', 'title':None,'heading':label,'work_type':'diploma_project','year':2020,**source(1)} for i,label in enumerate(['Diplomový projekt 2020','Diplomový projekt 2 2020'])] s=FakeSession({gs.CATALOG_QUERY:catalog,gs.WORK_FACTS_QUERY:records}) raw=gs.search_graph_with_session(s,'Aký názov má diplomový projekt Jána Holpa z roku 2020?') assert raw['status']=='ok' and len(raw['facts'])==2 context=graph_context(raw);a=selected(context) result=generated(a,context) assert result['accepted_for_delivery'] assert result['answer']['answer'].count('názov v zdroji nie je uvedený')==2 assert len(result['answer']['claims'])==2 and not any(c.get('title') for c in result['answer']['claims']) bad=copy.deepcopy(a);bad['claims'][0]['title']='Vymyslený názov' assert 'unsupported_academic_binding' in answer_context_issues(bad,context,check_text=False) bad=copy.deepcopy(a);bad['claims'].pop() assert 'missing_academic_record_details' in answer_context_issues(bad,context,check_text=False) def shared_context(three=False): people=PEOPLE[:3] if three else PEOPLE[:2] raw={'status':'ok','reason':None,'complete':True,'truncated':False,'aggregation':None, 'query':'Aké spoločné kľúčové slová majú '+' a '.join(people)+'?', 'constraints':gs.extract_query_constraints('spoločné kľúčové slová'), 'resolved_entities':[person(i,p) for i,p in enumerate(people,1)], 'sources':[source(i) for i in range(1,len(people)+1)],'facts':[]} for keyword,left,right in [('NLP',1,2),('RAG',1,2)]+([('NLP',1,3)] if three else []): raw['facts'].append({'fact_type':'shared_keyword','keyword':keyword, 'left_person':people[left-1],'right_person':people[right-1], 'left_document_path':source(left)['document_path'],'right_document_path':source(right)['document_path'], 'left_source_url':source(left)['source_url'],'right_source_url':source(right)['source_url']}) return graph_context(raw) def test_shared_tags_are_summarized_once_and_require_all_peoples_evidence(): context=shared_context();a=selected(context) result=generated(a,context) assert result['accepted_for_delivery'] text=result['answer']['answer'] assert 'Spoločné kľúčové slová: **NLP**, **RAG**.' in text assert text.count('NLP')==1 and text.count('RAG')==1 and 'Pri osobe' not in text bad=copy.deepcopy(a);bad['claims'].pop() assert 'missing_shared_property_evidence' in answer_context_issues(bad,context,check_text=False) def test_pairwise_shared_tag_does_not_mean_shared_by_everyone(): context=shared_context(three=True) assert {r['value'] for r in context['shared_properties']}=={'NLP'} text=render_graph_answer(selected(context),context)['answer'] assert 'Spoločné kľúčové slová: **NLP**.' in text and '**RAG**' not in text def test_incomplete_shared_tags_cannot_claim_exhaustive_results(): context=shared_context();context.update(complete=False,truncated=True,truncation_reasons=['fact_limit']) result=generated(selected(context),context) assert result['accepted_for_delivery'] assert 'Doložené spoločné kľúčové slová' in result['answer']['answer'] assert 'Výpis je neúplný.' in result['answer']['answer']