mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
feat(memory): scope management API by agent (#5565)
* feat(memory): scope management API by agent Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> * fix(memory): preserve scoped management isolation Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> --------- Signed-off-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com> Co-authored-by: PeaceMaker-best <221849497+PeaceMaker-best@users.noreply.github.com>
This commit is contained in:
PeaceMaker-best
GitHub
parent
af67738521
commit
b73a4e79bf
@ -9,6 +9,7 @@ from pydantic import BaseModel, Field, field_validator
|
||||
from app.gateway.authz import require_permission
|
||||
from app.gateway.internal_auth import get_trusted_internal_owner_user_id
|
||||
from deerflow.agents.memory import MemoryConflictError, MemoryCorruptionError, MemoryManager, get_memory_manager
|
||||
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
||||
from deerflow.config.memory_config import get_memory_config
|
||||
from deerflow.config.paths import make_safe_user_id
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
@ -154,7 +155,46 @@ def _unsupported_501(manager: object, label: str) -> HTTPException:
|
||||
)
|
||||
|
||||
|
||||
async def _get_memory_or_501(manager: MemoryManager, user_id: str, label: str) -> dict[str, Any]:
|
||||
def _management_agent_name_or_501(
|
||||
manager: MemoryManager,
|
||||
agent_name: str | None,
|
||||
) -> str | None:
|
||||
"""Validate an optional agent-scoped management target.
|
||||
|
||||
A backend must explicitly declare that its management operations bind to
|
||||
``agent_name``. Merely accepting the keyword is not enough: adapters may
|
||||
expose user-global memory and ignore it, which would turn a scoped edit
|
||||
into a mutation of the wrong bucket. Preserve the caller's spelling here:
|
||||
remote backends such as Mem0 use the agent name as a case-sensitive storage
|
||||
identity, while DeerMem performs its own lowercase canonicalization.
|
||||
"""
|
||||
if agent_name is None:
|
||||
return None
|
||||
if not AGENT_NAME_PATTERN.fullmatch(agent_name):
|
||||
raise HTTPException(
|
||||
status_code=422,
|
||||
detail=(f"Invalid agent name '{agent_name}'. Must match {AGENT_NAME_PATTERN.pattern} (letters, digits, and hyphens only)."),
|
||||
)
|
||||
if getattr(manager, "supports_agent_scoped_management", False) is not True:
|
||||
raise HTTPException(
|
||||
status_code=501,
|
||||
detail=(f"Operation 'agent-scoped management' not supported by memory backend '{type(manager).__name__}'."),
|
||||
)
|
||||
return agent_name
|
||||
|
||||
|
||||
def _agent_scope_kwargs(agent_name: str | None) -> dict[str, str]:
|
||||
"""Pass an agent scope only when the caller explicitly selected one."""
|
||||
return {"agent_name": agent_name} if agent_name is not None else {}
|
||||
|
||||
|
||||
async def _get_memory_or_501(
|
||||
manager: MemoryManager,
|
||||
user_id: str,
|
||||
label: str,
|
||||
*,
|
||||
agent_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Read the full memory doc; 501 if the backend doesn't expose one.
|
||||
|
||||
``get_memory`` is tier-2 (default ``raise NotImplementedError``); a minimal
|
||||
@ -164,8 +204,13 @@ async def _get_memory_or_501(manager: MemoryManager, user_id: str, label: str) -
|
||||
of a raw 500. ``label`` is the operation name in the 501 detail (the
|
||||
endpoint's verb, e.g. "get memory" / "export memory" / "reload memory").
|
||||
"""
|
||||
scope_kwargs = _agent_scope_kwargs(agent_name)
|
||||
try:
|
||||
return await asyncio.to_thread(manager.get_memory, user_id=user_id)
|
||||
return await asyncio.to_thread(
|
||||
manager.get_memory,
|
||||
user_id=user_id,
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, label) from None
|
||||
except (MemoryConflictError, MemoryCorruptionError) as exc:
|
||||
@ -211,11 +256,11 @@ class MemoryStatusResponse(BaseModel):
|
||||
response_model=MemoryResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Get Memory Data",
|
||||
description="Retrieve the current global memory data including user context, history, and facts.",
|
||||
description="Retrieve one selected agent's memory, or the default bucket when agent_name is omitted.",
|
||||
)
|
||||
@require_permission("memory", "read")
|
||||
async def get_memory(request: Request) -> MemoryResponse:
|
||||
"""Get the current global memory data.
|
||||
async def get_memory(request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Get memory data for one agent scope or the default bucket.
|
||||
|
||||
Returns:
|
||||
The current memory data with user context, history, and facts.
|
||||
@ -249,7 +294,13 @@ async def get_memory(request: Request) -> MemoryResponse:
|
||||
```
|
||||
"""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
memory_data = await _get_memory_or_501(manager, _resolve_memory_user_id(request), "get memory")
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
memory_data = await _get_memory_or_501(
|
||||
manager,
|
||||
_resolve_memory_user_id(request),
|
||||
"get memory",
|
||||
agent_name=selected_agent,
|
||||
)
|
||||
return MemoryResponse(**memory_data)
|
||||
|
||||
|
||||
@ -258,10 +309,10 @@ async def get_memory(request: Request) -> MemoryResponse:
|
||||
response_model=MemoryResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Reload Memory Data",
|
||||
description="Reload memory data from the storage file, refreshing the in-memory cache.",
|
||||
description="Reload one selected agent's memory, or the default bucket when agent_name is omitted.",
|
||||
)
|
||||
@require_permission("memory", "read")
|
||||
async def reload_memory(request: Request) -> MemoryResponse:
|
||||
async def reload_memory(request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Reload memory data from file.
|
||||
|
||||
This forces a reload of the memory data from the storage file,
|
||||
@ -272,8 +323,14 @@ async def reload_memory(request: Request) -> MemoryResponse:
|
||||
"""
|
||||
user_id = _resolve_memory_user_id(request)
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data = await asyncio.to_thread(manager.reload_memory, user_id=user_id)
|
||||
memory_data = await asyncio.to_thread(
|
||||
manager.reload_memory,
|
||||
user_id=user_id,
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
# Non-DeerMem backends have no reload concept; fall back to get_memory
|
||||
# (read-only refresh, so degrading is safe and still useful -- vs fact
|
||||
@ -281,7 +338,12 @@ async def reload_memory(request: Request) -> MemoryResponse:
|
||||
# would hide data loss). If get_memory is also unsupported (a minimal
|
||||
# backend with no full doc), surface 501 rather than a raw 500: reads
|
||||
# degrade only when there is a doc to degrade to.
|
||||
memory_data = await _get_memory_or_501(manager, user_id, "reload memory")
|
||||
memory_data = await _get_memory_or_501(
|
||||
manager,
|
||||
user_id,
|
||||
"reload memory",
|
||||
agent_name=selected_agent,
|
||||
)
|
||||
except (MemoryConflictError, MemoryCorruptionError) as exc:
|
||||
raise _map_memory_manager_error(exc) from exc
|
||||
return MemoryResponse(**memory_data)
|
||||
@ -291,15 +353,21 @@ async def reload_memory(request: Request) -> MemoryResponse:
|
||||
"/memory",
|
||||
response_model=MemoryResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Clear All Memory Data",
|
||||
description="Delete all saved memory data and reset the memory structure to an empty state.",
|
||||
summary="Clear Memory Data",
|
||||
description="Delete one selected agent's memory, or all saved memory for the current user when agent_name is omitted.",
|
||||
)
|
||||
@require_permission("memory", "write")
|
||||
async def clear_memory(request: Request) -> MemoryResponse:
|
||||
"""Clear all persisted memory data."""
|
||||
async def clear_memory(request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Clear one agent bucket, or all user memory when no scope is selected."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data = await asyncio.to_thread(manager.clear_memory, user_id=_resolve_memory_user_id(request))
|
||||
memory_data = await asyncio.to_thread(
|
||||
manager.clear_memory,
|
||||
user_id=_resolve_memory_user_id(request),
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, "clear memory") from None
|
||||
except (MemoryConflictError, MemoryCorruptionError) as exc:
|
||||
@ -318,9 +386,11 @@ async def clear_memory(request: Request) -> MemoryResponse:
|
||||
description="Create a single saved memory fact manually.",
|
||||
)
|
||||
@require_permission("memory", "write")
|
||||
async def create_memory_fact_endpoint(body: FactCreateRequest, request: Request) -> MemoryResponse:
|
||||
async def create_memory_fact_endpoint(body: FactCreateRequest, request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Create a single fact manually."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data, fact_id = await asyncio.to_thread(
|
||||
manager.create_fact,
|
||||
@ -328,6 +398,7 @@ async def create_memory_fact_endpoint(body: FactCreateRequest, request: Request)
|
||||
category=body.category,
|
||||
confidence=body.confidence,
|
||||
user_id=_resolve_memory_user_id(request),
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, "create fact") from None
|
||||
@ -352,11 +423,18 @@ async def create_memory_fact_endpoint(body: FactCreateRequest, request: Request)
|
||||
description="Delete a single saved memory fact by its fact id.",
|
||||
)
|
||||
@require_permission("memory", "write")
|
||||
async def delete_memory_fact_endpoint(fact_id: str, request: Request) -> MemoryResponse:
|
||||
async def delete_memory_fact_endpoint(fact_id: str, request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Delete a single fact from memory by fact id."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data = await asyncio.to_thread(manager.delete_fact, fact_id, user_id=_resolve_memory_user_id(request))
|
||||
memory_data = await asyncio.to_thread(
|
||||
manager.delete_fact,
|
||||
fact_id,
|
||||
user_id=_resolve_memory_user_id(request),
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, "delete fact") from None
|
||||
except KeyError as exc:
|
||||
@ -377,9 +455,11 @@ async def delete_memory_fact_endpoint(fact_id: str, request: Request) -> MemoryR
|
||||
description="Partially update a single saved memory fact by its fact id while preserving omitted fields.",
|
||||
)
|
||||
@require_permission("memory", "write")
|
||||
async def update_memory_fact_endpoint(fact_id: str, body: FactPatchRequest, request: Request) -> MemoryResponse:
|
||||
async def update_memory_fact_endpoint(fact_id: str, body: FactPatchRequest, request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Partially update a single fact manually."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data = await asyncio.to_thread(
|
||||
manager.update_fact,
|
||||
@ -388,6 +468,7 @@ async def update_memory_fact_endpoint(fact_id: str, body: FactPatchRequest, requ
|
||||
category=body.category,
|
||||
confidence=body.confidence,
|
||||
user_id=_resolve_memory_user_id(request),
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, "update fact") from None
|
||||
@ -408,13 +489,19 @@ async def update_memory_fact_endpoint(fact_id: str, body: FactPatchRequest, requ
|
||||
response_model=MemoryResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Export Memory Data",
|
||||
description="Export the current global memory data as JSON for backup or transfer.",
|
||||
description="Export one selected agent's memory as JSON, or the default bucket when agent_name is omitted.",
|
||||
)
|
||||
@require_permission("memory", "read")
|
||||
async def export_memory(request: Request) -> MemoryResponse:
|
||||
async def export_memory(request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Export the current memory data."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
memory_data = await _get_memory_or_501(manager, _resolve_memory_user_id(request), "export memory")
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
memory_data = await _get_memory_or_501(
|
||||
manager,
|
||||
_resolve_memory_user_id(request),
|
||||
"export memory",
|
||||
agent_name=selected_agent,
|
||||
)
|
||||
return MemoryResponse(**memory_data)
|
||||
|
||||
|
||||
@ -423,17 +510,20 @@ async def export_memory(request: Request) -> MemoryResponse:
|
||||
response_model=MemoryResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Import Memory Data",
|
||||
description="Import and overwrite the current global memory data from a JSON payload.",
|
||||
description="Import the default memory document, or replace only one selected agent's facts while preserving shared summaries.",
|
||||
)
|
||||
@require_permission("memory", "write")
|
||||
async def import_memory(body: MemoryResponse, request: Request) -> MemoryResponse:
|
||||
"""Import and persist memory data."""
|
||||
async def import_memory(body: MemoryResponse, request: Request, agent_name: str | None = None) -> MemoryResponse:
|
||||
"""Import memory data, preserving shared summaries for a selected agent."""
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
scope_kwargs = _agent_scope_kwargs(selected_agent)
|
||||
try:
|
||||
memory_data = await asyncio.to_thread(
|
||||
manager.import_memory,
|
||||
body.model_dump(exclude_none=True),
|
||||
user_id=_resolve_memory_user_id(request),
|
||||
**scope_kwargs,
|
||||
)
|
||||
except NotImplementedError:
|
||||
raise _unsupported_501(manager, "import memory") from None
|
||||
@ -502,10 +592,10 @@ async def get_memory_config_endpoint(request: Request) -> MemoryConfigResponse:
|
||||
response_model=MemoryStatusResponse,
|
||||
response_model_exclude_none=True,
|
||||
summary="Get Memory Status",
|
||||
description="Retrieve both memory configuration and current data in a single request.",
|
||||
description="Retrieve memory configuration and one selected agent's data, using the default bucket when agent_name is omitted.",
|
||||
)
|
||||
@require_permission("memory", "read")
|
||||
async def get_memory_status(request: Request) -> MemoryStatusResponse:
|
||||
async def get_memory_status(request: Request, agent_name: str | None = None) -> MemoryStatusResponse:
|
||||
"""Get the memory system status including configuration and data.
|
||||
|
||||
Returns:
|
||||
@ -513,7 +603,13 @@ async def get_memory_status(request: Request) -> MemoryStatusResponse:
|
||||
"""
|
||||
config = get_memory_config()
|
||||
manager = await asyncio.to_thread(get_memory_manager)
|
||||
memory_data = await _get_memory_or_501(manager, _resolve_memory_user_id(request), "get memory status")
|
||||
selected_agent = _management_agent_name_or_501(manager, agent_name)
|
||||
memory_data = await _get_memory_or_501(
|
||||
manager,
|
||||
_resolve_memory_user_id(request),
|
||||
"get memory status",
|
||||
agent_name=selected_agent,
|
||||
)
|
||||
|
||||
return MemoryStatusResponse(
|
||||
config=MemoryConfigResponse(
|
||||
|
||||
@ -63,7 +63,22 @@ The legacy shared agent layout is read-only fallback data.
|
||||
|
||||
DeerMem maps a missing agent name to `__default__`.
|
||||
That name is reserved and cannot identify a custom agent.
|
||||
Public agent names use lowercase canonical form.
|
||||
DeerMem canonicalizes public agent names to lowercase for local storage.
|
||||
Remote backends may preserve the case-sensitive identity used when facts were
|
||||
written.
|
||||
|
||||
Gateway management reads, reload, import/export, clear, and single-fact CRUD
|
||||
accept an optional `agent_name`. A backend must opt in with
|
||||
`supports_agent_scoped_management = True`; otherwise a scoped request returns
|
||||
501 instead of silently operating on user-global or default-bucket data.
|
||||
Omitting the parameter preserves the legacy default bucket for reads, reload,
|
||||
import/export, and fact CRUD; omitting it from clear preserves the legacy
|
||||
user-wide clear. Gateway validates the public agent-name grammar but preserves
|
||||
the caller's spelling so case-sensitive remote identities remain reachable;
|
||||
each backend owns any storage-specific canonicalization.
|
||||
Scoped import replaces only the selected agent's facts. It always preserves the
|
||||
user's shared `user` and `history` summaries, including when an older or
|
||||
fact-only import payload supplies empty summary defaults.
|
||||
|
||||
#### Operating modes
|
||||
|
||||
|
||||
@ -111,6 +111,7 @@ class DeerMem(MemoryManager):
|
||||
# for tool mode). Backends without real search inherit the False default and
|
||||
# cannot be used with mode="tool".
|
||||
supports_search: ClassVar[bool] = True
|
||||
supports_agent_scoped_management: ClassVar[bool] = True
|
||||
|
||||
def model_post_init(self, __context: Any) -> None:
|
||||
"""Construct DeerMem's dependencies from ``self.backend_config``.
|
||||
@ -575,6 +576,7 @@ class DeerMem(MemoryManager):
|
||||
memory_data,
|
||||
agent_name=_resolve_agent_name(agent_name),
|
||||
user_id=user_id,
|
||||
replace_shared_summaries=agent_name is None,
|
||||
)
|
||||
)
|
||||
return _compat_document(imported)
|
||||
|
||||
@ -960,7 +960,14 @@ class MemoryUpdater:
|
||||
except Exception:
|
||||
logger.warning("Failed to record capacity-eviction audit", exc_info=True)
|
||||
|
||||
def import_memory_data(self, memory_data: dict[str, Any], agent_name: str | None = None, *, user_id: str | None = None) -> dict[str, Any]:
|
||||
def import_memory_data(
|
||||
self,
|
||||
memory_data: dict[str, Any],
|
||||
agent_name: str | None = None,
|
||||
*,
|
||||
user_id: str | None = None,
|
||||
replace_shared_summaries: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Persist imported memory data via the injected storage."""
|
||||
if not isinstance(memory_data, dict):
|
||||
raise ValueError("memory_data")
|
||||
@ -985,14 +992,19 @@ class MemoryUpdater:
|
||||
)
|
||||
current_by_id = {str(fact.get("id")): fact for fact in current.get("facts", []) if isinstance(fact, dict)}
|
||||
incoming_ids = {str(fact.get("id")) for fact in incoming_facts}
|
||||
change_set = {
|
||||
"upserts": incoming_facts,
|
||||
"upsertRevisions": {str(fact.get("id")): (int(current_by_id[str(fact.get("id"))].get("revision") or 1) if str(fact.get("id")) in current_by_id else None) for fact in incoming_facts},
|
||||
"deletes": [fact_id for fact_id in current_by_id if fact_id not in incoming_ids],
|
||||
"deleteRevisions": {fact_id: int(fact.get("revision") or 1) for fact_id, fact in current_by_id.items() if fact_id not in incoming_ids},
|
||||
}
|
||||
if replace_shared_summaries:
|
||||
change_set["summaries"] = {
|
||||
"user": copy.deepcopy(memory_data.get("user", {})),
|
||||
"history": copy.deepcopy(memory_data.get("history", {})),
|
||||
}
|
||||
self._storage.apply_changes(
|
||||
{
|
||||
"upserts": incoming_facts,
|
||||
"upsertRevisions": {str(fact.get("id")): (int(current_by_id[str(fact.get("id"))].get("revision") or 1) if str(fact.get("id")) in current_by_id else None) for fact in incoming_facts},
|
||||
"deletes": [fact_id for fact_id in current_by_id if fact_id not in incoming_ids],
|
||||
"deleteRevisions": {fact_id: int(fact.get("revision") or 1) for fact_id, fact in current_by_id.items() if fact_id not in incoming_ids},
|
||||
"summaries": {"user": copy.deepcopy(memory_data.get("user", {})), "history": copy.deepcopy(memory_data.get("history", {}))},
|
||||
},
|
||||
change_set,
|
||||
agent_name=agent_name,
|
||||
user_id=user_id,
|
||||
expected_manifest_revision=int(current.get("revision") or 0),
|
||||
|
||||
@ -72,6 +72,9 @@ class Mem0Manager(MemoryManager):
|
||||
# search() is overridden below -> flag must be True (contract invariant);
|
||||
# this also enables memory mode="tool".
|
||||
supports_search: ClassVar[bool] = True
|
||||
# Mem0 binds management reads/clears to ``agent_id`` in the remote filter.
|
||||
# Fact CRUD remains unsupported and continues to return the base 501.
|
||||
supports_agent_scoped_management: ClassVar[bool] = True
|
||||
# mem0 extracts/deduplicates facts from full conversations through add();
|
||||
# its fact CRUD hooks are intentionally unsupported, so tool mode retains
|
||||
# passive writes while exposing query-aware search.
|
||||
|
||||
@ -192,6 +192,14 @@ class MemoryManager(BaseModel):
|
||||
# that fails fast at instantiation rather than silently returning empty
|
||||
# results). Default False: a new backend must explicitly opt in to tool mode.
|
||||
supports_search: ClassVar[bool] = False
|
||||
# Opt-in capability for Gateway management calls that name an agent scope.
|
||||
# The HTTP layer rejects a scoped read/write unless the backend declares
|
||||
# this flag, because accepting ``agent_name`` in a Python signature does
|
||||
# not prove that an adapter actually binds storage and mutations to it.
|
||||
# Backends that leave it False retain the unscoped management API and get
|
||||
# an explicit 501 for scoped management instead of silently operating on
|
||||
# the user's default/global bucket.
|
||||
supports_agent_scoped_management: ClassVar[bool] = False
|
||||
# Backends that rely on conversation-level extraction instead of fact CRUD
|
||||
# can retain MemoryMiddleware writes while tool mode supplies query-aware
|
||||
# search. Most backends keep tool mode fully model-directed.
|
||||
@ -417,7 +425,11 @@ class MemoryManager(BaseModel):
|
||||
agent_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Import a memory document into the bucket; return the merged result.
|
||||
Default: unsupported."""
|
||||
|
||||
An explicit ``agent_name`` replaces only that agent's facts. Shared
|
||||
user/history summaries must remain unchanged even when the incoming
|
||||
document contains summary fields. Default: unsupported.
|
||||
"""
|
||||
raise NotImplementedError(f"import_memory not supported by {type(self).__name__}")
|
||||
|
||||
def export_memory(
|
||||
|
||||
@ -640,6 +640,25 @@ class TestMem0ManagerManage:
|
||||
mgr, _fake = _manager()
|
||||
assert mgr.get_memory() == {"facts": []}
|
||||
|
||||
def test_mixed_case_agent_identity_matches_write_read_and_clear(self) -> None:
|
||||
"""Management must keep the exact agent_id used by existing writes."""
|
||||
mgr, fake = _manager()
|
||||
agent_name = "Research-Agent"
|
||||
mgr.add(
|
||||
"thread-1",
|
||||
[HumanMessage(content="I prefer concise reports"), AIMessage(content="Noted.")],
|
||||
user_id="u1",
|
||||
agent_name=agent_name,
|
||||
)
|
||||
fake.list_results = [{"id": "m1", "memory": "prefers concise reports"}]
|
||||
|
||||
assert mgr.get_memory(user_id="u1", agent_name=agent_name)["facts"][0]["id"] == "m1"
|
||||
assert mgr.clear_memory(user_id="u1", agent_name=agent_name) == {"facts": []}
|
||||
|
||||
assert fake.added[0]["agent_id"] == agent_name
|
||||
assert fake.list_calls[0]["filters"] == {"AND": [{"user_id": "u1"}, {"agent_id": agent_name}]}
|
||||
assert fake.deleted == [{"user_id": "u1", "agent_id": agent_name, "run_id": None}]
|
||||
|
||||
def test_clear_memory_deletes_bucket_and_returns_empty(self) -> None:
|
||||
mgr, fake = _manager()
|
||||
assert mgr.clear_memory(user_id="u1", agent_name="lead_agent") == {"facts": []}
|
||||
|
||||
@ -109,6 +109,7 @@ def test_import_memory_route_returns_imported_memory() -> None:
|
||||
response = client.post("/api/memory/import", json=imported_memory)
|
||||
assert response.status_code == 200
|
||||
assert response.json()["facts"] == imported_memory["facts"]
|
||||
assert "agent_name" not in mock_mgr.import_memory.call_args.kwargs
|
||||
|
||||
|
||||
def test_import_route_without_agent_name_persists_default_bucket_markdown(tmp_path) -> None:
|
||||
@ -180,6 +181,7 @@ def test_clear_memory_route_returns_cleared_memory() -> None:
|
||||
response = client.delete("/api/memory")
|
||||
assert response.status_code == 200
|
||||
assert response.json()["facts"] == []
|
||||
assert "agent_name" not in mock_mgr.clear_memory.call_args.kwargs
|
||||
|
||||
|
||||
# ── fact CRUD (normal / error) ─────────────────────────────────────────────
|
||||
@ -199,6 +201,225 @@ def test_create_memory_fact_route_returns_updated_memory() -> None:
|
||||
assert response.json()["facts"] == updated_memory["facts"]
|
||||
|
||||
|
||||
def test_scoped_memory_read_and_fact_crud_preserve_requested_agent_name() -> None:
|
||||
"""The gateway must not remap a case-sensitive remote backend identity."""
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
selected_memory = _sample_memory(
|
||||
facts=[
|
||||
{
|
||||
"id": "fact_shared-id",
|
||||
"content": "Research agent preference",
|
||||
"category": "preference",
|
||||
"confidence": 0.9,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "manual",
|
||||
}
|
||||
]
|
||||
)
|
||||
manager = MagicMock()
|
||||
manager.supports_agent_scoped_management = True
|
||||
manager.get_memory.return_value = selected_memory
|
||||
manager.create_fact.return_value = (selected_memory, "fact_shared-id")
|
||||
manager.update_fact.return_value = selected_memory
|
||||
manager.delete_fact.return_value = _sample_memory()
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
fetched = client.get("/api/memory?agent_name=Research-Agent")
|
||||
created = client.post(
|
||||
"/api/memory/facts?agent_name=Research-Agent",
|
||||
json={"content": "Research agent preference"},
|
||||
)
|
||||
updated = client.patch(
|
||||
"/api/memory/facts/fact_shared-id?agent_name=Research-Agent",
|
||||
json={"confidence": 0.95},
|
||||
)
|
||||
deleted = client.delete("/api/memory/facts/fact_shared-id?agent_name=Research-Agent")
|
||||
|
||||
assert fetched.status_code == 200
|
||||
assert created.status_code == 200
|
||||
assert updated.status_code == 200
|
||||
assert deleted.status_code == 200
|
||||
manager.get_memory.assert_called_once_with(user_id="alice", agent_name="Research-Agent")
|
||||
assert manager.create_fact.call_args.kwargs["agent_name"] == "Research-Agent"
|
||||
assert manager.update_fact.call_args.kwargs["agent_name"] == "Research-Agent"
|
||||
assert manager.delete_fact.call_args.kwargs["agent_name"] == "Research-Agent"
|
||||
|
||||
|
||||
def test_scoped_import_and_clear_preserve_requested_agent_name() -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
selected_memory = _sample_memory()
|
||||
manager = MagicMock()
|
||||
manager.supports_agent_scoped_management = True
|
||||
manager.import_memory.return_value = selected_memory
|
||||
manager.clear_memory.return_value = selected_memory
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
imported = client.post("/api/memory/import?agent_name=Research-Agent", json=selected_memory)
|
||||
cleared = client.delete("/api/memory?agent_name=Research-Agent")
|
||||
|
||||
assert imported.status_code == 200
|
||||
assert cleared.status_code == 200
|
||||
assert manager.import_memory.call_args.kwargs == {"user_id": "alice", "agent_name": "Research-Agent"}
|
||||
assert manager.clear_memory.call_args.kwargs == {"user_id": "alice", "agent_name": "Research-Agent"}
|
||||
|
||||
|
||||
def test_scoped_fact_import_preserves_shared_summaries(tmp_path) -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
|
||||
shared_memory = _sample_memory()
|
||||
shared_memory["user"]["workContext"] = {
|
||||
"summary": "Shared work context",
|
||||
"updatedAt": "2026-09-30T00:00:00Z",
|
||||
}
|
||||
shared_memory["history"]["recentMonths"] = {
|
||||
"summary": "Shared recent history",
|
||||
"updatedAt": "2026-09-30T00:00:00Z",
|
||||
}
|
||||
manager.import_memory(shared_memory, user_id="alice")
|
||||
fact_only_import = {
|
||||
"facts": [
|
||||
{
|
||||
"id": "fact-agent-a",
|
||||
"content": "Agent A preference",
|
||||
"category": "preference",
|
||||
"confidence": 0.9,
|
||||
"createdAt": "2026-09-30T00:00:00Z",
|
||||
"source": "import",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
response = client.post(
|
||||
"/api/memory/import?agent_name=agent-a",
|
||||
json=fact_only_import,
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
imported = response.json()
|
||||
assert [fact["id"] for fact in imported["facts"]] == ["fact-agent-a"]
|
||||
assert imported["user"]["workContext"]["summary"] == "Shared work context"
|
||||
assert imported["history"]["recentMonths"]["summary"] == "Shared recent history"
|
||||
|
||||
|
||||
def test_scoped_import_export_and_clear_are_isolated_between_agent_buckets(tmp_path) -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
|
||||
agent_a = _sample_memory(
|
||||
facts=[
|
||||
{
|
||||
"id": "fact-agent-a",
|
||||
"content": "Agent A preference",
|
||||
"category": "preference",
|
||||
"confidence": 0.9,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "import",
|
||||
}
|
||||
]
|
||||
)
|
||||
agent_b = _sample_memory(
|
||||
facts=[
|
||||
{
|
||||
"id": "fact-agent-b",
|
||||
"content": "Agent B preference",
|
||||
"category": "preference",
|
||||
"confidence": 0.9,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "import",
|
||||
}
|
||||
]
|
||||
)
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
assert client.post("/api/memory/import?agent_name=agent-a", json=agent_a).status_code == 200
|
||||
assert client.post("/api/memory/import?agent_name=agent-b", json=agent_b).status_code == 200
|
||||
exported_a = client.get("/api/memory/export?agent_name=agent-a")
|
||||
cleared_a = client.delete("/api/memory?agent_name=agent-a")
|
||||
exported_b = client.get("/api/memory/export?agent_name=agent-b")
|
||||
|
||||
assert [fact["id"] for fact in exported_a.json()["facts"]] == ["fact-agent-a"]
|
||||
assert cleared_a.status_code == 200
|
||||
assert cleared_a.json()["facts"] == []
|
||||
assert [fact["id"] for fact in exported_b.json()["facts"]] == ["fact-agent-b"]
|
||||
|
||||
|
||||
def test_scoped_management_rejects_backend_without_scope_capability() -> None:
|
||||
"""A backend that ignores agent_name must not silently expose the default bucket."""
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = MagicMock()
|
||||
manager.supports_agent_scoped_management = False
|
||||
|
||||
with patch("app.gateway.routers.memory.get_memory_manager", return_value=manager):
|
||||
with TestClient(app) as client:
|
||||
read_response = client.get("/api/memory?agent_name=research-agent")
|
||||
import_response = client.post("/api/memory/import?agent_name=research-agent", json=_sample_memory())
|
||||
clear_response = client.delete("/api/memory?agent_name=research-agent")
|
||||
|
||||
assert read_response.status_code == 501
|
||||
assert import_response.status_code == 501
|
||||
assert clear_response.status_code == 501
|
||||
assert "agent-scoped management" in read_response.json()["detail"]
|
||||
manager.get_memory.assert_not_called()
|
||||
manager.import_memory.assert_not_called()
|
||||
manager.clear_memory.assert_not_called()
|
||||
|
||||
|
||||
def test_scoped_memory_read_rejects_malformed_agent_name() -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = MagicMock()
|
||||
manager.supports_agent_scoped_management = True
|
||||
|
||||
with patch("app.gateway.routers.memory.get_memory_manager", return_value=manager):
|
||||
with TestClient(app) as client:
|
||||
response = client.get("/api/memory?agent_name=../other-user")
|
||||
|
||||
assert response.status_code == 422
|
||||
manager.get_memory.assert_not_called()
|
||||
|
||||
|
||||
def test_scoped_reload_fallback_reads_the_same_agent_bucket() -> None:
|
||||
"""A backend without reload support must not lose scope on read fallback."""
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = MagicMock()
|
||||
manager.supports_agent_scoped_management = True
|
||||
manager.reload_memory.side_effect = NotImplementedError("no cache")
|
||||
manager.get_memory.return_value = _sample_memory()
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
response = client.post("/api/memory/reload?agent_name=Research-Agent")
|
||||
|
||||
assert response.status_code == 200
|
||||
manager.reload_memory.assert_called_once_with(user_id="alice", agent_name="Research-Agent")
|
||||
manager.get_memory.assert_called_once_with(user_id="alice", agent_name="Research-Agent")
|
||||
|
||||
|
||||
def test_create_memory_fact_route_maps_conflict_to_409() -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
@ -338,6 +559,55 @@ def test_settings_fact_crud_without_agent_name_uses_default_agent(tmp_path) -> N
|
||||
assert "facts" not in json.loads(memory_path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def test_scoped_fact_mutation_does_not_cross_agent_bucket_with_same_fact_id(
|
||||
tmp_path,
|
||||
) -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
manager = DeerMem(backend_config={"storage_path": str(tmp_path)})
|
||||
shared_id = "fact_01HZZZZZZZZZZZZZZZZZZZZZZZ"
|
||||
agent_a = _sample_memory(
|
||||
facts=[
|
||||
{
|
||||
"id": shared_id,
|
||||
"content": "Agent A original",
|
||||
"category": "context",
|
||||
"confidence": 0.8,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "manual",
|
||||
}
|
||||
]
|
||||
)
|
||||
agent_b = _sample_memory(
|
||||
facts=[
|
||||
{
|
||||
"id": shared_id,
|
||||
"content": "Agent B original",
|
||||
"category": "context",
|
||||
"confidence": 0.8,
|
||||
"createdAt": "2026-03-20T00:00:00Z",
|
||||
"source": "manual",
|
||||
}
|
||||
]
|
||||
)
|
||||
manager.import_memory(agent_a, user_id="alice", agent_name="agent-a")
|
||||
manager.import_memory(agent_b, user_id="alice", agent_name="agent-b")
|
||||
|
||||
with (
|
||||
patch("app.gateway.routers.memory.get_memory_manager", return_value=manager),
|
||||
patch("app.gateway.routers.memory.get_effective_user_id", return_value="alice"),
|
||||
TestClient(app) as client,
|
||||
):
|
||||
response = client.patch(
|
||||
f"/api/memory/facts/{shared_id}?agent_name=agent-a",
|
||||
json={"content": "Agent A updated"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert manager.get_memory(user_id="alice", agent_name="agent-a")["facts"][0]["content"] == "Agent A updated"
|
||||
assert manager.get_memory(user_id="alice", agent_name="agent-b")["facts"][0]["content"] == "Agent B original"
|
||||
|
||||
|
||||
def test_update_memory_fact_route_preserves_omitted_fields() -> None:
|
||||
app = make_authed_test_app()
|
||||
app.include_router(memory.router)
|
||||
|
||||
@ -3,8 +3,10 @@ import type { NextRequest } from "next/server";
|
||||
const BACKEND_BASE_URL =
|
||||
process.env.NEXT_PUBLIC_BACKEND_BASE_URL ?? "http://127.0.0.1:8001";
|
||||
|
||||
function buildBackendUrl(pathname: string) {
|
||||
return new URL(pathname, BACKEND_BASE_URL);
|
||||
function buildBackendUrl(request: NextRequest, pathname: string) {
|
||||
const url = new URL(pathname, BACKEND_BASE_URL);
|
||||
url.search = request.nextUrl.search;
|
||||
return url;
|
||||
}
|
||||
|
||||
async function proxyRequest(request: NextRequest, pathname: string) {
|
||||
@ -14,7 +16,7 @@ async function proxyRequest(request: NextRequest, pathname: string) {
|
||||
headers.delete("content-length");
|
||||
|
||||
const hasBody = !["GET", "HEAD"].includes(request.method);
|
||||
const response = await fetch(buildBackendUrl(pathname), {
|
||||
const response = await fetch(buildBackendUrl(request, pathname), {
|
||||
method: request.method,
|
||||
headers,
|
||||
body: hasBody ? await request.arrayBuffer() : undefined,
|
||||
|
||||
@ -3,8 +3,10 @@ import type { NextRequest } from "next/server";
|
||||
const BACKEND_BASE_URL =
|
||||
process.env.NEXT_PUBLIC_BACKEND_BASE_URL ?? "http://127.0.0.1:8001";
|
||||
|
||||
function buildBackendUrl(pathname: string) {
|
||||
return new URL(pathname, BACKEND_BASE_URL);
|
||||
function buildBackendUrl(request: NextRequest, pathname: string) {
|
||||
const url = new URL(pathname, BACKEND_BASE_URL);
|
||||
url.search = request.nextUrl.search;
|
||||
return url;
|
||||
}
|
||||
|
||||
async function proxyRequest(request: NextRequest, pathname: string) {
|
||||
@ -14,7 +16,7 @@ async function proxyRequest(request: NextRequest, pathname: string) {
|
||||
headers.delete("content-length");
|
||||
|
||||
const hasBody = !["GET", "HEAD"].includes(request.method);
|
||||
const response = await fetch(buildBackendUrl(pathname), {
|
||||
const response = await fetch(buildBackendUrl(request, pathname), {
|
||||
method: request.method,
|
||||
headers,
|
||||
body: hasBody ? await request.arrayBuffer() : undefined,
|
||||
|
||||
58
frontend/tests/unit/app/api/memory-route.test.ts
Normal file
58
frontend/tests/unit/app/api/memory-route.test.ts
Normal file
@ -0,0 +1,58 @@
|
||||
import { afterEach, describe, expect, it, rs } from "@rstest/core";
|
||||
import { NextRequest } from "next/server";
|
||||
|
||||
import { POST as postMemoryPath } from "@/app/api/memory/[...path]/route";
|
||||
import { DELETE as deleteMemory } from "@/app/api/memory/route";
|
||||
|
||||
describe("memory API proxy", () => {
|
||||
afterEach(() => {
|
||||
rs.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("preserves the selected agent on root destructive requests", async () => {
|
||||
const fetchMock = rs
|
||||
.spyOn(globalThis, "fetch")
|
||||
.mockResolvedValue(new Response("{}", { status: 200 }));
|
||||
const request = new NextRequest(
|
||||
"http://deer-flow.test/api/memory?agent_name=Research-Agent",
|
||||
{ method: "DELETE" },
|
||||
);
|
||||
|
||||
await deleteMemory(request);
|
||||
|
||||
const [target] = fetchMock.mock.calls[0]!;
|
||||
expect(target).toBeInstanceOf(URL);
|
||||
if (!(target instanceof URL)) {
|
||||
throw new Error("Memory proxy target must be a URL");
|
||||
}
|
||||
expect(target.href).toBe(
|
||||
"http://127.0.0.1:8001/api/memory?agent_name=Research-Agent",
|
||||
);
|
||||
});
|
||||
|
||||
it("preserves the full query string on nested memory requests", async () => {
|
||||
const fetchMock = rs
|
||||
.spyOn(globalThis, "fetch")
|
||||
.mockResolvedValue(new Response("{}", { status: 200 }));
|
||||
const request = new NextRequest(
|
||||
"http://deer-flow.test/api/memory/import?agent_name=Research-Agent&mode=facts%2Bmetadata",
|
||||
{
|
||||
method: "POST",
|
||||
body: JSON.stringify({ facts: [] }),
|
||||
},
|
||||
);
|
||||
|
||||
await postMemoryPath(request, {
|
||||
params: Promise.resolve({ path: ["import"] }),
|
||||
});
|
||||
|
||||
const [target] = fetchMock.mock.calls[0]!;
|
||||
expect(target).toBeInstanceOf(URL);
|
||||
if (!(target instanceof URL)) {
|
||||
throw new Error("Memory proxy target must be a URL");
|
||||
}
|
||||
expect(target.href).toBe(
|
||||
"http://127.0.0.1:8001/api/memory/import?agent_name=Research-Agent&mode=facts%2Bmetadata",
|
||||
);
|
||||
});
|
||||
});
|
||||
Loading…
x
Reference in New Issue
Block a user