981 lines
32 KiB
Python
981 lines
32 KiB
Python
"""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",
|
|
)
|