feat(sdk): rerank switch, size-gated MMR, scene="all" search and list, productivity edit/delete

search:
- rerank=True (default) keeps today's behavior; rerank=False skips the
  local cross-encoder, graph rerank and the coding facts' BM25 rerank for
  that call (carried on ReadRequest, so concurrent calls are independent).
- MMR now runs only when the caller's limit is below MMR_MAX_TOP_K (20).
  It is O(k^2 * pool) Jaccard and made large result pages slow.
- scene="all" searches chat, working and coding concurrently with one
  shared embedding and fuses them by Reciprocal Rank Fusion (k=60) over
  each lane's own rank, so no lane is buried by its score scale. Items
  carry scene, kind, memory_id, content, fused score, lane score/rank,
  source and times. A failed or disabled lane is reported in "errors".

list_memories:
- scene="coding"|"working" honour include_sources/exclude_sources with
  the filtered total.
- scene="all" lists the three scenes newest first with limit/offset over
  the merged order, per-scene totals and the same unified item fields.

update/delete:
- kind/variant disambiguate like get(); an experience update replaces
  its solution (task= also replaces the task and its search vectors), a
  fact update replaces the statement; delete removes experiences and
  facts with their vectors. Chat calls are unchanged.
- The HTTP server maps an ambiguous id to 409 and a bad namespace to 400
  instead of 500, passes kind/variant through, and strips the query
  string from PUT/DELETE paths, which leaked it into the memory id.

memory_stats now reads productivity times (written as naive UTC) as UTC.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -4,6 +4,33 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- SDK: `search(..., rerank=False)` skips every reranking stage for that one
+  call (the local cross-encoder, the memory-graph rerank, and the BM25 rerank of
+  working/coding facts), so results come back in retrieval-score order. The
+  flag travels on the read request, so concurrent searches do not affect each
+  other. The default (`True`) keeps current behavior; `/api/v1/search` accepts
+  it too.
+- SDK: MMR diversity selection now runs only when the requested `limit` is
+  below 20 (`MMR_MAX_TOP_K`) and is skipped for larger limits, where its
+  quadratic token-overlap pass is slow and gains little. Previously it ran for
+  every limit. `MEMORY_HYBRID_V2_MMR_LAMBDA` still sets the weight.
+- SDK: `search(scene="all")` searches chat, working and coding at the same time
+  and merges them with Reciprocal Rank Fusion (k=60) over each scene's own
+  ranking, so no scene is crowded out by the other scenes' score scales. Each
+  result says which scene and kind it came from. A scene that fails or is turned
+  off is reported in `errors` and does not fail the search. Source filters apply
+  to all three.
+- SDK: `list_memories(scene="all")` lists chat, working and coding memories
+  together, newest update first, with `limit`/`offset` paging over the merged
+  order and `total` summing all three. `scene="coding"` and `"working"` now
+  honor `include_sources` / `exclude_sources`, and `total` counts only the
+  filtered memories.
+- SDK: `update()` and `delete()` now work on working/coding experiences and
+  facts, not just chat memories. They take the same optional `kind` / `variant`
+  as `get()`, and an ID found in more than one place raises
+  `AmbiguousMemoryIdError` instead of guessing. Updating an experience replaces
+  its solution and keeps its task, unless a new `task=` is also passed; updating
+  a fact replaces its statement. Deleting removes the stored row and its vectors.
 - SDK: `list_memories(scene="chat" | "coding" | "working")` lists one kind of
   memory. `coding` and `working` list that scene's whole store, newest update
   first (`workspace_id` still narrows it), and `total` is now the real count

hy_memory/coding/store.py

@@ -217,6 +217,21 @@ def _in_clause(column: str, values: Optional[List[str]]) -> Tuple[str, List[Any]
     return f"{column} IN ({placeholders})", list(values)
 
 
+def _source_where(
+    include_sources: Optional[List[str]], exclude_sources: Optional[List[str]],
+) -> Tuple[str, List[Any]]:
+    """Caller-source filter as SQL (``1`` when unfiltered); exclusion wins."""
+    clauses: List[str] = []
+    params: List[Any] = []
+    if include_sources:
+        clauses.append(f"source IN ({','.join('?' for _ in include_sources)})")
+        params.extend(include_sources)
+    if exclude_sources:
+        clauses.append(f"source NOT IN ({','.join('?' for _ in exclude_sources)})")
+        params.extend(exclude_sources)
+    return (" AND ".join(clauses) or "1"), params
+
+
 def _scope_matches(
     row_workspace_id: Optional[str],
     row_branch: Optional[str],
@@ -603,6 +618,86 @@ class CodingMemoryStore:
         logger.info(f"[coding-store] deleted experience_id={experience_id}")
         return True
 
+    async def edit_experience(
+        self,
+        experience_id: str,
+        *,
+        user_id: Optional[str] = None,
+        task: Optional[str] = None,
+        solution: Optional[str] = None,
+        search_keys: Optional[List[str]] = None,
+    ) -> Optional[CodingMemory]:
+        """调用方手动改一条经验（client.update 用）。返回改后的经验；不存在返回 None。
+
+        与 update() 的区别：update 是写入链路的覆盖式重写，带写授权校验（同
+        agent/session/workspace 才能改）；这里是用户对已有条目的直接编辑，只动
+        给定的字段，其余原样保留，因此不做授权校验。
+
+        向量行只嵌 task + search_keys（solution 不进向量），所以只改 solution 时
+        不重嵌、不碰向量库；改了 task / search_keys 才删旧 key 行重写。
+        updated_at 总是刷新。
+        """
+        prev = await self.get_by_id(experience_id, user_id=user_id)
+        if prev is None:
+            return None
+        updated = deepcopy(prev)
+        if task is not None:
+            updated.task = task
+        if solution is not None:
+            updated.solution = solution
+        if search_keys is not None:
+            updated.search_keys = list(search_keys)
+        updated.updated_at = datetime.utcnow()
+        await _run_in_pool(self._upsert_sql_sync, updated)
+        if updated.task != prev.task or updated.search_keys != prev.search_keys:
+            await self._delete_vec_keys_by_count(experience_id, 1 + len(prev.search_keys))
+            await self._upsert_vec_keys(updated)
+        logger.info(f"[coding-store] edited experience_id={experience_id}")
+        return updated
+
+    async def edit_fact(
+        self,
+        fact_id: str,
+        *,
+        statement: str,
+        user_id: Optional[str] = None,
+    ) -> Optional[WorkingFact]:
+        """调用方手动改一条事实的 statement。返回改后的事实；不存在返回 None。
+
+        statement 是事实层的检索文本，所以 active 事实会重嵌并覆盖它那一行向量
+        （node_id 就是 fact_id，upsert 即覆盖）。superseded 事实的向量行早已撤掉，
+        这里不把它加回去，否则会让过期结论重新占召回名额。
+        """
+        prev = await self.get_fact(fact_id, user_id=user_id)
+        if prev is None:
+            return None
+        updated = deepcopy(prev)
+        updated.statement = statement
+        await _run_in_pool(self._upsert_fact_sql_sync, updated)
+        if updated.status == "active":
+            await self._upsert_fact_vec(updated)
+        logger.info(f"[coding-store] edited fact_id={fact_id}")
+        return updated
+
+    async def delete_fact(self, fact_id: str, user_id: Optional[str] = None) -> bool:
+        """物理删除一条事实（SQLite 行 + 事实向量行），返回是否真有删到。
+
+        别处留下的引用（经验的 source_fact_ids、后继事实的 fact_link）不清：
+        召回侧本来就按存在性过滤死链（get_many / get_facts 查不到自然掉队）。
+        """
+        prev = await self.get_fact(fact_id, user_id=user_id)
+        if prev is None:
+            logger.info(f"[coding-store] delete fact skipped (not found): {fact_id}")
+            return False
+        if self._fact_vec is not None:
+            try:
+                await self._fact_vec.delete(fact_id)
+            except Exception as e:
+                logger.debug(f"[coding-store] delete fact vec {fact_id} ignored: {e}")
+        await _run_in_pool(self._delete_fact_sql_sync, fact_id, prev.user_id)
+        logger.info(f"[coding-store] deleted fact_id={fact_id}")
+        return True
+
     # ------------------------------------------------------------
     # 读取
     # ------------------------------------------------------------
@@ -687,26 +782,35 @@ class CodingMemoryStore:
         branch: Optional[str] = None,
         limit: int = 100,
         offset: int = 0,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
     ) -> List["CodingMemory"]:
         """按 workspace_id（+可选 branch）纯过滤列出该 user 的完整 coding memory。
 
         作用域规则见 _scope_where（与 search 路径共用同一套）。无语义排序，按
-        updated_at 倒序（最近修改优先）。
+        updated_at 倒序（最近修改优先）。include_sources / exclude_sources 按
+        source 列在 SQL 里过滤（排除优先），所以 LIMIT/OFFSET 在过滤之后生效。
         """
         return await _run_in_pool(
-            self._list_by_workspace_sync, user_id, workspace_id, branch, limit, offset
+            self._list_by_workspace_sync, user_id, workspace_id, branch, limit, offset,
+            include_sources, exclude_sources,
         )
 
     def _list_by_workspace_sync(
         self, user_id: str, workspace_id: str, branch: Optional[str],
         limit: int, offset: int,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
     ) -> List["CodingMemory"]:
         scope_sql, scope_params = _scope_where(workspace_id, branch)
+        source_sql, source_params = _source_where(include_sources, exclude_sources)
         with self._lock:
             assert self._conn is not None
             sql = (f"SELECT * FROM coding_memory_meta WHERE user_id = ? AND {scope_sql} "
-                   f"ORDER BY updated_at DESC LIMIT ? OFFSET ?")
-            params: List[Any] = [user_id] + scope_params + [limit, offset]
+                   f"AND {source_sql} "
+                   f"ORDER BY updated_at DESC, experience_id ASC LIMIT ? OFFSET ?")
+            params: List[Any] = ([user_id] + scope_params + source_params
+                                 + [limit, offset])
             cur = self._conn.execute(sql, params)
             return [_row_to_memory(r) for r in cur.fetchall()]
 
