fix(search): agent-agentic 500 + regenerate stale e2e seed/tests

src/everos/memory/search/agentic_agent.py

@@ -20,13 +20,15 @@
 
 from __future__ import annotations
 
+import datetime as _dt
 from collections.abc import Awaitable, Callable
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from everalgo.rank.agentic import aagentic_retrieve
 from everalgo.rank.hybrid import ahybrid_retrieve
 from everalgo.types import Candidate
 
+from everos.component.utils.datetime import from_timestamp, to_timestamp_ms
 from everos.core.observability.tracing import memory_span
 from everos.memory.search.callbacks import build_rerank_fn
 from everos.memory.search.shaper import (
@@ -56,6 +58,48 @@
 _REFINEMENT_STRATEGY: str = "multi_query"
 
 
+def _to_everalgo_doc_metadata(
+    metadata: dict[str, Any], *, text_field: str
+) -> dict[str, Any]:
+    """Bridge agent recall metadata to the everalgo ``_format_docs`` contract.
+
+    ``aagentic_retrieve`` renders round-1 candidates into the sufficiency /
+    multi-query LLM prompt via ``everalgo.rank.agentic._format_docs``, which
+    reads ``metadata["episode"]`` as a dict with ``subject`` + ``content`` and
+    a ms-epoch ``metadata["timestamp"]``. Agent-kind rows carry their body in
+    ``text_field`` (``task_intent`` / ``skill``) and the time in ``timestamp``
+    (datetime); without this bridge ``_format_docs`` raises ``TypeError``.
+    Mirrors the episode path's bridge in ``agentic.py``.
+    ``_restore_shaper_metadata`` reverts it before DTO shaping.
+    """
+    bridged = dict(metadata)
+    content = metadata.get(text_field)
+    if isinstance(content, str):
+        bridged["episode"] = {
+            "subject": metadata.get("subject", ""),
+            "content": content,
+        }
+    timestamp = metadata.get("timestamp")
+    if isinstance(timestamp, _dt.datetime):
+        bridged["timestamp"] = to_timestamp_ms(timestamp)
+    return bridged
+
+
+def _restore_shaper_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
+    """Revert ``_to_everalgo_doc_metadata`` before agent-DTO shaping.
+
+    The shaper reads ``timestamp`` as a ``datetime``; the bridged ms-epoch
+    int must be reverted. The injected ``episode`` dict is inert for the
+    agent shapers (they read case/skill fields), so it is simply dropped.
+    """
+    reverted = dict(metadata)
+    timestamp = metadata.get("timestamp")
+    if isinstance(timestamp, (int, float)):
+        reverted["timestamp"] = from_timestamp(timestamp)
+    reverted.pop("episode", None)
+    return reverted
+
+
 async def search_agent_cases_agentic(
     query: str,
     *,
@@ -171,7 +215,7 @@ async def hybrid_full(q: str, k: int) -> list[Candidate]:
             observation_type="retriever",
             metadata={"phase": "agentic_hybrid"},
         ):
-            return await ahybrid_retrieve(
+            hits = await ahybrid_retrieve(
                 q,
                 dense_retrieve=_dense,
                 sparse_retrieve=_sparse,
@@ -180,6 +224,19 @@ async def hybrid_full(q: str, k: int) -> list[Candidate]:
                 sparse_candidates=_SPARSE_CANDIDATES,
                 rrf_k=_HYBRID_RRF_K,
             )
+        # Bridge to the everalgo doc contract so ``_format_docs`` (the LLM
+        # sufficiency / multi-query prompt) sees an episode dict + ms
+        # timestamp; agent-kind rows otherwise lack it and _format_docs raises.
+        return [
+            c.model_copy(
+                update={
+                    "metadata": _to_everalgo_doc_metadata(
+                        c.metadata, text_field=recaller.text_field
+                    )
+                }
+            )
+            for c in hits
+        ]
 
     rerank_fn = build_rerank_fn(reranker, text_field=recaller.text_field)
 
@@ -197,4 +254,9 @@ async def hybrid_full(q: str, k: int) -> list[Candidate]:
         multi_query_count=_MULTI_QUERY_COUNT,
         rrf_k=_HYBRID_RRF_K,
     )
-    return candidates
+    # Revert the doc-contract bridge so the agent DTO shapers see the
+    # original metadata shape (timestamp as datetime, no ``episode`` dict).
+    return [
+        c.model_copy(update={"metadata": _restore_shaper_metadata(c.metadata)})
+        for c in candidates
+    ]

tests/e2e/conftest.py

@@ -136,6 +136,17 @@ async def core_pipeline_runtime(
     monkeypatch.setattr(client_mod, "_llm_client", None, raising=False)
     _reset_strategy_singletons(monkeypatch)
 
+    # The full-app lifespan starts OME, whose config reloader requires an
+    # ``ome.toml`` in the memory root (normally created by ``everos init``).
+    # Provision the packaged default so the pipeline e2e can boot; OME
+    # strategies are code-registered (see api/lifespans/ome.py), so the
+    # default config is sufficient for extraction.
+    import shutil
+    from importlib.resources import files
+
+    default_ome = files("everos.config") / "default_ome.toml"
+    shutil.copyfile(str(default_ome), str(tmp_path / "ome.toml"))
+
     yield tmp_path
 
 

tests/e2e/test_search_endpoint_e2e.py

@@ -155,35 +155,38 @@ async def _seed_user_profiles(rows: list[dict[str, Any]]) -> list[UserProfile]:
 
 
 async def _seed_user_memory_cluster(eps: list[dict], *, owner_id: str) -> None:
-    """Seed one ``user_memory`` cluster covering every memcell in ``eps``.
+    """Seed one ``user_memory`` cluster covering every episode in ``eps``.
 
     The AGENTIC episode path goes through ``acluster_retrieve`` (see
     ``memory/search/agentic.py``), which narrows hybrid candidates to the
-    union of cluster member memcell ids. Tests that exercise the AGENTIC
+    union of cluster member ids. Production user_memory clusters store
+    episode ``entry_id`` members (``member_type="episode"``; see
+    ``strategies/trigger_profile_clustering.py``), matching the entry_id
+    keying of ``fetch_all_for_owner``. Tests that exercise the AGENTIC
     method therefore need at least one cluster whose members cover the
-    seeded episodes' ``parent_id``s — otherwise ``cluster_scoped`` yields
+    seeded episodes' ``entry_id``s — otherwise ``cluster_scoped`` yields
     nothing and the agentic pipeline returns ``[]``.
 
     Centroid is embedded from one of the episode bodies via the live
     embedder; with a single cluster the cosine ranking against the query
     is trivial (only one candidate), so any reasonable anchor works.
     """
-    memcell_ids = list({ep["parent_id"] for ep in eps})
+    entry_ids = list({ep["entry_id"] for ep in eps})
     centroid_text = eps[0]["episode"]
     centroid_vec = await get_embedder().embed(centroid_text)
     await cluster_repo.upsert_with_members(
         AlgoCluster(
             id=mint_cluster_id(),
             centroid=np.asarray(centroid_vec, dtype=np.float32),
-            count=len(memcell_ids),
+            count=len(entry_ids),
             last_ts=int(time.time() * 1000),
             preview=[ep["episode"][:80] for ep in eps[:3]],
-            members=memcell_ids,
+            members=entry_ids,
         ),
         owner_id=owner_id,
         owner_type="user",
         kind="user_memory",
-        member_type="memcell",
+        member_type="episode",
     )
 
 
@@ -1073,27 +1076,28 @@ async def test_search_hybrid_hierarchical_eviction_with_memcell_facts(
     """
     eps = _eps_for_owner(search_seed, "caroline")
     await _seed_episodes(eps)
-    ep_parent_ids = {r["parent_id"] for r in eps}
+    ep_entry_ids = {r["entry_id"] for r in eps}
     matching_facts = [
-        r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_parent_ids
+        r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_entry_ids
     ]
-    assert matching_facts, "seed should have at least one fact sharing a memcell"
+    assert matching_facts, "seed should have at least one fact bridging an episode"
     await _seed_atomic_facts(matching_facts)
 
     resp = await _post(client, query="counseling", method="hybrid", top_k=5)
     assert resp.status_code == 200
     data = resp.json()["data"]
     assert data["episodes"], "hybrid should return at least one episode"
-    # Whichever facts *do* get embedded must share parent_id with their
-    # host episode (the memcell-bridge invariant).
+    # Whichever facts *do* get embedded must bridge to their host episode
+    # via ``parent_id == episode.entry_id`` (the current fact-linkage
+    # invariant; see extract_atomic_facts).
     for ep in data["episodes"]:
         if not ep["atomic_facts"]:
             continue
-        host_parent = next((e["parent_id"] for e in eps if e["id"] == ep["id"]), None)
+        host_entry = next((e["entry_id"] for e in eps if e["id"] == ep["id"]), None)
         for fact in ep["atomic_facts"]:
             seed_fact = next((r for r in matching_facts if r["id"] == fact["id"]), None)
             if seed_fact is not None:
-                assert seed_fact["parent_id"] == host_parent
+                assert seed_fact["parent_id"] == host_entry
 
 
 @pytest.mark.slow
@@ -1125,11 +1129,11 @@ class _AlphaZeroConfig(mgr.RankConfig):
 
     eps = _eps_for_owner(search_seed, "caroline")
     await _seed_episodes(eps)
-    ep_parent_ids = {r["parent_id"] for r in eps}
+    ep_entry_ids = {r["entry_id"] for r in eps}
     matching_facts = [
-        r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_parent_ids
+        r for r in search_seed["atomic_fact"] if r["parent_id"] in ep_entry_ids
     ]
-    assert matching_facts, "seed should have at least one fact sharing a memcell"
+    assert matching_facts, "seed should have at least one fact bridging an episode"
     await _seed_atomic_facts(matching_facts)
 
     resp = await _post(client, query="counseling", method="hybrid", top_k=10)
@@ -1142,8 +1146,8 @@ class _AlphaZeroConfig(mgr.RankConfig):
         "alpha=0 should let hierarchical eviction promote >=1 fact"
     )
 
-    # Memcell-bridge invariant — every attached fact's parent_id must
-    # match its host episode's parent_id.
+    # Fact-linkage invariant — every attached fact's parent_id must match
+    # its host episode's entry_id (current fact→episode bridge).
     eps_by_id = {e["id"]: e for e in eps}
     for ep in data["episodes"]:
         host = eps_by_id.get(ep["id"])
@@ -1152,7 +1156,7 @@ class _AlphaZeroConfig(mgr.RankConfig):
         for fact in ep["atomic_facts"]:
             seed_fact = next((r for r in matching_facts if r["id"] == fact["id"]), None)
             if seed_fact is not None:
-                assert seed_fact["parent_id"] == host["parent_id"]
+                assert seed_fact["parent_id"] == host["entry_id"]
 
 
 # ═══════════════════════════════════════════════════════════════════════
@@ -1287,7 +1291,7 @@ async def test_search_filter_error_returns_422(
     # FastAPI's default ``{"detail": ...}``). The FilterError text
     # lands in ``error.message``.
     body = resp.json()
-    assert body["error"]["code"] == "HTTP_ERROR"
+    assert body["error"]["code"] == "INVALID_INPUT"
     assert "this_field_does_not_exist" in body["error"]["message"]
 
 
@@ -1315,7 +1319,7 @@ async def test_vector_search_with_session_filter(
     base = _eps_for_owner(search_seed, "caroline")
     facts = _facts_for_owner(search_seed, "caroline")
     half = len(base) // 2
-    target_parent_ids = {r["parent_id"] for r in base[:half]}
+    target_parent_ids = {r["entry_id"] for r in base[:half]}
 
     await _seed_episodes(
         [{**r, "session_id": "sess_target"} for r in base[:half]]
@@ -1379,7 +1383,7 @@ async def test_agentic_search_with_timestamp_filter(
         [
             {
                 **f,
-                "parent_id": eps_post[i % len(eps_post)]["parent_id"],
+                "parent_id": eps_post[i % len(eps_post)]["entry_id"],
                 "timestamp": eps_post[i % len(eps_post)]["timestamp"],
             }
             for i, f in enumerate(facts)

tests/fixtures/_dump_search_seed.py

@@ -8,13 +8,14 @@
 
 Sampling rules:
 
-- **episode**: first 8 rows per owner (caroline + melanie). Captures
-  the parent_id (= memcell_id) set so downstream tables can be
-  bridge-consistent.
-- **atomic_fact**: every row whose ``parent_id`` is in the episode-
-  parent set above, capped at 50 to keep the seed compact. This
-  guarantees hierarchical-eviction testing can verify "facts sharing a
-  memcell with the matched episode get embedded".
+- **episode + atomic_fact**: sampled together for fact↔episode
+  coherence. Facts link to their host episode via
+  ``atomic_fact.parent_id == episode.entry_id`` (parent_type="episode";
+  see ``memory/strategies/extract_atomic_facts.py``). Episodes that host
+  facts are picked first (up to 8/owner) so hierarchical-eviction and the
+  fact-first paths (vector MaxSim, agentic) have a non-trivial,
+  multi-episode corpus; facts are then kept iff their host episode made
+  the cut (≤15/owner), guaranteeing every kept fact bridges back.
 - **foresight**: 5 per owner. Archived for future use; current
   ``/search`` does not query foresight, so the seed only exists so
   downstream tests can opt in without re-cutting the corpus.
@@ -74,35 +75,43 @@ def main() -> None:
     OUT_DIR.mkdir(parents=True, exist_ok=True)
     db = lancedb.connect(str(CORPUS))
 
-    # 1) episodes — first 8 per owner.
+    # 1) episodes + 2) atomic_facts — sampled together so the slice is
+    #    fact↔episode coherent AND rich (facts spread across several
+    #    episodes, not piled on one).
+    #
+    # Current extraction links a fact to its host episode via
+    # ``atomic_fact.parent_id == episode.entry_id`` (parent_type="episode";
+    # see ``memory/strategies/extract_atomic_facts.py``). The fact-first
+    # search paths (vector MaxSim, agentic) fetch episodes by that
+    # entry_id, so the seed must preserve that linkage. Sample the
+    # episodes that actually HAVE facts first (up to 8/owner), so the
+    # agentic LLM-sufficiency step and hierarchical eviction have a
+    # non-trivial, multi-episode corpus to work with; only then fall back
+    # to fact-less episodes to top up owner coverage.
     eps_all = _read(db, "episode")
-    eps: list[dict[str, Any]] = []
-    parent_memcells: set[str] = set()
-    for owner in ALL_OWNERS:
-        owned = [r for r in eps_all if r["owner_id"] == owner][:8]
-        eps.extend(owned)
-        for r in owned:
-            parent_memcells.add(r["parent_id"])
-
-    # 2) atomic_facts — every fact whose parent_id is in the episode
-    #    parent set, capped to keep the seed compact (and so hierarchical
-    #    ``facts_for_episodes`` has a useful but bounded pool to
-    #    bucket back into episodes).
     afs_all = _read(db, "atomic_fact")
-    # Atomic facts fan out per-owner (a single fact about a memcell that
-    # mentions two users gets two rows, one for each owner) — sampling
-    # naively can leave one owner with zero facts. Take per-owner caps
-    # so both caroline and melanie have facts whose parent_id matches
-    # their own episodes' parent_id (memcell bridge).
+
+    eps: list[dict[str, Any]] = []
     afs: list[dict[str, Any]] = []
     for owner in ALL_OWNERS:
-        afs.extend(
-            [
-                r
-                for r in afs_all
-                if r["owner_id"] == owner and r["parent_id"] in parent_memcells
-            ][:10]
-        )
+        owner_eps = [r for r in eps_all if r["owner_id"] == owner]
+        owner_facts = [r for r in afs_all if r["owner_id"] == owner]
+        facts_by_ep: dict[str, list[dict[str, Any]]] = {}
+        for f in owner_facts:
+            facts_by_ep.setdefault(f["parent_id"], []).append(f)
+        # Episodes that host facts come first (richest bridges), then the
+        # rest — capped at 8 to keep the seed compact.
+        with_facts = [e for e in owner_eps if e["entry_id"] in facts_by_ep]
+        without_facts = [e for e in owner_eps if e["entry_id"] not in facts_by_ep]
+        chosen = (with_facts + without_facts)[:8]
+        eps.extend(chosen)
+        # Spread facts across episodes (<=3 per host) so several episodes are
+        # bridged, not one — richer corpus for agentic + hierarchical
+        # eviction. Every kept fact bridges back via its host entry_id.
+        owner_afs: list[dict[str, Any]] = []
+        for e in chosen:
+            owner_afs.extend(facts_by_ep.get(e["entry_id"], [])[:3])
+        afs.extend(owner_afs[:15])
 
     # 3) foresights — 5 per owner, archived for future use.
     fss_all = _read(db, "foresight")
@@ -128,7 +137,8 @@ def main() -> None:
 
     for name, count, size in written:
         print(f"  {name:14s}: {count:3d} rows  ({size // 1024} KB)")
-    print(f"  parent_memcells captured: {len(parent_memcells)}")
+    bridged = len({f["parent_id"] for f in afs})
+    print(f"  fact-bridged episodes: {bridged}")
 
 
 if __name__ == "__main__":