0afe787c95
* fix: prevent authority-invalid HY retries Co-authored-by: Cursor <cursoragent@cursor.com> * fix: align coding authority with conversations Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: alvinfei <alvinfei@tencent.com> Co-authored-by: Cursor <cursoragent@cursor.com>
686 lines
29 KiB
Python
686 lines
29 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""CodingMemoryStore 测试"""
|
||
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
from datetime import datetime
|
||
|
||
import pytest
|
||
|
||
from hy_memory.coding.store import _vec_node_id, _derive_coding_collection_name
|
||
from hy_memory.coding.types import CodingMemory, WorkingFact
|
||
|
||
|
||
# ================================================================
|
||
# Helpers
|
||
# ================================================================
|
||
|
||
class TestHelpers:
|
||
def test_vec_node_id_format(self):
|
||
assert _vec_node_id("e1", 0) == "e1:0"
|
||
assert _vec_node_id("e1", 5) == "e1:5"
|
||
|
||
def test_derive_collection_name(self):
|
||
assert _derive_coding_collection_name("agent_memories") == "agent_memories_coding_keys"
|
||
assert _derive_coding_collection_name(None) == "agent_memories_coding_keys"
|
||
assert _derive_coding_collection_name("custom") == "custom_coding_keys"
|
||
|
||
|
||
# ================================================================
|
||
# Store e2e（用真实 chroma + mock embed）
|
||
# ================================================================
|
||
|
||
@pytest.mark.asyncio
|
||
class TestStoreCRUD:
|
||
async def test_insert_get(self, coding_store):
|
||
m = CodingMemory(
|
||
experience_id="e1", user_id="u1", agent_id="a1",
|
||
task="打包发布", search_keys=["凭证位置"], solution="...",
|
||
boundary_envs="twine>=6", workspace_id="ws/a",
|
||
files=["s.py"], confidence=0.9,
|
||
)
|
||
await coding_store.insert(m)
|
||
got = await coding_store.get_by_id("e1", user_id="u1")
|
||
assert got is not None
|
||
assert got.task == "打包发布" and got.search_keys == ["凭证位置"]
|
||
assert got.boundary_envs == "twine>=6"
|
||
assert got.files == ["s.py"]
|
||
|
||
async def test_get_by_id_wrong_user(self, coding_store):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="T", solution="S")
|
||
await coding_store.insert(m)
|
||
# 错的 user_id 应拿不到
|
||
got = await coding_store.get_by_id("e1", user_id="u_wrong")
|
||
assert got is None
|
||
|
||
async def test_list_user_tasks(self, coding_store):
|
||
await coding_store.insert(CodingMemory(experience_id="e1", user_id="u1",
|
||
task="T1", solution="S"))
|
||
await coding_store.insert(CodingMemory(experience_id="e2", user_id="u1",
|
||
task="T2", solution="S"))
|
||
await coding_store.insert(CodingMemory(experience_id="e3", user_id="u_other",
|
||
task="OTHER", solution="S"))
|
||
tasks = await coding_store.list_user_tasks("u1", limit=10)
|
||
assert sorted(tasks) == ["T1", "T2"]
|
||
assert "OTHER" not in tasks
|
||
|
||
async def test_update_overrides(self, coding_store, mock_embed):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="T",
|
||
search_keys=["old1", "old2"], solution="old",
|
||
session_id="conversation-1")
|
||
await coding_store.insert(m)
|
||
|
||
# 旧 aspect 应可命中
|
||
emb = (await mock_embed.embed_batch(["old1"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(emb, user_id="u1", top=5)
|
||
assert any(h["experience_id"] == "e1" for h in hits)
|
||
|
||
# update
|
||
m.search_keys = ["new1"]
|
||
m.solution = "new"
|
||
await coding_store.update(m)
|
||
|
||
# 新 aspect 应命中
|
||
emb_new = (await mock_embed.embed_batch(["new1"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(emb_new, user_id="u1", top=5)
|
||
assert any(h["experience_id"] == "e1" for h in hits)
|
||
|
||
# SQLite 端正确
|
||
got = await coding_store.get_by_id("e1", user_id="u1")
|
||
assert got.solution == "new"
|
||
assert got.search_keys == ["new1"]
|
||
|
||
async def test_update_rejects_authority_scope_change(self, coding_store):
|
||
await coding_store.insert(CodingMemory(
|
||
experience_id="e1", user_id="u1", agent_id="scope-a",
|
||
workspace_id="repo-a", session_id="conversation-1",
|
||
task="T", solution="old",
|
||
))
|
||
replacement = CodingMemory(
|
||
experience_id="e1", user_id="u1", agent_id="scope-b",
|
||
workspace_id="repo-a", session_id="conversation-1",
|
||
task="T", solution="new",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="coding_memory_update_authority_mismatch"):
|
||
await coding_store.update(replacement)
|
||
|
||
stored = await coding_store.get_by_id("e1", user_id="u1")
|
||
assert stored is not None
|
||
assert stored.agent_id == "scope-a"
|
||
assert stored.solution == "old"
|
||
|
||
async def test_update_rejects_conversation_change_and_legacy_empty_session(
|
||
self, coding_store
|
||
):
|
||
for experience_id, old_session in (
|
||
("different-conversation", "conversation-1"),
|
||
("legacy-empty-session", None),
|
||
):
|
||
await coding_store.insert(CodingMemory(
|
||
experience_id=experience_id,
|
||
user_id="u1",
|
||
agent_id="scope-a",
|
||
workspace_id="repo-a",
|
||
session_id=old_session,
|
||
task="T",
|
||
solution="old",
|
||
))
|
||
replacement = CodingMemory(
|
||
experience_id=experience_id,
|
||
user_id="u1",
|
||
agent_id="scope-a",
|
||
workspace_id="repo-a",
|
||
session_id="conversation-2",
|
||
task="T",
|
||
solution="new",
|
||
)
|
||
|
||
with pytest.raises(ValueError, match="coding_memory_update_authority_mismatch"):
|
||
await coding_store.update(replacement)
|
||
|
||
stored = await coding_store.get_by_id(experience_id, user_id="u1")
|
||
assert stored is not None
|
||
assert stored.solution == "old"
|
||
|
||
async def test_delete(self, coding_store):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="T",
|
||
search_keys=["k"], solution="S")
|
||
await coding_store.insert(m)
|
||
assert await coding_store.delete("e1", user_id="u1") is True
|
||
assert await coding_store.get_by_id("e1", user_id="u1") is None
|
||
# 删不存在的返回 False
|
||
assert await coding_store.delete("nope", user_id="u1") is False
|
||
|
||
async def test_delete_wrong_user(self, coding_store):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="T", solution="S")
|
||
await coding_store.insert(m)
|
||
# 错 user_id 不能删
|
||
deleted = await coding_store.delete("e1", user_id="u_wrong")
|
||
assert deleted is False
|
||
# 仍然能拿到
|
||
got = await coding_store.get_by_id("e1", user_id="u1")
|
||
assert got is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestStoreSearch:
|
||
async def test_search_hit_task_key(self, coding_store, mock_embed):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="打包发布",
|
||
search_keys=["凭证位置", "上传地址"], solution="S",
|
||
workspace_id="ws/a")
|
||
await coding_store.insert(m)
|
||
emb = (await mock_embed.embed_batch(["打包发布"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(
|
||
emb, user_id="u1", workspace_id="ws/a", top=10,
|
||
)
|
||
assert len(hits) == 1
|
||
assert hits[0]["experience_id"] == "e1"
|
||
assert hits[0]["matched_key_kind"] == "task"
|
||
|
||
async def test_search_hit_aspect_key(self, coding_store, mock_embed):
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="打包发布",
|
||
search_keys=["凭证位置"], solution="S")
|
||
await coding_store.insert(m)
|
||
emb = (await mock_embed.embed_batch(["凭证位置"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(emb, user_id="u1", top=10)
|
||
assert hits[0]["matched_key_kind"] == "aspect"
|
||
assert hits[0]["matched_key_text"] == "凭证位置"
|
||
|
||
async def test_search_filters_by_workspace(self, coding_store, mock_embed):
|
||
"""workspace 传了就按作用域筛，不传则全量召回。"""
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="T", solution="S",
|
||
workspace_id="ws/a")
|
||
await coding_store.insert(m)
|
||
emb = (await mock_embed.embed_batch(["T"]))[0]
|
||
for ws, expect in (("ws/a", 1), ("ws/wrong", 0), (None, 1)):
|
||
hits = await coding_store.search_by_query_embedding(
|
||
emb, user_id="u1", workspace_id=ws, top=5,
|
||
)
|
||
assert len(hits) == expect, f"workspace_id={ws!r}"
|
||
|
||
async def test_search_isolates_by_user(self, coding_store, mock_embed):
|
||
await coding_store.insert(CodingMemory(experience_id="e1", user_id="u1",
|
||
task="T", solution="S"))
|
||
await coding_store.insert(CodingMemory(experience_id="e2", user_id="u_other",
|
||
task="T", solution="S"))
|
||
emb = (await mock_embed.embed_batch(["T"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(emb, user_id="u1", top=10)
|
||
assert [h["experience_id"] for h in hits] == ["e1"]
|
||
|
||
async def test_search_dedup_by_experience_id(self, coding_store, mock_embed):
|
||
# 同一经验多个 keys 都可能命中，dedup 应保证每条经验最多一行
|
||
m = CodingMemory(experience_id="e1", user_id="u1", task="aaa",
|
||
search_keys=["aaa", "aaa"], solution="S")
|
||
await coding_store.insert(m)
|
||
emb = (await mock_embed.embed_batch(["aaa"]))[0]
|
||
hits = await coding_store.search_by_query_embedding(emb, user_id="u1", top=10)
|
||
unique = {h["experience_id"] for h in hits}
|
||
assert len(hits) == len(unique)
|
||
assert len(hits) <= 1
|
||
|
||
|
||
# ================================================================
|
||
# 旧 schema 迁移
|
||
# ================================================================
|
||
|
||
_LEGACY_DDL = """
|
||
CREATE TABLE coding_memory_meta (
|
||
memory_id TEXT PRIMARY KEY,
|
||
user_id TEXT NOT NULL,
|
||
agent_id TEXT NOT NULL DEFAULT 'default_agent',
|
||
task TEXT NOT NULL,
|
||
search_keys TEXT NOT NULL DEFAULT '[]',
|
||
solution TEXT NOT NULL,
|
||
boundary_envs TEXT NOT NULL DEFAULT '',
|
||
boundary_scope TEXT NOT NULL,
|
||
workspace_id TEXT,
|
||
branch TEXT,
|
||
session_id TEXT,
|
||
files TEXT NOT NULL DEFAULT '[]',
|
||
confidence REAL NOT NULL DEFAULT 0.7,
|
||
source TEXT NOT NULL DEFAULT 'auto_extract',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
"""
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestLegacySchemaMigration:
|
||
async def test_migrates_and_keeps_rows(self, coding_config, mock_embed):
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
|
||
db_path = coding_config.coding.db_path
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.executescript(_LEGACY_DDL)
|
||
conn.execute("CREATE INDEX idx_coding_user ON coding_memory_meta(user_id)")
|
||
conn.execute(
|
||
"INSERT INTO coding_memory_meta VALUES "
|
||
"(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
("m_old", "u1", "a1", "老 task", json.dumps(["k1"]), "老 solution",
|
||
"twine>=6", "project", "ws/a", "main", "s1", json.dumps(["f.py"]),
|
||
0.9, "auto_extract", "2026-01-01T00:00:00", "2026-01-02T00:00:00"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
try:
|
||
got = await store.get_by_id("m_old", user_id="u1")
|
||
assert got is not None
|
||
# id 沿用旧值，因此向量库 node_id 不受影响
|
||
assert got.experience_id == "m_old"
|
||
assert got.task == "老 task"
|
||
assert got.search_keys == ["k1"]
|
||
assert got.boundary_envs == "twine>=6"
|
||
assert got.files == ["f.py"]
|
||
# 迁移后仍可正常写入
|
||
await store.insert(CodingMemory(experience_id="e_new", user_id="u1",
|
||
task="新 task", solution="S"))
|
||
assert sorted(await store.list_user_tasks("u1")) == ["新 task", "老 task"]
|
||
finally:
|
||
await store.close()
|
||
|
||
async def test_migration_is_idempotent(self, coding_config, mock_embed):
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
|
||
for _ in range(2):
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
await store.close()
|
||
|
||
conn = sqlite3.connect(coding_config.coding.db_path)
|
||
cols = [r[1] for r in conn.execute("PRAGMA table_info(coding_memory_meta)")]
|
||
conn.close()
|
||
assert "experience_id" in cols
|
||
assert "memory_id" not in cols
|
||
assert "boundary_scope" not in cols
|
||
|
||
async def test_legacy_rows_get_empty_source_fact_ids(self, coding_config, mock_embed):
|
||
"""老库没有 source_fact_ids 列，迁移后应该是空链而不是炸掉。"""
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
|
||
db_path = coding_config.coding.db_path
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.executescript(_LEGACY_DDL)
|
||
conn.execute(
|
||
"INSERT INTO coding_memory_meta VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
("m_old", "u1", "a1", "老 task", "[]", "老 solution",
|
||
"", "project", "ws/a", "main", "s1", "[]",
|
||
0.9, "auto_extract", "2026-01-01T00:00:00", "2026-01-02T00:00:00"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
try:
|
||
got = await store.get_by_id("m_old", user_id="u1")
|
||
assert got.source_fact_ids == []
|
||
await store.set_experience_source_facts("m_old", ["f1"])
|
||
got = await store.get_by_id("m_old", user_id="u1")
|
||
assert got.source_fact_ids == ["f1"]
|
||
finally:
|
||
await store.close()
|
||
|
||
async def test_adds_source_fact_ids_to_renamed_schema(self, coding_config, mock_embed):
|
||
"""已经改过名、但建于加 source_fact_ids 之前的库，走 ADD COLUMN 补列。"""
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
|
||
db_path = coding_config.coding.db_path
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.executescript(_PRE_FACT_DDL)
|
||
conn.execute(
|
||
"INSERT INTO coding_memory_meta VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
("e1", "u1", "a1", "task", "[]", "solution", "", "ws/a", "main", "s1",
|
||
"[]", 0.9, "auto_extract", "2026-01-01T00:00:00", "2026-01-02T00:00:00"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
try:
|
||
got = await store.get_by_id("e1", user_id="u1")
|
||
assert got.source_fact_ids == []
|
||
finally:
|
||
await store.close()
|
||
|
||
async def test_adds_chain_summary_to_pre_chain_schema(self, coding_config, mock_embed):
|
||
"""建于加演变总结之前的库，走 ADD COLUMN 补列，老事实总结为空。"""
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
from hy_memory.coding.types import WorkingFact
|
||
|
||
db_path = coding_config.coding.db_path
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.executescript(_PRE_CHAIN_FACT_DDL)
|
||
conn.execute(
|
||
"INSERT INTO working_fact VALUES (?,?,?,?,?,?,?,?,?,?,?)",
|
||
("f_old", "u1", "a1", "s1", "ws/a", "老结论", "[]", None, "[]",
|
||
"active", "2026-01-01T00:00:00"),
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
try:
|
||
got = await store.get_fact("f_old", user_id="u1")
|
||
assert got is not None
|
||
assert got.fact_chain_summary == ""
|
||
# 补列之后照常能写总结
|
||
await store.insert_fact(WorkingFact(
|
||
fact_id="f_new", user_id="u1", statement="新结论",
|
||
fact_chain_summary="从老结论改成了新结论。",
|
||
))
|
||
back = await store.get_fact("f_new", user_id="u1")
|
||
assert back.fact_chain_summary == "从老结论改成了新结论。"
|
||
finally:
|
||
await store.close()
|
||
|
||
async def test_writes_ignore_abandoned_fact_chain_column(
|
||
self, coding_config, mock_embed
|
||
):
|
||
"""短命的 fact_chain 列已废弃，但老库里还留着，而且是 NOT NULL。
|
||
|
||
现在的 INSERT 不再提这一列，靠它的 DEFAULT 兜住；兜不住的话所有写入都会
|
||
直接报 NOT NULL constraint failed。
|
||
"""
|
||
from hy_memory.coding.store import CodingMemoryStore
|
||
from hy_memory.coding.types import WorkingFact
|
||
|
||
db_path = coding_config.coding.db_path
|
||
os.makedirs(os.path.dirname(db_path), exist_ok=True)
|
||
conn = sqlite3.connect(db_path)
|
||
conn.executescript(_ABANDONED_CHAIN_FACT_DDL)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
store = CodingMemoryStore(coding_config, embed_service=mock_embed)
|
||
await store.initialize()
|
||
try:
|
||
await store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", statement="新结论",
|
||
fact_chain_summary="一句总结。",
|
||
))
|
||
got = await store.get_fact("f1", user_id="u1")
|
||
assert got.statement == "新结论"
|
||
assert got.fact_chain_summary == "一句总结。"
|
||
# upsert 走的是另一条 SQL 分支，同样不该被那一列绊住
|
||
await store.supersede_fact("f1", "f2")
|
||
assert (await store.get_fact("f1", user_id="u1")).status == "superseded"
|
||
finally:
|
||
await store.close()
|
||
|
||
|
||
# 加演变总结之前的事实表
|
||
_PRE_CHAIN_FACT_DDL = """
|
||
CREATE TABLE IF NOT EXISTS working_fact (
|
||
fact_id TEXT PRIMARY KEY,
|
||
user_id TEXT NOT NULL,
|
||
agent_id TEXT NOT NULL DEFAULT 'default_agent',
|
||
session_id TEXT,
|
||
workspace_id TEXT,
|
||
statement TEXT NOT NULL,
|
||
files TEXT NOT NULL DEFAULT '[]',
|
||
fact_link TEXT,
|
||
derived_experience_ids TEXT NOT NULL DEFAULT '[]',
|
||
status TEXT NOT NULL DEFAULT 'active',
|
||
created_at TEXT NOT NULL
|
||
);
|
||
"""
|
||
|
||
|
||
# 短暂存在过 fact_chain 列的那一版事实表（现已废弃，老库里还留着）
|
||
_ABANDONED_CHAIN_FACT_DDL = """
|
||
CREATE TABLE IF NOT EXISTS working_fact (
|
||
fact_id TEXT PRIMARY KEY,
|
||
user_id TEXT NOT NULL,
|
||
agent_id TEXT NOT NULL DEFAULT 'default_agent',
|
||
session_id TEXT,
|
||
workspace_id TEXT,
|
||
statement TEXT NOT NULL,
|
||
files TEXT NOT NULL DEFAULT '[]',
|
||
fact_link TEXT,
|
||
derived_experience_ids TEXT NOT NULL DEFAULT '[]',
|
||
status TEXT NOT NULL DEFAULT 'active',
|
||
fact_chain TEXT NOT NULL DEFAULT '[]',
|
||
fact_chain_summary TEXT NOT NULL DEFAULT '',
|
||
created_at TEXT NOT NULL
|
||
);
|
||
"""
|
||
|
||
|
||
# ================================================================
|
||
# 事实层存储
|
||
# ================================================================
|
||
|
||
_PRE_FACT_DDL = """
|
||
CREATE TABLE IF NOT EXISTS coding_memory_meta (
|
||
experience_id TEXT PRIMARY KEY,
|
||
user_id TEXT NOT NULL,
|
||
agent_id TEXT NOT NULL DEFAULT 'default_agent',
|
||
task TEXT NOT NULL,
|
||
search_keys TEXT NOT NULL DEFAULT '[]',
|
||
solution TEXT NOT NULL,
|
||
boundary_envs TEXT NOT NULL DEFAULT '',
|
||
workspace_id TEXT,
|
||
branch TEXT,
|
||
session_id TEXT,
|
||
files TEXT NOT NULL DEFAULT '[]',
|
||
confidence REAL NOT NULL DEFAULT 0.7,
|
||
source TEXT NOT NULL DEFAULT 'auto_extract',
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
);
|
||
"""
|
||
|
||
|
||
class TestFactCollectionName:
|
||
def test_derive(self):
|
||
from hy_memory.coding.store import _derive_fact_collection_name
|
||
assert _derive_fact_collection_name("agent_memories") == "agent_memories_working_facts"
|
||
assert _derive_fact_collection_name(None) == "agent_memories_working_facts"
|
||
|
||
def test_separate_from_experience_collection(self):
|
||
"""两层各自独立 collection，不能撞名。"""
|
||
from hy_memory.coding.store import (
|
||
_derive_coding_collection_name, _derive_fact_collection_name,
|
||
)
|
||
assert _derive_coding_collection_name("x") != _derive_fact_collection_name("x")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestFactStore:
|
||
async def test_insert_and_get(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", agent_id="a1", session_id="s1",
|
||
workspace_id="ws/a", statement="事实正文", files=["a.py"],
|
||
derived_experience_ids=["e1"],
|
||
))
|
||
got = await coding_store.get_fact("f1", user_id="u1")
|
||
assert got.statement == "事实正文"
|
||
assert got.files == ["a.py"]
|
||
assert got.derived_experience_ids == ["e1"]
|
||
assert got.status == "active"
|
||
assert got.fact_link is None
|
||
# created_at 自动补
|
||
assert got.created_at is not None
|
||
|
||
async def test_get_fact_user_isolation(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", statement="S",
|
||
))
|
||
assert await coding_store.get_fact("f1", user_id="u_other") is None
|
||
|
||
async def test_get_facts_preserves_order(self, coding_store):
|
||
for i in range(3):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id=f"f{i}", user_id="u1", statement=f"事实 {i}",
|
||
))
|
||
got = await coding_store.get_facts(["f2", "f0", "f_missing"], user_id="u1")
|
||
assert [f.fact_id for f in got] == ["f2", "f0"]
|
||
|
||
async def test_search_facts_scores_and_isolation(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", statement="打包发布到私有包索引",
|
||
))
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f2", user_id="u_other", statement="打包发布到私有包索引",
|
||
))
|
||
emb = await coding_store.embed_service.embed("打包发布到私有包索引")
|
||
|
||
hits = await coding_store.search_facts(emb, user_id="u1", top=5)
|
||
assert [f.fact_id for f, _ in hits] == ["f1"]
|
||
assert hits[0][1] > 0.9
|
||
|
||
async def test_supersede_soft_deletes(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_old", user_id="u1", statement="旧结论",
|
||
))
|
||
assert await coding_store.supersede_fact("f_old", "f_new") is True
|
||
|
||
got = await coding_store.get_fact("f_old", user_id="u1")
|
||
# 行还在，链路可回溯
|
||
assert got.status == "superseded"
|
||
assert got.fact_link == "f_new"
|
||
# 但已经不占召回名额
|
||
emb = await coding_store.embed_service.embed("旧结论")
|
||
assert await coding_store.search_facts(emb, user_id="u1") == []
|
||
|
||
async def test_search_respects_top(self, coding_store):
|
||
for i in range(6):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id=f"f{i}", user_id="u1", statement=f"事实 {i}",
|
||
))
|
||
emb = await coding_store.embed_service.embed("事实 0")
|
||
assert len(await coding_store.search_facts(emb, user_id="u1", top=3)) == 3
|
||
|
||
async def test_orphan_index_rows_dont_starve_recall(self, coding_store):
|
||
"""向量行还在、SQLite 行没了的孤儿不能吃掉候选名额。
|
||
|
||
两边不是一个事务里写的，中间挂掉就会留下这种残留。不过采样的话前 top 条
|
||
全是孤儿时召回会直接空掉。
|
||
"""
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_real", user_id="u1", statement="真实事实",
|
||
))
|
||
# 造 5 条孤儿：向量库有，SQLite 删掉
|
||
for i in range(5):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id=f"f_orphan{i}", user_id="u1", statement="真实事实",
|
||
))
|
||
with coding_store._lock:
|
||
coding_store._conn.execute(
|
||
"DELETE FROM working_fact WHERE fact_id LIKE 'f_orphan%'"
|
||
)
|
||
|
||
emb = await coding_store.embed_service.embed("真实事实")
|
||
hits = await coding_store.search_facts(emb, user_id="u1", top=5)
|
||
assert [f.fact_id for f, _ in hits] == ["f_real"]
|
||
|
||
async def test_supersede_twice_is_noop(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_old", user_id="u1", statement="旧结论",
|
||
))
|
||
assert await coding_store.supersede_fact("f_old", "f_a") is True
|
||
assert await coding_store.supersede_fact("f_old", "f_b") is False
|
||
got = await coding_store.get_fact("f_old", user_id="u1")
|
||
assert got.fact_link == "f_a"
|
||
|
||
async def test_supersede_missing_fact(self, coding_store):
|
||
assert await coding_store.supersede_fact("nope", "f_new") is False
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
class TestFactAncestors:
|
||
"""历史不另存，顺 fact_link 反查还原。"""
|
||
|
||
async def _chain(self, store, n, *, user_id="u1"):
|
||
"""建 n 代：f0 被 f1 顶掉，f1 被 f2 顶掉……返回 id 列表，最新在末尾。"""
|
||
ids = []
|
||
for i in range(n):
|
||
fid = f"f{i}"
|
||
await store.insert_fact(WorkingFact(
|
||
fact_id=fid, user_id=user_id, statement=f"第 {i} 代",
|
||
created_at=datetime(2026, 1, i + 1),
|
||
))
|
||
if ids:
|
||
await store.supersede_fact(ids[-1], fid)
|
||
ids.append(fid)
|
||
return ids
|
||
|
||
async def test_walks_back_in_time_order(self, coding_store):
|
||
ids = await self._chain(coding_store, 4)
|
||
got = await coding_store.get_fact_ancestors(ids[-1], user_id="u1")
|
||
assert [f.fact_id for f in got] == ["f0", "f1", "f2"], "时间正序，不含自己"
|
||
assert [f.statement for f in got] == ["第 0 代", "第 1 代", "第 2 代"]
|
||
|
||
async def test_no_ancestors_for_fresh_fact(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_solo", user_id="u1", statement="全新结论",
|
||
))
|
||
assert await coding_store.get_fact_ancestors("f_solo", user_id="u1") == []
|
||
|
||
async def test_respects_max_depth(self, coding_store):
|
||
"""长命事实不该把整条链都拖进提示词。"""
|
||
ids = await self._chain(coding_store, 6)
|
||
got = await coding_store.get_fact_ancestors(
|
||
ids[-1], user_id="u1", max_depth=2
|
||
)
|
||
assert [f.fact_id for f in got] == ["f3", "f4"], "截断留最近的几代"
|
||
|
||
async def test_scoped_by_user(self, coding_store):
|
||
"""别人的事实不该出现在我的血统里。"""
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_theirs", user_id="u2", statement="别人的旧结论",
|
||
))
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f_mine", user_id="u1", statement="我的结论",
|
||
))
|
||
await coding_store.supersede_fact("f_theirs", "f_mine")
|
||
assert await coding_store.get_fact_ancestors("f_mine", user_id="u1") == []
|
||
|
||
async def test_cycle_does_not_hang(self, coding_store):
|
||
"""fact_link 成环理论上不该发生，真发生了也不能死循环。"""
|
||
for fid in ("f_a", "f_b"):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id=fid, user_id="u1", statement=fid,
|
||
))
|
||
await coding_store.supersede_fact("f_a", "f_b")
|
||
await coding_store.supersede_fact("f_b", "f_a")
|
||
|
||
got = await coding_store.get_fact_ancestors("f_a", user_id="u1")
|
||
assert [f.fact_id for f in got] == ["f_b"], "撞回起点就停"
|
||
|
||
async def test_set_derived_experiences(self, coding_store):
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", statement="S",
|
||
))
|
||
await coding_store.set_fact_derived_experiences("f1", ["e1", "e2"])
|
||
got = await coding_store.get_fact("f1", user_id="u1")
|
||
assert got.derived_experience_ids == ["e1", "e2"]
|
||
|
||
async def test_experience_and_fact_dont_cross_contaminate(self, coding_store):
|
||
"""两层各自独立 collection：搜事实不该搜出经验，反之亦然。"""
|
||
await coding_store.insert(CodingMemory(
|
||
experience_id="e1", user_id="u1", task="同样的文本", solution="S",
|
||
))
|
||
await coding_store.insert_fact(WorkingFact(
|
||
fact_id="f1", user_id="u1", statement="同样的文本",
|
||
))
|
||
emb = await coding_store.embed_service.embed("同样的文本")
|
||
|
||
fact_hits = await coding_store.search_facts(emb, user_id="u1", top=10)
|
||
assert [f.fact_id for f, _ in fact_hits] == ["f1"]
|
||
|
||
exp_hits = await coding_store.search_by_query_embedding(emb, user_id="u1", top=10)
|
||
assert [h["experience_id"] for h in exp_hits] == ["e1"]
|