134 lines
7.3 KiB
Python
134 lines
7.3 KiB
Python
"""Whole discovery evidence reaches the same production provenance guard."""
|
|
import json
|
|
import pytest
|
|
from tools.answers.question_plan import build_question_plan, QuestionPlan
|
|
from tools.answers.delivery_contract import mechanical_check, ContractError
|
|
from tools.knowledge.sources import profile_units
|
|
from tools.knowledge.query import answer_from_index
|
|
from tools.knowledge.evidence import PROVENANCE_FIELDS
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Kto zo študentov robil na chatbotovi a čo mu už fungovalo?',
|
|
'Ktorí študenti pripravovali alebo ručne označovali dáta? Čo označovali?',
|
|
'Kto skúšal model niečo nové naučiť alebo ho dotrénovať? Ako zisťoval, či sa model zlepšil?',
|
|
'Kto naprogramoval hlasového asistenta?',
|
|
'Ktorý študent sa zaoberal spracovaním dokumentov?',
|
|
'Kto sa venoval tomu, aby počítač v texte našiel mená ľudí alebo organizácií?',
|
|
])
|
|
def test_unknown_student_is_the_answer_not_a_required_input(question):
|
|
plan=build_question_plan(question,[])
|
|
assert not plan.clarification
|
|
assert plan.spec.intent=='discovery' and not plan.spec.people
|
|
|
|
|
|
@pytest.mark.parametrize('question',[
|
|
'Ktorí študenti majú tag RAG?',
|
|
'Kto má diplomovú prácu z roku 2026?',
|
|
'Ktorí študenti majú aspoň tri práce?',
|
|
])
|
|
def test_property_queries_keep_the_existing_metadata_route(question):
|
|
assert build_question_plan(question,[]).spec.intent=='metadata'
|
|
|
|
|
|
@pytest.mark.parametrize('engine',['rag','graphrag'])
|
|
def test_dated_discovery_projection_preserves_source_identity_for_final_guard(engine):
|
|
units=profile_units({'title':'Eva Nová','path':'pages/students/2020/eva/README.md',
|
|
'content':'# Diplomová práca 2027\nStretnutie 5.5.2026\nStav:\n- Chatbot funguje pre agendu mesta.'},include_records=True)
|
|
index={'version':'content','index_version':{'version':'v'},'units':units}
|
|
def select(messages):
|
|
payload=json.loads(messages[-1]['content'])
|
|
if 'scopes' in payload:
|
|
return {'choices':[{'message':{'content':json.dumps({'scopes':[{
|
|
'scope_id':r['scope_id'],'matches_activity':True,'primary_ids':r['seed_ids'],'parts':[]
|
|
} for r in payload['scopes']]})}}]}
|
|
supplied=[u for g in payload['evidence'] for u in g['units']]
|
|
return {'choices':[{'message':{'content':json.dumps({'evidence_ids':[r['id'] for r in supplied]})}}]}
|
|
question='Kto vytvoril chatbota?'
|
|
result=answer_from_index(question,index,engine=engine,mode='discovery',complete=select)
|
|
assert result['status']=='ok' and result['evidence']
|
|
by_id={r['id']:r for r in units}
|
|
for row in result['evidence']:
|
|
assert all(row.get(k)==by_id[row['id']].get(k) for k in PROVENANCE_FIELDS)
|
|
plan=QuestionPlan(question=question,intent='discovery')
|
|
mechanical_check(plan,result,index)
|
|
dated=next(r for r in result['evidence'] if r.get('meeting_id'))
|
|
dated.pop('meeting_id')
|
|
with pytest.raises(ContractError,match='altered_evidence'):
|
|
mechanical_check(plan,result,index)
|
|
|
|
|
|
def test_selection_failure_diagnostics_survive_mcp_without_leaking_internal_fields():
|
|
from tools.answers.delivery_contract import diagnose
|
|
from packages.claude.client import validated_delivery_check
|
|
from tools.answers.answer_errors import public_failure
|
|
for stage in ['planning','evidence_selection']:
|
|
r=diagnose(public_failure({'reason':'answer_rejected','engine':'rag','_failure_stage':stage}),'Kto pracoval na téme?','rag')
|
|
check=validated_delivery_check(r,'Kto pracoval na téme?','rag')
|
|
assert check['layer']==stage and check['code']=='invalid_model_selection'
|
|
assert '_failure_stage' not in r
|
|
|
|
|
|
def test_ranked_paraphrase_reaches_selection_without_absolute_cosine_gate():
|
|
units=profile_units({'title':'Eva Nová','path':'pages/students/2020/eva/README.md',
|
|
'content':'# Diplomová práca 2027\nStav:\n- Dokončená anotácia korpusu.'},include_records=True)
|
|
row=next(r for r in units if 'anotácia' in r['text'])
|
|
seen=[]
|
|
def select(messages):
|
|
payload=json.loads(messages[-1]['content'])
|
|
if 'scopes' in payload:
|
|
choices=[{'scope_id':r['scope_id'],'matches_activity':True,
|
|
'primary_ids':r['seed_ids'],'parts':[]} for r in payload['scopes']]
|
|
value={'scopes':choices}
|
|
else:
|
|
units=[u for g in payload['evidence'] for u in g['units']]
|
|
seen.extend(units)
|
|
value={'evidence_ids':[r['id'] for r in units]}
|
|
return {'choices':[{'message':{'content':json.dumps(value)}}]}
|
|
result=answer_from_index('Kto ručne označoval textové dáta?',
|
|
{'version':'c','index_version':{'version':'v'},'units':units},engine='rag',
|
|
mode='discovery',units=[row],relevance={row['id']:0.63},complete=select)
|
|
assert seen and result['status']=='ok'
|
|
assert result['evidence'][0]['text']==row['text']
|
|
|
|
|
|
def _chatbot_run(summary):
|
|
units=profile_units({'title':'Eva Nová','path':'pages/students/2020/eva/README.md',
|
|
'content':'# Diplomová práca 2027\nStretnutie 5.5.2026\nStav:\n- Chatbot funguje pre agendu mesta.'},include_records=True)
|
|
index={'version':'content','index_version':{'version':'v'},'units':units}
|
|
def select(messages):
|
|
payload=json.loads(messages[-1]['content'])
|
|
if 'scopes' in payload:
|
|
return {'choices':[{'message':{'content':json.dumps({'scopes':[{
|
|
'scope_id':r['scope_id'],'matches_activity':True,'primary_ids':r['seed_ids'],'parts':[]
|
|
} for r in payload['scopes']]})}}]}
|
|
if 'units' in payload: # wording of isolated units
|
|
return {'choices':[{'message':{'content':json.dumps({'sentences':[
|
|
{'id':u['id'],'text':summary} for u in payload['units']] if summary else []})}}]}
|
|
return {'choices':[{'message':{'content':json.dumps({'evidence_ids':[u['id'] for g in payload['evidence'] for u in g['units']]})}}]}
|
|
question='Kto vytvoril chatbota?'
|
|
plan=build_question_plan(question,[])
|
|
return answer_from_index(question,index,engine='rag',mode='discovery',complete=select,plan=plan)
|
|
|
|
|
|
def test_readable_discovery_wording_keeps_the_verbatim_answer_for_restoration():
|
|
from tools.answers.delivery_contract import verbatim
|
|
plain=_chatbot_run(None)
|
|
readable=_chatbot_run('Chatbot funguje pre mestskú agendu.')
|
|
assert 'Chatbot funguje pre mestskú agendu (záznam' in readable['answer']
|
|
assert readable['evidence']==plain['evidence'] and readable['source_urls']==plain['source_urls']
|
|
# Restoring the swaps yields exactly the verbatim answer, byte for byte.
|
|
assert verbatim(readable,readable['_prose_swaps'])['answer']==plain['answer']
|
|
assert '_prose_swaps' not in plain
|
|
|
|
|
|
def test_broad_topic_may_select_more_ids_than_requested():
|
|
from tools.knowledge.query import select_topic_evidence
|
|
candidates=[{'id':f'u{i}','person':f'Meno{i} Priezvisko','work_label':'Diplomová práca 2027',
|
|
'phase':'result','text':f'- Chatbot verzia {i} funguje.'} for i in range(55)]
|
|
def select(messages):
|
|
payload=json.loads(messages[-1]['content'])
|
|
return {'choices':[{'message':{'content':json.dumps({'evidence_ids':[u['id'] for g in payload['evidence'] for u in g['units']]})}}]}
|
|
# 55 valid IDs exceed the requested 40; the selection is kept, not failed.
|
|
assert len(select_topic_evidence('Kto pracoval na chatbote?',candidates,select))==55
|