@@ -715,29 +819,39 @@ class CodingMemoryStore:
         user_id: str,
         workspace_id: Optional[str] = None,
         branch: Optional[str] = None,
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
     ) -> Dict[str, Any]:
         """经验数、最新创建/修改时间、active 事实数。纯聚合，不读正文。
 
-        作用域与 list_by_workspace 相同：workspace_id 为空表示该 user 全部。
+        作用域与 list_by_workspace 相同：workspace_id 为空表示该 user 全部；
+        来源过滤同样与 list_by_workspace 一致，count 因此恒等于分页可翻到的总数。
         """
-        return await _run_in_pool(self._stats_sync, user_id, workspace_id, branch)
+        return await _run_in_pool(
+            self._stats_sync, user_id, workspace_id, branch,
+            include_sources, exclude_sources,
+        )
 
     def _stats_sync(
         self, user_id: str, workspace_id: Optional[str], branch: Optional[str],
+        include_sources: Optional[List[str]] = None,
+        exclude_sources: Optional[List[str]] = None,
     ) -> Dict[str, Any]:
         scope_sql, scope_params = _scope_where(workspace_id, branch)
+        source_sql, source_params = _source_where(include_sources, exclude_sources)
         with self._lock:
             assert self._conn is not None
             row = self._conn.execute(
                 f"SELECT COUNT(*) AS n, MAX(created_at) AS created, "
                 f"MAX(updated_at) AS updated FROM coding_memory_meta "
-                f"WHERE user_id = ? AND {scope_sql}",
-                [user_id] + scope_params,
+                f"WHERE user_id = ? AND {scope_sql} AND {source_sql}",
+                [user_id] + scope_params + source_params,
             ).fetchone()
             facts = self._conn.execute(
                 f"SELECT COUNT(*) AS n FROM working_fact "
-                f"WHERE user_id = ? AND status = 'active' AND {scope_sql}",
-                [user_id] + scope_params,
+                f"WHERE user_id = ? AND status = 'active' AND {scope_sql} "
+                f"AND {source_sql}",
+                [user_id] + scope_params + source_params,
             ).fetchone()
         return {
             "count": int(row["n"] or 0),
@@ -1124,6 +1238,14 @@ class CodingMemoryStore:
                 (experience_id, user_id),
             )
 
+    def _delete_fact_sql_sync(self, fact_id: str, user_id: str) -> None:
+        with self._lock:
+            assert self._conn is not None
+            self._conn.execute(
+                "DELETE FROM working_fact WHERE fact_id = ? AND user_id = ?",
+                (fact_id, user_id),
+            )
+
     def _get_by_id_sync(
         self, experience_id: str, user_id: Optional[str]
     ) -> Optional[CodingMemory]:

hy_memory/pipelines/_retrieval/diversity.py

