"""降權要會動，而且不能殺掉「存起來很久以後才用」的東西。 Pan 選了最保守的退場規則，理由是「我自己常常存一些我真的很久以後才會用到的 東西」。那個顧慮是這份測試的第一個 class：規則必須在**定義上**碰不到那種卡片， 不是靠門檻調得夠高。 退場在這個專案有前例可循，而前例都是壞的：excluded 旗標做成單行道，害一張被 解析 bug 誤排除的好卡片只能手動改檔案救回來（2026-09-11）；消化的 --apply 用 結論取代筆記，丟掉四成內容還讓壓縮比看起來變好（記憶 xkb-synthesis-silent-loss）。 兩次都是「做了不可逆的事，而且看不出來」。所以這裡每一條測的都是不可逆性： 降權的東西還在結果裡嗎 它還會被量測嗎（不然永遠回不來） 它這次真的相關的時候，會不會還是被壓掉 統計讀不到的時候，召回會不會跟著壞 """ from __future__ import annotations import sys import tempfile import unittest from unittest import mock from pathlib import Path ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import xkb_eviction as ev # noqa: E402 from xkb_evidence import identity_key import xkb_score # noqa: E402 from xkb_memory_service import KnowledgeCatalog, Store, tag_demoted # noqa: E402 class StoredForLaterSurvives(unittest.TestCase): """Pan 的顧慮。規則要在定義上碰不到這種卡片。""" def test_a_card_nobody_has_searched_for_is_never_demoted(self): """存起來以後才用的卡片，特徵是**沒被撈出來過**。 judged_count 低，所以不管 relevant 是 0 還是什麼，都不該降權。 這是這條規則跟「很久沒用到就退場」最大的差別——後者正好會殺掉它。 """ for considered in range(0, ev.DEMOTE_AFTER_CONSIDERED): with self.subTest(considered=considered): self.assertFalse(ev.is_demoted(considered, 0)) def test_demotion_needs_repeated_uselessness_not_age(self): """規則裡沒有任何時間欄位。久沒被碰過不是降權理由。""" import inspect src = inspect.getsource(ev.is_demoted) for word in ("days", "last_", "age", "timestamp", "now"): self.assertNotIn(word, src, "降權不該看時間——看時間就會殺掉存起來以後才用的東西") def test_one_hit_ever_is_enough_to_stay(self): """有一次明確的相關判斷就夠了，不需要維持命中率。 保守在這裡：這不是 gain 那套「比平均差就退」，是「從來沒有用過」。 """ self.assertFalse(ev.is_demoted(999, 1)) class DemotionIsNotRemoval(unittest.TestCase): def test_a_demoted_record_is_still_in_the_results(self): """壓到最後，不是不見。不然索引壞掉會長得像知識庫是空的。""" results = xkb_score.rank([ {"title": "好卡片", "score": 0.88, "score_scale": "card_semantic"}, {"title": "被降權的", "score": 0.10, "score_scale": "card_semantic", "demoted": True}, ]) self.assertEqual(len(results), 2) self.assertEqual(results[-1]["title"], "被降權的") def test_demoted_sinks_below_merely_weak(self): """被證實沒用過的，要排在只是這次弱的之後——limit 先切掉它。""" results = xkb_score.rank([ {"title": "這次弱", "score": 0.10, "score_scale": "card_semantic"}, {"title": "從來沒用過", "score": 0.12, "score_scale": "card_semantic", "demoted": True}, ]) self.assertEqual([r["title"] for r in results], ["這次弱", "從來沒用過"]) def test_rank_only_obeys_the_flag_it_does_not_decide(self): """排序不判斷該不該降權，只照旗標排。 判斷在合併召回處依 judge verdict 與 namespace 統計進行，不能拿檢索分數代替。 """ import inspect src = inspect.getsource(xkb_score.rank) self.assertNotIn("RELEVANCE_FLOOR) and", src) results = xkb_score.rank([ {"title": "高分但被降權", "score": 0.88, "score_scale": "card_semantic", "demoted": True}, {"title": "低分沒被降權", "score": 0.60, "score_scale": "card_semantic"}, ]) self.assertEqual(results[-1]["title"], "高分但被降權") class RevivalUsesExplicitVerdicts(unittest.TestCase): def test_retrieval_scores_do_not_override_rejections(self): for scale, score in (("card_semantic", .99), ("card_keyword", 99)): records = [{"id": "noise", "score": score, "score_scale": scale}] tag_demoted(records, lambda: {identity_key({"id": "noise"})}, relevant_ids=set()) self.assertTrue(records[0]["demoted"]) def test_current_relevance_clears_all_legs_and_stale_flags(self): records = [{"id": "noise", "score_scale": scale, "demoted": True} for scale in ("card_keyword", "card_semantic")] tag_demoted(records, lambda: {identity_key({"id": "noise"})}, relevant_ids={identity_key({"id": "noise"})}) self.assertFalse(any(r.get("demoted") for r in records)) records[0]["demoted"] = True tag_demoted(records, lambda: set()) self.assertNotIn("demoted", records[0]) class RevivalNeedsNoIntervention(unittest.TestCase): """這是跟 excluded 旗標那個單行道的差別。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.store = Store(Path(self.tmp.name) / "memory.sqlite") def _consider(self, record_id: str, similarity: float, injected: bool, times: int): for _ in range(times): item = {"id": record_id, "score": similarity, "judge": .9 if injected else .01} self.store.record_recall_usage("private", [item], [item] if injected else [], {"status": "judged"}) def test_it_becomes_demoted_after_enough_useless_considerations(self): n = ev.DEMOTE_AFTER_CONSIDERED self._consider("cards/noise.md", 0.51, False, n - 1) self.assertEqual(self.store.demoted_ids(), set(), "還沒到門檻就降權的話，保守的意義就沒了") self._consider("cards/noise.md", 0.51, False, 1) self.assertEqual(self.store.demoted_ids(), {identity_key({"id": "cards/noise.md"})}) def test_one_real_hit_lifts_the_demotion_with_nobody_deciding(self): """被降權之後還是會被量測，所以它自己能回來。 excluded 旗標做不到這件事：它把卡片從索引拿掉，於是再也不會被 judged，relevant_count 永遠是 0，永遠回不來。 """ self._consider("cards/borderline.md", 0.53, False, ev.DEMOTE_AFTER_CONSIDERED) self.assertIn(identity_key({"id": "cards/borderline.md"}), self.store.demoted_ids()) # 有一天某個問法讓它通過地板。 self._consider("cards/borderline.md", 0.71, True, 1) self.assertEqual(self.store.demoted_ids(), set(), "降權必須自動解除，不能要人手動改檔案") def test_the_threshold_has_one_definition(self): """門檻不准有第二份拷貝——尺度與門檻在這個專案各自被複製過一次。""" self.assertEqual( self.store.demoted_ids.__defaults__, (ev.DEMOTE_AFTER_CONSIDERED,)) class ItMustNotDependOnWhichLegFoundIt(unittest.TestCase): """第一版掛在語意腿尾端，真實資料一驗就破。 每條召回腿都自己新建 dict（語意、關鍵字、wiki 各有一份欄位清單），所以掛在 其中一條上的標記會被其他腿繞過。那張被撈出 8 次、從來沒用上的卡片，在日文 查詢下是**關鍵字腿**撈到的——標記完全沒掛上，而單元測試全綠。 這是記憶 xkb-many-writers-one-reader 的同一個形狀：保護要放在讀的那一端。 這裡用關鍵字腿的欄位形狀（line 666 那一長串，沒有經過 _drop_irrelevant） 來證明標記與腿無關。 """ def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.store = Store(Path(self.tmp.name) / "memory.sqlite") for _ in range(ev.DEMOTE_AFTER_CONSIDERED): self.store.record_recall_usage("private", [{"id": "02-seo-geo/999", "judge": .01}], [], {"status": "judged"}) def _keyword_leg_record(self, record_id: str, score: float) -> dict: """關鍵字腿產出的形狀。它不經過 _drop_irrelevant。""" return {"record_type": "knowledge_card", "id": record_id, "title": record_id, "summary": "", "source_file": f"cards/{record_id}.md", "source_type": "card", "score_scale": "card_keyword", "score": score, "retrieval": "keyword"} def test_a_keyword_leg_hit_is_still_demoted(self): demoted_id = "02-seo-geo/999" records = [self._keyword_leg_record(demoted_id, 3.0), self._keyword_leg_record("01-good/111", 1.0)] self.store.catalog.search = lambda *a, **k: { "records": records, "filtered_counts": {"by_layer": {}}} self.store.recall = lambda *a, **k: {"memories": []} with mock.patch("xkb_memory_service.xkb_jev.relevance", return_value=None): out = self.store.knowledge_recall("ヘアメイク プロンプト", limit=10) got = {r["id"]: r for r in out["records"]} self.assertTrue(got[demoted_id].get("demoted"), "關鍵字腿撈到的也要被降權——不然換一條腿就繞過了") self.assertFalse(got["01-good/111"].get("demoted")) def test_no_leg_tags_on_its_own(self): """標記只能在合併點發生。多一處就多一條可以不一致的路。""" import inspect from xkb_memory_service import KnowledgeCatalog as KC, Store as St per_leg = (KC._semantic_search, KC._wiki_search, KC._drop_irrelevant, KC.search) for fn in per_leg: with self.subTest(fn=fn.__name__): self.assertNotIn("tag_demoted", inspect.getsource(fn), "降權不要掛在單一條腿上") self.assertIn("tag_demoted", inspect.getsource(St.knowledge_recall), "合併點沒有標記，降權就完全不會發生") class LosingTheStatsMustNotBreakRecall(unittest.TestCase): """降權是最佳化，不是正確性。""" def test_no_sink_tags_nothing(self): records = [{"id": "cards/a.md"}] tag_demoted(records, None) self.assertNotIn("demoted", records[0]) def test_a_failing_sink_is_swallowed(self): def boom(): raise RuntimeError("統計表不見了") records = [{"id": "cards/a.md"}] tag_demoted(records, boom) # 不可以炸 self.assertNotIn("demoted", records[0]) def test_it_tags_only_the_matching_record(self): records = [{"id": "cards/noise.md"}, {"id": "cards/good.md"}] tag_demoted(records, lambda: {identity_key({"id": "cards/noise.md"})}) self.assertTrue(records[0].get("demoted")) self.assertNotIn("demoted", records[1]) def test_the_catalog_does_not_need_to_know_about_demotion(self): """降權跟「去哪裡找知識」無關——catalog 替身不該為它多長一個方法。""" self.assertFalse(hasattr(KnowledgeCatalog(), "_tag_demoted")) self.assertFalse(hasattr(KnowledgeCatalog(), "demoted_sink")) if __name__ == "__main__": unittest.main()