From 9fef3a1caab97169be081df6d938ba15dad6c496 Mon Sep 17 00:00:00 2001 From: Alexander Matviychuk Date: Wed, 2 Sep 2026 22:37:46 +0900 Subject: [PATCH] fix(cascade): scope LanceDB row ids by md path to stop cross-project overwrites MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every daily-log handler built row ids as f"{owner_id}_{entry_id}" and user_profile as bare owner_id. Entry numbering is per-file (__NNNNNNNN), so any two projects with a same-date file produce identical row ids and merge_insert("id") replaces the previous project's rows wholesale — only the last-processed file per date survives. Proven in production (1.2.3) by version-walk: one transition swapping 54 rows between two projects' same-date files at identical entry_ids; user_profile collapsed 16 project profiles to 2. Row id is now f"{md_path}#{entry_id}" for the four daily-log kinds and md_path for user_profile (path unique by construction, robust to future layout dimensions). Profile recall derives the same path from app/project/owner. Adds a parametrized coexistence regression test (fails on <= 1.2.3, passes with the fix). Migration note: existing deployments need a full cascade rebuild — sha-skip keeps old-id rows otherwise. Fixes #320 --- .../memory/cascade/handlers/agent_case.py | 2 +- .../memory/cascade/handlers/atomic_fact.py | 2 +- src/everos/memory/cascade/handlers/episode.py | 2 +- .../memory/cascade/handlers/foresight.py | 2 +- .../memory/cascade/handlers/user_profile.py | 2 +- src/everos/memory/search/manager.py | 2 +- src/everos/memory/search/recall/profile.py | 14 +- .../test_handler_row_id_project_scoping.py | 181 ++++++++++++++++++ 8 files changed, 199 insertions(+), 8 deletions(-) create mode 100644 tests/unit/test_memory/test_cascade/test_handler_row_id_project_scoping.py diff --git a/src/everos/memory/cascade/handlers/agent_case.py b/src/everos/memory/cascade/handlers/agent_case.py index a6fa9c5b7..70ca3c68c 100644 --- a/src/everos/memory/cascade/handlers/agent_case.py +++ b/src/everos/memory/cascade/handlers/agent_case.py @@ -77,7 +77,7 @@ async def _build_row( reason="embedding_capability_unavailable", ) return AgentCase( - id=f"{owner_id}_{entry.entry_id}", + id=f"{md_path}#{entry.entry_id}", entry_id=entry.entry_id, owner_id=owner_id, owner_type=owner_type, diff --git a/src/everos/memory/cascade/handlers/atomic_fact.py b/src/everos/memory/cascade/handlers/atomic_fact.py index 6dea35e4e..d72232f6e 100644 --- a/src/everos/memory/cascade/handlers/atomic_fact.py +++ b/src/everos/memory/cascade/handlers/atomic_fact.py @@ -62,7 +62,7 @@ async def _build_row( reason="embedding_capability_unavailable", ) return AtomicFact( - id=f"{owner_id}_{entry.entry_id}", + id=f"{md_path}#{entry.entry_id}", entry_id=entry.entry_id, owner_id=owner_id, owner_type=owner_type, diff --git a/src/everos/memory/cascade/handlers/episode.py b/src/everos/memory/cascade/handlers/episode.py index 84878d302..b284a9ed1 100644 --- a/src/everos/memory/cascade/handlers/episode.py +++ b/src/everos/memory/cascade/handlers/episode.py @@ -102,7 +102,7 @@ async def _build_row( tokens = self._deps.tokenizer.tokenize(tokenize_source) return Episode( - id=f"{owner_id}_{entry.entry_id}", + id=f"{md_path}#{entry.entry_id}", entry_id=entry.entry_id, owner_id=owner_id, owner_type=owner_type, diff --git a/src/everos/memory/cascade/handlers/foresight.py b/src/everos/memory/cascade/handlers/foresight.py index adbefe020..24fc6e4f8 100644 --- a/src/everos/memory/cascade/handlers/foresight.py +++ b/src/everos/memory/cascade/handlers/foresight.py @@ -86,7 +86,7 @@ async def _build_row( " ".join(self._deps.tokenizer.tokenize(evidence)) if evidence else None ) return Foresight( - id=f"{owner_id}_{entry.entry_id}", + id=f"{md_path}#{entry.entry_id}", entry_id=entry.entry_id, owner_id=owner_id, owner_type=owner_type, diff --git a/src/everos/memory/cascade/handlers/user_profile.py b/src/everos/memory/cascade/handlers/user_profile.py index 4842986d0..5da5a33fa 100644 --- a/src/everos/memory/cascade/handlers/user_profile.py +++ b/src/everos/memory/cascade/handlers/user_profile.py @@ -80,7 +80,7 @@ async def handle_added_or_modified(self, md_path: str) -> HandlerOutcome: } ) - row_id = owner_id + row_id = md_path prior = await user_profile_repo.get_by_id(row_id) if prior is not None and prior.content_sha256 == digest: return HandlerOutcome( diff --git a/src/everos/memory/search/manager.py b/src/everos/memory/search/manager.py index 1ae5bfd91..1b8dc8a5d 100644 --- a/src/everos/memory/search/manager.py +++ b/src/everos/memory/search/manager.py @@ -576,7 +576,7 @@ async def _search_agent_skills( async def _fetch_profile(self, req: SearchRequest) -> list[SearchProfileItem]: if not req.include_profile or req.owner_type != "user": return [] - return await self._profile.fetch(req.owner_id) + return await self._profile.fetch(req.owner_id, req.app_id, req.project_id) # ── Recall helpers ────────────────────────────────────────────── diff --git a/src/everos/memory/search/recall/profile.py b/src/everos/memory/search/recall/profile.py index be149b927..2a929f86a 100644 --- a/src/everos/memory/search/recall/profile.py +++ b/src/everos/memory/search/recall/profile.py @@ -29,16 +29,26 @@ class ProfileRecaller: """Fetch the owner's profile row from LanceDB, return at most one item.""" - async def fetch(self, owner_id: str) -> list[SearchProfileItem]: + async def fetch( + self, + owner_id: str, + app_id: str = "default", + project_id: str = "default", + ) -> list[SearchProfileItem]: """Return ``[item]`` if a profile row exists, otherwise ``[]``. Empty list (rather than 404) lets the caller emit a normal response with ``profiles=[]`` while the user is still in their cold-start window (no profile synthesised yet). + + The row id is the profile's md path (one profile per + ``//users//user.md`` since 2026-09-02 — + owner-only ids collapsed every project onto one row). """ if not owner_id: return [] - row = await user_profile_repo.get_by_id(owner_id) + row_id = f"{app_id}/{project_id}/users/{owner_id}/user.md" + row = await user_profile_repo.get_by_id(row_id) if row is None: logger.debug("profile_fetch_miss", owner_id=owner_id) return [] diff --git a/tests/unit/test_memory/test_cascade/test_handler_row_id_project_scoping.py b/tests/unit/test_memory/test_cascade/test_handler_row_id_project_scoping.py new file mode 100644 index 000000000..c99bd7961 --- /dev/null +++ b/tests/unit/test_memory/test_cascade/test_handler_row_id_project_scoping.py @@ -0,0 +1,181 @@ +"""Regression test for EverOS #320 — cross-project LanceDB row-id collision. + +Every daily-log cascade kind allocates ``entry_id`` per file +(``__NNNNNNNN``), so two projects with a same-date file +produce identical entry ids. Row ids were built as +``f"{owner_id}_{entry_id}"`` — identical across projects — and +``merge_insert("id")`` treated the other project's rows as matches and +replaced them: only the last-processed file per date survived. + +The fix derives the row id from the file's relative path plus entry_id +(``f"{md_path}#{entry_id}"``) — unique by construction. This test +writes the same-date entry for two different projects through each +daily-log writer + handler pair and asserts both projects' rows +coexist under distinct ids. It fails on every version <= 1.2.3. +""" + +from __future__ import annotations + +import datetime as _dt +from pathlib import Path + +import pytest + +from everos.component.embedding import EmbeddingCapability, EmbeddingProvider +from everos.component.tokenizer import Tokenizer +from everos.core.persistence import MemoryRoot +from everos.infra.persistence.markdown import ( + AgentCaseWriter, + AtomicFactWriter, + EpisodeWriter, + ForesightWriter, +) +from everos.memory.cascade.handlers import HandlerDeps +from everos.memory.cascade.handlers.agent_case import AgentCaseHandler +from everos.memory.cascade.handlers.atomic_fact import AtomicFactHandler +from everos.memory.cascade.handlers.episode import EpisodeHandler +from everos.memory.cascade.handlers.foresight import ForesightHandler + +_DATE = _dt.date(2026, 8, 27) +_OWNER = "alex" + + +class _StubTokenizer(Tokenizer): + def tokenize(self, text: str) -> list[str]: + return [tok for tok in text.split() if tok] + + def tokenize_batch(self, texts): # type: ignore[no-untyped-def] + return [self.tokenize(t) for t in texts] + + +class _StubEmbedder(EmbeddingProvider): + dim = 1024 + + async def embed(self, text: str) -> list[float]: + return [0.1] * self.dim + + async def embed_batch(self, texts): # type: ignore[no-untyped-def] + return [await self.embed(t) for t in texts] + + +class _FakeRepo: + """In-memory repo with the merge-by-id semantics of merge_insert.""" + + def __init__(self) -> None: + self.rows: dict[str, object] = {} + + async def find_where(self, where: str, *, limit: int = 100) -> list: + prefix = "md_path = '" + if where.startswith(prefix): + md_path = where[len(prefix) :].rstrip("'") + return [r for r in self.rows.values() if r.md_path == md_path] + return [] + + async def upsert(self, rows: list) -> None: + for r in rows: + self.rows[r.id] = r # merge_insert: same id => replace + + async def delete_by_md_path(self, md_path: str) -> int: + before = len(self.rows) + self.rows = { + k: r for k, r in self.rows.items() if r.md_path != md_path + } + return before - len(self.rows) + + +@pytest.fixture +def memory_root(tmp_path: Path) -> MemoryRoot: + mr = MemoryRoot(tmp_path) + mr.ensure() + return mr + + +@pytest.fixture(autouse=True) +def stub_embedder(monkeypatch: pytest.MonkeyPatch) -> None: + import everos.component.embedding.accessor as acc + + monkeypatch.setattr( + acc, "_capability", EmbeddingCapability(provider=_StubEmbedder()) + ) + + +def _inline() -> dict: + return { + "owner_id": _OWNER, + "session_id": "s1", + "timestamp": "2026-08-27T10:00:00+00:00", + "parent_type": "memcell", + "parent_id": "mc_parent", + "sender_ids": [_OWNER], + } + + +# kind -> (writer cls, handler cls, sections, relative md path template) +_KINDS = { + "episode": ( + EpisodeWriter, + EpisodeHandler, + {"Subject": "S", "Summary": "Stub", "Content": "body"}, + "dsh/{project}/users/alex/episodes/episode-2026-08-27.md", + ), + "atomic_fact": ( + AtomicFactWriter, + AtomicFactHandler, + {"Fact": "a fact"}, + "dsh/{project}/users/alex/.atomic_facts/atomic_fact-2026-08-27.md", + ), + "foresight": ( + ForesightWriter, + ForesightHandler, + {"Foresight": "a foresight"}, + "dsh/{project}/users/alex/.foresights/foresight-2026-08-27.md", + ), + "agent_case": ( + AgentCaseWriter, + AgentCaseHandler, + {"TaskIntent": "do a thing", "Approach": "carefully"}, + "dsh/{project}/agents/dsh/.cases/agent_case-2026-08-27.md", + ), +} + + +@pytest.mark.parametrize("kind", sorted(_KINDS)) +async def test_same_date_entries_coexist_across_projects( + kind: str, memory_root: MemoryRoot, monkeypatch: pytest.MonkeyPatch +) -> None: + writer_cls, handler_cls, sections, path_tpl = _KINDS[kind] + repo = _FakeRepo() + monkeypatch.setattr(handler_cls, "lance_repo", repo) + + writer = writer_cls(memory_root) + handler = handler_cls( + HandlerDeps(memory_root=memory_root, tokenizer=_StubTokenizer()) + ) + scope_id = _OWNER if kind != "agent_case" else "dsh" + + paths: list[str] = [] + for project in ("proj-a", "proj-b"): + await writer.append_entry( + scope_id, + inline=_inline() if kind != "agent_case" else {**_inline(), "agent_id": "dsh", "quality_score": "0.9"}, + sections=sections, + date=_DATE, + app_id="dsh", + project_id=project, + ) + md_path = path_tpl.format(project=project) + paths.append(md_path) + await handler.handle_added_or_modified(md_path) + + # Both projects' rows must coexist (pre-fix: project B's upsert + # matched project A's row id and replaced it — 1 row, proj-b's). + rows_a = await repo.find_where(f"md_path = '{paths[0]}'") + rows_b = await repo.find_where(f"md_path = '{paths[1]}'") + assert len(rows_a) == 1, ( + f"{kind}: project A's row was replaced by project B's " + f"(ids collide across projects — EverOS #320)" + ) + assert len(rows_b) == 1 + assert rows_a[0].id != rows_b[0].id + assert rows_a[0].project_id == "proj-a" + assert rows_b[0].project_id == "proj-b"