sync update from LAN repo (source dhk-1 @ 3e87be55)
This commit is contained in:
gozeros committed
1 parent
ab6068a7c1
commit
42f2a26fa8
1482 files changed
+227759
-9584
No files matched your search
@@ -0,0 +1,980 @@
|
||||
"""Runtime boundaries for project-scoped cross-conversation memory."""
|
||||
|
||||
import json
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from langchain_core.messages import AIMessage, HumanMessage
|
||||
|
||||
from app.gateway import analysis_project_memory as runtime
|
||||
from app.gateway.routers import threads as thread_routes
|
||||
from deerflow.config import system_settings
|
||||
from deerflow.persistence.analysis_project_memory import AnalysisProjectMemoryStore
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_model_extraction_projects_existing_memory_to_json_safe_fields(monkeypatch):
|
||||
store = SimpleNamespace(
|
||||
recall=AsyncMock(
|
||||
return_value={
|
||||
"snapshot": {"summary": "已有摘要"},
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "fact:region",
|
||||
"kind": "fact",
|
||||
"content": "目标区域是华东",
|
||||
"confidence": 0.9,
|
||||
"importance": 0.8,
|
||||
"pinned": False,
|
||||
"source_thread_id": "thread-a",
|
||||
"created_at": datetime.now(UTC),
|
||||
"embedding": [0.1] * 1_024,
|
||||
"embedding_json": "[0.1]",
|
||||
}
|
||||
],
|
||||
}
|
||||
)
|
||||
)
|
||||
captured = {}
|
||||
|
||||
class Model:
|
||||
async def ainvoke(self, messages, *, config):
|
||||
captured["system"] = messages[0].content
|
||||
captured["source"] = json.loads(messages[1].content)
|
||||
captured["config"] = config
|
||||
return SimpleNamespace(
|
||||
content=json.dumps(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "fact:region",
|
||||
"kind": "fact",
|
||||
"content": "目标区域是华东",
|
||||
"confidence": 0.9,
|
||||
"importance": 0.8,
|
||||
"evidence_ids": ["m2"],
|
||||
}
|
||||
],
|
||||
"snapshot": {"summary": "目标区域是华东"},
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
)
|
||||
|
||||
monkeypatch.setattr(runtime, "create_chat_model", lambda **_kwargs: Model())
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
config=SimpleNamespace(memory=SimpleNamespace(v1=SimpleNamespace(model_name=None))),
|
||||
)
|
||||
)
|
||||
result = await runtime.AnalysisProjectMemoryExecutor(app)._model_extraction(
|
||||
{
|
||||
"project_id": "project-a",
|
||||
"owner_user_id": "alice",
|
||||
"payload": {
|
||||
"project_name": "区域规划",
|
||||
"project_background": "年度规划",
|
||||
"messages": [
|
||||
{"role": "user", "content": "我们的目标区域是什么？"},
|
||||
{"role": "assistant", "content": "华东。"},
|
||||
],
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
existing = captured["source"]["existing_memory"]["items"][0]
|
||||
assert existing == {
|
||||
"logical_key": "fact:region",
|
||||
"kind": "fact",
|
||||
"content": "目标区域是华东",
|
||||
"confidence": 0.9,
|
||||
"importance": 0.8,
|
||||
"pinned": False,
|
||||
"source_thread_id": "thread-a",
|
||||
"evidence_authority": "unknown",
|
||||
}
|
||||
assert captured["source"]["conversation"] == [
|
||||
{
|
||||
"evidence_id": "m1",
|
||||
"id": "",
|
||||
"role": "user",
|
||||
"content": "我们的目标区域是什么？",
|
||||
},
|
||||
{
|
||||
"evidence_id": "m2",
|
||||
"id": "",
|
||||
"role": "assistant",
|
||||
"content": "华东。",
|
||||
},
|
||||
]
|
||||
assert captured["config"] == {"run_name": "analysis_project_memory_extract"}
|
||||
assert "仅过滤助手自行报告的真实运行时当前日期/时间" in captured["system"]
|
||||
assert "本项目假定今天为某日" in captured["system"]
|
||||
assert result["items"][0]["content"] == "目标区域是华东"
|
||||
assert result["items"][0]["confidence"] == 0.55
|
||||
assert result["items"][0]["metadata"]["evidence_authority"] == "assistant"
|
||||
|
||||
|
||||
def test_extraction_redacts_credentials_and_privacy_data():
|
||||
cleaned = runtime._clean_extraction(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "fact:contact",
|
||||
"kind": "fact",
|
||||
"content": "token=super-secret；手机号 13800138000",
|
||||
"confidence": 0.8,
|
||||
"importance": 0.7,
|
||||
}
|
||||
],
|
||||
"snapshot": {"summary": "password=hunter2；联系 13800138000"},
|
||||
}
|
||||
)
|
||||
|
||||
serialized = json.dumps(cleaned, ensure_ascii=False)
|
||||
assert "super-secret" not in serialized
|
||||
assert "hunter2" not in serialized
|
||||
assert "13800138000" not in serialized
|
||||
assert "REDACTED" in serialized or "脱敏" in serialized
|
||||
|
||||
|
||||
def test_fallback_requires_complete_exchange_and_is_bounded():
|
||||
user_only = runtime._fallback_extraction(
|
||||
{"messages": [{"role": "user", "content": "只有问题"}]}
|
||||
)
|
||||
assert user_only["items"][0]["content"] == "历史用户消息：只有问题"
|
||||
|
||||
result = runtime._fallback_extraction(
|
||||
{
|
||||
"messages": [
|
||||
{"role": "user", "content": "项目代号是什么？"},
|
||||
{"role": "assistant", "content": "项目代号是星河。"},
|
||||
]
|
||||
}
|
||||
)
|
||||
assert len(result["items"]) == 1
|
||||
assert result["items"][0]["kind"] == "episode"
|
||||
assert "项目代号是什么" in result["items"][0]["content"]
|
||||
assert "星河" not in result["items"][0]["content"]
|
||||
assert result["items"][0]["metadata"]["evidence_authority"] == "user"
|
||||
|
||||
|
||||
def test_fallback_preserves_earlier_explicit_user_setting_not_only_last_question():
|
||||
result = runtime._fallback_extraction(
|
||||
{
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "修改一下设定，记忆日期为2026年12月31日。",
|
||||
},
|
||||
{"role": "assistant", "content": "已记录。"},
|
||||
{"role": "user", "content": "今天日期多少号？"},
|
||||
{"role": "assistant", "content": "今天是2026年9月24日。"},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
assert [item["content"] for item in result["items"]] == [
|
||||
"历史用户消息：修改一下设定，记忆日期为2026年12月31日。",
|
||||
"历史用户消息：今天日期多少号？",
|
||||
]
|
||||
assert all("2026年9月24日" not in item["content"] for item in result["items"])
|
||||
assert all(
|
||||
item["metadata"]["evidence_authority"] == "user"
|
||||
for item in result["items"]
|
||||
)
|
||||
|
||||
|
||||
def test_ground_extraction_rejects_assistant_capability_and_volatile_date_claims():
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[
|
||||
{"id": "u1", "role": "user", "content": "今天日期多少号？"},
|
||||
{
|
||||
"id": "a1",
|
||||
"role": "assistant",
|
||||
"content": "当前系统日期为2026年9月24日，星期四。",
|
||||
},
|
||||
{"id": "u2", "role": "user", "content": "你能查看其他对话吗？"},
|
||||
{"id": "a2", "role": "assistant", "content": "我无法访问其他对话。"},
|
||||
]
|
||||
)
|
||||
extraction = runtime._clean_extraction(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "memory-access",
|
||||
"kind": "constraint",
|
||||
"content": "助手无法访问其他对话",
|
||||
"confidence": 0.95,
|
||||
"importance": 0.8,
|
||||
# The model incorrectly cites the user's lookup request as
|
||||
# if it confirmed the assistant's capability statement.
|
||||
"evidence_ids": ["m3", "m4"],
|
||||
},
|
||||
{
|
||||
"logical_key": "memory-date",
|
||||
"kind": "fact",
|
||||
"content": "当前系统日期为2026年9月24日，星期四",
|
||||
"confidence": 0.95,
|
||||
"importance": 0.8,
|
||||
# The user only asked the question and never supplied this
|
||||
# date, so m1 must not elevate the assistant answer.
|
||||
"evidence_ids": ["m1", "m2"],
|
||||
},
|
||||
{
|
||||
"logical_key": "memory-date-en",
|
||||
"kind": "fact",
|
||||
"content": "System date is 2026-09-24 (Thursday)",
|
||||
"confidence": 0.95,
|
||||
"importance": 0.8,
|
||||
# The same unsupported runtime claim may be translated by
|
||||
# the extractor; it must be rejected in English too.
|
||||
"evidence_ids": ["m1", "m2"],
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
grounded = runtime._ground_extraction(extraction, conversation)
|
||||
|
||||
assert grounded["items"] == []
|
||||
|
||||
|
||||
def test_ground_extraction_keeps_explicit_user_setting_with_user_authority():
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[
|
||||
{
|
||||
"id": "u1",
|
||||
"role": "user",
|
||||
"content": "本项目假定今天为2026年12月31日。",
|
||||
},
|
||||
{"id": "a1", "role": "assistant", "content": "已经修改。"},
|
||||
]
|
||||
)
|
||||
extraction = runtime._clean_extraction(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "memory-date",
|
||||
"kind": "fact",
|
||||
"content": "本项目模拟日期为2026年12月31日",
|
||||
"confidence": 0.98,
|
||||
"importance": 0.9,
|
||||
"evidence_ids": ["m1", "m2"],
|
||||
}
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
[item] = runtime._ground_extraction(extraction, conversation)["items"]
|
||||
|
||||
assert item["content"] == "本项目模拟日期为2026年12月31日"
|
||||
assert item["source_message_ids"] == ["u1"]
|
||||
assert item["metadata"] == {
|
||||
"evidence_authority": "user",
|
||||
"evidence_roles": ["user"],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("user_text", "nickname"),
|
||||
[
|
||||
("你叫小爱同学", "小爱同学"),
|
||||
("你现在叫铁蛋，我给你起的新名字", "铁蛋"),
|
||||
("对，你的昵称叫铁蛋", "铁蛋"),
|
||||
("以后就叫你 Tiedan", "Tiedan"),
|
||||
("我给你起了个新昵称：小鹿", "小鹿"),
|
||||
],
|
||||
)
|
||||
def test_ground_extraction_deterministically_keeps_user_assigned_nickname(
|
||||
user_text, nickname
|
||||
):
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[
|
||||
{"id": "u1", "role": "user", "content": user_text},
|
||||
{
|
||||
"id": "a1",
|
||||
"role": "assistant",
|
||||
"content": "系统身份不能修改。",
|
||||
},
|
||||
]
|
||||
)
|
||||
|
||||
[item] = runtime._ground_extraction(
|
||||
{"items": [], "snapshot": None}, conversation
|
||||
)["items"]
|
||||
|
||||
assert item["logical_key"] == "assistant.nickname"
|
||||
assert item["kind"] == "preference"
|
||||
assert item["content"] == f"本项目中用户给助手设定的昵称是“{nickname}”"
|
||||
assert item["source_message_ids"] == ["u1"]
|
||||
assert item["metadata"] == {
|
||||
"evidence_authority": "user",
|
||||
"evidence_roles": ["user"],
|
||||
"deterministic": "assistant_nickname",
|
||||
}
|
||||
|
||||
|
||||
def test_ground_extraction_latest_nickname_wins_and_rejects_model_semantic_reversal():
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[
|
||||
{"id": "u1", "role": "user", "content": "你叫小爱同学"},
|
||||
{
|
||||
"id": "a1",
|
||||
"role": "assistant",
|
||||
"content": "我不是小爱同学，我叫CM助手。",
|
||||
},
|
||||
{"id": "u2", "role": "user", "content": "对，你的昵称叫铁蛋"},
|
||||
{"id": "a2", "role": "assistant", "content": "你可以叫我铁蛋。"},
|
||||
]
|
||||
)
|
||||
extraction = runtime._clean_extraction(
|
||||
{
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "cm-assistant-identity",
|
||||
"kind": "fact",
|
||||
"content": "CM助手是项目内部智能体，不是小爱同学",
|
||||
"confidence": 1,
|
||||
"importance": 0.8,
|
||||
"evidence_ids": ["m1", "m2"],
|
||||
},
|
||||
{
|
||||
"logical_key": "assistant-name",
|
||||
"kind": "fact",
|
||||
"content": "助手的名字是CM助手",
|
||||
"confidence": 1,
|
||||
"importance": 0.8,
|
||||
"evidence_ids": ["m3", "m4"],
|
||||
},
|
||||
]
|
||||
}
|
||||
)
|
||||
|
||||
[item] = runtime._ground_extraction(extraction, conversation)["items"]
|
||||
|
||||
assert item["logical_key"] == "assistant.nickname"
|
||||
assert item["content"] == "本项目中用户给助手设定的昵称是“铁蛋”"
|
||||
assert item["source_message_ids"] == ["u2"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"user_text",
|
||||
[
|
||||
"你的昵称叫什么？",
|
||||
"你叫铁蛋吗？",
|
||||
"你叫铁蛋？",
|
||||
"你的昵称叫铁蛋吗？",
|
||||
"你叫 Tiedan 吗？",
|
||||
"你的昵称叫 Tiedan 吗",
|
||||
"我不是给你起了一个新的昵称",
|
||||
"不要叫铁蛋",
|
||||
"取消昵称铁蛋",
|
||||
],
|
||||
)
|
||||
def test_ground_extraction_does_not_invent_nickname_from_question_or_negation(
|
||||
user_text,
|
||||
):
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[{"id": "u1", "role": "user", "content": user_text}]
|
||||
)
|
||||
|
||||
assert runtime._ground_extraction(
|
||||
{"items": [], "snapshot": None}, conversation
|
||||
)["items"] == []
|
||||
|
||||
|
||||
def test_conversation_episode_keeps_bounded_raw_exchange_for_other_project_threads():
|
||||
conversation = runtime._conversation_with_evidence_ids(
|
||||
[
|
||||
{"id": "u1", "role": "user", "content": "项目代号是什么？"},
|
||||
{"id": "a1", "role": "assistant", "content": "项目代号是星河。"},
|
||||
]
|
||||
)
|
||||
|
||||
[item] = runtime._conversation_episode_items(conversation)
|
||||
|
||||
assert item["kind"] == "episode"
|
||||
assert "用户：项目代号是什么？" in item["content"]
|
||||
assert "助手：项目代号是星河。" in item["content"]
|
||||
assert item["source_message_ids"] == ["u1", "a1"]
|
||||
assert item["metadata"]["raw_project_conversation"] is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_project_pause_skips_extractor_model(monkeypatch):
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": False}),
|
||||
complete_job=AsyncMock(return_value=None),
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(analysis_project_memory_store=store)
|
||||
)
|
||||
executor = runtime.AnalysisProjectMemoryExecutor(app)
|
||||
extraction = AsyncMock()
|
||||
monkeypatch.setattr(executor, "_model_extraction", extraction)
|
||||
await executor.execute(
|
||||
{"id": "job-a", "project_id": "project-a", "owner_user_id": "alice"},
|
||||
lease_owner="worker-a",
|
||||
)
|
||||
|
||||
extraction.assert_not_awaited()
|
||||
store.complete_job.assert_awaited_once_with(
|
||||
"job-a",
|
||||
lease_owner="worker-a",
|
||||
result={"skipped": "project_memory_disabled"},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_executor_redacts_persisted_failure_details(monkeypatch):
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
apply_extraction=AsyncMock(
|
||||
side_effect=RuntimeError(
|
||||
"upstream failed with token=super-secret and phone 13800138000"
|
||||
)
|
||||
),
|
||||
fail_job=AsyncMock(return_value=None),
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
llmwiki_embedding=None,
|
||||
)
|
||||
)
|
||||
executor = runtime.AnalysisProjectMemoryExecutor(app)
|
||||
monkeypatch.setattr(
|
||||
executor,
|
||||
"_model_extraction",
|
||||
AsyncMock(return_value={"items": [], "snapshot": None}),
|
||||
)
|
||||
|
||||
await executor.execute(
|
||||
{
|
||||
"id": "job-a",
|
||||
"project_id": "project-a",
|
||||
"owner_user_id": "alice",
|
||||
"payload": {"messages": []},
|
||||
"attempt": 1,
|
||||
},
|
||||
lease_owner="worker-a",
|
||||
)
|
||||
|
||||
error = store.fail_job.await_args.kwargs["error"]
|
||||
assert "super-secret" not in error
|
||||
assert "13800138000" not in error
|
||||
assert "REDACTED" in error or "脱敏" in error
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_queued_payload_is_redacted_before_durable_storage(monkeypatch):
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
enqueue_job=AsyncMock(
|
||||
return_value=({"id": "job-a", "status": "queued"}, True)
|
||||
),
|
||||
)
|
||||
|
||||
class Checkpointer:
|
||||
async def aget_tuple(self, _config):
|
||||
return SimpleNamespace(
|
||||
checkpoint={
|
||||
"channel_values": {
|
||||
"messages": [
|
||||
HumanMessage(
|
||||
id="message-user",
|
||||
content="token=super-secret；手机号 13800138000",
|
||||
),
|
||||
AIMessage(
|
||||
id="message-assistant",
|
||||
content="password=hunter2；已经记录",
|
||||
),
|
||||
]
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
analysis_project_memory_dispatcher=None,
|
||||
checkpointer=Checkpointer(),
|
||||
)
|
||||
)
|
||||
result = await runtime.enqueue_completed_project_run(
|
||||
app=app,
|
||||
project={
|
||||
"id": "project-a",
|
||||
"user_id": "alice",
|
||||
"mode": "qa",
|
||||
"name": "password=project-secret",
|
||||
"background": "联系人 13800138000",
|
||||
},
|
||||
thread_id="thread-a",
|
||||
run_id="run-a",
|
||||
)
|
||||
|
||||
assert result == {"id": "job-a", "status": "queued"}
|
||||
payload = store.enqueue_job.await_args.kwargs["payload"]
|
||||
serialized = json.dumps(payload, ensure_ascii=False)
|
||||
for secret in (
|
||||
"super-secret",
|
||||
"hunter2",
|
||||
"project-secret",
|
||||
"13800138000",
|
||||
):
|
||||
assert secret not in serialized
|
||||
assert "REDACTED" in serialized or "脱敏" in serialized
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatcher_does_not_claim_beyond_concurrency():
|
||||
store = SimpleNamespace(claim_jobs=AsyncMock(return_value=[]))
|
||||
dispatcher = runtime.AnalysisProjectMemoryDispatcher(
|
||||
store,
|
||||
SimpleNamespace(),
|
||||
concurrency=2,
|
||||
)
|
||||
dispatcher._running = {"job-a": object(), "job-b": object()}
|
||||
assert await dispatcher.dispatch_once() == 0
|
||||
store.claim_jobs.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generic_memory_switch_off_does_not_block_project_recall(monkeypatch):
|
||||
monkeypatch.setattr(
|
||||
system_settings,
|
||||
"load_system_settings",
|
||||
lambda: SimpleNamespace(memory=SimpleNamespace(enabled=False)),
|
||||
)
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
recall=AsyncMock(
|
||||
return_value={
|
||||
"state": {"status": "ready", "revision": 1},
|
||||
"snapshot": {},
|
||||
"items": [{"content": "项目日期为2026年12月31日"}],
|
||||
}
|
||||
),
|
||||
)
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
llmwiki_embedding=None,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
scope, recalled = await runtime.prepare_project_memory(
|
||||
request=request,
|
||||
project={
|
||||
"id": "project-a",
|
||||
"user_id": "alice",
|
||||
"mode": "qa",
|
||||
},
|
||||
thread_id="thread-b",
|
||||
query="上次结论是什么？",
|
||||
)
|
||||
|
||||
assert scope is not None and scope["enabled"] is True
|
||||
assert recalled is not None
|
||||
assert recalled["items"][0]["content"] == "项目日期为2026年12月31日"
|
||||
store.get_settings.assert_awaited_once()
|
||||
store.recall.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_project_recall_does_not_wait_for_query_embedding_before_stream():
|
||||
embedding = SimpleNamespace(
|
||||
embed_query=AsyncMock(return_value=[0.1, 0.2]),
|
||||
fingerprint="remote-encoder-v1",
|
||||
)
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
recall=AsyncMock(
|
||||
return_value={
|
||||
"state": {"status": "ready", "revision": 1},
|
||||
"snapshot": {},
|
||||
"items": [{"content": "本项目中助手昵称是铁蛋"}],
|
||||
}
|
||||
),
|
||||
)
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
llmwiki_embedding=embedding,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
_scope, recalled = await runtime.prepare_project_memory(
|
||||
request=request,
|
||||
project={"id": "project-a", "user_id": "alice", "mode": "qa"},
|
||||
thread_id="thread-b",
|
||||
query="我给你起的昵称是什么？",
|
||||
)
|
||||
|
||||
assert recalled is not None
|
||||
embedding.embed_query.assert_not_awaited()
|
||||
assert store.recall.await_args.kwargs["query_embedding"] is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_incremental_building_state_keeps_indexed_memory_recallable():
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
recall=AsyncMock(
|
||||
return_value={
|
||||
"state": {"status": "building", "revision": 1},
|
||||
"snapshot": {},
|
||||
"items": [
|
||||
{
|
||||
"logical_key": "assistant.nickname",
|
||||
"kind": "preference",
|
||||
"content": "本项目中用户给助手设定的昵称是“铁蛋”",
|
||||
}
|
||||
],
|
||||
}
|
||||
),
|
||||
)
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
llmwiki_embedding=None,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
scope, recalled = await runtime.prepare_project_memory(
|
||||
request=request,
|
||||
project={"id": "project-a", "user_id": "alice", "mode": "qa"},
|
||||
thread_id="thread-b",
|
||||
query="我给你起的昵称是什么？",
|
||||
)
|
||||
|
||||
assert scope is not None and scope["enabled"] is True
|
||||
assert recalled is not None
|
||||
assert recalled["revision"] == 1
|
||||
assert recalled["items"][0]["content"] == "本项目中用户给助手设定的昵称是“铁蛋”"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", ["rebuilding", "recovering"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_destructive_rebuild_state_is_fail_closed_for_prompt_recall(
|
||||
status, monkeypatch
|
||||
):
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
recall=AsyncMock(
|
||||
return_value={
|
||||
"state": {"status": status, "revision": 9},
|
||||
"snapshot": {"summary": "must not leak during recovery"},
|
||||
"items": [{"content": "stale fact"}],
|
||||
}
|
||||
),
|
||||
)
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
llmwiki_embedding=None,
|
||||
)
|
||||
)
|
||||
)
|
||||
scope, recalled = await runtime.prepare_project_memory(
|
||||
request=request,
|
||||
project={"id": "project-a", "user_id": "alice", "mode": "qa"},
|
||||
thread_id="thread-b",
|
||||
query="之前的结论是什么？",
|
||||
)
|
||||
|
||||
assert scope is not None and scope["enabled"] is True
|
||||
assert recalled is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_source_invalidation_failure_propagates_before_mutation():
|
||||
project = {"id": "project-a", "user_id": "alice", "mode": "qa"}
|
||||
projects = SimpleNamespace(project_for_thread=AsyncMock(return_value=project))
|
||||
store = SimpleNamespace(
|
||||
invalidate_source=AsyncMock(side_effect=RuntimeError("database unavailable"))
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_store=projects,
|
||||
analysis_project_memory_store=store,
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(RuntimeError, match="database unavailable"):
|
||||
await runtime.invalidate_project_memory_before_thread_mutation(
|
||||
app=app,
|
||||
thread_id="thread-a",
|
||||
source_message_ids=["message-a"],
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_generic_memory_switch_off_still_rebuilds_project_after_source_mutation(
|
||||
monkeypatch,
|
||||
):
|
||||
project = {"id": "project-a", "user_id": "alice", "mode": "qa"}
|
||||
projects = SimpleNamespace(
|
||||
thread_ids=AsyncMock(return_value=["thread-a", "thread-b"])
|
||||
)
|
||||
store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_store=projects,
|
||||
analysis_project_memory_store=store,
|
||||
)
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
system_settings,
|
||||
"load_system_settings",
|
||||
lambda: SimpleNamespace(memory=SimpleNamespace(enabled=False)),
|
||||
)
|
||||
enqueue = AsyncMock(return_value=2)
|
||||
monkeypatch.setattr(runtime, "enqueue_project_rebuild", enqueue)
|
||||
|
||||
await runtime.finalize_project_memory_after_thread_mutation(
|
||||
app=app,
|
||||
project=project,
|
||||
thread_id="thread-a",
|
||||
)
|
||||
|
||||
projects.thread_ids.assert_awaited_once_with("project-a", "alice")
|
||||
enqueue.assert_awaited_once_with(
|
||||
app=app,
|
||||
project=project,
|
||||
thread_ids=["thread-a", "thread-b"],
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rebuild_enumeration_crash_remains_durably_recoverable(monkeypatch):
|
||||
store = AnalysisProjectMemoryStore()
|
||||
|
||||
class BrokenCheckpointer:
|
||||
async def aget_tuple(self, _config):
|
||||
raise RuntimeError("checkpoint unavailable")
|
||||
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_memory_store=store,
|
||||
analysis_project_memory_dispatcher=None,
|
||||
checkpointer=BrokenCheckpointer(),
|
||||
)
|
||||
)
|
||||
project = {
|
||||
"id": "project-a",
|
||||
"user_id": "alice",
|
||||
"mode": "qa",
|
||||
"name": "Project A",
|
||||
"background": "",
|
||||
}
|
||||
with pytest.raises(RuntimeError, match="checkpoint unavailable"):
|
||||
await runtime.enqueue_project_rebuild(
|
||||
app=app,
|
||||
project=project,
|
||||
thread_ids=["thread-a"],
|
||||
)
|
||||
|
||||
state = await store.get_state("project-a", "alice")
|
||||
assert state is not None and state["status"] == "rebuilding"
|
||||
future = datetime.now(UTC) + timedelta(days=365)
|
||||
[claimed] = await store.claim_stalled_rebuilds(
|
||||
stale_before=future,
|
||||
claimed_at=future,
|
||||
)
|
||||
assert claimed["status"] == "recovering"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rebuild_enqueue_honors_authoritative_project_switch():
|
||||
project = {
|
||||
"id": "project-a",
|
||||
"user_id": "alice",
|
||||
"mode": "qa",
|
||||
"name": "Project A",
|
||||
"background": "",
|
||||
}
|
||||
paused_store = SimpleNamespace(
|
||||
get_settings=AsyncMock(return_value={"enabled": False}),
|
||||
start_rebuild=AsyncMock(side_effect=AssertionError("must not rebuild")),
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(analysis_project_memory_store=paused_store)
|
||||
)
|
||||
assert await runtime.enqueue_project_rebuild(
|
||||
app=app, project=project, thread_ids=["thread-a"]
|
||||
) == 0
|
||||
paused_store.start_rebuild.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatcher_replays_claimed_stalled_rebuild(monkeypatch):
|
||||
project = {
|
||||
"id": "project-a",
|
||||
"user_id": "alice",
|
||||
"mode": "qa",
|
||||
"name": "Project A",
|
||||
"background": "",
|
||||
}
|
||||
store = SimpleNamespace(
|
||||
claim_stalled_rebuilds=AsyncMock(
|
||||
return_value=[
|
||||
{
|
||||
"project_id": "project-a",
|
||||
"owner_user_id": "alice",
|
||||
"status": "recovering",
|
||||
}
|
||||
]
|
||||
),
|
||||
get_settings=AsyncMock(return_value={"enabled": True}),
|
||||
)
|
||||
projects = SimpleNamespace(
|
||||
get=AsyncMock(return_value=project),
|
||||
thread_ids=AsyncMock(return_value=["thread-a", "thread-b"]),
|
||||
)
|
||||
app = SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
analysis_project_store=projects,
|
||||
analysis_project_memory_store=store,
|
||||
)
|
||||
)
|
||||
enqueue = AsyncMock(return_value=2)
|
||||
monkeypatch.setattr(runtime, "enqueue_project_rebuild", enqueue)
|
||||
dispatcher = runtime.AnalysisProjectMemoryDispatcher(
|
||||
store,
|
||||
SimpleNamespace(app=app),
|
||||
)
|
||||
|
||||
assert await dispatcher.recover_stalled_rebuilds_once() == 1
|
||||
enqueue.assert_awaited_once_with(
|
||||
app=app,
|
||||
project=project,
|
||||
thread_ids=["thread-a", "thread-b"],
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_route_does_not_mutate_checkpoint_when_memory_fence_fails():
|
||||
class Checkpointer:
|
||||
def __init__(self):
|
||||
self.aput = AsyncMock()
|
||||
|
||||
async def aget_tuple(self, _config):
|
||||
return SimpleNamespace(
|
||||
checkpoint={
|
||||
"channel_values": {
|
||||
"messages": [HumanMessage(id="message-a", content="旧内容")]
|
||||
},
|
||||
"channel_versions": {"messages": 1},
|
||||
},
|
||||
metadata={},
|
||||
)
|
||||
|
||||
def get_next_version(self, value, _channel):
|
||||
return int(value or 0) + 1
|
||||
|
||||
checkpointer = Checkpointer()
|
||||
project = {"id": "project-a", "user_id": "alice", "mode": "qa"}
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
checkpointer=checkpointer,
|
||||
analysis_project_store=SimpleNamespace(
|
||||
project_for_thread=AsyncMock(return_value=project)
|
||||
),
|
||||
analysis_project_memory_store=SimpleNamespace(
|
||||
invalidate_source=AsyncMock(
|
||||
side_effect=RuntimeError("memory database unavailable")
|
||||
)
|
||||
),
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await thread_routes.edit_thread_message.__wrapped__(
|
||||
"thread-a",
|
||||
"message-a",
|
||||
thread_routes.EditMessageRequest(content="新内容"),
|
||||
request,
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 503
|
||||
checkpointer.aput.assert_not_awaited()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_edit_validation_after_invalidation_rebuilds_unchanged_checkpoint(
|
||||
monkeypatch,
|
||||
):
|
||||
class Checkpointer:
|
||||
def __init__(self):
|
||||
self.aput = AsyncMock()
|
||||
|
||||
async def aget_tuple(self, _config):
|
||||
return SimpleNamespace(
|
||||
checkpoint={
|
||||
"channel_values": {
|
||||
"messages": [HumanMessage(id="message-a", content="旧内容")]
|
||||
},
|
||||
"channel_versions": {"messages": 1},
|
||||
},
|
||||
metadata={},
|
||||
)
|
||||
|
||||
def get_next_version(self, value, _channel):
|
||||
return int(value or 0) + 1
|
||||
|
||||
project = {"id": "project-a", "user_id": "alice", "mode": "qa"}
|
||||
invalidate = AsyncMock(return_value=project)
|
||||
finalize = AsyncMock()
|
||||
monkeypatch.setattr(
|
||||
thread_routes,
|
||||
"_invalidate_project_memory_source_before_mutation",
|
||||
invalidate,
|
||||
)
|
||||
monkeypatch.setattr(
|
||||
thread_routes,
|
||||
"_finalize_project_memory_source_mutation",
|
||||
finalize,
|
||||
)
|
||||
|
||||
def fail_edit(_message, _content):
|
||||
raise TypeError("provider message is immutable")
|
||||
|
||||
monkeypatch.setattr(thread_routes, "_set_message_content", fail_edit)
|
||||
checkpointer = Checkpointer()
|
||||
request = SimpleNamespace(
|
||||
app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer))
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await thread_routes.edit_thread_message.__wrapped__(
|
||||
"thread-a",
|
||||
"message-a",
|
||||
thread_routes.EditMessageRequest(content="新内容"),
|
||||
request,
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 422
|
||||
checkpointer.aput.assert_not_awaited()
|
||||
finalize.assert_awaited_once_with(
|
||||
request,
|
||||
project=project,
|
||||
thread_id="thread-a",
|
||||
)
|
||||
Reference in new issue
Block a user