"""Ranking and proactive delivery: Jev verdicts plus local rules, no generation model.""" import copy import os import sys import tempfile import unittest from pathlib import Path from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) import xkb_delivery as delivery import xkb_memory_service as service from xkb_eval import score_case def record(key, judge, body='', score=.9, scale='wiki_semantic'): return {'id': key, 'judge': judge, 'summary': body or key, 'score': score, 'score_scale': scale, 'namespace': 'private'} class ProactiveDelivery(unittest.TestCase): def select(self, records, judge, conversation, delivered=frozenset()): return delivery.select(records, judge, conversation, query='Help with the current task', delivered=delivered) def test_current_need_overrides_old_topic_retrieval_rank(self): source = [record('old-video', .36, score=.95), record('old-shot', .22, score=.9), record('carbon', .89, score=.8)] ranked = delivery.rank(source, {'status': 'judged'}) self.assertEqual([r['id'] for r in ranked], ['carbon', 'old-video', 'old-shot']) selected = self.select(ranked, {'status': 'judged'}, []) self.assertEqual([r['id'] for r in selected['records']], ['carbon']) self.assertEqual(len(ranked), 3) # Selection does not erase candidates. def test_at_most_three_and_no_generation_model_is_called(self): records = [record(str(i), .9, f'distinct step {i}') for i in range(10)] result = self.select(records, {'status': 'judged'}, []) self.assertEqual([r['id'] for r in result['records']], ['0', '1', '2']) self.assertEqual(result['status'], 'ready') self.assertFalse(hasattr(delivery, '_llm')) def test_partial_judge_delivers_positive_verdicts_and_stays_visible(self): source = [record('weak', .2, score=.98), record('unknown', None, score=.95), record('useful', .9, score=.9)] ranked = delivery.rank(source, {'status': 'partial'}) self.assertEqual([r['id'] for r in ranked], ['useful', 'unknown', 'weak']) selected = self.select(ranked, {'status': 'partial'}, []) self.assertEqual([r['id'] for r in selected['records']], ['useful']) self.assertEqual(selected['status'], 'degraded') fallback = delivery.rank(copy.deepcopy(source), {'status': 'unavailable'}) self.assertEqual([r['id'] for r in fallback], ['weak', 'unknown', 'useful']) unavailable = self.select(fallback, {'status': 'unavailable'}, []) self.assertEqual(unavailable['mode'], 'background') self.assertEqual(unavailable['status'], 'degraded') def test_exact_duplicates_and_text_already_in_a_reply_are_not_delivered(self): claim = 'Keep original source, reporting period, units and coefficient version for every calculation.' history = [{'role': 'assistant', 'content': claim}] records = [record('already', .9, claim), record('first', .9, 'A distinct useful step'), record('copy', .8, 'A distinct useful step'), record('weak', .2, 'Another step')] selected = self.select(records, {'status': 'judged'}, history) self.assertEqual([r['id'] for r in selected['records']], ['first']) self.assertEqual([r['reason'] for r in selected['withheld']], ['already_in_conversation', 'duplicate_claim', 'weak_relevance']) self.assertEqual(len(selected['previously_presented']), 1) def test_a_user_statement_does_not_count_as_already_presented(self): claim = 'Keep original source, reporting period, units and coefficient version for every calculation.' result = self.select([record('evidence', .9, claim)], {'status': 'judged'}, [{'role': 'user', 'content': claim}]) self.assertEqual(result['mode'], 'suggest') def test_display_only_backend_snippets_are_never_injected(self): candidate = {**record('a', .9, 'Delete the backup'), 'excerpt_truncated': True} result = self.select([candidate], {'status': 'judged'}, []) self.assertEqual(result['records'], []) self.assertEqual(result['withheld'][0]['reason'], 'incomplete_source_excerpt') def test_id_only_titles_are_replaced_by_the_first_sentence(self): card = {**record('2044072293383712878', .9, '單一鏡頭只安排一個核心動作。其餘拆成下一鏡。'), 'title': '2044072293383712878'} shown = self.select([card], {'status': 'judged'}, [])['records'][0] self.assertEqual(shown['title'], '單一鏡頭只安排一個核心動作') named = {**record('x', .9, 'Body.'), 'title': 'Seedance 2.0 workflow'} self.assertEqual(self.select([named], {'status': 'judged'}, [])['records'][0]['title'], 'Seedance 2.0 workflow') def history(self, key, judge, kind='memory_note'): return {**record(key, judge, f'history {key}'), 'record_type': kind} def card(self, key, judge): return {**record(key, judge, f'card {key}'), 'record_type': 'knowledge_card'} def test_reusable_knowledge_comes_before_a_more_relevant_history_note(self): records = [self.history('old-sync', .9), self.card('a', .6), self.card('b', .55), self.card('c', .52)] result = self.select(records, {'status': 'judged'}, []) self.assertEqual([r['id'] for r in result['records']], ['a', 'b', 'c']) self.assertEqual(result['withheld'], [{'evidence_key': delivery.identity_key(records[0]), 'reason': 'limit'}]) def test_history_needs_strong_relevance_and_gets_at_most_one_slot(self): records = [self.history('note', .9), self.history('trace', .85, 'conversation_trace'), self.history('weak-note', .7), self.card('a', .6)] result = self.select(records, {'status': 'judged'}, []) self.assertEqual([r['id'] for r in result['records']], ['a', 'note']) self.assertEqual([w['reason'] for w in result['withheld']], ['history_limit', 'history_below_floor']) def test_prefixed_hash_titles_are_replaced_too(self): for title in ('Xkb Case 6769746875625f737461722d', 'Local Readme 064efe10'): card = {**record('x', .9, 'One main action per shot. More.'), 'title': title} self.assertEqual(self.select([card], {'status': 'judged'}, [])['records'][0]['title'], 'One main action per shot') for title in ('Seedance 2.0 workflow', 'Release v1.2 notes'): card = {**record('x', .9, 'Body.'), 'title': title} self.assertEqual(self.select([card], {'status': 'judged'}, [])['records'][0]['title'], title) def test_card_form_fields_are_not_suggested(self): chunk = lambda key, body: {**record(key, .82, body), 'record_type': 'knowledge_chunk', 'title': '2101602734004953114'} fields_only = [chunk('claim', '## 2. Claim 等級\n- **等級**：Attested\n - Attested：原文直接引用'), chunk('rel', '## 6. 與現有知識的關係\n- **2043234633408798932**：無法判定關聯。'), chunk('rel-tail', '- **2023473551412801929**：看似相關。\n- **2030931033693241543**：無法判定')] knowledge = chunk('k', '## 3. 關鍵論點\n- **單一主要動作**：一個鏡頭只放一個動作，比較穩定。') result = self.select(fields_only + [knowledge], {'status': 'judged'}, []) self.assertEqual([r['id'] for r in result['records']], ['k']) self.assertEqual(result['records'][0]['title'], '單一主要動作：一個鏡頭只放一個動作，比較穩定') self.assertEqual({w['reason'] for w in result['withheld']}, {'card_metadata_section'}) def test_long_bodies_keep_whole_paragraphs_and_are_labelled(self): body = 'First complete paragraph.\n\n' + 'later detail. ' * 200 shown = delivery.excerpt(body) self.assertTrue(shown.startswith('First complete paragraph.')) self.assertTrue(shown.endswith('（節錄）')) single = 'One long sentence. ' * 100 cut = delivery.excerpt(single) self.assertTrue(cut.endswith('（節錄）')) self.assertLessEqual(len(cut), delivery.EXCERPT_CHARS + 10) self.assertEqual(delivery.excerpt('short'), 'short') def test_fusion_preserves_all_legs_and_strongest_judged_text(self): source = [record('same', .2, 'weak excerpt', .8), record('other', .7, score=.95), record('same', .9, 'actionable excerpt', 2, 'card_keyword')] ranked = delivery.rank(source, {'status': 'judged'}) self.assertEqual(ranked[0]['id'], 'same') self.assertEqual(ranked[0]['summary'], 'actionable excerpt') self.assertEqual(set(ranked[0]['matched_by']), {'wiki_semantic', 'card_keyword'}) self.assertEqual(len(ranked), 2) anonymous = [record('', .7, 'one'), record('', .8, 'two')] self.assertEqual(len(delivery.rank(anonymous, {'status': 'judged'})), 2) def test_delivery_evaluation_does_not_relabel_candidate_results(self): packet = {'records': [record('useful', .9), record('noise', .2)], 'retrieval_mode': 'keyword', 'judge': {'status': 'judged'}} packet['delivery'] = self.select(packet['records'], packet['judge'], []) case = {'id': 'one', 'expected_ids': ['useful'], 'expected_delivery': 'evidence'} self.assertFalse(score_case(case, packet)['ok']) self.assertTrue(score_case({**case, 'surface': 'delivery'}, packet)['ok']) def test_noise_has_explicit_empty_delivery_and_does_not_call_search(self): with tempfile.TemporaryDirectory() as tmp: store = service.Store(Path(tmp)/'test.sqlite') with patch.object(store.catalog, 'search') as search: packet = store.knowledge_recall('ok') search.assert_not_called() self.assertEqual(packet['delivery']['mode'], 'none') def test_shared_judge_sees_bounded_wiki_candidates_below_legacy_cosine_floor(self): import continuity_recall as cr with tempfile.TemporaryDirectory() as tmp, \ patch.object(cr, '_load_semantic_vectors', return_value={'wiki/topics/role.md#Guide': [.6,.8]}), \ patch.object(cr, '_embed_query', return_value=[1.,0.]), \ patch.object(cr, '_section_text', return_value='Use the same character reference in every shot.'), \ patch.object(cr, 'WIKI_MIN_SIMILARITY', .65): store = service.Store(Path(tmp)/'test.sqlite') store.catalog.wiki_topics_dir = Path(tmp)/'topics' self.assertEqual(cr.recall_semantic('The actor changes between shots'), []) with patch.dict(os.environ, {'XKB_JEV_DECIDE': '1'}): hits = store.catalog._wiki_search('The actor changes between shots', 1) self.assertEqual(len(hits), 1) self.assertEqual(hits[0]['score'], .6) with patch.dict(os.environ, {'XKB_JEV_DECIDE': '0'}): self.assertEqual(store.catalog._wiki_search('The actor changes between shots', 1), []) class GroundedSelection(unittest.TestCase): def test_upstream_wiki_boundaries_preserve_late_qualifiers(self): import continuity_recall as cr long_body = 'Deployment detail. ' * 24 + 'Delete the backup only after verified recovery. Never delete an unverified backup.' with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root/'guide.md').write_text('# Deployment\n\n'+long_body, encoding='utf-8') store = service.Store(root/'test.sqlite') store.catalog.wiki_topics_dir = root packet = store.catalog.search('deployment', 5, options={'semantic': False, 'cards': False, 'conversations': False}) body = packet['records'][0]['summary'] self.assertIn('Never delete an unverified backup.', body) marked = ('Operational detail. ' + ' ')*18 + 'Delete the backup only after verified recovery.' with patch.object(cr, '_load_semantic_vectors', return_value={'wiki/topics/guide.md#Guide': [1.,0.]}), \ patch.object(cr, '_embed_query', return_value=[1.,0.]), \ patch.object(cr, '_section_text', return_value=marked): hit = cr.recall_semantic('deployment', sections_per_document=2, excerpt_limit=2000)[0] self.assertIn('only after verified recovery.', hit.excerpt) def test_explicit_attribute_can_recover_a_complementary_section(self): import continuity_recall as cr vectors = {'wiki/topics/guide.md#General deployment': [1., 0.], 'wiki/topics/guide.md#More general advice': [.99, .01], 'wiki/topics/guide.md#Record artifact version and source': [.8, .6]} with patch.object(cr, '_load_semantic_vectors', return_value=vectors), \ patch.object(cr, '_embed_query', return_value=[1., 0.]), \ patch.object(cr, '_base_heading', side_effect=lambda rest, section: section), \ patch.object(cr, '_section_text', side_effect=lambda rest, section: section): hits = cr.recall_semantic('deployment version source', top_k=2, sections_per_document=2) self.assertEqual([h.section for h in hits], ['General deployment', 'Record artifact version and source']) self.assertEqual(hits[1].score, .8) def test_semantic_recall_can_return_action_beside_same_document_intro(self): import continuity_recall as cr vectors = {'wiki/topics/guide.md#Intro': [1., 0.], 'wiki/topics/guide.md#Procedure': [.99, .01]} with patch.object(cr, '_load_semantic_vectors', return_value=vectors), \ patch.object(cr, '_embed_query', return_value=[1., 0.]), \ patch.object(cr, '_base_heading', side_effect=lambda rest, section: section), \ patch.object(cr, '_section_text', side_effect=lambda rest, section: section + '\n\n' + 'step '*60): self.assertEqual(len(cr.recall_semantic('workflow', top_k=4)), 1) hits = cr.recall_semantic('workflow', top_k=4, sections_per_document=2, excerpt_limit=2000) self.assertEqual([hit.section for hit in hits], ['Intro', 'Procedure']) self.assertGreater(len(hits[1].excerpt), 200) self.assertIn('\n\n', hits[1].excerpt) def test_evaluation_rejects_document_hit_with_useless_excerpt(self): case = {'id': 'lineage', 'surface': 'delivery', 'expected_ids': ['a'], 'required_excerpt_patterns': ['source', 'version'], 'forbidden_excerpt_patterns': ['page covers']} packet = {'records': [], 'retrieval_mode': 'fixture', 'delivery': {'status': 'ready', 'records': [record('a', .9, '1.')]}} self.assertFalse(score_case(case, packet)['ok']) packet['delivery']['records'][0]['summary'] = 'This page covers source and version.' self.assertFalse(score_case(case, packet)['ok']) packet['delivery']['records'][0]['summary'] = 'Keep version. *(self-derived · source: notes.md)*' self.assertFalse(score_case(case, packet)['ok']) packet['delivery']['records'][0]['summary'] = 'For every result preserve source and version.' self.assertTrue(score_case(case, packet)['ok']) if __name__ == '__main__': unittest.main()