fix(memory): compose visibility + key filters via explicit AND

Greptile P1 (filter-fragility): `where.update(vis)` was semantically
correct today, but dict-merging by key meant any future visibility
filter that grew a new top-level "OR" would silently clobber the
existing key filter.

Compose explicitly instead:

    where = {"AND": [key_filter, vis]}

Applied to both `list_memory` and `_find_memory_for_caller`. When
either side is empty (admin has no visibility filter; list has no
key filter), skip the wrapper and use the non-empty side directly
to keep the generated SQL clean.

Test fake's `_matches` now understands top-level `AND` too.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Krrish Dholakia 2026-04-24 09:08:53 -07:00
parent 6402711452
commit c5fafe56bb
2 changed files with 23 additions and 12 deletions

View file

@ -223,17 +223,25 @@ async def list_memory(
"""List memory entries visible to the caller."""
prisma_client = _require_prisma()
where: dict = {}
vis = _visibility_filter(user_api_key_dict)
if vis is not None:
where.update(vis)
# key_prefix takes precedence over exact `key` if both are passed.
# Both sit under the visibility filter (ANDed), so prefix scans can
# never leak across user/team scopes.
# Build the key filter first (prefix wins if both `key` and `key_prefix`
# are passed). Then AND it with the visibility filter via an explicit
# top-level "AND" — safer than `dict.update` since future visibility
# filters could grow an "OR" key that would clobber this one if merged
# by key.
key_filter: dict = {}
if key_prefix is not None:
where["key"] = {"startsWith": key_prefix}
key_filter["key"] = {"startsWith": key_prefix}
elif key is not None:
where["key"] = key
key_filter["key"] = key
vis = _visibility_filter(user_api_key_dict)
where: dict
if vis is None:
where = key_filter
elif not key_filter:
where = vis
else:
where = {"AND": [key_filter, vis]}
try:
total = await prisma_client.db.litellm_memorytable.count(where=where)
@ -254,10 +262,9 @@ async def _find_memory_for_caller(
prisma_client: Any, key: str, user_api_key_dict: UserAPIKeyAuth
) -> Any:
"""Look up a memory row by key, scoped to the caller's visibility."""
where: dict = {"key": key}
key_filter = {"key": key}
vis = _visibility_filter(user_api_key_dict)
if vis is not None:
where.update(vis)
where = key_filter if vis is None else {"AND": [key_filter, vis]}
rows = await prisma_client.db.litellm_memorytable.find_many(
where=where, take=1, order={"updated_at": "desc"}
)

View file

@ -55,6 +55,10 @@ class _InMemoryMemoryTable:
def _matches(self, row: MagicMock, where: Dict[str, Any]) -> bool:
for k, v in where.items():
if k == "AND":
if not all(self._matches(row, clause) for clause in v):
return False
continue
if k == "OR":
if not any(self._matches(row, clause) for clause in v):
return False