@@ -160,6 +160,17 @@ def near_dup_demote(
     return out
 
 
+# MMR 只在调用方请求的 top-k（search 的 limit）小于这个值时自动启用。
+# mmr_select 的贪心是 O(k^2 * pool) 次 Jaccard，k 大时代价陡增，而大 k 本身已
+# 足够容纳多样性，收益很小。权重仍由 MEMORY_HYBRID_V2_MMR_LAMBDA 控制。
+MMR_MAX_TOP_K = 20
+
+
+def mmr_applies(top_k: int, lambda_: float) -> bool:
+    """本次检索是否走 MMR：λ 在有效区间内，且 top_k < MMR_MAX_TOP_K。"""
+    return 0.0 < float(lambda_) < 0.999 and int(top_k) < MMR_MAX_TOP_K
+
+
 def mmr_select(
     hits: Sequence[Dict[str, Any]],
     limit: int,

hy_memory/pipelines/base.py

@@ -401,6 +401,11 @@ class ReadRequest:
     intention_limit: int = 0           # Intention 路的 topk，默认 0 关闭
 
 
+    # 重排开关（按次生效）。False 时跳过全部重排阶段：本地 cross-encoder
+    # (recall.local_rerank_enabled) 与记忆关系图重排 (_apply_graph_rerank)，
+    # 结果按召回分排序。走请求对象而非环境变量，并发请求互不影响。
+    rerank: bool = True
+
     # 时间过滤
     created_after: Optional[float] = None  # Unix timestamp (float)，只返回 gmt_created >= 此值的记忆
 

hy_memory/pipelines/reader_base.py

@@ -58,6 +58,7 @@ from ._retrieval.ner import (
 )
 from ._retrieval.diversity import (
     near_dup_demote,
+    mmr_applies,
     mmr_select,
     social_boilerplate_demote,
     pool_df_demote,
@@ -1279,7 +1280,10 @@ class BaseReadPipeline(ReadPipeline):
             SAME_SERIES_TIMELINE_ON = bool(
                 getattr(self.config.recall, "same_series_timeline_enabled", False)
             )
-            LOCAL_RERANK_ON = bool(
+            # 按次的重排开关（search(rerank=False)）：关掉本地 cross-encoder 与图重排，
+            # 结果按召回分排序。只读请求对象，不碰共享状态，并发请求互不影响。
+            RERANK_ON = bool(getattr(request, "rerank", True))
+            LOCAL_RERANK_ON = RERANK_ON and bool(
                 getattr(self.config.recall, "local_rerank_enabled", False)
             )
             LOCAL_RERANK_POOL_FACTOR = max(1, int(
@@ -1431,6 +1435,9 @@ class BaseReadPipeline(ReadPipeline):
             MIN_SCORE = _resolve_normal_min_score(request.min_score)
 
             final_limit = request.limit if request.limit > 0 else 10
+            # MMR 只在调用方的 top-k（final_limit，即 search 的 limit）< MMR_MAX_TOP_K
+            # 时启用；池子（LOCAL_RERANK 下是 final_limit 的若干倍）不参与判定。
+            MMR_ON = mmr_applies(final_limit, MMR_LAMBDA)
             vdb_sem_limit = max(final_limit * 3, 30)
             vdb_kw_limit = max(final_limit * 5, 80)
             graph_limit = max(
@@ -2462,7 +2469,7 @@ class BaseReadPipeline(ReadPipeline):
             # 只把已选出的 top-N 换个顺序收益很小。
             relation_paths: List[Dict[str, Any]] = []
             _graph_stats: Dict[str, Any] = {}
-            if self._expansion_on():
+            if RERANK_ON and self._expansion_on():
                 _t_graph = datetime.now()
                 _pool_before = len(normal_results)
                 normal_results, relation_paths, _graph_stats = await self._apply_graph_rerank(
@@ -2538,7 +2545,7 @@ class BaseReadPipeline(ReadPipeline):
                     ))
                     if LOCAL_RERANK_ON else final_limit
                 )
-                if 0.0 < MMR_LAMBDA < 0.999 and len(normal_results) > _pre_rerank_limit:
+                if MMR_ON and len(normal_results) > _pre_rerank_limit:
                     normal_results = mmr_select(
                         normal_results,
                         _pre_rerank_limit,
@@ -2715,7 +2722,7 @@ class BaseReadPipeline(ReadPipeline):
                         reverse=True,
                     )
                     if (
-                        0.0 < MMR_LAMBDA < 0.999
+                        MMR_ON
                         and len(multi_seed_pool) > _pre_rerank_limit
                     ):
                         multi_seed_pool = mmr_select(
@@ -3544,6 +3551,8 @@ class BaseReadPipeline(ReadPipeline):
                 "episode_capsule_render_mode": EPISODE_CAPSULE_RENDER_MODE,
                 "same_series_timeline_enabled": SAME_SERIES_TIMELINE_ON,
                 "local_rerank_enabled": LOCAL_RERANK_ON,
+                "rerank": RERANK_ON,
+                "mmr_applied": MMR_ON,
                 "local_rerank_model": getattr(
                     self.config.recall,
                     "local_rerank_model",

hy_memory/server.py

@@ -33,9 +33,28 @@ import threading
 from datetime import datetime
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import Any, Dict, Optional
+from urllib.parse import parse_qs, urlparse
+
+from .retrieval import AmbiguousMemoryIdError
 
 logger = logging.getLogger("hy_memory.server")
 
+
+def _namespace_args(values: Dict[str, Any]) -> Dict[str, Any]:
+    """kind / variant from a query string (lists) or a JSON body (scalars).
+
+    They disambiguate an id that exists in more than one namespace (chat,
+    experience, fact; working or coding). Absent means "resolve it".
+    """
+    args: Dict[str, Any] = {}
+    for key in ("kind", "variant"):
+        value = values.get(key)
+        if isinstance(value, list):
+            value = value[0] if value else None
+        if value:
+            args[key] = str(value)
+    return args
+
 # Lazy-init global client
 _client = None
 _client_lock = None
@@ -148,7 +167,6 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
             if path == "/api/v1/metrics":
                 minutes = 5
                 if "?" in self.path:
-                    from urllib.parse import parse_qs, urlparse
                     qs = parse_qs(urlparse(self.path).query)
                     minutes = int(qs.get("minutes", ["5"])[0])
                 client = _get_client()
@@ -161,7 +179,8 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
             if m:
                 memory_id = m.group(1)
                 client = _get_client()
-                result = client.get(memory_id)
+                query = parse_qs(urlparse(self.path).query)
+                result = client.get(memory_id, **_namespace_args(query))
                 if result is None:
                     _json_response(self, 404, {"error": "not_found", "memory_id": memory_id})
                 else:
@@ -170,6 +189,10 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
 
             _json_response(self, 404, {"error": "not_found", "path": path})
 
+        except AmbiguousMemoryIdError as e:
+            _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)})
+        except ValueError as e:
+            _json_response(self, 400, {"error": str(e)})
         except Exception as e:
             logger.error(f"[server] GET {path} error: {e}", exc_info=True)
             _json_response(self, 500, {"error": str(e)})
@@ -213,7 +236,7 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
             _json_response(self, 500, {"error": str(e)})
 
     def do_PUT(self):
-        path = self.path.rstrip("/")
+        path = self.path.split("?")[0].rstrip("/")
         body = _read_json_body(self)
         if body is None:
             _json_response(self, 400, {"error": "invalid JSON body"})
@@ -228,29 +251,43 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
                     _json_response(self, 400, {"error": "content is required"})
                     return
                 client = _get_client()
-                result = client.update(memory_id, content)
+                result = client.update(
+                    memory_id,
+                    content,
+                    **_namespace_args(body),
+                    **({"task": body["task"]} if body.get("task") else {}),
+                )
                 _json_response(self, 200, result)
                 return
 
             _json_response(self, 404, {"error": "not_found", "path": path})
 
+        except AmbiguousMemoryIdError as e:
+            _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)})
+        except ValueError as e:
+            _json_response(self, 400, {"error": str(e)})
         except Exception as e:
             logger.error(f"[server] PUT {path} error: {e}", exc_info=True)
             _json_response(self, 500, {"error": str(e)})
 
     def do_DELETE(self):
-        path = self.path.rstrip("/")
+        path = self.path.split("?")[0].rstrip("/")
         try:
             m = re.match(r"^/api/v1/memories/([^/]+)$", path)
             if m:
                 memory_id = m.group(1)
                 client = _get_client()
-                result = client.delete(memory_id)
+                query = parse_qs(urlparse(self.path).query)
+                result = client.delete(memory_id, **_namespace_args(query))
                 _json_response(self, 200, result)
                 return
 
             _json_response(self, 404, {"error": "not_found", "path": path})
 
+        except AmbiguousMemoryIdError as e:
+            _json_response(self, 409, {"error": "ambiguous_memory_id", "detail": str(e)})
+        except ValueError as e:
+            _json_response(self, 400, {"error": str(e)})
         except Exception as e:
             logger.error(f"[server] DELETE {path} error: {e}", exc_info=True)
             _json_response(self, 500, {"error": str(e)})
@@ -455,7 +492,8 @@ class MemoryHTTPHandler(BaseHTTPRequestHandler):
             if key in body:
                 kwargs[key] = body[key]
 
-        for key in ("limit", "min_score", "schema_min_score", "schema_limit", "reader"):
+        for key in ("limit", "min_score", "schema_min_score", "schema_limit", "reader",
+                    "rerank"):
             if key in body:
                 kwargs[key] = body[key]
 

tests/coding/test_all_scenes.py

@@ -0,0 +1,665 @@
+# -*- coding: utf-8 -*-
+"""scene="all"（search / list）与生产力条目的 update / delete。
+
+客户端用 object.__new__(HyMemoryClient) 搭最小骨架，公开 async_* 方法通过
+__wrapped__ 直接调（不需要 client 自己的 loop 线程），与
+test_list_and_stats_by_scene.py 同一套写法。
+"""
+
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from hy_memory.client import _SCENE_RRF_K, HyMemoryClient
+from hy_memory.coding.store import CodingMemoryStore
+from hy_memory.coding.types import CodingMemory, WorkingFact
+from hy_memory.models.memory import MemoryLayer, MemoryStatus
+from hy_memory.retrieval import AmbiguousMemoryIdError
+
+BASE = datetime(2026, 10, 20, 8, 0, 0)
+
+
+def _unwrapped(name):
+    return getattr(HyMemoryClient, name).__wrapped__
+
+
+async def _store(coding_config, mock_embed, variant):
+    store = CodingMemoryStore(coding_config, embed_service=mock_embed, variant=variant)
+    await store.initialize()
+    return store
+
+
+async def _noop(*_args, **_kwargs):
+    return None
+
+
+def _client(coding_config, mock_embed, stores, *, nodes=()):
+    client = object.__new__(HyMemoryClient)
+    client._config = coding_config
+    client._coding_stores = dict(stores)
+    client._embed_service = mock_embed
+    client._cache = None
+    client._graph_store = None
+    client._history_store = None
+    client._vector_store = _FakeVectorStore(list(nodes))
+    client._get_coding_writer = _noop
+    client._maybe_store_op_log = _noop
+    client._decorate_with_chain = lambda *a, **k: ("", [])
+    return client
+
+
+class _FakeVectorStore:
+    def __init__(self, nodes):
+        self.nodes = {n.node_id: n for n in nodes}
+
+    async def list_by_user(self, *, layers=None, **_):
+        if layers == [MemoryLayer.L0_BASIC_INFO]:
+            return []
+        return list(self.nodes.values())
+
+    async def get_by_id(self, memory_id):
+        return self.nodes.get(memory_id)
+
+
+def _chat_node(node_id, *, created, modified=None, source="cursor"):
+    return SimpleNamespace(
+        node_id=node_id, layer=MemoryLayer.L2_FACT, status=MemoryStatus.ACTIVE,
+        content=f"chat {node_id}", memory_at=None,
+        gmt_created=created, gmt_modified=modified, valid_from=None,
+        user_id="u", agent_id="a", session_id="s", source=source,
+        speculate=None, tags=[], custom={},
+    )
+
+
+# ================================================================
+# search(scene="all")
+# ================================================================
+
+def _chat_resp(*ids):
+    return {
+        "request_id": "r",
+        "memories": {
+            "normal": [
+                {"memory_id": i, "content": f"chat {i}", "score": 0.9 - n * 0.1,
+                 "gmt_created": 1_700_000_000 + n, "source": "cursor"}
+                for n, i in enumerate(ids)
+            ],
+            "schema": [],
+            "proactive": [],
+        },
+    }
+
+
+def _prod_resp(facts=(), exps=()):
+    return {
+        "request_id": "r",
+        "scene": "productivity",
+        "facts": [
+            {"fact_id": f, "statement": f"stmt {f}", "score": 0.5,
+             "created_at": "2026-10-20T08:00:00", "source": "claude"}
+            for f in facts
+        ],
+        "experiences": [
+            {"experience_id": e, "task": f"task {e}", "solution": f"sol {e}",
+             "score": 0.7, "updated_at": "2026-10-20T09:00:00", "source": "claude"}
+            for e in exps
+        ],
+    }
+
+
+def _fake_all_client(*, chat, working=None, coding=None, fail=(), enable=True,
+                     seen=None):
+    """_search_all_scenes 的依赖全部换成桩：只测融合与容错。"""
+    client = object.__new__(HyMemoryClient)
+    client._config = SimpleNamespace(coding=SimpleNamespace(enable=enable))
+
+    class _Embed:
+        calls = 0
+
+        async def embed(self, text):
+            _Embed.calls += 1
+            return [0.1, 0.2]
+
+    client._embed_service = _Embed()
+    seen = seen if seen is not None else {}
+
+    async def _impl(query, **kw):
+        seen["chat"] = kw
+        if "chat" in fail:
+            raise RuntimeError("chat backend down")
+        return chat
+
+    async def _coding_store(variant):
+        return object() if {"working": working, "coding": coding}[variant] is not None else None
+
+    async def _search_coding(target_query, **kw):
+        seen[kw["variant"]] = kw
+        if kw["variant"] in fail:
+            raise RuntimeError(f"{kw['variant']} store broken")
+        return {"working": working, "coding": coding}[kw["variant"]]
+
+    client._async_search_impl = _impl
+    client._coding_store_for_exact_get = _coding_store
+    client._search_coding = _search_coding
+    return client, seen
+
+
+async def _search_all(client, **kw):
+    params = dict(
+        queries=None, search_mode="default", user_ids=["u"], agent_ids=[],
+        session_ids=[], include_sources=["claude"], exclude_sources=["mixed"],
+        limit=10, min_score=0.4, working_fact_min_score=None,
+        working_experience_min_score=None, schema_min_score=0.4, schema_limit=10,
+        intention_limit=0, created_after=None, reader="", workspace_id=None,
+        branch=None, request_id="req", layers=None, query_embedding=None,
+        rerank=True, target_query="q", lite_mode=False, t0=0.0,
+    )
+    params.update(kw)
+    return await HyMemoryClient._search_all_scenes(client, "q", **params)
+
+
+async def test_search_all_fuses_by_lane_rank_with_rrf():
+    client, seen = _fake_all_client(
+        chat=_chat_resp("c1", "c2", "c3"),
+        working=_prod_resp(facts=["wf1"], exps=["we1", "we2"]),
+        coding=_prod_resp(exps=["ce1"]),
+    )
+
+    out = await _search_all(client, rerank=False)
+
+    assert out["scene"] == "all"
+    assert out["request_id"] == "req"
+    assert out["lanes"] == {"chat": 3, "working": 3, "coding": 1}
+    assert out["errors"] == {}
+    order = [(m["scene"], m["memory_id"]) for m in out["memories"]]
+    # 名次 1：chat → working → coding（同分按路序），名次 2：chat / working ...
+    assert order == [
+        ("chat", "c1"), ("working", "wf1"), ("coding", "ce1"),
+        ("chat", "c2"), ("working", "we1"),
+        ("chat", "c3"), ("working", "we2"),
+    ]
+    first = out["memories"][0]
+    assert first["score"] == round(1.0 / (_SCENE_RRF_K + 1), 6)
+    assert first["lane_score"] == pytest.approx(0.9)
+    assert first["lane_rank"] == 1 and first["kind"] == "chat"
+    fact = out["memories"][1]
+    assert fact["kind"] == "fact" and fact["content"] == "stmt wf1"
+    assert fact["lane_score"] == 0.5
+    exp = out["memories"][4]
+    assert exp["kind"] == "experience"
+    assert exp["memory_id"] == "we1"
+    assert exp["content"] == "task we1\nsol we1"
+    assert exp["score"] == round(1.0 / (_SCENE_RRF_K + 2), 6)
+    # 生产力时间戳是 naive UTC，按 UTC 换算成 epoch
+    assert exp["updated_at"] == int(
+        datetime(2026, 10, 20, 9, tzinfo=timezone.utc).timestamp()
+    )
+    # rerank / 来源过滤 / 阈值传到每一路，query 向量只算一次
+    for lane in ("chat", "working", "coding"):
+        assert seen[lane]["rerank"] is False
+        assert seen[lane]["include_sources"] == ["claude"]
+        assert seen[lane]["query_embedding"] == [0.1, 0.2]
+    assert seen["chat"]["scene"] == "chat"
+    assert client._embed_service.calls == 1
+
+
+async def test_search_all_truncates_after_fusion():
+    client, _ = _fake_all_client(
+        chat=_chat_resp("c1", "c2", "c3"),
+        working=_prod_resp(exps=["we1", "we2"]),
+        coding=None,
+    )
+    out = await _search_all(client, limit=3)
+    assert [m["memory_id"] for m in out["memories"]] == ["c1", "we1", "c2"]
+    # coding 没写过：不贡献、不算错误
+    assert out["lanes"]["coding"] == 0
+    assert "coding" not in out["errors"]
+
+
+async def test_search_all_isolates_a_failing_lane():
+    client, _ = _fake_all_client(
+        chat=_chat_resp("c1"),
+        working=_prod_resp(exps=["we1"]),
+        coding=_prod_resp(exps=["ce1"]),
+        fail=("working",),
+    )
+    out = await _search_all(client)
+    assert [m["memory_id"] for m in out["memories"]] == ["c1", "ce1"]
+    assert out["lanes"] == {"chat": 1, "working": 0, "coding": 1}
+    assert "store broken" in out["errors"]["working"]
+
+    client, _ = _fake_all_client(
+        chat=_chat_resp("c1"), working=_prod_resp(exps=["we1"]), fail=("chat",),
+    )
+    out = await _search_all(client)
+    assert [m["memory_id"] for m in out["memories"]] == ["we1"]
+    assert "chat backend down" in out["errors"]["chat"]
+
+
+async def test_search_all_reports_disabled_productivity():
+    client, seen = _fake_all_client(
+        chat=_chat_resp("c1"), working=_prod_resp(exps=["we1"]), enable=False,
+    )
+    out = await _search_all(client)
+    assert [m["memory_id"] for m in out["memories"]] == ["c1"]
+    assert out["lanes"] == {"chat": 1, "working": 0, "coding": 0}
+    assert set(out["errors"]) == {"working", "coding"}
+    assert "working" not in seen
+
+
+async def test_async_search_routes_scene_all(coding_config):
+    seen = {}
+
+    async def _spy(query, **kw):
+        seen.update(kw)
+        return {"scene": "all"}
+
+    client = object.__new__(HyMemoryClient)
+    client._config = coding_config
+    client._search_all_scenes = _spy
+    out = await _unwrapped("async_search")(
+        client, "q", scene=" ALL ", user_ids=["u"], include_sources=["Claude"],
+        rerank=False, limit=7,
+    )
+    assert out == {"scene": "all"}
+    assert seen["rerank"] is False and seen["limit"] == 7
+    assert seen["include_sources"] == ["claude"]
+    assert seen["exclude_sources"] == ["mixed"]
+
+    with pytest.raises(ValueError, match="all"):
+        await _unwrapped("async_search")(client, "q", scene="every", user_ids=["u"])
+
+
+async def test_search_all_productivity_lanes_respect_sources(coding_config, mock_embed):
+    """真 store：两路生产力检索按 source 过滤，经验与事实都过。"""
+    coding_config.coding.enable = True
+    working = await _store(coding_config, mock_embed, "working")
+    coding = await _store(coding_config, mock_embed, "coding")
+    try:
+        await working.insert(CodingMemory(
+            experience_id="w-claude", user_id="u", task="deploy service",
+            solution="kubectl apply", source="claude",
+        ))
+        await working.insert(CodingMemory(
+            experience_id="w-cursor", user_id="u", task="deploy service",
+            solution="helm upgrade", source="cursor",
+        ))
+        await working.insert_fact(WorkingFact(
+            fact_id="f-cursor", user_id="u", statement="deploy service",
+            source="cursor",
+        ))
+        await coding.insert(CodingMemory(
+            experience_id="c-claude", user_id="u", task="deploy service",
+            solution="make deploy", source="claude",
+        ))
+        client = _client(coding_config, mock_embed,
+                         {"working": working, "coding": coding})
+
+        async def _chat(query, **kw):
+            return _chat_resp()
+
+        client._async_search_impl = _chat
+        out = await _search_all(
+            client, target_query="deploy service",
+            include_sources=["claude"], exclude_sources=["mixed"],
+        )
+        ids = {(m["scene"], m["memory_id"]) for m in out["memories"]}
+        assert ids == {("working", "w-claude"), ("coding", "c-claude")}
+        assert all(m["source"] == "claude" for m in out["memories"])
+    finally:
+        await working.close()
+        await coding.close()
+
+
+async def test_productivity_search_rerank_false_skips_fact_lexical_rerank(
+    coding_config, mock_embed, monkeypatch,
+):
+    from hy_memory.coding import lexical
+
+    calls = []
+    real = lexical.rerank
+
+    def _spy(*args, **kwargs):
+        calls.append(args)
+        return real(*args, **kwargs)
+
+    monkeypatch.setattr(lexical, "rerank", _spy)
+    working = await _store(coding_config, mock_embed, "working")
+    try:
+        for index in range(8):
+            await working.insert_fact(WorkingFact(
+                fact_id=f"f{index}", user_id="u", statement="deploy service",
+            ))
+        client = _client(coding_config, mock_embed, {"working": working})
+
+        async def _run(rerank):
+            return await HyMemoryClient._search_coding(
+                client, "deploy service", user_ids=["u"], workspace_id=None,
+                branch=None, limit=2, request_id="r", t0=0.0,
+                variant="working", rerank=rerank,
+            )
+
+        await _run(False)
+        assert calls == []
+        await _run(True)
+        assert len(calls) == 1
+    finally:
+        await working.close()
+
+
+# ================================================================
+# list_memories(scene="all") 与生产力 list 的来源过滤
+# ================================================================
+
+async def test_list_productivity_scene_filters_sources_with_true_total(
+    coding_config, mock_embed,
+):
+    coding = await _store(coding_config, mock_embed, "coding")
+    try:
+        for index, source in enumerate(["claude", "cursor", "claude", "Claude"]):
+            await coding.insert(CodingMemory(
+                experience_id=f"c{index}", user_id="u", task="T", solution="S",
+                source=source,
+            ))
+        client = _client(coding_config, mock_embed, {"coding": coding})
+        list_memories = _unwrapped("async_list_memories")
+
+        page = await list_memories(
+            client, user_id="u", scene="coding", include_sources=["CLAUDE"], limit=2,
+        )
+        assert page["total"] == 3
+        assert len(page["experiences"]) == 2
+        assert all(e["source"] == "claude" for e in page["experiences"])
+
+        page = await list_memories(
+            client, user_id="u", scene="coding", exclude_sources=["claude"],
+        )
+        assert page["total"] == 1
+        assert [e["experience_id"] for e in page["experiences"]] == ["c1"]
+
+        unfiltered = await list_memories(client, user_id="u", scene="coding")
+        assert unfiltered["total"] == 4
+    finally:
+        await coding.close()
+
+
+async def _seed_all(coding_config, mock_embed):
+    coding = await _store(coding_config, mock_embed, "coding")
+    working = await _store(coding_config, mock_embed, "working")
+    # 生产力时间戳按 UTC 写；这里直接给 updated_at 再改回去（insert 会刷新）
+    for eid, hours, source in (("c-old", 1, "claude"), ("c-new", 5, "cursor")):
+        await coding.insert(CodingMemory(
+            experience_id=eid, user_id="u", task=f"task {eid}", solution="S",
+            source=source,
+        ))
+        await _set_updated(coding, eid, BASE + timedelta(hours=hours))
+    await working.insert(CodingMemory(
+        experience_id="w-mid", user_id="u", task="task w", solution="W",
+        source="claude",
+    ))
+    await _set_updated(working, "w-mid", BASE + timedelta(hours=3))
+    # chat 节点是本地时间 naive；与 UTC 的经验比较时要换算到同一 epoch
+    utc_offset = datetime.now() - datetime.utcnow()
+    nodes = [
+        _chat_node("m-newest", created=BASE + utc_offset + timedelta(hours=6),
+                   source="claude"),
+        _chat_node("m-edited", created=BASE + utc_offset,
+                   modified=BASE + utc_offset + timedelta(hours=4), source="cursor"),
+        _chat_node("m-oldest", created=BASE + utc_offset, source="claude"),
+    ]
+    return coding, working, nodes
+
+
+async def _set_updated(store, experience_id, when):
+    def _sync():
+        with store._lock:
+            store._conn.execute(
+                "UPDATE coding_memory_meta SET updated_at = ? WHERE experience_id = ?",
+                (when.isoformat(), experience_id),
+            )
+
+    _sync()
+
+
+async def test_list_all_merges_newest_first_and_paginates(coding_config, mock_embed):
+    coding, working, nodes = await _seed_all(coding_config, mock_embed)
+    try:
+        client = _client(coding_config, mock_embed,
+                         {"coding": coding, "working": working}, nodes=nodes)
+        list_memories = _unwrapped("async_list_memories")
+
+        out = await list_memories(client, user_id="u", scene="all", limit=10)
+        assert out["scene"] == "all"
+        assert out["total"] == 6
+        assert out["totals"] == {"chat": 3, "working": 1, "coding": 2}
+        assert [(m["scene"], m["memory_id"]) for m in out["memories"]] == [
+            ("chat", "m-newest"),
+            ("coding", "c-new"),
+            ("chat", "m-edited"),
+            ("working", "w-mid"),
+            ("coding", "c-old"),
+            ("chat", "m-oldest"),
+        ]
+        exp = out["memories"][1]
+        assert exp["kind"] == "experience" and exp["content"] == "task c-new\nS"
+        assert exp["status"] == "active"
+        assert exp["updated_at"] > out["memories"][2]["updated_at"]
+        assert out["memories"][0]["kind"] == "chat"
+
+        pages = []
+        for offset in (0, 2, 4):
+            page = await list_memories(
+                client, user_id="u", scene="all", limit=2, offset=offset,
+            )
+            assert page["total"] == 6
+            pages.extend(m["memory_id"] for m in page["memories"])
+        assert pages == [m["memory_id"] for m in out["memories"]]
+
+        with pytest.raises(ValueError, match="desc"):
+            await list_memories(client, user_id="u", scene="all", order="asc")
+    finally:
+        await coding.close()
+        await working.close()
+
+
+async def test_list_all_applies_source_and_status_filters(coding_config, mock_embed):
+    coding, working, nodes = await _seed_all(coding_config, mock_embed)
+    try:
+        client = _client(coding_config, mock_embed,
+                         {"coding": coding, "working": working}, nodes=nodes)
+        list_memories = _unwrapped("async_list_memories")
+
+        out = await list_memories(
+            client, user_id="u", scene="all", include_sources=["claude"],
+        )
+        assert [m["memory_id"] for m in out["memories"]] == [
+            "m-newest", "w-mid", "c-old", "m-oldest",
+        ]
+        assert out["totals"] == {"chat": 2, "working": 1, "coding": 1}
+        assert out["total"] == 4
+
+        out = await list_memories(
+            client, user_id="u", scene="all", statuses=["archived"],
+        )
+        # 经验只有 ACTIVE：两路生产力不贡献；chat 假库不按状态过滤，原样返回
+        assert out["totals"]["working"] == 0 and out["totals"]["coding"] == 0
+        assert all(m["scene"] == "chat" for m in out["memories"])
+    finally:
+        await coding.close()
+        await working.close()
+
+
+async def test_list_all_does_not_create_unwritten_scene_databases(coding_config, mock_embed):
+    client = _client(coding_config, mock_embed, {}, nodes=[
+        _chat_node("m1", created=BASE),
+    ])
+    out = await _unwrapped("async_list_memories")(client, user_id="u", scene="all")
+    assert out["total"] == 1
+    assert out["totals"] == {"chat": 1, "working": 0, "coding": 0}
+    for variant in ("working", "coding"):
+        assert not CodingMemoryStore.database_exists(
+            coding_config, db_path=coding_config.coding.db_path, variant=variant,
+        )
+
+
+# ================================================================
+# update / delete 生产力条目
+# ================================================================
+
+async def test_update_experience_replaces_solution_and_keeps_task(coding_config, mock_embed):
+    working = await _store(coding_config, mock_embed, "working")
+    try:
+        await working.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="deploy service",
+            search_keys=["rollout"], solution="old way",
+        ))
+        await _set_updated(working, "e1", BASE)
+        client = _client(coding_config, mock_embed, {"working": working})
+
+        out = await _unwrapped("async_update")(client, "e1", "new way")
+        assert out["success"] is True
+        assert out["memory_id"] == "e1"
+        assert out["kind"] == "experience" and out["variant"] == "working"
+        assert {"request_id", "elapsed_ms"} <= set(out)
+
+        stored = await working.get_by_id("e1")
+        assert stored.task == "deploy service"
+        assert stored.solution == "new way"
+        assert stored.updated_at > BASE
+        # solution 不进向量：仍能按原 task 召回
+        hits = await working.search_by_query_embedding(
+            await mock_embed.embed("deploy service"), user_id="u",
+        )
+        assert hits and hits[0]["experience_id"] == "e1"
+
+        out = await _unwrapped("async_update")(
+            client, "e1", "newer way", task="ship service",
+        )
+        assert out["success"] is True
+        stored = await working.get_by_id("e1")
+        assert stored.task == "ship service"
+        hits = await working.search_by_query_embedding(
+            await mock_embed.embed("ship service"), user_id="u",
+        )
+        assert hits[0]["experience_id"] == "e1"
+        assert hits[0]["score"] == pytest.approx(1.0, abs=1e-4)
+    finally:
+        await working.close()
+
+
+async def test_update_and_delete_fact(coding_config, mock_embed):
+    coding = await _store(coding_config, mock_embed, "coding")
+    try:
+        await coding.insert_fact(WorkingFact(
+            fact_id="f1", user_id="u", statement="build uses make",
+        ))
+        client = _client(coding_config, mock_embed, {"coding": coding})
+
+        out = await _unwrapped("async_update")(
+            client, "f1", "build uses bazel", kind="fact",
+        )
+        assert out["success"] is True and out["variant"] == "coding"
+        assert (await coding.get_fact("f1")).statement == "build uses bazel"
+        hits = await coding.search_facts(
+            await mock_embed.embed("build uses bazel"), user_id="u",
+        )
+        assert [f.fact_id for f, _ in hits] == ["f1"]
+
+        with pytest.raises(ValueError, match="task"):
+            await _unwrapped("async_update")(client, "f1", "x", task="t")
+
+        out = await _unwrapped("async_delete")(client, "f1")
+        assert out["success"] is True and out["deleted_count"] == 1
+        assert out["kind"] == "fact"
+        assert await coding.get_fact("f1") is None
+        assert await coding.search_facts(
+            await mock_embed.embed("build uses bazel"), user_id="u",
+        ) == []
+    finally:
+        await coding.close()
+
+
+async def test_delete_experience_removes_row_and_vectors(coding_config, mock_embed):
+    working = await _store(coding_config, mock_embed, "working")
+    try:
+        await working.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="deploy service",
+            search_keys=["rollout"], solution="S",
+        ))
+        client = _client(coding_config, mock_embed, {"working": working})
+
+        out = await _unwrapped("async_delete")(
+            client, "e1", kind="experience", variant="working",
+        )
+        assert out["success"] is True
+        assert out["deleted_count"] == 1
+        assert out["memory_id"] == "e1"
+        assert await working.get_by_id("e1") is None
+        assert await working._vec.get_by_id("e1:0") is None
+        assert await working._vec.get_by_id("e1:1") is None
+    finally:
+        await working.close()
+
+
+async def test_productivity_not_found_and_ambiguity(coding_config, mock_embed):
+    working = await _store(coding_config, mock_embed, "working")
+    coding = await _store(coding_config, mock_embed, "coding")
+    try:
+        for store in (working, coding):
+            await store.insert(CodingMemory(
+                experience_id="same", user_id="u", task="T", solution="S",
+            ))
+        client = _client(coding_config, mock_embed,
+                         {"working": working, "coding": coding})
+
+        with pytest.raises(AmbiguousMemoryIdError):
+            await _unwrapped("async_update")(client, "same", "x")
+        with pytest.raises(AmbiguousMemoryIdError):
+            await _unwrapped("async_delete")(client, "same")
+
+        out = await _unwrapped("async_delete")(client, "same", variant="coding")
+        assert out["success"] is True and out["variant"] == "coding"
+        assert await working.get_by_id("same") is not None
+
+        missing = await _unwrapped("async_update")(client, "nope", "x", kind="experience")
+        assert missing["success"] is False and "not found" in missing["error_message"]
+        missing = await _unwrapped("async_delete")(client, "nope", kind="fact")
+        assert missing["success"] is False and missing["deleted_count"] == 0
+        # variant 给了就不回落到 chat
+        missing = await _unwrapped("async_delete")(client, "nope", variant="working")
+        assert missing["success"] is False and missing["deleted_count"] == 0
+
+        with pytest.raises(ValueError):
+            await _unwrapped("async_delete")(client, "same", kind="chat", variant="coding")
+    finally:
+        await working.close()
+        await coding.close()
+
+
+async def test_update_and_delete_without_kind_still_handle_chat(coding_config, mock_embed):
+    """默认行为不变：chat ID 走原 chat 路径（store 都没建时也不建库）。"""
+    client = _client(coding_config, mock_embed, {})
+    seen = []
+
+    async def _chat_update(memory_id, content):
+        seen.append(("update", memory_id, content))
+        return {"success": True, "memory_id": memory_id}
+
+    async def _chat_delete(memory_id):
+        seen.append(("delete", memory_id))
+        return {"success": True, "deleted_count": 1}
+
+    client._update_chat = _chat_update
+    client._delete_chat = _chat_delete
+    client._vector_store = _FakeVectorStore([_chat_node("m1", created=BASE)])
+
+    assert (await _unwrapped("async_update")(client, "m1", "new"))["success"]
+    assert (await _unwrapped("async_delete")(client, "m1"))["success"]
+    # 哪都没有：仍交给 chat 路径给出原来的 not-found 结果
+    await _unwrapped("async_delete")(client, "ghost")
+    assert seen == [("update", "m1", "new"), ("delete", "m1"), ("delete", "ghost")]
+    for variant in ("working", "coding"):
+        assert not CodingMemoryStore.database_exists(
+            coding_config, db_path=coding_config.coding.db_path, variant=variant,
+        )

