Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion packages/core/src/agent_memory/core/recall.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import dataclasses
import datetime as dt
import pathlib
import sqlite3

from . import timestamp
Expand Down Expand Up @@ -113,7 +114,9 @@ def _eligible(
return eligible

def _in_scope(self, path: str, scope: str) -> bool:
return path.startswith(scope.strip("/"))
path_parts = pathlib.PurePath(path).parts
scope_parts = pathlib.PurePath(scope.strip("/\\")).parts
return bool(scope_parts) and path_parts[: len(scope_parts)] == scope_parts

def _current_at(self, row: sqlite3.Row, moment: dt.datetime) -> bool:
if timestamp.parse(str(row["valid_from"])) > moment:
Expand Down
24 changes: 24 additions & 0 deletions tests/unit/test_recall.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
from agent_memory.core.access_log import AccessLog
from agent_memory.core.database import Database
from agent_memory.core.recall import Recall
from agent_memory.core.schema import MemorySchema, render
from agent_memory.core.store import LEVEL_ABSTRACT, LEVEL_OUTLINE


Expand Down Expand Up @@ -67,6 +68,29 @@ def test_scope_restricts_by_path_prefix(seeded):
assert {hit.type for hit in hits} == {"experience"}


def test_scope_matches_a_complete_path_component(store):
extra = MemorySchema(
type="experience-archive",
description="Archived experiences used to test a colliding scope prefix.",
key=("subject",),
)
store.schemas.path_for(extra.type).write_text(render(extra), encoding="utf-8")
store.record(
abstract="Shared scope marker in the active experience domain",
type="experience",
name="active-scope-marker",
)
store.record(
abstract="Shared scope marker in the archive domain",
type=extra.type,
fields={"subject": "archive-scope-marker"},
)

hits = Recall(store).recall("shared scope marker", scope="experience")

assert _names(hits) == ["active-scope-marker"]


def test_l0_entries_carry_the_full_contract(seeded):
hit = Recall(seeded).recall("E4021")[0]
payload = hit.as_dict()
Expand Down