tests/test_xkb_memory_service.py
Rama main · réplica de lectura del repositorio original.
from __future__ import annotations
import importlib.util
import tempfile
import unittest
import unittest.mock
from unittest import mock
from pathlib import Path
SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "xkb_memory_service.py"
spec = importlib.util.spec_from_file_location("xkb_memory_service", SCRIPT)
assert spec and spec.loader
module = importlib.util.module_from_spec(spec)
import sys
sys.modules[spec.name] = module
spec.loader.exec_module(module)
Store = module.Store
_wiki_patch = None
def setUpModule() -> None:
"""讓測試不要碰到真實的知識庫。
recall 加入 wiki 搜尋之後，任何呼叫 knowledge_recall 的測試都會去載入
使用者真實的語意索引（本機 1,872 筆），再對 Gemini 打一次 embedding。
那讓測試變慢、要花錢、依賴網路，而且結果隨著知識庫內容改變——
在這台機器上甚至會 segfault。
預設回空，個別測試要驗 wiki 行為時再自己 patch。
"""
global _wiki_patch
_wiki_patch = mock.patch.object(module.KnowledgeCatalog, "_wiki_search", return_value=[])
_wiki_patch.start()
def tearDownModule() -> None:
if _wiki_patch is not None:
_wiki_patch.stop()
class XKBMemoryServiceTests(unittest.TestCase):
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
self.store = Store(Path(self.tmp.name) / "memory.sqlite")
self.store.catalog.index_file = Path(self.tmp.name) / "search_index.json"
self.store.catalog.cards_dir = Path(self.tmp.name) / "cards"
self.store.catalog.wiki_topics_dir = Path(self.tmp.name) / "wiki" / "topics"
def tearDown(self) -> None:
self.tmp.cleanup()
def test_session_turn_complete_recall_and_idempotency(self) -> None:
session = self.store.open_session(
{
"source": "fixture",
"agent_id": "fixture",
"namespace": "private",
"session_key": "session-1",
}
)
resumed = self.store.open_session(
{
"source": "fixture",
"agent_id": "fixture",
"namespace": "private",
"session_key": "session-1",
}
)
self.assertFalse(session["resumed"])
self.assertTrue(resumed["resumed"])
turn = self.store.start_turn(
{
"session_id": session["session_id"],
"turn_id": "turn-1",
"query": "shared memory query",
}
)
self.assertFalse(turn["resumed"])
completed = self.store.complete_turn(
"turn-1",
{
"session_id": session["session_id"],
"query": "shared memory query",
"answer": "shared memory answer",
"status": "succeeded",
"content": {"answer": "shared memory answer"},
},
)
duplicate = self.store.complete_turn(
"turn-1",
{
"session_id": session["session_id"],
"query": "shared memory query",
"answer": "shared memory answer",
"status": "succeeded",
"content": {"answer": "shared memory answer"},
},
)
self.assertTrue(completed["stored"])
self.assertEqual(completed["retrieval"]["query"], "shared memory query")
with self.store.connect() as db:
persisted = db.execute("SELECT status, retrieval_json, trace_id FROM turns WHERE turn_id=?", ("turn-1",)).fetchone()
self.assertEqual(persisted["status"], "succeeded")
self.assertEqual(persisted["trace_id"], completed["trace_id"])
self.assertEqual(__import__("json").loads(persisted["retrieval_json"])["query"], "shared memory query")
self.assertTrue(duplicate["deduplicated"])
recall = self.store.recall("shared memory")
self.assertEqual(recall["count"], 1)
self.assertIn("shared memory answer", recall["context"])
artifact = self.store.artifact(completed["trace_id"])
self.assertIsNotNone(artifact)
self.assertEqual(artifact["retrieval"]["query"], "shared memory query")
def test_input_validation_rejects_empty_query_bad_limits_and_status(self) -> None:
with self.assertRaisesRegex(ValueError, "query is required"):
self.store.knowledge_recall(" ")
with self.assertRaisesRegex(ValueError, "limit must be an integer"):
self.store.knowledge_recall("query", limit="10")
with self.assertRaisesRegex(ValueError, "between 1 and 50"):
self.store.knowledge_recall("query", limit=0)
session = self.store.open_session({"source": "fixture", "session_key": "validation"})
with self.assertRaisesRegex(ValueError, "namespace is required"):
self.store.open_session({"source": "fixture", "session_key": "bad", "namespace": ""})
self.store.start_turn({"session_id": session["session_id"], "turn_id": "validation-turn", "query": "query"})
with self.assertRaisesRegex(ValueError, "status must be"):
self.store.complete_turn("validation-turn", {"query": "query", "answer": "answer", "status": "running"})
def test_cancelled_turn_is_not_stored(self) -> None:
session = self.store.open_session({"source": "fixture", "session_key": "session-2"})
self.store.start_turn({"session_id": session["session_id"], "turn_id": "turn-2", "query": "cancel me"})
result = self.store.complete_turn("turn-2", {"status": "cancelled"})
self.assertFalse(result["stored"])
self.assertEqual(self.store.recall("cancel me")["count"], 0)
def test_unified_recall_includes_external_knowledge_records(self) -> None:
original = self.store.catalog
class FixtureCatalog:
def search(self, query, limit=10, namespace="private"):
return {
"records": [{
"record_type": "knowledge_card",
"id": "fixture-card",
"title": "Fixture knowledge",
"summary": "shared knowledge evidence",
"score": 3,
}]
}
self.store.catalog = FixtureCatalog()
try:
result = self.store.knowledge_recall("shared knowledge")
finally:
self.store.catalog = original
self.assertEqual(result["count"], 1)
self.assertEqual(result["records"][0]["record_type"], "knowledge_card")
self.assertIn("Fixture knowledge", result["context"])
def test_catalog_fails_closed_across_namespaces_and_exposes_metadata(self) -> None:
catalog = module.KnowledgeCatalog()
root = Path(self.tmp.name)
catalog.cards_dir = root / "cards"
catalog.wiki_topics_dir = root / "topics"
catalog.index_file = root / "index.json"
catalog.cards_dir.mkdir()
catalog.wiki_topics_dir.mkdir()
(catalog.cards_dir / "secret.md").write_text("---\nnamespace: team-a\nsensitivity: internal\n---\nsecret", encoding="utf-8")
(catalog.index_file).write_text('{"items":[{"id":"secret","title":"Secret"}]}', encoding="utf-8")
self.assertIsNone(catalog.card("secret", "team-b"))
visible = catalog.card("secret", "team-a")
self.assertEqual(visible["namespace"], "team-a")
self.assertEqual(catalog.search("secret", namespace="team-b")["count"], 0)
def test_semantic_backend_is_used_and_declared(self) -> None:
original = module.xbrain_query
def fixture_query(query, *, limit, no_expand, semantic, diagnostics=None):
self.assertEqual(query, "semantic fixture")
self.assertTrue(semantic)
return [{
"slug": "fixture-semantic-card",
"title": "Semantic fixture",
"chunk_text": "vector evidence",
"source_url": "https://example.test/source",
"type": "knowledge-card",
"score": 0.91,
}]
module.xbrain_query = fixture_query
try:
# 前提要寫出來。原本這個測試能過，只是因為 fixture 的 slug 剛好
# 不在這台機器的 cards_index 裡——換一台機器、或哪天真的建了同名
# 的卡片，斷言就會翻面。測試不該從磁碟繼承它的前提。
import xkb_relevance
with unittest.mock.patch.object(xkb_relevance, "similarities",
return_value=None):
result = self.store.catalog.search("semantic fixture", 3)
finally:
module.xbrain_query = original
self.assertEqual(result["retrieval_mode"], "xbrain_hybrid")
self.assertEqual(result["records"][0]["retrieval"], "xbrain_hybrid")
record = result["records"][0]
# 一個鍵都對不到向量索引時，這裡不做任何比較，也就不該假裝比過：
# 後端的分數原樣帶著，只加上「沒驗證過」的標籤。
# 跨層可比是合併點（knowledge_recall → xkb_score.rank）的事，
# 不是這裡編一個位置出來——我在這個檔案上來回改了三輪才把它放對地方。
self.assertEqual(record["score"], 0.91)
self.assertEqual(record.get("score_basis"), "unverified")
# 沒被驗證過的記錄有自己的尺度。這件事以前是靠在 xkb_relevance 裡
# 改寫分數做到的，那等於捏造一個測量值；現在 score 保持是後端自己
# 算的數字，排序交給 xkb_score 的尺度表。
self.assertEqual(record.get("score_scale"), "unverified")
self.assertNotIn("_unverified", record)
def test_layers_are_ordered_by_a_comparable_score_not_raw_numbers(self) -> None:
"""合併之後要照跨層可比的分數排，不是照各層自己的原始數字。
三層的尺度互不相同：卡片是 gbrain 的 RRF、wiki 是餘弦、對話是關鍵字
比例（上限 0.65）。合併點原本直接照 score 排，等於拿公分比英吋。
這是外部審查連續六輪繞著打轉的那個缺陷——每一輪我都在 xkb_relevance
裡替驗不出相似度的項目找「壓到哪裡才對」，而不管壓到哪裡都會撞上另一層
的尺度：壓高壓過 wiki 的真餘弦，壓低掉到對話軌跡之下，而後者會讓索引
壞掉長得像「知識庫裡沒東西」。真正該修的是合併點。
測試要走真正的路徑。我第一版直接呼叫 xkb_score.rank()，於是把合併點
改回原始排序它照樣是綠的——測到的是那個函式，不是那條路。
"""
def packet_for(card_score: float) -> list[str]:
knowledge = {
"records": [
{"title": "卡片", "score": card_score, "score_scale": "card_semantic",
"record_type": "knowledge_chunk"},
{"title": "wiki 結論", "score": 0.68, "score_scale": "wiki_semantic",
"record_type": "wiki_topic"},
],
"retrieval_mode": "xbrain_hybrid", "filtered_counts": {},
"semantic_retrieval_attempted": True,
"semantic_backend": {"status": "used"},
"dropped_as_irrelevant": 0,
}
conversation = {"memories": [{"query": "對話軌跡", "answer": "", "score": 0.65}]}
with mock.patch.object(type(self.store.catalog), "search",
return_value=knowledge), \
mock.patch.object(type(self.store), "recall", return_value=conversation):
packet = self.store.knowledge_recall("任意問題", 5, "private")
return [r.get("title") or r.get("query") for r in packet["records"]]
strong = packet_for(0.75)
self.assertEqual(strong[0], "wiki 結論",
"消化過的結論該排在相近強度的原始素材之前")
self.assertLess(strong.index("卡片"), strong.index("對話軌跡"),
"語意命中的卡片該排在關鍵字命中的對話軌跡之前")
# 對話軌跡仍然在封包裡——不是被系統性地丟掉。RRF 之下它就是另一條腿，
# 照名次貢獻，不再被雙重折扣（呼叫端 0.65 折扣 + 尺度表未折扣錨點）
# 壓到門檻之下、一填滿 limit 就整段消失。
marginal = packet_for(0.56)
self.assertIn("對話軌跡", marginal,
"對話層被整段擠出封包")
# 截斷防護：知識層填滿 limit 時，最強的對話軌跡仍要進封包（不是被
# marginal 卡片擠到 records[:limit] 之外）。這是舊測試那條順序斷言真正
# 要守的東西——RRF 之下順序斷言不再成立，但「不被整段截掉」還是要守。
many_cards = {
"records": [
{"title": f"卡片{n}", "score": 0.56, "score_scale": "card_semantic",
"record_type": "knowledge_chunk"} for n in range(8)
],
"retrieval_mode": "xbrain_hybrid", "filtered_counts": {},
"semantic_retrieval_attempted": True,
"semantic_backend": {"status": "used"}, "dropped_as_irrelevant": 0,
}
conversation = {"memories": [{"query": "強對話軌跡", "answer": "", "score": 0.65}]}
with mock.patch.object(type(self.store.catalog), "search", return_value=many_cards), mock.patch.object(type(self.store), "recall", return_value=conversation):
packet = self.store.knowledge_recall("任意", 5, "private")
titles = [r.get("title") or r.get("query") for r in packet["records"]]
self.assertIn("強對話軌跡", titles, "知識層填滿 limit 後對話軌跡被整段截掉")
# scale mixing 結構上不可能：關鍵字尺度的分數（6~20）跟餘弦尺度的
# 分數（0~1）在 RRF 裡都只是「腿內第 N 名」，不會因為原始數字大就碾壓。
import xkb_score
mixed = xkb_score.rank([
{"title": "關鍵字命中的書籤", "score": 15.0, "score_scale": "card_keyword"},
{"title": "餘弦命中的卡片", "score": 0.62, "score_scale": "card_semantic"},
])
self.assertLess(abs(mixed[0]["unified_score"] - mixed[1]["unified_score"]),
0.001,
"15 分的關鍵字書籤把 0.62 的餘弦卡片碾壓了——scale mixing 回來了")
def test_each_producer_declares_a_scale_xkb_score_knows(self) -> None:
"""算出分數的地方要標對尺度，而且是 xkb_score 認得的那些。
source_type 是「這是什麼東西」（會回給 API 用戶），score_scale 是
「這個數字怎麼來的」。標錯或沒標，整層就會用錯錨點對齊，而且不會有
任何錯誤訊息——不認得的尺度會靜靜掉到 0.5 的 fallback。
我第一版這個測試是在原始碼文字裡找 `score_scale` 這個子字串，於是
它對「值是錯的」完全無感——寫這個測試的當下就有兩個標籤是錯的，
它照樣是綠的。改成檢查實際產出的值。
"""
import xkb_score
known = set(xkb_score.DEFAULT_ANCHORS) & set(xkb_score.DEFAULT_WEIGHTS)
def fixture_query(query, *, limit, no_expand, semantic, diagnostics=None):
return [{"slug": "s", "title": "t", "chunk_text": "c",
"source_url": "", "type": "knowledge-card", "score": 0.8}]
original = module.xbrain_query
module.xbrain_query = fixture_query
try:
packet = self.store.knowledge_recall("尺度測試", 5, "private")
finally:
module.xbrain_query = original
scales = {r.get("score_scale") for r in packet["records"]}
self.assertTrue(scales, "一筆記錄都沒有，測不到尺度")
self.assertNotIn(None, scales, "有記錄沒有標尺度")
for scale in scales:
self.assertIn(scale, known,
f"{scale} 不在 xkb_score 的尺度表裡，會靜靜掉到 fallback")
def test_every_record_template_declares_its_specific_scale(self) -> None:
"""每一個產生分數的樣板都要標對尺度——包括這個測試檔案 mock 掉的那些。
動態的那個測試只走得到 fixture 那一筆：setUpModule 把 _wiki_search 整個
換成回空陣列、store 也是空的，所以 wiki 與對話那兩個樣板沒有任何測試
看著。我刪掉的靜態測試本來守著它們，替換的版本反而涵蓋更少。
而靜態檢查要看**值**。我更早的版本只找 score_scale 這個欄位名，
於是四個標籤裡有兩個是錯的時候它照樣是綠的。
"""
source = (Path(module.__file__).parent / "xkb_memory_service.py").read_text(
encoding="utf-8")
expected = {
'"retrieval": "wiki_semantic"': "wiki_semantic",
'"record_type": "conversation_trace"': "conversation",
}
# GBrain declares its scale at the adapter boundary; the service must
# preserve it. Runtime coverage is in test_recall_backend_contract.
self.assertIn('"score_scale": hit.get("score_scale", "card")', source)
for marker, scale in expected.items():
idx = source.index(marker)
window = source[max(0, idx - 700):idx + 300]
self.assertIn(f'"score_scale": "{scale}"', window,
f"{marker} 的樣板沒有標成 {scale}")
def test_the_keyword_fallback_declares_a_keyword_scale(self) -> None:
"""關鍵字退路的分數是 0~1 的覆蓋率，不是 RRF 也不是餘弦。
這條路徑上的記錄一度被標成 card / wiki_semantic——用餘弦與 RRF 的錨點
去對齊覆蓋率，會把真實的先後順序翻過來，正是收斂想消滅的那件事，
只是搬到了退路上。上一版的測試連這兩個樣板都沒看到。
"""
source = (Path(module.__file__).parent / "xkb_memory_service.py").read_text(
encoding="utf-8")
for marker in ('"retrieval": "keyword"',):
idx = 0
found = 0
while True:
idx = source.find(marker, idx + 1)
if idx < 0:
break
found += 1
template = source[max(0, idx - 700):idx]
self.assertRegex(
template, r'"score_scale": "(card_keyword|wiki_keyword)"',
"關鍵字退路的記錄要標關鍵字尺度")
self.assertGreater(found, 0, "找不到關鍵字退路的記錄樣板")
def test_recall_packet_explains_acl_namespace_filtering_and_backend(self) -> None:
original = module.xbrain_query
def fixture_query(query, *, limit, no_expand, semantic, diagnostics=None):
return [
{"slug": "allowed", "title": "Allowed", "chunk_text": "visible", "namespace": "team-a"},
{"slug": "hidden", "title": "Hidden", "chunk_text": "secret", "namespace": "team-b"},
]
module.xbrain_query = fixture_query
try:
result = self.store.knowledge_recall("acl packet", namespace="team-a")
finally:
module.xbrain_query = original
self.assertEqual(result["request_namespace"], "team-a")
self.assertEqual(result["acl_policy"]["decision"], "allow_public_or_matching_namespace")
self.assertEqual(result["filtered_counts"]["semantic"], 1)
self.assertIn("records_filtered_by_acl", result["warnings"])
self.assertEqual(result["semantic_backend"]["status"], "used")
# The breakdown must say which layer was filtered, otherwise "nothing
# came back" and "the wiki layer was filtered out" look identical.
self.assertEqual(result["filtered_counts"]["total"], 1)
self.assertEqual(result["filtered_counts"]["by_layer"]["semantic"], 1)
self.assertEqual(result["filtered_counts"]["by_layer"]["conversation"], 0)
def test_acl_filter_counts_are_attributed_per_layer(self) -> None:
"""A quiet result must say which layer the ACL removed."""
module = importlib.import_module("xkb_memory_service")
original = module.xbrain_query
module.xbrain_query = None # force the keyword path, where layers are known
try:
result = self.store.knowledge_recall("anything", namespace="team-a")
finally:
module.xbrain_query = original
counts = result["filtered_counts"]
self.assertEqual(result["retrieval_mode"], "keyword_fallback")
self.assertEqual(set(counts["by_layer"]), {"card", "wiki", "semantic", "conversation"})
self.assertEqual(counts["total"], sum(counts["by_layer"].values()))
# No semantic backend ran, so nothing can be attributed to it.
self.assertEqual(counts["by_layer"]["semantic"], 0)
# Legacy channel keys stay consistent with the layer breakdown.
self.assertEqual(counts["keyword"], counts["by_layer"]["card"] + counts["by_layer"]["wiki"])
def test_turn_start_persists_retrieval_packet_and_source_ids(self) -> None:
original = self.store.knowledge_recall
packet = {
"schema": "xkb-knowledge-service.v1",
"query": "turn query",
"records": [{"id": "knowledge-1", "record_type": "knowledge_chunk", "score": 0.8}],
"context": "historical evidence",
"retrieval_mode": "xbrain_hybrid",
"semantic_retrieval_attempted": True,
"warnings": [],
}
self.store.knowledge_recall = lambda query, limit=10, namespace="private", conversation=None, **_: packet
try:
session = self.store.open_session({"source": "fixture", "session_key": "retrieval-session"})
started = self.store.start_turn({"session_id": session["session_id"], "turn_id": "retrieval-turn", "query": "turn query"})
resumed = self.store.start_turn({"session_id": session["session_id"], "turn_id": "retrieval-turn", "query": "turn query"})
finally:
self.store.knowledge_recall = original
self.assertEqual(started["source_memory_ids"], ["knowledge-1"])
self.assertEqual(started["retrieval"]["retrieval_mode"], "xbrain_hybrid")
self.assertTrue(resumed["resumed"])
self.assertEqual(resumed["retrieval"]["query"], "turn query")
def test_turn_start_rejects_idempotency_conflicts(self) -> None:
first = self.store.open_session({"source": "fixture", "session_key": "identity-1", "namespace": "private"})
other = self.store.open_session({"source": "fixture", "session_key": "identity-2", "namespace": "private"})
self.store.start_turn({"session_id": first["session_id"], "turn_id": "identity-turn", "query": "original"})
with self.assertRaisesRegex(ValueError, "session_id"):
self.store.start_turn({"session_id": other["session_id"], "turn_id": "identity-turn", "query": "original"})
with self.assertRaisesRegex(ValueError, "query"):
self.store.start_turn({"session_id": first["session_id"], "turn_id": "identity-turn", "query": "changed"})
with self.assertRaisesRegex(ValueError, "namespace"):
self.store.start_turn({"session_id": first["session_id"], "turn_id": "identity-turn", "query": "original", "namespace": "shared"})
def test_recall_is_scoped_to_session_namespace_and_start_requires_matching_namespace(self) -> None:
private = self.store.open_session({"source": "fixture", "session_key": "private", "namespace": "private"})
shared = self.store.open_session({"source": "fixture", "session_key": "shared", "namespace": "shared"})
self.store.start_turn({"session_id": private["session_id"], "turn_id": "private-turn", "query": "same namespace secret"})
self.store.complete_turn("private-turn", {"query": "same namespace secret", "answer": "private answer"})
self.store.start_turn({"session_id": shared["session_id"], "turn_id": "shared-turn", "query": "same namespace secret"})
self.store.complete_turn("shared-turn", {"query": "same namespace secret", "answer": "shared answer"})
self.assertEqual([m["answer"] for m in self.store.recall("same namespace", namespace="private")["memories"]], ["private answer"])
self.assertEqual([m["answer"] for m in self.store.recall("same namespace", namespace="shared")["memories"]], ["shared answer"])
with self.assertRaisesRegex(ValueError, "namespace conflicts with session"):
self.store.start_turn({"session_id": private["session_id"], "turn_id": "bad-namespace", "query": "q", "namespace": "shared"})
def test_turn_start_rejects_unknown_session(self) -> None:
with self.assertRaisesRegex(ValueError, "unknown session_id"):
self.store.start_turn({"session_id": "missing", "turn_id": "missing-turn", "query": "q"})
def test_pipeline_snapshot_is_read_only_and_lists_known_stages(self) -> None:
snapshot = self.store.catalog.pipeline_snapshot(days=7)
self.assertTrue(snapshot["read_only"])
self.assertEqual(snapshot["control_plane"], "observed_status_only")
stages = {item["stage"] for item in snapshot["stages"]}
self.assertEqual(
stages,
{"ingest", "card_generation", "index", "distill", "promotion", "publish"},
)
self.assertTrue(all(item["control"] == "read_only" for item in snapshot["stages"]))
self.assertIn("summary", snapshot)
def test_job_event_is_persisted_and_filterable_without_running_worker(self) -> None:
created = self.store.record_job_event({
"job_id": "job-1",
"stage": "index",
"worker": "build_vector_index.py",
"status": "running",
"input_ref": "cards:batch-1",
"metadata": {"count": 3},
})
self.assertTrue(created["stored"])
self.assertTrue(created["observed_only"])
completed = self.store.record_job_event({
"job_id": "job-1",
"stage": "index",
"worker": "build_vector_index.py",
"status": "succeeded",
"output_ref": "vector-index:batch-1",
})
self.assertEqual(completed["status"], "succeeded")
result = self.store.list_jobs(stage="index", status="succeeded")
self.assertEqual(result["count"], 1)
self.assertEqual(result["jobs"][0]["output_ref"], "vector-index:batch-1")
self.assertFalse(self.store.list_jobs(stage="distill")["count"])
class RelevanceAndIntentTests(unittest.TestCase):
"""Injecting ten results into every turn is what makes an unrelated
question cost as much as a real one."""
def setUp(self) -> None:
self.module = module
def test_acknowledgements_and_greetings_skip_retrieval_entirely(self) -> None:
skip = self.module.Store._skip_reason
self.assertEqual(skip("好"), "acknowledgement")
self.assertEqual(skip("謝謝"), "acknowledgement")
self.assertEqual(skip("早安"), "greeting")
self.assertEqual(skip("hi"), "greeting")
def test_real_questions_are_never_skipped(self) -> None:
"""Eight Chinese characters is a question, not chitchat."""
skip = self.module.Store._skip_reason
self.assertEqual(skip("碳盤查的計算方式"), "")
self.assertEqual(skip("XKB 召回"), "")
self.assertEqual(skip("我們之前怎麼處理碳盤查的"), "")
def test_vector_key_maps_slug_onto_index_key(self) -> None:
import xkb_relevance
self.assertEqual(xkb_relevance.vector_key("01-topic/12345"), "01-topic/12345.md")
self.assertEqual(xkb_relevance.vector_key("01-topic/12345.md"), "01-topic/12345.md")
self.assertEqual(xkb_relevance.vector_key("https://x.com/i/status/1"), "")
self.assertEqual(xkb_relevance.vector_key(""), "")
def test_rank_score_is_replaced_by_measured_similarity(self) -> None:
"""The backend's score says "ranked first", not "is relevant"."""
catalog = self.module.KnowledgeCatalog()
records = [{"id": "a/1", "score": 0.88}, {"id": "a/2", "score": 0.87}]
with mock.patch("xkb_relevance.similarities",
return_value={"a/1.md": 0.72, "a/2.md": 0.30}):
kept, dropped = catalog._drop_irrelevant("q", records)
self.assertEqual(dropped, 1)
self.assertEqual([item["id"] for item in kept], ["a/1"])
self.assertEqual(kept[0]["score"], 0.72)
self.assertEqual(kept[0]["rank_score"], 0.88)
def test_unavailable_similarity_passes_records_through(self) -> None:
"""A missing index must not silently delete every result."""
catalog = self.module.KnowledgeCatalog()
records = [{"id": "a/1", "score": 0.88}]
with mock.patch("xkb_relevance.similarities", return_value=None):
kept, dropped = catalog._drop_irrelevant("q", records)
self.assertEqual((len(kept), dropped), (1, 0))
def test_warnings_separate_broken_backend_from_nothing_relevant(self) -> None:
broken = self.module._recall_warnings(
{"retrieval_mode": "keyword_fallback", "dropped_as_irrelevant": 0}, {})
self.assertIn("semantic_backend_unavailable_or_empty", broken[0])
nothing_relevant = self.module._recall_warnings(
{"retrieval_mode": "keyword_fallback", "dropped_as_irrelevant": 10}, {})
self.assertIn("relevance floor", nothing_relevant[0])
self.assertNotIn("unavailable", nothing_relevant[0])
class KnowledgeRetirementTests(unittest.TestCase):
"""Wiki and cards only ever grew; nothing ever aged out."""
def setUp(self) -> None:
self.tmp = tempfile.TemporaryDirectory()
self.store = Store(Path(self.tmp.name) / "memory.sqlite")
self.addCleanup(self.tmp.cleanup)
def test_records_that_never_clear_the_floor_are_reported_as_cold(self) -> None:
for _ in range(6):
self.store.record_usage([("cold/1", 0.31, False), ("warm/1", 0.72, True)])
report = self.store.cold_knowledge(min_considered=5)
self.assertEqual([r["record_id"] for r in report["records"]], ["cold/1"])
self.assertEqual(report["tracked_records"], 2)
self.assertEqual(report["ever_useful"], 1)
def test_useful_records_are_never_listed_however_often_retrieved(self) -> None:
for _ in range(20):
self.store.record_usage([("warm/1", 0.40, False)])
self.store.record_usage([("warm/1", 0.80, True)])
report = self.store.cold_knowledge(min_considered=5)
self.assertEqual(report["records"], [])
def test_counters_accumulate_and_keep_the_best_similarity(self) -> None:
self.store.record_usage([("a/1", 0.30, False)])
self.store.record_usage([("a/1", 0.51, False)])
row = self.store.cold_knowledge(min_considered=1)["records"][0]
self.assertEqual(row["considered_count"], 2)
self.assertEqual(row["injected_count"], 0)
self.assertAlmostEqual(row["best_similarity"], 0.51, places=4)
self.assertIsNone(row["last_injected_at"])
def test_retirement_is_advisory_only(self) -> None:
"""Provenance is the point; nothing may be deleted automatically."""
self.store.record_usage([("cold/1", 0.10, False)] * 1)
report = self.store.cold_knowledge(min_considered=1)
self.assertFalse(report["automatic_retirement"])
self.assertTrue(report["read_only"])
def test_usage_accounting_never_breaks_recall(self) -> None:
catalog = module.KnowledgeCatalog()
catalog.usage_sink = lambda observations: (_ for _ in ()).throw(RuntimeError("db down"))
with mock.patch("xkb_relevance.similarities", return_value={"a/1.md": 0.9}):
kept, dropped = catalog._drop_irrelevant("q", [{"id": "a/1", "score": 0.88}])
self.assertEqual((len(kept), dropped), (1, 0))
class QueryEmbeddingCacheTests(unittest.TestCase):
"""The same query is embedded twice per recall: once by hybrid search,
once by the relevance filter. The second call is a wasted round trip."""
def test_repeated_query_is_embedded_once(self) -> None:
import continuity_recall as cr
cr._QUERY_VECTORS.clear()
calls = []
def fake(query):
calls.append(query)
return [0.1, 0.2]
with mock.patch.object(cr, "_embed_query_uncached", side_effect=fake):
first = cr._embed_query("same question")
second = cr._embed_query("same question")
cr._embed_query("different question")
self.assertEqual(first, second)
self.assertEqual(calls, ["same question", "different question"])
cr._QUERY_VECTORS.clear()
def test_failure_is_cached_too_so_a_dead_provider_is_not_retried(self) -> None:
import continuity_recall as cr
cr._QUERY_VECTORS.clear()
calls = []
with mock.patch.object(cr, "_embed_query_uncached", side_effect=lambda q: calls.append(q) or None):
self.assertIsNone(cr._embed_query("q"))
self.assertIsNone(cr._embed_query("q"))
self.assertEqual(len(calls), 1)
cr._QUERY_VECTORS.clear()
def test_cache_is_bounded(self) -> None:
import continuity_recall as cr
cr._QUERY_VECTORS.clear()
with mock.patch.object(cr, "_embed_query_uncached", return_value=[0.0]):
for i in range(cr._QUERY_VECTOR_LIMIT + 5):
cr._embed_query(f"q{i}")
self.assertLessEqual(len(cr._QUERY_VECTORS), cr._QUERY_VECTOR_LIMIT)
cr._QUERY_VECTORS.clear()
if __name__ == "__main__":
unittest.main()