tests/coding/test_list_and_stats_by_scene.py

@@ -6,7 +6,7 @@ working 两类经验既列不出也数不到：一个 45 条记忆的库在界
 """
 
 import asyncio
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 
 import pytest
@@ -97,13 +97,14 @@ async def test_stats_counts_every_scene_with_its_latest_times(coding_config, moc
             "latest_updated_at": int((BASE + timedelta(days=1)).timestamp()),
         }
         assert stats["scenes"]["coding"]["count"] == 2
+        # Productivity stores write naive UTC (utcnow), so naive means UTC.
         assert stats["scenes"]["coding"]["latest_created_at"] == int(
-            (BASE + timedelta(hours=2)).timestamp()
+            (BASE + timedelta(hours=2)).replace(tzinfo=timezone.utc).timestamp()
         )
         # The store stamps updated_at on write; the stat is the newest it holds.
         stored = [await coding.get_by_id(eid, user_id="u") for eid in ("c1", "c2")]
         assert stats["scenes"]["coding"]["latest_updated_at"] == max(
-            int(item.updated_at.timestamp()) for item in stored
+            int(item.updated_at.replace(tzinfo=timezone.utc).timestamp()) for item in stored
         )
         assert stats["scenes"]["working"]["count"] == 1
         assert stats["scenes"]["working"]["facts"] == 1

tests/test_search_rerank_mmr_gates.py

@@ -0,0 +1,258 @@
+# -*- coding: utf-8 -*-
+"""search(rerank=...) 与 MMR 的按次开关。
+
+- rerank=False：本地 cross-encoder 与图重排都不跑（看 spy 是否被调到）。
+- MMR：调用方的 limit < MMR_MAX_TOP_K 时启用，>= 时跳过。
+- 默认值（rerank=True、limit=10）保持原行为。
+"""
+
+import asyncio
+import math
+import uuid
+
+import pytest
+
+from hy_memory.config import MemoryConfig
+from hy_memory.models.memory import MemoryLayer, MemoryNode, MemoryStatus
+from hy_memory.pipelines import reader_base
+from hy_memory.pipelines._retrieval.diversity import MMR_MAX_TOP_K, mmr_applies
+from hy_memory.pipelines.base import ReadRequest
+
+DIMS = 32
+
+
+def _unit(vec):
+    n = sum(x * x for x in vec) ** 0.5 or 1.0
+    return [x / n for x in vec]
+
+
+def _query_vec():
+    return _unit([1.0] + [0.0] * (DIMS - 1))
+
+
+def _vec(cos):
+    return _unit([cos, math.sqrt(1 - cos * cos)] + [0.0] * (DIMS - 2))
+
+
+@pytest.fixture
+def config(tmp_path):
+    return MemoryConfig.from_dict({"vector_store": {
+        "provider": "chroma",
+        "collection_name": f"t_gates_{uuid.uuid4().hex[:8]}",
+        "persist_directory": str(tmp_path / "chroma"),
+        "embedding_dims": DIMS,
+    }})
+
+
+async def _reader_with(config, count):
+    from hy_memory.data.vector_store_chroma import ChromaVectorStore
+    from hy_memory.pipelines.reader_legacy import LegacyReadPipeline
+
+    store = ChromaVectorStore(config)
+    await store.initialize()
+    for index in range(count):
+        node = MemoryNode(
+            node_id=f"m{index:02d}", user_id="u1", agent_id="a1", session_id="s1",
+            content=f"memory number {index} about topic{index} and detail{index}",
+            layer=MemoryLayer.L2_FACT, status=MemoryStatus.ACTIVE, is_latest=True,
+        )
+        node.embedding = _vec(0.95 - index * 0.01)
+        await store.upsert(node)
+    reader = LegacyReadPipeline(config, vector_store=store)
+    await reader.initialize()
+    return reader, store
+
+
+def _request(limit, **kw):
+    return ReadRequest(
+        query="memory topic", query_embedding=_query_vec(),
+        user_ids=["u1"], agent_ids=["a1"],
+        limit=limit, min_score=0.2, schema_min_score=0.2, **kw,
+    )
+
+
+@pytest.fixture
+def mmr_spy(monkeypatch):
+    calls = []
+    real = reader_base.mmr_select
+
+    def _spy(hits, limit, **kw):
+        calls.append(limit)
+        return real(hits, limit, **kw)
+
+    monkeypatch.setattr(reader_base, "mmr_select", _spy)
+    return calls
+
+
+def test_mmr_threshold_is_defined_once():
+    assert MMR_MAX_TOP_K == 20
+    assert mmr_applies(19, 0.55)
+    assert not mmr_applies(20, 0.55)
+    # 权重仍由 MMR_LAMBDA 控制：关掉 λ 时任何 top-k 都不走 MMR
+    assert not mmr_applies(5, 0.0)
+    assert not mmr_applies(5, 1.0)
+
+
+def test_mmr_runs_when_limit_below_threshold(config, mmr_spy):
+    async def _run():
+        reader, store = await _reader_with(config, 12)
+        try:
+            resp = await reader.read(_request(5))
+            assert resp.success
+            assert resp.extra["config"]["mmr_applied"] is True
+            assert len(resp.memories) == 5
+        finally:
+            await store.close()
+
+    asyncio.run(_run())
+    assert mmr_spy == [5]
+
+
+def test_mmr_skipped_when_limit_reaches_threshold(config, mmr_spy):
+    async def _run():
+        reader, store = await _reader_with(config, MMR_MAX_TOP_K + 6)
+        try:
+            resp = await reader.read(_request(MMR_MAX_TOP_K))
+            assert resp.success
+            assert resp.extra["config"]["mmr_applied"] is False
+            ids = [m["memory_id"] for m in resp.memories]
+            assert len(ids) == MMR_MAX_TOP_K
+            # 无 MMR：按召回分（构造时单调递减）原序截断
+            assert ids == [f"m{index:02d}" for index in range(MMR_MAX_TOP_K)]
+        finally:
+            await store.close()
+
+    asyncio.run(_run())
+    assert mmr_spy == []
+
+
+class _FakeReranker:
+    model_name = "fake-cross-encoder"
+    device = "cpu"
+    load_error = ""
+
+    def __init__(self):
+        self.calls = 0
+
+    def score(self, query, documents):
+        self.calls += 1
+        # 反转召回序：分数与原名次相反
+        return [float(i) for i in range(len(documents))]
+
+
+@pytest.fixture
+def rerank_spies(config, monkeypatch):
+    config.recall.local_rerank_enabled = True
+    reranker = _FakeReranker()
+    factory_calls = []
+
+    def _get_local_reranker(*args, **kwargs):
+        factory_calls.append(kwargs)
+        return reranker
+
+    monkeypatch.setattr(reader_base, "get_local_reranker", _get_local_reranker)
+    graph_calls = []
+
+    async def _graph_rerank(self, *, hits, **kwargs):
+        graph_calls.append(len(hits))
+        return hits, [], {}
+
+    monkeypatch.setattr(
+        reader_base.BaseReadPipeline, "_apply_graph_rerank", _graph_rerank,
+    )
+    monkeypatch.setattr(
+        reader_base.BaseReadPipeline, "_expansion_on", lambda self: True,
+    )
+    return {"reranker": reranker, "factory": factory_calls, "graph": graph_calls}
+
+
+def test_rerank_false_skips_local_and_graph_rerank(config, rerank_spies):
+    async def _run():
+        reader, store = await _reader_with(config, 8)
+        try:
+            resp = await reader.read(_request(5, rerank=False))
+            assert resp.success
+            assert resp.extra["config"]["rerank"] is False
+            assert resp.extra["config"]["local_rerank_enabled"] is False
+            assert resp.extra["local_rerank"]["applied"] is False
+            return [m["memory_id"] for m in resp.memories]
+        finally:
+            await store.close()
+
+    ids = asyncio.run(_run())
+    assert rerank_spies["factory"] == []
+    assert rerank_spies["reranker"].calls == 0
+    assert rerank_spies["graph"] == []
+    assert len(ids) == 5
+
+
+def test_rerank_default_true_keeps_rerank_stages(config, rerank_spies):
+    async def _run():
+        reader, store = await _reader_with(config, 8)
+        try:
+            req = _request(5)
+            assert req.rerank is True
+            resp = await reader.read(req)
+            assert resp.success
+            assert resp.extra["config"]["rerank"] is True
+        finally:
+            await store.close()
+
+    asyncio.run(_run())
+    assert len(rerank_spies["factory"]) == 1
+    assert rerank_spies["reranker"].calls >= 1
+    assert len(rerank_spies["graph"]) == 1
+
+
+def test_rerank_is_per_request_under_concurrency(config, rerank_spies):
+    """并发的两次 read 各按自己的 rerank 走，不互相影响。"""
+    async def _run():
+        reader, store = await _reader_with(config, 8)
+        try:
+            off, on = await asyncio.gather(
+                reader.read(_request(5, rerank=False)),
+                reader.read(_request(5, rerank=True)),
+            )
+            assert off.extra["config"]["rerank"] is False
+            assert on.extra["config"]["rerank"] is True
+            assert off.extra["local_rerank"]["applied"] is False
+        finally:
+            await store.close()
+
+    asyncio.run(_run())
+    assert len(rerank_spies["factory"]) == 1
+    assert len(rerank_spies["graph"]) == 1
+
+
+@pytest.mark.parametrize("kwargs, expected", [({}, True), ({"rerank": False}, False)])
+def test_client_search_threads_rerank_into_read_request(kwargs, expected):
+    """client.async_search → ReadRequest.rerank；默认 True（原行为）。"""
+    from hy_memory.client import HyMemoryClient
+    from hy_memory.pipelines.base import ReadResponse
+
+    seen = {}
+
+    class _Reader:
+        VERSION = "fake"
+
+        async def read(self, request):
+            seen["rerank"] = request.rerank
+            seen["limit"] = request.limit
+            return ReadResponse()
+
+    class _Registry:
+        async def get_reader(self, reader_name=""):
+            return _Reader()
+
+    client = object.__new__(HyMemoryClient)
+    client._config = MemoryConfig.from_env()
+    client._cache = None
+    client._history_store = None
+    client._registry = _Registry()
+    client._mode = "pro"
+
+    out = asyncio.run(HyMemoryClient.async_search.__wrapped__(
+        client, "q", user_ids=["u"], limit=7, **kwargs,
+    ))
+    assert "memories" in out
+    assert seen == {"rerank": expected, "limit": 7}

tests/test_server_memory_namespace.py

@@ -0,0 +1,103 @@
+"""The memory routes report an ambiguous id as 409 and pass kind/variant on.
+
+An id can exist in more than one namespace (chat, experience, fact). The
+client fails closed with AmbiguousMemoryIdError; over HTTP that was a 500.
+"""
+
+import json
+import threading
+from http.server import ThreadingHTTPServer
+from urllib.request import Request, urlopen
+from urllib.error import HTTPError
+
+import pytest
+
+from hy_memory import server
+from hy_memory.retrieval import AmbiguousMemoryIdError
+
+
+class _Client:
+    def __init__(self):
+        self.calls = []
+
+    def get(self, memory_id, **kwargs):
+        self.calls.append(("get", memory_id, kwargs))
+        if not kwargs:
+            raise AmbiguousMemoryIdError(memory_id, ("chat", "working/experience"))
+        return {"memory_id": memory_id}
+
+    def update(self, memory_id, content, **kwargs):
+        self.calls.append(("update", memory_id, kwargs))
+        return {"success": True, "memory_id": memory_id}
+
+    def delete(self, memory_id, **kwargs):
+        self.calls.append(("delete", memory_id, kwargs))
+        if kwargs.get("kind") == "bogus":
+            raise ValueError("kind must be one of: chat, experience, fact")
+        return {"success": True, "deleted_count": 1}
+
+
+@pytest.fixture
+def running(monkeypatch):
+    client = _Client()
+    monkeypatch.setattr(server, "_get_client", lambda: client)
+    httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.MemoryHTTPHandler)
+    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield client, f"http://127.0.0.1:{httpd.server_address[1]}"
+    finally:
+        httpd.shutdown()
+        httpd.server_close()
+
+
+def _call(method, url, body=None):
+    data = json.dumps(body).encode() if body is not None else None
+    request = Request(url, data=data, method=method, headers={"Content-Type": "application/json"})
+    try:
+        with urlopen(request) as response:
+            return response.status, json.loads(response.read())
+    except HTTPError as error:
+        return error.code, json.loads(error.read())
+
+
+def test_an_ambiguous_id_is_409_not_500(running):
+    _, base = running
+
+    status, body = _call("GET", f"{base}/api/v1/memories/m1")
+
+    assert status == 409
+    assert body["error"] == "ambiguous_memory_id"
+
+
+def test_kind_and_variant_disambiguate_get_update_and_delete(running):
+    client, base = running
+
+    assert _call("GET", f"{base}/api/v1/memories/m1?kind=experience&variant=coding")[0] == 200
+    assert _call(
+        "PUT", f"{base}/api/v1/memories/m1",
+        {"content": "new", "kind": "experience", "variant": "working", "task": "T"},
+    )[0] == 200
+    assert _call("DELETE", f"{base}/api/v1/memories/m1?kind=fact&variant=coding")[0] == 200
+
+    assert client.calls == [
+        ("get", "m1", {"kind": "experience", "variant": "coding"}),
+        ("update", "m1", {"kind": "experience", "variant": "working", "task": "T"}),
+        ("delete", "m1", {"kind": "fact", "variant": "coding"}),
+    ]
+
+
+def test_a_bad_namespace_is_400(running):
+    _, base = running
+
+    status, _ = _call("DELETE", f"{base}/api/v1/memories/m1?kind=bogus")
+
+    assert status == 400
+
+
+def test_chat_calls_without_a_namespace_are_unchanged(running):
+    client, base = running
+
+    _call("PUT", f"{base}/api/v1/memories/m1", {"content": "new"})
+
+    assert client.calls == [("update", "m1", {})]