535 lines
20 KiB
Python
535 lines
20 KiB
Python
"""End-to-end effects tests for project-scoped cross-conversation memory.
|
|
|
|
These tests deliberately use the real SQLite repositories and the real run
|
|
policy. Only infrastructure outside this feature's boundary is substituted:
|
|
the completed-thread checkpointer, the extraction LLM, and the unrelated
|
|
generic/personal memory switch used by independence regressions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
from sqlalchemy import event
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
|
|
|
|
from app.gateway import analysis_project_memory as memory_runtime
|
|
from app.gateway import analysis_projects as project_policy
|
|
from app.gateway.routers.thread_runs import RunCreateRequest
|
|
from deerflow.agents.analysis_project import project_system_prompt
|
|
from deerflow.config import system_settings
|
|
from deerflow.persistence.analysis_project_memory import (
|
|
AnalysisProjectMemoryItemRow,
|
|
AnalysisProjectMemoryJobRow,
|
|
AnalysisProjectMemoryStateRow,
|
|
AnalysisProjectMemoryStore,
|
|
)
|
|
from deerflow.persistence.analysis_projects import (
|
|
AnalysisProjectConfigRow,
|
|
AnalysisProjectRow,
|
|
AnalysisProjectStore,
|
|
AnalysisProjectThreadRow,
|
|
)
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
|
|
|
|
class _Checkpoint:
|
|
"""Small mutable checkpointer implementing the one read used by memory."""
|
|
|
|
def __init__(self) -> None:
|
|
self._messages: dict[str, list[object]] = {}
|
|
|
|
def set_messages(self, thread_id: str, messages: list[object]) -> None:
|
|
self._messages[thread_id] = messages
|
|
|
|
async def aget_tuple(self, config: dict) -> SimpleNamespace | None:
|
|
thread_id = str(config.get("configurable", {}).get("thread_id") or "")
|
|
messages = self._messages.get(thread_id)
|
|
if messages is None:
|
|
return None
|
|
return SimpleNamespace(
|
|
checkpoint={"channel_values": {"messages": messages}}
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class _Harness:
|
|
engine: AsyncEngine
|
|
projects: AnalysisProjectStore
|
|
memories: AnalysisProjectMemoryStore
|
|
checkpointer: _Checkpoint
|
|
app: FastAPI
|
|
request: Request
|
|
|
|
async def close(self) -> None:
|
|
await self.engine.dispose()
|
|
|
|
|
|
async def _make_harness(tmp_path) -> _Harness:
|
|
engine = create_async_engine(
|
|
f"sqlite+aiosqlite:///{tmp_path / 'project-memory-integration.db'}"
|
|
)
|
|
|
|
@event.listens_for(engine.sync_engine, "connect")
|
|
def _enable_foreign_keys(dbapi_connection, _connection_record): # noqa: ANN001
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys=ON")
|
|
cursor.close()
|
|
|
|
# Create only the production tables exercised by this feature. This keeps
|
|
# the test independent from unrelated schemas while preserving all real
|
|
# foreign keys, constraints, transactions, and repository code paths.
|
|
async with engine.begin() as connection:
|
|
for table in (
|
|
AnalysisProjectConfigRow.__table__,
|
|
AnalysisProjectRow.__table__,
|
|
AnalysisProjectThreadRow.__table__,
|
|
AnalysisProjectMemoryStateRow.__table__,
|
|
AnalysisProjectMemoryItemRow.__table__,
|
|
AnalysisProjectMemoryJobRow.__table__,
|
|
):
|
|
await connection.run_sync(table.create)
|
|
|
|
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
|
projects = AnalysisProjectStore(session_factory)
|
|
memories = AnalysisProjectMemoryStore(session_factory)
|
|
checkpointer = _Checkpoint()
|
|
app = FastAPI()
|
|
app.state.analysis_project_store = projects
|
|
app.state.analysis_project_memory_store = memories
|
|
app.state.analysis_project_memory_dispatcher = None
|
|
app.state.checkpointer = checkpointer
|
|
# Absence of an encoder is a supported production degradation path.
|
|
app.state.llmwiki_embedding = None
|
|
request = Request(
|
|
{
|
|
"type": "http",
|
|
"app": app,
|
|
"headers": [],
|
|
"method": "POST",
|
|
"path": "/api/threads/test/runs",
|
|
"query_string": b"",
|
|
"root_path": "",
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": ("testclient", 50000),
|
|
}
|
|
)
|
|
return _Harness(engine, projects, memories, checkpointer, app, request)
|
|
|
|
|
|
@contextmanager
|
|
def _as_user(user_id: str):
|
|
token = set_current_user(SimpleNamespace(id=user_id))
|
|
try:
|
|
yield
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
|
|
def _set_master_switch(monkeypatch: pytest.MonkeyPatch, *, enabled: bool) -> None:
|
|
"""Set the unrelated generic-memory switch for independence regressions."""
|
|
|
|
monkeypatch.setattr(
|
|
system_settings,
|
|
"load_system_settings",
|
|
lambda: SimpleNamespace(memory=SimpleNamespace(enabled=enabled)),
|
|
)
|
|
|
|
|
|
def _run_body(question: str) -> RunCreateRequest:
|
|
return RunCreateRequest(
|
|
input={"messages": [{"role": "user", "content": question}]},
|
|
context={},
|
|
config={"configurable": {}, "context": {}},
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thread_a_extraction_is_recalled_by_thread_b_and_enters_prompt(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Exercise the complete durable path users depend on across conversations."""
|
|
|
|
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
|
# Project memory must remain fully functional while generic/personal
|
|
# memory is disabled deployment-wide.
|
|
_set_master_switch(monkeypatch, enabled=False)
|
|
harness = await _make_harness(tmp_path)
|
|
try:
|
|
project = await harness.projects.create(
|
|
"alice", mode="qa", name="火星登陆", background="制定登陆方案"
|
|
)
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-a")
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-b")
|
|
harness.checkpointer.set_messages(
|
|
"thread-a",
|
|
[
|
|
HumanMessage(id="message-a-user", content="最终着陆点定在哪里？"),
|
|
AIMessage(
|
|
id="message-a-assistant",
|
|
content="已经确认：最终着陆点为乌托邦平原。",
|
|
),
|
|
],
|
|
)
|
|
|
|
queued = await memory_runtime.enqueue_completed_project_run(
|
|
app=harness.app,
|
|
project=project,
|
|
thread_id="thread-a",
|
|
run_id="run-a",
|
|
)
|
|
assert queued is not None
|
|
assert queued["status"] == "queued"
|
|
|
|
[job] = await harness.memories.claim_jobs(
|
|
lease_owner="integration-worker",
|
|
lease_until=datetime.now(UTC) + timedelta(minutes=3),
|
|
limit=1,
|
|
)
|
|
executor = memory_runtime.AnalysisProjectMemoryExecutor(harness.app)
|
|
extraction_llm = AsyncMock(
|
|
return_value={
|
|
"items": [
|
|
{
|
|
"logical_key": "landing-site",
|
|
"kind": "decision",
|
|
"content": "最终着陆点为乌托邦平原",
|
|
"confidence": 0.98,
|
|
"importance": 0.95,
|
|
"pinned": False,
|
|
}
|
|
],
|
|
"snapshot": {"summary": "登陆地点已经确认"},
|
|
}
|
|
)
|
|
monkeypatch.setattr(executor, "_model_extraction", extraction_llm)
|
|
await executor.execute(job, lease_owner="integration-worker")
|
|
|
|
extraction_llm.assert_awaited_once()
|
|
stored_items = await harness.memories.list_items(project["id"], "alice")
|
|
stored = next(
|
|
item for item in stored_items if item["logical_key"] == "landing-site"
|
|
)
|
|
assert stored["content"] == "最终着陆点为乌托邦平原"
|
|
assert stored["source_thread_id"] == "thread-a"
|
|
assert stored["source_message_ids"] == [
|
|
"message-a-user",
|
|
"message-a-assistant",
|
|
]
|
|
assert isinstance(stored["metadata"]["source_order"], float)
|
|
assert any(
|
|
item["kind"] == "episode"
|
|
and item["metadata"].get("raw_project_conversation") is True
|
|
for item in stored_items
|
|
)
|
|
[completed_job] = await harness.memories.list_jobs(project["id"], "alice")
|
|
assert completed_job["status"] == "completed"
|
|
assert completed_job["result"]["fallback"] is False
|
|
|
|
with _as_user("alice"):
|
|
prepared, resolved = await project_policy.prepare_project_run(
|
|
_run_body("上次确定的着陆点是什么？"),
|
|
"thread-b",
|
|
harness.request,
|
|
)
|
|
|
|
assert resolved is not None and resolved["id"] == project["id"]
|
|
assert prepared.context["project_memory_scope"] == {
|
|
"kind": "analysis_project",
|
|
"project_id": project["id"],
|
|
"owner_user_id": "alice",
|
|
"enabled": True,
|
|
}
|
|
recalled = prepared.context["project_memory"]
|
|
# Revision 1 is the immediate raw-exchange index written when the run
|
|
# completes; the extractor merge advances the durable project state
|
|
# once more.
|
|
assert recalled["revision"] == 2
|
|
assert recalled["items"][0]["content"] == "最终着陆点为乌托邦平原"
|
|
assert recalled["items"][0]["source_thread_id"] == "thread-a"
|
|
|
|
prompt = project_system_prompt(prepared.context)
|
|
assert "<project_memory>" in prompt
|
|
assert "最终着陆点为乌托邦平原" in prompt
|
|
assert "同一项目历史对话整理出的背景数据" in prompt
|
|
finally:
|
|
await harness.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_thread_b_recalls_immediate_nickname_while_incremental_job_is_running(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""An ordinary extractor job must not hide its already indexed context."""
|
|
|
|
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
|
harness = await _make_harness(tmp_path)
|
|
try:
|
|
project = await harness.projects.create(
|
|
"alice", mode="qa", name="昵称记忆测试"
|
|
)
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-a")
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-b")
|
|
harness.checkpointer.set_messages(
|
|
"thread-a",
|
|
[
|
|
HumanMessage(
|
|
id="nickname-user",
|
|
content="你现在叫铁蛋，我给你起的新名字。",
|
|
),
|
|
AIMessage(id="nickname-assistant", content="好的。"),
|
|
],
|
|
)
|
|
|
|
queued = await memory_runtime.enqueue_completed_project_run(
|
|
app=harness.app,
|
|
project=project,
|
|
thread_id="thread-a",
|
|
run_id="nickname-run",
|
|
)
|
|
assert queued is not None
|
|
immediate_items = await harness.memories.list_items(project["id"], "alice")
|
|
assert any(
|
|
item["logical_key"] == "assistant.nickname"
|
|
and "铁蛋" in item["content"]
|
|
for item in immediate_items
|
|
)
|
|
|
|
[running] = await harness.memories.claim_jobs(
|
|
lease_owner="slow-incremental-worker",
|
|
lease_until=datetime.now(UTC) + timedelta(minutes=3),
|
|
limit=1,
|
|
)
|
|
assert running["id"] == queued["id"]
|
|
state = await harness.memories.get_state(project["id"], "alice")
|
|
assert state is not None and state["status"] == "building"
|
|
|
|
with _as_user("alice"):
|
|
prepared, resolved = await project_policy.prepare_project_run(
|
|
_run_body("我给你起的新昵称叫什么？"),
|
|
"thread-b",
|
|
harness.request,
|
|
)
|
|
|
|
assert resolved is not None and resolved["id"] == project["id"]
|
|
recalled = prepared.context["project_memory"]
|
|
assert recalled is not None
|
|
assert any("铁蛋" in item["content"] for item in recalled["items"])
|
|
assert "铁蛋" in project_system_prompt(prepared.context)
|
|
finally:
|
|
await harness.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_project_and_owner_boundaries_prevent_memory_leakage(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
|
_set_master_switch(monkeypatch, enabled=True)
|
|
harness = await _make_harness(tmp_path)
|
|
try:
|
|
alpha = await harness.projects.create("alice", mode="qa", name="Alpha")
|
|
beta = await harness.projects.create("alice", mode="qa", name="Beta")
|
|
bob_project = await harness.projects.create("bob", mode="qa", name="Bob")
|
|
await harness.projects.bind_thread(alpha["id"], "alice", "alpha-thread")
|
|
await harness.projects.bind_thread(beta["id"], "alice", "beta-thread")
|
|
await harness.projects.bind_thread(bob_project["id"], "bob", "bob-thread")
|
|
|
|
await harness.memories.upsert_item(
|
|
alpha["id"],
|
|
"alice",
|
|
logical_key="alpha-secret",
|
|
kind="fact",
|
|
content="Alpha 项目的代号是朱雀",
|
|
source_thread_id="alpha-source",
|
|
)
|
|
await harness.memories.upsert_item(
|
|
beta["id"],
|
|
"alice",
|
|
logical_key="beta-secret",
|
|
kind="fact",
|
|
content="Beta 项目的代号是玄武",
|
|
source_thread_id="beta-source",
|
|
)
|
|
await harness.memories.upsert_item(
|
|
bob_project["id"],
|
|
"bob",
|
|
logical_key="bob-secret",
|
|
kind="fact",
|
|
content="Bob 项目的代号是白虎",
|
|
source_thread_id="bob-source",
|
|
)
|
|
|
|
with _as_user("alice"):
|
|
prepared, _ = await project_policy.prepare_project_run(
|
|
_run_body("项目代号是什么？"), "beta-thread", harness.request
|
|
)
|
|
serialized = str(prepared.context["project_memory"])
|
|
assert "Beta 项目的代号是玄武" in serialized
|
|
assert "Alpha 项目的代号是朱雀" not in serialized
|
|
assert "Bob 项目的代号是白虎" not in serialized
|
|
|
|
# Repository defence-in-depth requires both project and owner; knowing
|
|
# a project id alone cannot reveal its memory.
|
|
wrong_owner = await harness.memories.recall(
|
|
alpha["id"], "bob", query="朱雀"
|
|
)
|
|
assert wrong_owner["state"] is None
|
|
assert wrong_owner["items"] == []
|
|
|
|
# The run-policy boundary also rejects an owner trying to use another
|
|
# user's bound thread before any memory can be injected.
|
|
with _as_user("bob"), pytest.raises(HTTPException) as exc_info:
|
|
await project_policy.prepare_project_run(
|
|
_run_body("Alpha 项目的代号是什么？"),
|
|
"alpha-thread",
|
|
harness.request,
|
|
)
|
|
assert exc_info.value.status_code == 404
|
|
finally:
|
|
await harness.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_generic_memory_switch_off_does_not_block_project_memory(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
|
_set_master_switch(monkeypatch, enabled=False)
|
|
harness = await _make_harness(tmp_path)
|
|
try:
|
|
project = await harness.projects.create("alice", mode="qa", name="关闭记忆")
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-a")
|
|
await harness.projects.bind_thread(project["id"], "alice", "thread-b")
|
|
await harness.memories.upsert_item(
|
|
project["id"],
|
|
"alice",
|
|
logical_key="project-date",
|
|
kind="fact",
|
|
content="项目约定日期为2026年12月31日",
|
|
source_thread_id="thread-a",
|
|
)
|
|
harness.checkpointer.set_messages(
|
|
"thread-a",
|
|
[
|
|
HumanMessage(id="u", content="项目约定日期改为2026年12月31日"),
|
|
AIMessage(id="a", content="已确认项目约定日期为2026年12月31日"),
|
|
],
|
|
)
|
|
|
|
with _as_user("alice"):
|
|
prepared, _ = await project_policy.prepare_project_run(
|
|
_run_body("项目约定日期是哪天？"), "thread-b", harness.request
|
|
)
|
|
assert prepared.context["project_memory_scope"]["enabled"] is True
|
|
assert prepared.context["project_memory"]["items"][0]["content"] == (
|
|
"项目约定日期为2026年12月31日"
|
|
)
|
|
assert "项目约定日期为2026年12月31日" in project_system_prompt(
|
|
prepared.context
|
|
)
|
|
|
|
queued = await memory_runtime.enqueue_completed_project_run(
|
|
app=harness.app,
|
|
project=project,
|
|
thread_id="thread-a",
|
|
run_id="run-a",
|
|
)
|
|
assert queued is not None and queued["status"] == "queued"
|
|
|
|
class CompletingExecutor:
|
|
app = harness.app
|
|
|
|
async def execute(self, job, *, lease_owner):
|
|
await harness.memories.complete_job(
|
|
job["id"],
|
|
lease_owner=lease_owner,
|
|
result={"project_memory_independent": True},
|
|
)
|
|
|
|
dispatcher = memory_runtime.AnalysisProjectMemoryDispatcher(
|
|
harness.memories,
|
|
CompletingExecutor(),
|
|
concurrency=1,
|
|
)
|
|
assert await dispatcher.dispatch_once() == 1
|
|
tasks = list(dispatcher._running.values())
|
|
await asyncio.gather(*tasks)
|
|
[completed] = await harness.memories.list_jobs(project["id"], "alice")
|
|
assert completed["id"] == queued["id"]
|
|
assert completed["status"] == "completed"
|
|
assert completed["result"] == {"project_memory_independent": True}
|
|
finally:
|
|
await harness.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_keyword_recall_is_effective_without_vector_encoder(
|
|
tmp_path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Vector unavailability must degrade to useful deterministic recall."""
|
|
|
|
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
|
_set_master_switch(monkeypatch, enabled=True)
|
|
harness = await _make_harness(tmp_path)
|
|
try:
|
|
project = await harness.projects.create("alice", mode="qa", name="预算项目")
|
|
await harness.projects.bind_thread(project["id"], "alice", "query-thread")
|
|
await harness.memories.upsert_item(
|
|
project["id"],
|
|
"alice",
|
|
logical_key="project-budget",
|
|
kind="fact",
|
|
content="The confirmed project budget is 500000 yuan",
|
|
importance=0.2,
|
|
source_thread_id="budget-thread",
|
|
)
|
|
await harness.memories.upsert_item(
|
|
project["id"],
|
|
"alice",
|
|
logical_key="project-venue",
|
|
kind="fact",
|
|
content="The conference venue is Shanghai",
|
|
importance=1.0,
|
|
source_thread_id="venue-thread",
|
|
)
|
|
|
|
direct = await harness.memories.recall(
|
|
project["id"],
|
|
"alice",
|
|
query="What is the confirmed project budget in yuan?",
|
|
)
|
|
assert direct["items"][0]["logical_key"] == "project-budget"
|
|
assert direct["items"][0]["recall_signals"]["lexical"] > 0
|
|
assert all(
|
|
item["recall_signals"]["vector"] == 0 for item in direct["items"]
|
|
)
|
|
assert all(
|
|
item["recall_signals"]["vector_compatible"] is False
|
|
for item in direct["items"]
|
|
)
|
|
|
|
with _as_user("alice"):
|
|
prepared, _ = await project_policy.prepare_project_run(
|
|
_run_body("What is the confirmed project budget in yuan?"),
|
|
"query-thread",
|
|
harness.request,
|
|
)
|
|
assert prepared.context["project_memory"]["items"][0]["content"] == (
|
|
"The confirmed project budget is 500000 yuan"
|
|
)
|
|
assert "The confirmed project budget is 500000 yuan" in project_system_prompt(
|
|
prepared.context
|
|
)
|
|
finally:
|
|
await harness.close()
|