這段 HTML 要怎麼改", "<3", "<= 這個符號怎麼念"): with self.subTest(text=text): self.assertFalse(is_harness_text(text)) class TheHookDoesNotOpenATurnForIt(unittest.TestCase): """不是「開了 turn 但不召回」——它根本不是一輪對話。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) def _run(self, prompt: str) -> list[str]: calls: list[str] = [] def spy(path, body, cfg): calls.append(path) return {"session_id": "s", "turn_id": "t", "retrieval": {"records": []}} state = Path(self.tmp.name) / "state.json" with mock.patch.object(xkb_agent_hook, "call", spy), \ mock.patch.object(xkb_agent_hook, "session_key", return_value="k"), \ mock.patch.object(xkb_agent_hook, "next_ordinal", return_value=1), \ mock.patch.object(xkb_agent_hook, "STATE_DIR", Path(self.tmp.name)), \ mock.patch.object(xkb_agent_hook, "state_path", return_value=state), \ mock.patch.object(xkb_agent_hook, "emit"): xkb_agent_hook.on_prompt({"prompt": prompt}, {"source": "test"}) return calls def test_harness_text_leaves_no_state_behind(self): """不開 turn，也不該留下狀態——留下的話會讓下一輪對不上。""" self._run(REAL_TASK_NOTIFICATION) self.assertEqual(list(Path(self.tmp.name).glob("*.json")), []) def test_a_task_notification_opens_nothing(self): self.assertEqual(self._run(REAL_TASK_NOTIFICATION), [], "連 session 都不該開——它不是一輪對話") def test_the_openclaw_internal_context_opens_nothing(self): self.assertEqual(self._run(REAL_OPENCLAW_CONTEXT), []) def test_a_real_prompt_still_opens_a_turn(self): calls = self._run(REAL_QUESTION) self.assertIn("/v1/sessions/open", calls) self.assertIn("/v1/turns/start", calls) def test_pasted_content_still_opens_a_turn(self): calls = self._run(REAL_PASTED_CONTENT) self.assertIn("/v1/turns/start", calls) class OneDefinitionForBothEnds(unittest.TestCase): """hook 擋一次、召回再擋一次，但判斷只能有一份。兩邊各寫一份就會各自漂移——這個專案的「多寫入者一讀取者」記錄在案，而 _skip_reason 的註解裡也寫著它從 parser 拿清單的理由：它自己那份拷貝曾經漏掉複合應答，於是「ok 收到」撈了十筆進一個什麼都沒問的對話。 """ def test_the_hook_imports_the_definition(self): import inspect source = inspect.getsource(xkb_agent_hook) self.assertIn("from conversation_state_parser import is_harness_text", source) self.assertNotIn("task-notification", source, "標記清單不要在 hook 裡再寫一份") def test_recall_skips_it_through_the_same_function(self): import inspect from xkb_memory_service import Store source = inspect.getsource(Store._skip_reason) self.assertIn("_noise_kind", source) self.assertEqual(noise_kind(REAL_TASK_NOTIFICATION), "harness_text", "召回端的 skip_reason 會直接變成這個字串") if __name__ == "__main__": unittest.main() class TheOldRowsStillCrowdTheRecallWindow(unittest.TestCase): """在 hook 端擋掉之前累積的那些列，還躺在資料庫裡。對話召回只看**最近 500 筆** turn。2026-09-24 的資料庫裡有 137 筆 turn 的 query 是背景任務通知——也就是視窗的 27% 被佔掉，真正的對話歷史被擠出去。在 Python 端過濾救不了這件事：它們已經佔掉名額了，所以排除要發生在 SQL 裡。刻意不刪那些列：它們的 answer 是真實的回答內容。代價是那些答案從此不會被對話召回撈到——真有知識價值的內容，它的家是卡片或 wiki，不是一個問句是系統通知的對話軌跡。 """ def setUp(self): from xkb_memory_service import Store self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.store = Store(Path(self.tmp.name) / "memory.sqlite") # 用真正的 API 建資料，不要自己寫 INSERT——第一版我猜了 sessions 的欄位， # 猜錯了。走公開路徑的話，schema 改了測試會跟著壞，那是對的。 session = self.store.open_session({ "source": "test", "agent_id": "test", "session_key": "k", "namespace": "private"}) rows = [ ("t1", "食品展的客戶通常怎麼找", "展前拿名單、展中面對面、展後跟催"), ("t2", REAL_TASK_NOTIFICATION, "食品展的客戶開發我剛做完了"), ("t3", REAL_OPENCLAW_CONTEXT, "食品展相關的內部通知"), ] for turn_id, query, answer in rows: self.store.start_turn({"session_id": session["session_id"], "turn_id": turn_id, "query": query}) self.store.complete_turn(turn_id, { "session_id": session["session_id"], "query": query, "answer": answer}) def test_harness_turns_never_enter_the_window(self): result = self.store.recall("食品展的客戶通常怎麼找", limit=10, namespace="private") queries = [m["query"] for m in result["memories"]] self.assertEqual(queries, ["食品展的客戶通常怎麼找"], "harness 產生的列不該佔用最近 500 筆的名額") def test_a_real_turn_is_still_recalled(self): result = self.store.recall("展前拿名單", limit=10, namespace="private") self.assertEqual(len(result["memories"]), 1) def test_the_rows_are_not_deleted(self): """排除的是召回，不是資料。那些 answer 是真實的回答內容。""" with self.store.connect() as db: count = db.execute("SELECT COUNT(*) FROM turns").fetchone()[0] self.assertEqual(count, 3)