fix(sdk): keep model output out of the reconcile parse-failure warning

The final RECONCILE JSON parse failure warning embedded the raw response
whenever MEMORY_LOG_LLM_CONTENT was on. It now reports the length; the raw
output is already logged per attempt at DEBUG.

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

CHANGELOG.md

@@ -4,6 +4,14 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
+- MemHub shows when working and coding experiences were talked about. After
+  the SDK upgrade every one of them read "时间未知" (time unknown) because the
+  SDK never stored their conversation time; hy-memory 2.0.7 stores it and
+  Core now passes it through the memory list, detail, search and home
+  summary. An experience stored before 2.0.7 shows its write time, labelled
+  "写入于" / "写入时间", instead of "时间未知". Metadata repair also writes
+  the conversation time it recovers from the migration bundle into the SDK's
+  coding/working stores when they already have the column.
 - The SDK migration no longer fails its first run on a home with legacy
   vectors: reattaching them left a Chroma instance open with its own settings,
   and the readback that followed could not open the same store
@@ -68,6 +76,42 @@ Hy Memory and MemHub are versioned and released independently.
   `hy_node_permissions` proof table. Evidence, governed memories, model
   profiles, SDK stores, and source/host grants are preserved.
 
+## Hy Memory 2.0.8 — 2026-10-28
+
+- New `MEMORY_PIPELINE_LOG_ENABLED` (default `true`) turns the pipeline JSONL
+  log (`{MEMORY_LOG_DIR}/<subdir>/<date>.log`) off. With `false` no JSONL
+  file or directory is written. When `MEMORY_PIPELINE_TRACE_ENABLED` is also
+  `false`, `store_pipeline_log` becomes a no-op that returns `True` at once:
+  no timing, redaction, background task or database write. The init log line
+  now states whether the JSONL log is on or off.
+- Conversation and memory text no longer appears in INFO or WARNING logs. The
+  `[add]` input, search queries and rewrites, coding experience tasks, fact
+  statements and subjects, reconcile op details, extracted info, embedding
+  inputs and unparsable model output moved to DEBUG. INFO lines keep ids,
+  counts and character lengths; parse-failure warnings remain but report only
+  the length of the output, e.g. `failed to parse output (812 chars)`. The
+  `[embed]` INFO line now carries `request.input_chars` instead of
+  `request.input`. The `JSON_PARSE_FAILED` error message from extraction no
+  longer embeds the model output either.
+
+## Hy Memory 2.0.7 — 2026-10-28
+
+- Published
+  [`hy-memory==2.0.7`](https://pypi.org/project/hy-memory/2.0.7/).
+- Coding and working experiences and facts now keep their conversation time.
+  `add(..., memory_at=...)` reached only the chat path before; the
+  productivity path dropped it, so every experience had only its write time
+  and MemHub showed them as "时间未知" (time unknown). `coding_memory_meta`
+  and `working_fact` gain a nullable `memory_at` column (epoch seconds) in
+  both the working and coding stores; existing databases get it on open, and
+  their old rows stay `None`. Without `memory_at` a write uses the time the
+  request arrived, as chat does. When an experience is updated by a later
+  conversation it keeps the later of the two times.
+- `list_memories` (`scene="coding"`, `"working"` and `"all"`), `get` and
+  `search` return `memory_at` for experiences and facts in the same field and
+  format as chat (epoch seconds, `None` when unknown). Experience search
+  results now also carry `created_at`.
+
 ## Hy Memory 2.0.6 — 2026-10-27
 
 - Published

docs/env_reference.md

@@ -191,6 +191,7 @@ SDK 不设置 Rerank Search 并发上限。开启后模型在 `HyMemoryClient`
 | `MEMORY_LOG_FORMAT` | `text` | `text` / `json` |
 | `MEMORY_LOG_DIR` | — | 统一日志根目录 |
 | `MEMORY_TRACE_LOG_DIR` | — | JSONL 日志目录的兼容旧名；优先使用 `MEMORY_LOG_DIR` |
+| `MEMORY_PIPELINE_LOG_ENABLED` | `true` | 是否写 pipeline JSONL 日志（`{MEMORY_LOG_DIR}/<subdir>/<date>.log`）；`false` 完全不写 |
 | `MEMORY_PIPELINE_TRACE_ENABLED` | `true` | 是否写入 cache 的 pipeline trace 表 |
 | `MEMORY_MEMORY_OPERATIONS_ENABLED` | `false` | 是否写入 memory operation 审计记录 |
 | `MEMORY_HISTORY_ENABLE` | `true` | 是否启用历史记录 |

hy_memory/agent/extractor.py

@@ -1177,7 +1177,7 @@ class Extractor:
                         if not tc_id:
                             logger.error(
                                 f"[extractor] tool_call_id is empty! "
-                                f"raw tc: {json.dumps(tc_raw, ensure_ascii=False, default=str)[:500]}"
+                                f"raw_keys={list(tc_raw.keys())}"
                             )
                         try:
                             t_res = await tool_registry.dispatch(
@@ -1311,11 +1311,12 @@ class Extractor:
                 if parsed is None:
                     if not all_tool_calls:
                         logger.warning(
-                            f"[extractor] JSON parse failed: {raw_text[:200]}"
+                            f"[extractor] JSON parse failed ({len(raw_text)} chars)"
                         )
+                        logger.debug(f"[extractor] unparsable output: {raw_text[:200]!r}")
                         return ExtractResult(
                             success=False,
-                            error=f"JSON parse failed: {raw_text[:200]}",
+                            error=f"JSON parse failed ({len(raw_text)} chars)",
                             error_code="JSON_PARSE_FAILED",
                             raw_response=raw_text,
                             tokens_used=total_tokens,

hy_memory/agent/llm_provider.py

@@ -129,8 +129,7 @@ def _parse_tool_calls_list(raw_calls: list, _json) -> List[Dict[str, Any]]:
         tc_id = tc.get("id") or ""
         if not tc_id:
             logger.error(
-                f"[llm] tool_call missing 'id'! raw: "
-                f"{_json.dumps(tc, ensure_ascii=False, default=str)[:500]}"
+                f"[llm] tool_call missing 'id'! keys={list(tc.keys())}"
             )
 
         # 保留原始 tc 完整结构，只规范化 arguments

hy_memory/agent/mem_agent.py

@@ -234,7 +234,7 @@ class MemAgent:
                     f"identity={len(identity)} facts={len(facts)} "
                     f"basic_info={basic_field_count}fields"
                 )
-                logger.info(f"MemAgent: extracted_info={extract_result.info}")
+                logger.debug(f"MemAgent: extracted_info={extract_result.info}")
 
             # 3. 冲突检测（如果有已有记忆）
             if mode == ProcessMode.FULL and existing_memories:

hy_memory/agent/reconciler.py

@@ -50,7 +50,7 @@ def is_canonical_state_node(node: Any) -> bool:
 
 
 def _log_ops_detail(ops: List["ReconcileOp"]) -> None:
-    """以单条 INFO（JSON list）输出 reconcile 的逐条结果，便于线上 debug。"""
+    """以单条 DEBUG（JSON list）输出 reconcile 的逐条结果；含记忆正文，不进 INFO。"""
     if not ops:
         return
     view = [
@@ -68,7 +68,7 @@ def _log_ops_detail(ops: List["ReconcileOp"]) -> None:
         }
         for o in ops
     ]
-    logger.info(
+    logger.debug(
         f"[reconciler] ops detail: {json.dumps(view, ensure_ascii=False, default=str)}"
     )
 
@@ -1514,9 +1514,10 @@ class MemoryReconciler:
                 logger.warning(
                     f"[reconciler] RECONCILE JSON parse failed "
                     f"(attempt {_attempt+1}/{_MAX_PARSE_ATTEMPTS}), "
-                    f"retrying: raw="
-                    f"{raw_response[:200] if _log_llm_content_enabled() else '<redacted>'}"
+                    f"retrying: raw({len(raw_response or '')} chars)"
                 )
+                if _log_llm_content_enabled():
+                    logger.debug(f"[reconciler] unparsable raw={raw_response[:200]!r}")
 
             recon_elapsed = (_time.perf_counter() - t_recon) * 1000
 
@@ -1528,8 +1529,8 @@ class MemoryReconciler:
                     error_code = "LLM_UNAVAILABLE"
                 elif raw_response.strip():
                     error_msg = (
-                        f"RECONCILE JSON parse failed after {_n_attempts} attempts: "
-                        f"{raw_response[:200] if _log_llm_content_enabled() else '<redacted>'}"
+                        f"RECONCILE JSON parse failed after {_n_attempts} attempts "
+                        f"({len(raw_response)} chars)"
                     )
                     error_code = "JSON_PARSE_FAILED"
                 else:
@@ -1806,7 +1807,8 @@ class MemoryReconciler:
                 except Exception as _re:
                     logger.debug(f"[reconciler] json_repair failed: {_re}")
             if data is None:
-                logger.warning(f"[reconciler] failed to parse ops JSON: {text[:200]}")
+                logger.warning(f"[reconciler] failed to parse ops JSON ({len(text)} chars)")
+                logger.debug(f"[reconciler] unparsable ops output: {text[:200]!r}")
                 return []
 
         if not isinstance(data, list):
@@ -2018,11 +2020,11 @@ class MemoryReconciler:
                 if clash is not None:
                     logger.warning(
                         f"[reconciler] memory_id {clash} already touched by '{touched[clash]}', "
-                        f"current op '{op.op} ({op.reason[:60]})' dropped"
+                        f"current op '{op.op}' dropped"
                     )
                     continue
                 for t in target_ids:
-                    touched[t] = f"{op.op}({op.reason[:40]})"
+                    touched[t] = op.op
             validated.append(op)
 
         return validated

hy_memory/agent/tools/base.py

@@ -211,7 +211,9 @@ def parse_tool_calls_from_json(raw: Any) -> List[ToolCall]:
                 try:
                     args = json.loads(args_raw) if args_raw.strip() else {}
                 except json.JSONDecodeError:
-                    logger.warning(f"[tools] failed to parse arguments as JSON: {args_raw[:120]}")
+                    logger.warning(
+                        f"[tools] failed to parse arguments as JSON ({len(args_raw)} chars)"
+                    )
                     args = {}
             elif isinstance(args_raw, dict):
                 args = args_raw

hy_memory/client.py

@@ -744,7 +744,7 @@ class HyMemoryClient:
     def _install_pipeline_observability_hooks(self) -> None:
         """
         包装 cache.store_pipeline_log：
-        - Log：JSONL 文件，始终写入（环节级，不依赖 Trace 开关）
+        - Log：JSONL 文件，由 MEMORY_PIPELINE_LOG_ENABLED 控制（默认开启）
         - Trace：SQLite pipeline_logs，由 MEMORY_PIPELINE_TRACE_ENABLED 控制（Inspector）
         - 落盘 fire-and-forget，不阻塞 search/write 热路径
         """
@@ -958,7 +958,7 @@ class HyMemoryClient:
                 parsed_tool_calls = json.loads(raw_tool_calls) if raw_tool_calls else []
             except Exception:
                 logger.warning(
-                    "tool_calls 是非法 JSON 字符串，已忽略: %r", raw_tool_calls[:200]
+                    "tool_calls 是非法 JSON 字符串，已忽略 (%d chars)", len(raw_tool_calls)
                 )
                 parsed_tool_calls = []
             raw_tool_calls = parsed_tool_calls if isinstance(parsed_tool_calls, list) else []
@@ -1544,6 +1544,7 @@ class HyMemoryClient:
                 # LLM 写的"这个结论一路怎么改成现在这样"，直接进提示词；
                 # 没被覆盖过就是空串。不参与检索——召回认的是 statement 的当前值。
                 "fact_chain_summary": fact.fact_chain_summary,
+                "memory_at": _epoch_seconds(fact.memory_at),
                 "created_at": fact.created_at.isoformat() if fact.created_at else None,
                 "score": round(float(score), 4),
                 "source": fact.source,
@@ -1652,6 +1653,8 @@ class HyMemoryClient:
                 "score": (round(float(score_by_id[d.experience_id]), 4)
                           if d.experience_id in score_by_id else None),
                 "confidence": d.confidence,
+                "memory_at": _epoch_seconds(d.memory_at),
+                "created_at": d.created_at.isoformat() if d.created_at else None,
                 "updated_at": d.updated_at.isoformat() if d.updated_at else None,
                 "source": d.source,
                 "extraction_source": d.extraction_source,
@@ -1953,6 +1956,7 @@ class HyMemoryClient:
                   "content"    chat 正文；经验为 "task\nsolution"；事实为 statement
                   "score"      RRF 融合分；"lane_score" 为该路原始分；"lane_rank" 为该路名次
                   "source", "created_at", "updated_at"（秒级 epoch，取不到为 None）
+                  "memory_at"  对话时间（秒级 epoch），各路原有字段，三路同名同格式
                 chat 路内部名次：normal → schema → proactive 依次排；working/coding 路
                 内部名次：事实与经验各按原顺序交替（事实在前），两层谁也不压谁。
         """
@@ -2144,7 +2148,8 @@ class HyMemoryClient:
                           "limit", "offset", "request_id", "elapsed_ms"}
                 每条带 "scene"、"kind"（chat | experience）、"memory_id"、"content"
                 （经验为 "task\nsolution"）、"source"、"status"、"created_at"、
-                "updated_at"（秒级 epoch），以及该路原有字段。
+                "updated_at"（秒级 epoch），以及该路原有字段（含对话时间
+                "memory_at"，秒级 epoch，三路同名同格式）。
         """
         return self._loop_thread.run(
             self.async_list_memories(
@@ -2967,6 +2972,7 @@ class HyMemoryClient:
                     session_id=session_id,
                     request_id=request_id,
                     source=source,
+                    memory_at=memory_at,
                 )
                 # Public receipt always identifies the productivity namespace.
                 resp["variant"] = scene_variant
@@ -3002,13 +3008,13 @@ class HyMemoryClient:
                 f"[add] writing {preview} user={uid} "
                 f"agent_id={agent_id} session_id={session_id} mode={self._mode}"
             )
-            logger.info(f"[add] messages detail: {[m.to_dict() for m in chat_messages]}")
+            logger.debug(f"[add] messages detail: {[m.to_dict() for m in chat_messages]}")
         else:
             logger.info(
                 f"[add] writing text({len(content)} chars) user={uid} "
-                f"agent_id={agent_id} session_id={session_id} mode={self._mode} "
-                f"content={content}"
+                f"agent_id={agent_id} session_id={session_id} mode={self._mode}"
             )
+            logger.debug(f"[add] text content={content!r}")
 
         # mode → agent_mode 映射
         # lite: 不调 LLM agent
@@ -4028,14 +4034,14 @@ class HyMemoryClient:
             decided_variant = HyMemoryClient._scene_variant(scene)
             logger.info(
                 f"[search] scene explicit={decided_scene} (no LLM judge); "
-                f"target={effective_target_query!r}"
+                f"target({len(effective_target_query or '')} chars)"
             )
         elif not agentic_enabled and not queries_supplied:
             # 默认 light search 仅指 query="..."：固定 normal，绝不调用 LLM。
             decided_scene = "normal"
             logger.info(
                 f"[search] search_mode=default single-query scene=normal (no LLM); "
-                f"target={target_query!r}"
+                f"target({len(target_query or '')} chars)"
             )
         elif agentic_enabled:
             # agentic：单/多 query 都强制一次 LLM 完成三分类 + rewrite + 时间解析。
@@ -4070,14 +4076,18 @@ class HyMemoryClient:
                 rewrite_created_after = ar.get("created_after")
                 logger.info(
                     f"[search] agentic scene={decided_scene} via rewrite-LLM "
-                    f"(classify={want_classify}); orig={target_query!r} "
-                    f"effective={effective_target_query!r} rewrite_used={rewrite_used} "
+                    f"(classify={want_classify}); rewrite_used={rewrite_used} "
                     f"created_after={rewrite_created_after}"
                 )
+                logger.debug(
+                    f"[search] agentic rewrite: orig={target_query!r} "
+                    f"effective={effective_target_query!r}"
+                )
             else:
                 decided_scene = "normal"
                 logger.info(
-                    f"[search] agentic rewrite-LLM fallback=normal; target={target_query!r}"
+                    f"[search] agentic rewrite-LLM fallback=normal; "
+                    f"target({len(target_query or '')} chars)"
                 )
         elif rewrite_enabled and not _too_long:
             # default + queries= 的原逻辑：配置开启时一次 LLM 做 rewrite/time，
@@ -4156,7 +4166,7 @@ class HyMemoryClient:
             decided_scene = "normal"
             logger.info(
                 f"[search] default multi-query features disabled; no LLM; "
-                f"target={target_query!r}"
+                f"target({len(target_query or '')} chars)"
             )
 
         # ────────────────────────────────────────────────────────────
@@ -4209,7 +4219,7 @@ class HyMemoryClient:
             else:
                 logger.info(
                     f"[search] routing to PRODUCTIVITY path variant={decided_variant}: "
-                    f"target={effective_target_query!r} "
+                    f"target({len(effective_target_query or '')} chars) "
                     f"workspace_id={workspace_id!r} branch={branch!r}"
                 )
                 return await self._search_coding(
@@ -4237,9 +4247,11 @@ class HyMemoryClient:
         # Chat 召回链路（原有逻辑）
         # ────────────────────────────────────────────────────────────
         logger.info(
-            f"[search] query='{effective_target_query}' user_ids={effective_user_ids} "
+            f"[search] query({len(effective_target_query or '')} chars) "
+            f"user_ids={effective_user_ids} "
             f"agent_ids={effective_agent_ids} session_ids={effective_session_ids} limit={limit}"
         )
+        logger.debug(f"[search] query={effective_target_query!r}")
 
         # rewrite 抽出的时间下界与用户显式 created_after 取更晚者（更严格）
         effective_created_after = created_after
@@ -5920,6 +5932,7 @@ class HyMemoryClient:
                 "source_fact_ids": list(d.source_fact_ids),
                 "confidence": d.confidence,
                 "status": MemoryStatus.ACTIVE.value,
+                "memory_at": _epoch_seconds(d.memory_at),
                 "created_at": d.created_at.isoformat() if d.created_at else None,
                 "updated_at": d.updated_at.isoformat() if d.updated_at else None,
                 "source": d.source,

hy_memory/coding/extractor.py

@@ -1166,7 +1166,10 @@ class CodingMemoryExtractor:
                     obj = None
 
         if not isinstance(obj, dict):
-            logger.warning(f"[coding-extract] failed to parse output: {text!r}")
+            logger.warning(
+                f"[coding-extract] failed to parse output ({len(text or '')} chars)"
+            )
+            logger.debug(f"[coding-extract] unparsable output: {text!r}")
             return [], []
 
         # "memories" 是双层化之前的 key，老 prompt 缓存 / 旧模型可能还这么回

hy_memory/coding/fact_reconciler.py

@@ -454,9 +454,10 @@ class WorkingFactReconciler:
         workspace_id: Optional[str] = None,
         branch: Optional[str] = None,
         source: str = "default",
+        memory_at: Optional[datetime] = None,
     ) -> Tuple[List[WorkingFact], List[FactReconcileOp]]:
         """
-        执行 reconcile + 持久化。
+        执行 reconcile + 持久化。memory_at 是这段对话的时间，记到新事实上。
 
         Returns:
             (真正落库的新事实, 执行过的 ops)。LLM 挂掉时退化为全 INSERT 不覆盖，
@@ -493,6 +494,7 @@ class WorkingFactReconciler:
             workspace_id=workspace_id,
             branch=branch,
             source=source,
+            memory_at=memory_at,
         )
         logger.info(
             f"[fact-reconcile] drafts={len(drafts)} inserted={len(facts)} "
@@ -568,9 +570,12 @@ class WorkingFactReconciler:
                     kept.append(fact)
             if blocked:
                 logger.info(
-                    f"[fact-reconcile] subject gate blocked {len(blocked)} candidate(s) "
-                    f"for subject={draft.subject!r}: "
-                    + ", ".join(f"{f.fact_id}({f.subject!r})" for f in blocked)
+                    f"[fact-reconcile] subject gate blocked {len(blocked)} candidate(s): "
+                    + ", ".join(f.fact_id for f in blocked)
+                )
+                logger.debug(
+                    f"[fact-reconcile] subject gate: draft subject={draft.subject!r} "
+                    f"blocked subjects={[f.subject for f in blocked]!r}"
                 )
             result.append(kept)
         return result
@@ -743,7 +748,10 @@ class WorkingFactReconciler:
                     obj = None
 
         if not isinstance(obj, dict):
-            logger.warning(f"[fact-reconcile] failed to parse ops output: {text!r}")
+            logger.warning(
+                f"[fact-reconcile] failed to parse ops output ({len(text or '')} chars)"
+            )
+            logger.debug(f"[fact-reconcile] unparsable ops output: {text!r}")
             return {}
         ops_raw = obj.get("ops")
         if not isinstance(ops_raw, list):
@@ -781,6 +789,7 @@ class WorkingFactReconciler:
         workspace_id: Optional[str],
         branch: Optional[str] = None,
         source: str = "default",
+        memory_at: Optional[datetime] = None,
     ) -> Tuple[List[WorkingFact], List[FactReconcileOp]]:
         inserted: List[WorkingFact] = []
         ops: List[FactReconcileOp] = []
@@ -801,6 +810,7 @@ class WorkingFactReconciler:
                 subject=draft.subject,
                 details=draft.details,
                 status="active",
+                memory_at=memory_at,
             )
 
             allowed = {c.fact_id for c in (candidates[idx] if idx < len(candidates) else [])}
@@ -831,8 +841,11 @@ class WorkingFactReconciler:
                 # 别的主体的状态不是。闸门关掉时这道复查一并跳过，否则闸门就没关干净。
                 if self.subject_gate and _subject_conflict(draft.subject, old.subject):
                     logger.warning(
-                        f"[fact-reconcile] refused cross-subject SUPERSEDE: "
-                        f"{old.fact_id}({old.subject!r}) by draft subject {draft.subject!r}"
+                        f"[fact-reconcile] refused cross-subject SUPERSEDE: {old.fact_id}"
+                    )
+                    logger.debug(
+                        f"[fact-reconcile] cross-subject: old subject={old.subject!r} "
+                        f"draft subject={draft.subject!r}"
                     )
                     continue
                 try:

hy_memory/coding/judge.py

@@ -195,8 +195,10 @@ async def classify_messages_scene(
         result = _parse_scene_json(resp.content)
         if result is None:
             logger.warning(
-                f"[coding-judge] failed to parse classify_messages output: {resp.content!r}; defaulting to chat"
+                f"[coding-judge] failed to parse classify_messages output "
+                f"({len(resp.content or '')} chars); defaulting to chat"
             )
+            logger.debug(f"[coding-judge] unparsable classify_messages output: {resp.content!r}")
             await _store_coding_log(
                 cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
                 session_id=session_id, step="CODING_JUDGE", prompt=prompt,
@@ -207,9 +209,8 @@ async def classify_messages_scene(
             return "chat"
         scene = result["scene"]
         parsed = {**result, "is_coding": scene != "chat"}
-        logger.info(
-            f"[coding-judge] scene={scene} reason={result.get('reason', '')!r}"
-        )
+        logger.info(f"[coding-judge] scene={scene}")
+        logger.debug(f"[coding-judge] scene={scene} reason={result.get('reason', '')!r}")
         await _store_coding_log(
             cache, request_id=request_id, user_id=user_id, agent_id=agent_id,
             session_id=session_id, step="CODING_JUDGE", prompt=prompt,
@@ -295,8 +296,10 @@ async def classify_queries_scene(
         result = _parse_scene_json(resp.content)
         if result is None:
             logger.warning(
-                f"[coding-judge] failed to parse classify_queries output: {resp.content!r}; defaulting to chat"
+                f"[coding-judge] failed to parse classify_queries output "
+                f"({len(resp.content or '')} chars); defaulting to chat"
             )
+            logger.debug(f"[coding-judge] unparsable classify_queries output: {resp.content!r}")
             return "chat"
         scene = result["scene"]
         logger.info(f"[coding-judge] queries scene={scene}")
@@ -422,15 +425,18 @@ async def classify_and_rewrite_queries(
     scene = _parse_scene_label(head)
     if scene is None:
         logger.warning(
-            f"[coding-judge] classify_and_rewrite unrecognized head={head!r}; falling back to chat"
+            f"[coding-judge] classify_and_rewrite unrecognized head "
+            f"({len(head)} chars); falling back to chat"
         )
+        logger.debug(f"[coding-judge] classify_and_rewrite unrecognized head={head!r}")
         return {"scene": "chat", "is_coding": False, "rewrite_query": rewrite, "ok": False}
     is_coding = scene != "chat"
 
     logger.info(
         f"[coding-judge] classify_and_rewrite scene={scene} "
-        f"rewrite={rewrite!r}"
+        f"rewrite({len(rewrite)} chars)"
     )
+    logger.debug(f"[coding-judge] classify_and_rewrite rewrite={rewrite!r}")
     return {"scene": scene, "is_coding": is_coding, "rewrite_query": rewrite, "ok": True}
 
 
@@ -569,8 +575,9 @@ async def analyze_queries(
         parsed_scene = _parse_scene_label(head)
         if parsed_scene is None:
             logger.warning(
-                f"[coding-judge] analyze_queries unrecognized head={head!r}; ok=False"
+                f"[coding-judge] analyze_queries unrecognized head ({len(head)} chars); ok=False"
             )
+            logger.debug(f"[coding-judge] analyze_queries unrecognized head={head!r}")
             return {**fallback, "since_raw": ""}
         scene = parsed_scene
         idx = 1
@@ -589,9 +596,10 @@ async def analyze_queries(
     created_after = _since_to_created_after(since_raw, current_time)
 
     logger.info(
-        f"[coding-judge] analyze scene={scene} rewrite={rewrite!r} "
+        f"[coding-judge] analyze scene={scene} rewrite({len(rewrite)} chars) "
         f"since={since_raw!r} created_after={created_after}"
     )
+    logger.debug(f"[coding-judge] analyze rewrite={rewrite!r}")
     return {
         "scene": scene, "is_coding": scene != "chat", "rewrite_query": rewrite,
         "created_after": created_after, "since_raw": since_raw, "ok": True,
@@ -614,7 +622,10 @@ def _since_to_created_after(since_raw: str, current_time: str) -> Optional[float
             return dt.timestamp()
         except (ValueError, TypeError):
             continue
-    logger.warning(f"[coding-judge] analyze_queries unparsable SINCE={since_raw!r}")
+    logger.warning(
+        f"[coding-judge] analyze_queries unparsable SINCE ({len(since_raw)} chars)"
+    )
+    logger.debug(f"[coding-judge] analyze_queries unparsable SINCE={since_raw!r}")
     return None
 
 

hy_memory/coding/reconciler.py

@@ -388,7 +388,10 @@ class CodingMemoryReconciler:
                     obj = None
 
         if not obj or not isinstance(obj, dict):
-            logger.warning(f"[coding-reconcile] failed to parse ops output: {text!r}")
+            logger.warning(
+                f"[coding-reconcile] failed to parse ops output ({len(text or '')} chars)"
+            )
+            logger.debug(f"[coding-reconcile] unparsable ops output: {text!r}")
             return []
         ops_raw = obj.get("ops") or []
         if not isinstance(ops_raw, list):
@@ -441,10 +444,10 @@ class CodingMemoryReconciler:
         for op in ops:
             if op.action == "ADD":
                 if op.draft_idx is None or not (0 <= op.draft_idx < len(drafts)):
-                    logger.warning(f"[coding-reconcile] drop invalid ADD: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop invalid ADD: {_op_ref(op)}")
                     continue
                 if op.draft_idx in consumed:
-                    logger.warning(f"[coding-reconcile] drop ADD on consumed draft: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop ADD on consumed draft: {_op_ref(op)}")
                     continue
                 d = drafts[op.draft_idx]
                 memory = self._draft_to_new_memory(d)
@@ -458,18 +461,18 @@ class CodingMemoryReconciler:
 
             elif op.action == "UPDATE":
                 if op.draft_idx is None or not (0 <= op.draft_idx < len(drafts)):
-                    logger.warning(f"[coding-reconcile] drop invalid UPDATE: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop invalid UPDATE: {_op_ref(op)}")
                     continue
                 if not op.target_experience_id:
-                    logger.warning(f"[coding-reconcile] drop UPDATE missing target: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop UPDATE missing target: {_op_ref(op)}")
                     continue
                 if not self._target_was_offered(op, candidates_per_draft):
                     logger.warning(
-                        f"[coding-reconcile] drop UPDATE outside authority scope: {op.to_dict()}"
+                        f"[coding-reconcile] drop UPDATE outside authority scope: {_op_ref(op)}"
                     )
                     continue
                 if op.draft_idx in consumed:
-                    logger.warning(f"[coding-reconcile] drop UPDATE on consumed draft: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop UPDATE on consumed draft: {_op_ref(op)}")
                     continue
                 d = drafts[op.draft_idx]
                 memory = self._draft_to_new_memory(d, experience_id=op.target_experience_id)
@@ -488,17 +491,17 @@ class CodingMemoryReconciler:
 
             elif op.action == "DELETE":
                 if not op.target_experience_id:
-                    logger.warning(f"[coding-reconcile] drop DELETE missing target: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop DELETE missing target: {_op_ref(op)}")
                     continue
                 if not self._target_was_offered(op, candidates_per_draft):
                     logger.warning(
-                        f"[coding-reconcile] drop DELETE outside authority scope: {op.to_dict()}"
+                        f"[coding-reconcile] drop DELETE outside authority scope: {_op_ref(op)}"
                     )
                     continue
                 # DELETE 需要知道 user_id；通过 draft_idx（如有）或第一个 draft 推断
                 user_id = self._infer_user_id(op, drafts)
                 if not user_id:
-                    logger.warning(f"[coding-reconcile] drop DELETE no user_id context: {op.to_dict()}")
+                    logger.warning(f"[coding-reconcile] drop DELETE no user_id context: {_op_ref(op)}")
                     continue
                 try:
                     ok = await self.store.delete(op.target_experience_id, user_id=user_id)
@@ -577,6 +580,7 @@ class CodingMemoryReconciler:
             confidence=d.confidence,
             source=d.source,
             extraction_source=d.extraction_source,
+            memory_at=d.memory_at,
         )
 
 
@@ -584,6 +588,14 @@ class CodingMemoryReconciler:
 # Helpers
 # ================================================================
 
+def _op_ref(op: ReconcileOp) -> str:
+    """op 的日志引用：只带动作与编号，不带 LLM 写的 reason。"""
+    return (
+        f"action={op.action} draft_idx={op.draft_idx} "
+        f"target={op.target_experience_id}"
+    )
+
+
 def _truncate(s: str, n: int) -> str:
     if not s:
         return ""

hy_memory/coding/store.py

@@ -86,6 +86,7 @@ CREATE TABLE IF NOT EXISTS coding_memory_meta (
     confidence       REAL NOT NULL DEFAULT 0.7,
     source           TEXT NOT NULL DEFAULT 'default',
     extraction_source TEXT NOT NULL DEFAULT 'auto_extract',
+    memory_at        INTEGER,
     created_at       TEXT NOT NULL,
     updated_at       TEXT NOT NULL
 );
@@ -109,6 +110,7 @@ CREATE TABLE IF NOT EXISTS working_fact (
     derived_experience_ids TEXT NOT NULL DEFAULT '[]',
     status                 TEXT NOT NULL DEFAULT 'active',
     fact_chain_summary     TEXT NOT NULL DEFAULT '',
+    memory_at              INTEGER,
     created_at             TEXT NOT NULL
 );
 -- 反查前驱（谁的 fact_link 指向我）走这个索引，写演变总结时要顺着链往回走
@@ -164,6 +166,8 @@ _ADDED_COLUMNS = {
     "coding_memory_meta": {
         "source_fact_ids": "TEXT NOT NULL DEFAULT '[]'",
         "extraction_source": "TEXT NOT NULL DEFAULT 'auto_extract'",
+        # 对话时间，秒级 epoch（同 chat）。老库里的行是 NULL：那时没存过。
+        "memory_at": "INTEGER",
     },
     # 早期版本存过一列 fact_chain（机械累积的历代说法），已废弃：同样的信息顺着
     # fact_link 反查就有。老库里那一列留着不管，有默认值不影响插入。
@@ -177,6 +181,7 @@ _ADDED_COLUMNS = {
         "subject": "TEXT NOT NULL DEFAULT ''",
         "details": "TEXT NOT NULL DEFAULT ''",
         "source": "TEXT NOT NULL DEFAULT 'default'",
+        "memory_at": "INTEGER",
     },
 }
 
@@ -290,6 +295,29 @@ def _parse_dt(s: Optional[str]) -> Optional[datetime]:
         return None
 
 
+def _from_epoch(value: Any) -> Optional[datetime]:
+    """memory_at 列（秒级 epoch）→ datetime；NULL 或坏值返回 None。"""
+    if value is None:
+        return None
+    try:
+        return datetime.fromtimestamp(int(value))
+    except Exception:
+        return None
+
+
+def _to_epoch(value: Optional[datetime]) -> Optional[int]:
+    return int(value.timestamp()) if isinstance(value, datetime) else None
+
+
+def _later(a: Optional[datetime], b: Optional[datetime]) -> Optional[datetime]:
+    """两个对话时间取较晚的；一边缺就取另一边。"""
+    if a is None:
+        return b
+    if b is None:
+        return a
+    return a if _to_epoch(a) >= _to_epoch(b) else b
+
+
 def _row_to_memory(row: sqlite3.Row) -> CodingMemory:
     """SQLite row → CodingMemory"""
     return CodingMemory(
@@ -308,6 +336,7 @@ def _row_to_memory(row: sqlite3.Row) -> CodingMemory:
         confidence=row["confidence"],
         source=row["source"],
         extraction_source=row["extraction_source"],
+        memory_at=_from_epoch(row["memory_at"]),
         created_at=_parse_dt(row["created_at"]),
         updated_at=_parse_dt(row["updated_at"]),
     )
@@ -334,6 +363,7 @@ def _row_to_fact(row: sqlite3.Row) -> WorkingFact:
         # 老库刚 ADD COLUMN 时可能取不到这一列，用 keys() 兜一下
         fact_chain_summary=(row["fact_chain_summary"] or ""
                             if "fact_chain_summary" in row.keys() else ""),
+        memory_at=_from_epoch(row["memory_at"]),
         created_at=_parse_dt(row["created_at"]),
     )
 
@@ -575,7 +605,11 @@ class CodingMemoryStore:
         await self._upsert_vec_keys(memory)
         logger.info(
             f"[coding-store] inserted experience_id={memory.experience_id} "
-            f"task={memory.task!r} keys={1 + len(memory.search_keys)}"
+            f"keys={1 + len(memory.search_keys)}"
+        )
+        logger.debug(
+            f"[coding-store] inserted experience_id={memory.experience_id} "
+            f"task={memory.task!r}"
         )
 
     async def update(self, memory: CodingMemory) -> None:
@@ -597,6 +631,8 @@ class CodingMemoryStore:
         memory.updated_at = datetime.utcnow()
         if memory.created_at is None:
             memory.created_at = prev.created_at
+        # 合并后的经验讲的是两段对话，对话时间取较晚的那段
+        memory.memory_at = _later(prev.memory_at, memory.memory_at)
         await _run_in_pool(self._upsert_sql_sync, memory)
         # 删旧 VDB
         await self._delete_vec_keys_by_count(memory.experience_id, prev_key_count)
@@ -970,7 +1006,8 @@ class CodingMemoryStore:
             fact.created_at = datetime.utcnow()
         await _run_in_pool(self._upsert_fact_sql_sync, fact)
         await self._upsert_fact_vec(fact)
-        logger.info(
+        logger.info(f"[coding-store] inserted fact_id={fact.fact_id}")
+        logger.debug(
             f"[coding-store] inserted fact_id={fact.fact_id} "
             f"statement={fact.statement!r}"
         )
@@ -1199,8 +1236,8 @@ class CodingMemoryStore:
                     experience_id, user_id, agent_id, task, search_keys, solution,
                     boundary_envs, workspace_id, branch, session_id,
                     files, source_fact_ids, confidence, source, extraction_source,
-                    created_at, updated_at
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    memory_at, created_at, updated_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(experience_id) DO UPDATE SET
                     user_id         = excluded.user_id,
                     agent_id        = excluded.agent_id,
@@ -1216,6 +1253,7 @@ class CodingMemoryStore:
                     confidence      = excluded.confidence,
                     source          = excluded.source,
                     extraction_source = excluded.extraction_source,
+                    memory_at       = excluded.memory_at,
                     updated_at      = excluded.updated_at
                 """,
                 (
@@ -1226,6 +1264,7 @@ class CodingMemoryStore:
                     json.dumps(memory.files, ensure_ascii=False),
                     json.dumps(memory.source_fact_ids, ensure_ascii=False),
                     memory.confidence, memory.source, memory.extraction_source,
+                    _to_epoch(memory.memory_at),
                     (memory.created_at or datetime.utcnow()).isoformat(),
                     (memory.updated_at or datetime.utcnow()).isoformat(),
                 ),
@@ -1366,8 +1405,8 @@ class CodingMemoryStore:
                     fact_id, user_id, agent_id, session_id, workspace_id, branch,
                     source, statement, files, subject, details,
                     fact_link, derived_experience_ids,
-                    status, fact_chain_summary, created_at
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    status, fact_chain_summary, memory_at, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                 ON CONFLICT(fact_id) DO UPDATE SET
                     user_id                = excluded.user_id,
                     agent_id               = excluded.agent_id,
@@ -1382,7 +1421,8 @@ class CodingMemoryStore:
                     fact_link              = excluded.fact_link,
                     derived_experience_ids = excluded.derived_experience_ids,
                     status                 = excluded.status,
-                    fact_chain_summary     = excluded.fact_chain_summary
+                    fact_chain_summary     = excluded.fact_chain_summary,
+                    memory_at              = excluded.memory_at
                 """,
                 (
                     fact.fact_id, fact.user_id, fact.agent_id,
@@ -1395,6 +1435,7 @@ class CodingMemoryStore:
                     json.dumps(fact.derived_experience_ids, ensure_ascii=False),
                     fact.status,
                     fact.fact_chain_summary or "",
+                    _to_epoch(fact.memory_at),
                     (fact.created_at or datetime.utcnow()).isoformat(),
                 ),
             )

hy_memory/coding/types.py

@@ -25,6 +25,11 @@ FactStatus = Literal["active", "superseded"]
 MAX_FACT_CHAIN_ENTRIES = 8
 
 
+def _epoch(value: Optional[datetime]) -> Optional[int]:
+    """memory_at 对外是秒级 epoch，与 chat 一致。"""
+    return int(value.timestamp()) if isinstance(value, datetime) else None
+
+
 # ================================================================
 # 草稿（LLM extract 阶段输出）
 # ================================================================
@@ -53,6 +58,8 @@ class CodingMemoryDraft:
     files: List[str] = field(default_factory=list)
     source: str = "default"
     extraction_source: str = "auto_extract"
+    # 对话时间（SDK 注入，同 chat 的 memory_at）
+    memory_at: Optional[datetime] = None
 
     def __post_init__(self) -> None:
         from ..utils.source_filter import normalize_source
@@ -73,6 +80,7 @@ class CodingMemoryDraft:
             "files": list(self.files),
             "source": self.source,
             "extraction_source": self.extraction_source,
+            "memory_at": _epoch(self.memory_at),
         }
 
 
@@ -111,6 +119,9 @@ class CodingMemory:
     extraction_source: str = "auto_extract"
 
     # 时间
+    # memory_at 是对话时间（同 chat），created_at / updated_at 是写库时间。
+    # UPDATE 合并时取两者中较晚的那个对话时间，见 store.update。
+    memory_at: Optional[datetime] = None
     created_at: Optional[datetime] = None
     updated_at: Optional[datetime] = None
 
@@ -135,6 +146,7 @@ class CodingMemory:
             "confidence": self.confidence,
             "source": self.source,
             "extraction_source": self.extraction_source,
+            "memory_at": _epoch(self.memory_at),
             "created_at": self.created_at.isoformat() if self.created_at else None,
             "updated_at": self.updated_at.isoformat() if self.updated_at else None,
         }
@@ -249,6 +261,7 @@ class WorkingFact:
     fact_chain_summary: str = ""
 
     # 时间
+    memory_at: Optional[datetime] = None   # 对话时间（同 chat），没传时是收到请求的时间
     created_at: Optional[datetime] = None  # 产生时间，新的优先
 
     def __post_init__(self) -> None:
@@ -272,6 +285,7 @@ class WorkingFact:
             "derived_experience_ids": list(self.derived_experience_ids),
             "status": self.status,
             "fact_chain_summary": self.fact_chain_summary,
+            "memory_at": _epoch(self.memory_at),
             "created_at": self.created_at.isoformat() if self.created_at else None,
         }
 

hy_memory/coding/writer.py

@@ -18,6 +18,7 @@ CodingWriter - 写入端协调器
 import logging
 import time
 import uuid
+from datetime import datetime
 from typing import Any, Dict, List, Optional
 
 from ..agent.llm_provider import LLMProvider
@@ -79,9 +80,11 @@ class CodingWriter:
         session_id: Optional[str] = None,
         request_id: Optional[str] = None,
         source: str = "default",
+        memory_at: Optional[datetime] = None,
     ) -> Dict[str, Any]:
         """
-        working/coding 段写入端到端。返回响应 dict（兼容 client.add 返回结构）：
+        working/coding 段写入端到端。memory_at 是对话时间，记到这次写出的经验和
+        事实上；不传则用收到请求的时间（与 chat 链路一致）。返回响应 dict（兼容 client.add 返回结构）：
             {
                 "success": True,
                 "scene": "productivity",
@@ -101,6 +104,8 @@ class CodingWriter:
         """
         request_id = request_id or str(uuid.uuid4())
         t0 = time.perf_counter()
+        if memory_at is None:
+            memory_at = datetime.now()
 
         if not messages:
             return self._empty_response(request_id, t0)
@@ -172,6 +177,7 @@ class CodingWriter:
         # extractor 只负责内容；调用来源由 SDK 入口注入，不能交给 LLM 生成。
         for draft in drafts:
             draft.source = source
+            draft.memory_at = memory_at
 
         # 3. 经验层 reconcile + 持久化
         ops: List[ReconcileOp] = (
@@ -197,6 +203,7 @@ class CodingWriter:
                 workspace_id=workspace_id,
                 branch=branch,
                 source=source,
+                memory_at=memory_at,
             )
 
         experience_ids = [op.target_experience_id for op in ops

hy_memory/core/embed_service.py

@@ -80,7 +80,7 @@ def _count_tokens(text: str) -> int:
     try:
         import tiktoken
 
-        return len(tiktoken.get_encoding("cl100k_base").encode(text))
+        return len(tiktoken.get_encoding("cl100k_base").encode(text, disallowed_special=()))
     except Exception:
         return len(text) // _CHARS_PER_TOKEN_FALLBACK + 1
 
@@ -98,7 +98,8 @@ def _truncate_for_embedding(text: str, limit: int = _EMBED_MAX_TOKENS) -> str:
         import tiktoken
 
         enc = tiktoken.get_encoding("cl100k_base")
-        ids = enc.encode(text)
+        # Input is ordinary user text, not a tokenizer control-token stream.
+        ids = enc.encode(text, disallowed_special=())
         if len(ids) <= limit:
             return text
         dropped = len(ids) - limit
@@ -753,16 +754,22 @@ class EmbedService:
         """
         打一条 INFO 级 embedding 请求日志（含 request_id / provider / model / api / 入参 / 出参）。
 
-        出参只含 vector 维度，不含向量数值。request_id 同时写入 message，
+        出参只含 vector 维度，不含向量数值。入参的 input 是待向量化的原文
+        （对话 / 记忆 / 查询），INFO 只记每条的字符数（input_chars），原文仅在
+        DEBUG 单独输出。request_id 同时写入 message，
         并依赖 log formatter 的 %(request_id)s（setup_logging 后生效）。
         """
+        texts = request_payload.get("input")
+        info_request = {k: v for k, v in request_payload.items() if k != "input"}
+        if isinstance(texts, list):
+            info_request["input_chars"] = [len(str(t)) for t in texts]
         payload = {
             "request_id": get_request_id(),
             "status": status,
             "provider": self._embed_config.provider,
             "model": self._embed_config.model,
             "api": api_url,
-            "request": request_payload,
+            "request": info_request,
             "response": response_payload,
         }
         try:
@@ -770,6 +777,10 @@ class EmbedService:
         except Exception:
             body = str(payload)
         logger.info(f"[embed] {body}")
+        if texts is not None and logger.isEnabledFor(logging.DEBUG):
+            logger.debug(
+                f"[embed] input: {json.dumps(texts, ensure_ascii=False, default=str)}"
+            )
 
     async def _embed_openai(
         self, texts: List[str]

hy_memory/pipelines/_retrieval/reconcile_retrieval.py

@@ -115,11 +115,16 @@ class ReconcileHybridRetriever:
         fused_max = max(fused_all) if fused_all else 0.0
 
         base = (
-            f"[reconcile-hybrid][{req}] pool={pool_size} q_terms={q_terms!r} "
+            f"[reconcile-hybrid][{req}] pool={pool_size} q_terms={len(q_terms)} "
             f"raw_bm25 max={raw_max:.4f} nonzero={raw_nonzero}/{len(raw_bm25)} "
             f"bm25_norm max={norm_max:.4f} nonzero={norm_nonzero}/{len(bm25_norms)} "
             f"sem_max={sem_max:.4f} fused_max={fused_max:.4f} "
             f"kept={kept}/{len(fused_all)} min_score={self._cfg.min_score} "
+            f"query({len(query_text or '')} chars)"
+        )
+        # query 是待对账的新记忆原文，词项与预览只进 DEBUG
+        logger.debug(
+            f"[reconcile-hybrid][{req}] q_terms={q_terms!r} "
             f"query={q_preview!r} lemmatized={lem_preview!r}"
         )
 

hy_memory/pipelines/cross_domain_sweeper.py

@@ -692,9 +692,10 @@ class CrossDomainSweeper:
                 )
 
             logger.info(
-                f"[sweeper] Created L6 core {core_id[:12]}: '{core_content[:60]}' "
+                f"[sweeper] Created L6 core {core_id[:12]} "
                 f"from {len(cluster_ids)} basics, evidence_inherited={len(inherited_evidence_ids)}"
             )
+            logger.debug(f"[sweeper] L6 core {core_id[:12]} content={core_content[:60]!r}")
             return 1, 0, [core_id]
 
         except Exception as e:

hy_memory/pipelines/reader_base.py

@@ -3587,8 +3587,9 @@ class BaseReadPipeline(ReadPipeline):
                 f"hidden_l0={len(hidden_l0_hits)} fwd_l6={len(forward_l6)} rev_l6={len(reverse_l6)} "
                 f"fused_l6={len(fused_l6)} graph_intent={len(graph_intention_hits)} "
                 f"returned={len(final_results)} "
-                f"query='{request.query[:80]}'"
+                f"query({len(request.query or '')} chars)"
             )
+            logger.debug(f"[read/{self.VERSION}] query={request.query[:80]!r}")
 
             response.success = True
 

hy_memory/pipelines/system2_agent.py

@@ -2516,7 +2516,8 @@ def _parse_operations_json(text: str) -> Optional[List[Dict[str, Any]]]:
             return result
         return None
     except json.JSONDecodeError:
-        logger.warning(f"[S2-agent] JSON parse failed: {json_str[:200]}")
+        logger.warning(f"[S2-agent] JSON parse failed ({len(json_str)} chars)")
+        logger.debug(f"[S2-agent] unparsable output: {json_str[:200]!r}")
         return None
 
 

hy_memory/utils/async_executor.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import concurrent.futures
+import contextvars
 import functools
 from concurrent.futures import Executor
 from typing import Any, Callable, Optional
@@ -39,7 +40,9 @@ async def run_in_executor_guarded(
 
 async def to_thread_guarded(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
     """Guarded equivalent of ``asyncio.to_thread`` using the default pool."""
-    return await run_in_executor_guarded(None, func, *args, **kwargs)
+    context = contextvars.copy_context()
+    call = functools.partial(func, *args, **kwargs)
+    return await run_in_executor_guarded(None, context.run, call)
 
 
 def concurrent_result_guarded(

hy_memory/utils/pipeline_log_writer.py

@@ -1,6 +1,7 @@
 # -*- coding: utf-8 -*-
 """
-Pipeline Log Writer — 环节级 JSONL 日志（始终开启，与 SQLite Trace 分离）。
+Pipeline Log Writer — 环节级 JSONL 日志（默认开启，MEMORY_PIPELINE_LOG_ENABLED=false 关闭；
+与 SQLite Trace 分离）。
 
 目录: {log_dir}/{subdir}/{date}.log
 每行: request_id, namespace, user_id, agent_id, step, prompt, content, parsed,

hy_memory/utils/pipeline_observability.py

@@ -1,16 +1,20 @@
 # -*- coding: utf-8 -*-
 """
-Pipeline 可观测性：Log（文件 JSONL，始终开启）与 Trace（SQLite pipeline_logs，可关闭）。
+Pipeline 可观测性：Log（文件 JSONL，默认开启，可关闭）与 Trace（SQLite pipeline_logs，可关闭）。
 
 - Log：运维 tail/grep，每个环节即时一行，不依赖后续步骤成功与否。
 - Trace：Inspector / 调试，落库 pipeline_logs（需 MEMORY_CACHE_BACKEND=sqlite 等持久化后端）。
 
 环境变量：
+  MEMORY_PIPELINE_LOG_ENABLED    — Log 写 JSONL 文件，默认 true；false 时完全不写
   MEMORY_PIPELINE_TRACE_ENABLED  — Trace 落库，默认 true；OpenClaw C 端可设 false
   MEMORY_MEMORY_OPERATIONS_ENABLED — memory_operations 审计表落库，默认 false
   MEMORY_LOG_DIR                 — Log 目录（优先）
   MEMORY_TRACE_LOG_DIR           — Log 目录（兼容旧名）
 
+Log 与 Trace 都关闭时不包装 store_pipeline_log，而是换成立即返回 True 的
+空操作：热路径不做任何计时、脱敏或后台任务，也不写库。
+
 子目录（{log_dir}/{subdir}/{date}.log）由 PipelineLogWriter 从每条记录的
 user_id 前缀（第一个 "__" 左侧）切分；无前缀回落 "default"。该前缀是
 通用 namespace，可用于项目、部署或数据分桶。
@@ -94,6 +98,16 @@ def is_pipeline_trace_enabled() -> bool:
     )
 
 
+def is_pipeline_log_enabled() -> bool:
+    """是否把 pipeline step 写成 JSONL 文件（{log_dir}/<subdir>/<date>.log）。默认开启。"""
+    return os.getenv("MEMORY_PIPELINE_LOG_ENABLED", "true").strip().lower() in (
+        "true",
+        "1",
+        "yes",
+        "on",
+    )
+
+
 def is_memory_operations_enabled() -> bool:
     """是否将知识库变动写入 memory_operations 审计表。默认关闭（false）。"""
     return os.getenv("MEMORY_MEMORY_OPERATIONS_ENABLED", "false").lower() in (
@@ -125,7 +139,7 @@ def redact_content(value: str) -> str:
 
 def resolve_pipeline_log_dir() -> str:
     """
-    JSONL pipeline log 根目录（始终使用，有默认值）。
+    JSONL pipeline log 根目录（MEMORY_PIPELINE_LOG_ENABLED 开启时使用，有默认值）。
 
     优先级: MEMORY_LOG_DIR > MEMORY_TRACE_LOG_DIR > {MEMORY_DATA_DIR}/logs/pipeline
     """
@@ -228,6 +242,7 @@ def wrap_store_pipeline_log(
 ) -> Callable[..., Any]:
     """包装 cache.store_pipeline_log：计时同步、JSONL/DB 异步落盘，不堵主请求。
 
+    - log_writer 为 None 时不写 JSONL（MEMORY_PIPELINE_LOG_ENABLED=false）
     - step-to-step elapsed_ms / event_at_ms 在调用瞬间打点（与业务墙钟对齐）
     - event_at_ms：调用方传入非 0 值则直接复用（App 已在自己被调用的那一刻
       同步打点，比这里更接近真实事件时刻）；传 0 / 不传才用本地 time.time()
@@ -410,7 +425,8 @@ def wrap_store_pipeline_log(
         try:
             loop = asyncio.get_running_loop()
             # JSONL 用线程池，与 loop 无亲和，留在当前 loop。
-            _track_task(loop.create_task(_persist_jsonl()))
+            if log_writer is not None:
+                _track_task(loop.create_task(_persist_jsonl()))
             # DB trace 用 cache pool（可能 aiomysql，绑定 owner_loop）。若当前不在
             # owner_loop（如 search 走独立 loop），派发回 owner_loop 执行，避免跨 loop。
             if trace_enabled:
@@ -446,25 +462,26 @@ def wrap_store_pipeline_log(
                         )
         except RuntimeError:
             # 无 running loop（极少见）：退化为同步写，保证不丢日志
-            try:
-                log_writer.write_step(
-                    request_id=request_id,
-                    user_id=user_id,
-                    agent_id=agent_id,
-                    session_id=session_id,
-                    step=step,
-                    prompt=prompt,
-                    response=response,
-                    parsed=parsed,
-                    memory_ids=mem_ids,
-                    elapsed_ms=step_elapsed_ms,
-                    prompt_tokens=prompt_tokens,
-                    completion_tokens=completion_tokens,
-                    total_tokens=total_tokens,
-                    event_at_ms=event_at_ms,
-                )
-            except Exception:
-                pass
+            if log_writer is not None:
+                try:
+                    log_writer.write_step(
+                        request_id=request_id,
+                        user_id=user_id,
+                        agent_id=agent_id,
+                        session_id=session_id,
+                        step=step,
+                        prompt=prompt,
+                        response=response,
+                        parsed=parsed,
+                        memory_ids=mem_ids,
+                        elapsed_ms=step_elapsed_ms,
+                        prompt_tokens=prompt_tokens,
+                        completion_tokens=completion_tokens,
+                        total_tokens=total_tokens,
+                        event_at_ms=event_at_ms,
+                    )
+                except Exception:
+                    pass
             if trace_enabled:
                 try:
                     result = original_store(
@@ -496,6 +513,11 @@ def wrap_store_pipeline_log(
     return _store_pipeline_observability
 
 
+async def _noop_store_pipeline_log(*args: Any, **kwargs: Any) -> bool:
+    """Log 与 Trace 都关闭时的 store_pipeline_log：不计时、不写文件、不写库。"""
+    return True
+
+
 def install_pipeline_observability(
     cache: Any,
     *,
@@ -503,6 +525,9 @@ def install_pipeline_observability(
 ) -> None:
     """给 cache 安装 JSONL + Trace 包装（fire-and-forget）。
 
+    MEMORY_PIPELINE_LOG_ENABLED=false 时不写 JSONL；若 Trace 也关闭，
+    store_pipeline_log 被换成空操作（不包装、不写库）。
+
     在 cache 所属 loop（write loop / _LoopThread）上调用，本函数会捕获该 loop
     作为 owner_loop 传给包装器：search 走独立 loop 后，从 search loop 触发的 DB
     落库会被派发回这个 owner_loop，避免跨 loop 用 aiomysql 污染连接池。
@@ -514,15 +539,31 @@ def install_pipeline_observability(
     except RuntimeError:
         owner_loop = None
 
-    log_dir = resolve_pipeline_log_dir()
-    log_writer = PipelineLogWriter(log_dir)
+    log_enabled = is_pipeline_log_enabled()
     trace_enabled = is_pipeline_trace_enabled()
-    logger.info(
-        f"[{log_label}] Pipeline log (JSONL) enabled: {log_dir}/<subdir>/<date>.log "
-        f"(subdir from user_id prefix before '__', else default) ; "
-        f"pipeline trace (DB)={'on' if trace_enabled else 'off'} ; "
-        f"persist=async(fire-and-forget)"
-    )
+    if not log_enabled and not trace_enabled:
+        logger.info(
+            f"[{log_label}] Pipeline log (JSONL) disabled ; pipeline trace (DB)=off ; "
+            f"store_pipeline_log=no-op"
+        )
+        cache.store_pipeline_log = _noop_store_pipeline_log
+        return
+
+    log_writer = None
+    if log_enabled:
+        log_dir = resolve_pipeline_log_dir()
+        log_writer = PipelineLogWriter(log_dir)
+        logger.info(
+            f"[{log_label}] Pipeline log (JSONL) enabled: {log_dir}/<subdir>/<date>.log "
+            f"(subdir from user_id prefix before '__', else default) ; "
+            f"pipeline trace (DB)={'on' if trace_enabled else 'off'} ; "
+            f"persist=async(fire-and-forget)"
+        )
+    else:
+        logger.info(
+            f"[{log_label}] Pipeline log (JSONL) disabled ; "
+            f"pipeline trace (DB)=on ; persist=async(fire-and-forget)"
+        )
     cache.store_pipeline_log = wrap_store_pipeline_log(
         cache.store_pipeline_log,
         log_writer=log_writer,

hy_memory/utils/token_count.py

@@ -36,7 +36,8 @@ def count_tokens(text: str) -> int:
     enc = _get_encoder()
     if enc:
         try:
-            return len(enc.encode(text))
+            # Quoted control-token spellings are ordinary input text.
+            return len(enc.encode(text, disallowed_special=()))
         except Exception as e:
             logger.debug(f"[token_count] encode failed, char fallback: {e}")
     return max(1, len(text) // 4)

memhub/README.md

@@ -18,7 +18,7 @@ If you run into a problem, please open an Issue with the version, platform,repro
 
 ## 发行与平台合同
 
-源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.6`。MemHub 源码版本为 `1.0.5`
+源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.7`。MemHub 源码版本为 `1.0.5`
 （build 14），GitHub 仓库为
 [`Tencent-Hunyuan/Hy-Memory`](https://github.com/Tencent-Hunyuan/Hy-Memory)，npm 包名为
 `memhub-cli`。当前代码架构见
@@ -118,7 +118,7 @@ MemHub 的 SQLite 数据默认位于 `~/.one-memory`，不在 npm 包或 Runtime
 
 ## Hy Memory
 
-SDK 依赖固定为公开 PyPI 正式版 `hy-memory==2.0.6`。SDK 配置、权限边界及 search
+SDK 依赖固定为公开 PyPI 正式版 `hy-memory==2.0.7`。SDK 配置、权限边界及 search
 返回格式见 [`MEMORY.md`](MEMORY.md)。HY search 返回的 `content` 已包含 SDK 统一生成的
 逻辑时间和演化上下文；调用方不应再次转换时间戳或重复拼接 `When`。
 

memhub/SBOM.spdx.json

@@ -2352,7 +2352,7 @@
     {
       "SPDXID": "SPDXRef-Package-hy-memory",
       "name": "hy-memory",
-      "versionInfo": "2.0.6",
+      "versionInfo": "2.0.7",
       "downloadLocation": "NOASSERTION",
       "filesAnalyzed": false,
       "licenseConcluded": "Apache-2.0",

memhub/THIRD_PARTY_NOTICES.md

@@ -9,7 +9,7 @@ Per-package license texts are not reproduced here yet; the license
 column is filled in as part of release review, and the SBOM records
 `NOASSERTION` until then rather than guessing.
 
-## hy-memory 2.0.6
+## hy-memory 2.0.7
 
 License: Apache-2.0
 

memhub/apps/core/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "fastapi>=0.141.1",
     "filelock>=3.32.4",
     # Public SDK version validated against MemHub's permission and recall contracts.
-    "hy-memory==2.0.6",
+    "hy-memory==2.0.7",
     # chromadb arrives through the HY SDK; locked here so an unrelated upgrade
     # cannot move the SDK's store. Built-in embedding consumes numpy directly.
     "chromadb==1.5.9",

memhub/apps/core/src/one_memory/api.py

@@ -47,8 +47,8 @@ from one_memory.background_work import shutdown_bulk_pool, warm_bulk_pool
 from one_memory.config import Settings, load_settings
 from one_memory.context_routes import sdk_context_router
 from one_memory.conversation_processing import (
-    _conversation_chunks,
     memory_task_timeout_seconds,
+    turn_batches,
 )
 from one_memory.database_access import (
     MAIN_DATABASE_REPORT_SECONDS,
@@ -95,6 +95,8 @@ from one_memory.hy_sdk_runtime import (
     SdkClientHandle,
     builtin_retrieval_snapshot,
     close_all_shared_runtimes,
+    hold_stores_offline,
+    release_stores,
     sdk_relevance_floor,
 )
 from one_memory.hy_sdk_types import (
@@ -223,6 +225,7 @@ from one_memory.sdk_memory_qa import (
     SdkMemoryAnswerer,
 )
 from one_memory.sdk_memory_runtime import SdkMemoryRuntime
+from one_memory.sdk_migration_log import error_chain_text, error_frames
 from one_memory.sdk_write_processor import SdkWriteCoordinator
 from one_memory.secrets import (
     KeyringSecretStore,
@@ -710,7 +713,8 @@ def create_app(
                 raise SdkMigrationRequired()
             if app.state.memory_authority is MemoryAuthority.SDK:
                 # The metadata repair after a cutover reads every bundle: it is
-                # recovery work, so it runs after readiness (`recover_after_ready`).
+                # recovery work, so it runs after readiness (`recover_after_ready`)
+                # while the SDK's stores are held offline (see there).
                 app.state.memory_governance_store = RetiredGovernanceStore(database_path)
                 app.state.evidence_store = RetiredEvidenceStore(database_path)
             else:
@@ -1026,28 +1030,25 @@ def create_app(
 
             app.state.memory_write_intake = UnifiedWriteIntake(
                 memory_write_queue,
-                source_writes_permitted=lambda source_id: (
-                    source_id == "one-memory-manual" or _source_writes_on(app, source_id)
-                ),
+                source_writes_permitted=lambda source_id: _source_writes_on(app, source_id),
             )
             app.state.start_memory_backfill = lambda: _start_backfill(app)
 
             async def enqueue_stable_evidence(record: EvidenceRecord) -> None:
-                priority = "live" if live_sync_arrival.get() else "normal"
+                live = live_sync_arrival.get()
                 conversation = await asyncio.to_thread(
                     app.state.evidence_store.list_conversation, record.evidence_id
                 )
-                for chunk in _conversation_chunks(conversation, max_turns=12, completed_only=True):
-                    if (
-                        record not in chunk
-                        or not any(item.role == "user" for item in chunk)
-                        or not any(item.role == "assistant" for item in chunk)
-                    ):
-                        continue
+                # Legacy authority. The turn rule groups the whole conversation
+                # from its start, so a batch's records and key never change:
+                # history (end of file) writes it all, live writes complete
+                # batches and leaves the rest to catch-up (`_run_backfill`).
+                batches = turn_batches(conversation, flush=not live).batches
+                for batch in _batches_settled_by(batches, record):
                     await asyncio.to_thread(
                         app.state.memory_write_intake.submit_records,
-                        chunk,
-                        priority=priority,
+                        batch,
+                        priority="live" if live else "normal",
                     )
 
             app.state.memory_addition_store = MemoryAdditionStore(
@@ -1189,10 +1190,14 @@ def create_app(
                     except Exception as error:
                         failed.append(name)
                         app.state.recovery_failed_steps = tuple(failed)
+                        # Where it failed, as `file:line:function` frames of
+                        # the exception chain, never its message: a bare
+                        # `error_type=KeyError` named no location.
                         logger.warning(
-                            "core_recovery_step_failed step=%s error_type=%s",
+                            "core_recovery_step_failed step=%s error_type=%s chain=%s",
                             name,
                             type(error).__name__,
+                            error_chain_text(error),
                         )
                         diagnostic_event(
                             level="warn",
@@ -1200,20 +1205,42 @@ def create_app(
                             event="core.recovery_step_failed",
                             outcome="failed",
                             error_type=type(error).__name__,
+                            safe_stack=error_frames(error) or None,
                             duration_ms=elapsed_ms(step_started),
                             details={"phase": name},
                         )
 
                 async def repair_sdk_metadata() -> None:
+                    # `sdk-migrate` repairs right after its cutover and writes
+                    # the receipt, so this normally finds it and returns. It
+                    # stays for homes whose repair failed before that (no
+                    # receipt). It opens its own Chroma client on the SDK's
+                    # vector store, and Chroma's shared System is not safe
+                    # beside another client: a first library request that
+                    # opened the SDK client meanwhile lost it (KeyError).
+                    # The stores were held offline before readiness, so any
+                    # SDK client waits for this step.
                     from one_memory.hy_sdk_migration import (
                         repair_sdk_metadata_from_bundle,
                     )
 
-                    repair = await asyncio.to_thread(
-                        repair_sdk_metadata_from_bundle,
-                        data_dir=runtime_settings.data_dir,
-                        database_path=database_path,
+                    if not app.state.sdk_stores_held:
+                        logger.warning("sdk_metadata_repair_deferred reason=sdk_client_open")
+                        return
+                    # The hold is this step's from here; shutdown leaves it.
+                    app.state.sdk_stores_held = False
+                    work = asyncio.ensure_future(
+                        asyncio.to_thread(
+                            repair_sdk_metadata_from_bundle,
+                            data_dir=runtime_settings.data_dir,
+                            database_path=database_path,
+                        )
                     )
+                    # Released when the thread ends, not when this coroutine
+                    # is cancelled: a cancelled await leaves the thread
+                    # writing Chroma.
+                    work.add_done_callback(lambda _work: release_stores(runtime_settings.data_dir))
+                    repair = await asyncio.shield(work)
                     if repair.vector_rows or repair.coding_rows:
                         logger.info(
                             "sdk_metadata_repair_completed mapped=%d vector=%d coding=%d",
@@ -1337,7 +1364,12 @@ def create_app(
                 )
 
             app.state.recovery_task = None
+            app.state.sdk_stores_held = False
             try:
+                if app.state.memory_authority is MemoryAuthority.SDK:
+                    # Before the first request can open an SDK client; the
+                    # metadata repair step releases it.
+                    app.state.sdk_stores_held = hold_stores_offline(runtime_settings.data_dir)
                 diagnostic_event(
                     level="info",
                     component="core.lifecycle",
@@ -1394,6 +1426,10 @@ def create_app(
                 if app.state.live_sync_watch is not None:
                     await app.state.live_sync_watch.shutdown()
                 await app.state.live_sync_supervisor.shutdown()
+                if app.state.sdk_stores_held:
+                    # Recovery never reached the metadata repair step.
+                    release_stores(runtime_settings.data_dir)
+                    app.state.sdk_stores_held = False
                 await app.state.import_supervisor.shutdown()
                 app.state.memory_write_drain.stop()
                 if app.state.memory_write_drain_task is not None:
@@ -2824,11 +2860,24 @@ def create_app(
         dependencies=protected,
     )
     async def sdk_home_summary() -> SdkHomeSummary:
+        """The home card: the memory page's count and its newest memory.
+
+        The count is `memory_stats` total (chat + working + coding
+        experiences, facts not counted), the same set the memory page lists
+        with scene="all". Without a scene `list_memories` counts chat only,
+        which showed 55 on the home card against 94 on the memory page.
+        """
         try:
             client = await _sdk_client_handle(app).client()
+            stats = await asyncio.to_thread(
+                client.memory_stats,
+                user_id="local",
+                statuses=["active"],
+            )
             raw = await asyncio.to_thread(
                 client.list_memories,
                 user_id="local",
+                scene="all",
                 statuses=["active"],
                 limit=1,
                 offset=0,
@@ -2838,8 +2887,9 @@ def create_app(
             _log_skipped_sdk_memories(skipped)
         except Exception as error:
             raise _sdk_http_error(error, "hy_memory_list_failed") from None
+        total = stats.get("total") if isinstance(stats, dict) else None
         return SdkHomeSummary(
-            memory_count=page.total,
+            memory_count=total if isinstance(total, int) and total >= 0 else page.total,
             latest=page.memories[0] if page.memories else None,
             observed_at=datetime.now(UTC),
         )
@@ -4330,14 +4380,11 @@ async def _run_backfill(app: FastAPI) -> None:
                     _evidence_store(app).list_conversation,
                     evidence_id,
                 )
-                for chunk in _conversation_chunks(
-                    conversation,
-                    max_turns=12,
-                    completed_only=True,
-                ):
-                    if not any(item.evidence_id in evidence_ids for item in chunk):
+                # Catch-up is history: the last turn and a short batch are written.
+                for batch in turn_batches(conversation, flush=True).batches:
+                    if not any(item.evidence_id in evidence_ids for item in batch):
                         continue
-                    await asyncio.to_thread(intake.submit_records, chunk, priority="normal")
+                    await asyncio.to_thread(intake.submit_records, batch, priority="normal")
             cursor = evidence_ids[-1]
     except asyncio.CancelledError:
         raise
@@ -4356,6 +4403,22 @@ async def _run_backfill(app: FastAPI) -> None:
         logger.exception("memory catch-up failed, no further memories will be made")
 
 
+def _batches_settled_by(
+    batches: tuple[tuple[EvidenceRecord, ...], ...],
+    record: EvidenceRecord,
+) -> list[tuple[EvidenceRecord, ...]]:
+    """The batches one arriving record can have made writable.
+
+    Its own batch, and the last two: a user message completes the turn
+    before it, which can close that turn's batch and, at the 8000-character
+    cut, the batch before it. Submitting is idempotent, so an older batch
+    named again is only a hit.
+    """
+    chosen = [batch for batch in batches if record in batch]
+    chosen.extend(batch for batch in batches[-2:] if batch not in chosen)
+    return chosen
+
+
 def _sdk_memory_runtime(app: FastAPI) -> SdkMemoryRuntime:
     return cast(
         SdkMemoryRuntime,
@@ -4651,10 +4714,17 @@ def _evidence_store(app: FastAPI) -> EvidenceStore:
 
 
 def _source_writes_on(app: FastAPI, source_id: str) -> bool:
-    """Whether this source's conversations may become memories."""
+    """Whether this source's conversations may become memories.
+
+    A hand-added memory is allowed unless its grant was explicitly revoked;
+    a missing grant row is allowed. The SDK add path writes no grant row, so
+    requiring one left every SDK-authority manual add unclaimable: Pending
+    forever with attempts=0 while later jobs ran. Intake, claim and execute
+    all ask this one question.
+    """
     if source_id == "one-memory-manual":
         receipt = _grant_store(app).receipt(user_id="local", source_id=source_id)
-        return receipt is not None and receipt.revoked_at is None
+        return receipt is None or receipt.revoked_at is None
     return _source_processing_enabled(app, source_id)
 
 

memhub/apps/core/src/one_memory/conversation_processing.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 import os
 import re
+from collections.abc import Sequence
+from dataclasses import dataclass
 
-from one_memory.imports.models import EvidenceContentKind, EvidenceOrigin, EvidenceRecord
-from one_memory.memory_tasks import segment_evidence_record
+from one_memory.imports.models import EvidenceContentKind, EvidenceRecord
 
 MEMORY_TASK_TIMEOUT_MARGIN_SECONDS = 60.0
 MEMORY_TASK_TIMEOUT_MIN_SECONDS = 180.0
@@ -29,10 +30,13 @@ _HOST_ENVELOPE = re.compile(
     r")",
     re.IGNORECASE | re.DOTALL,
 )
-_WHOLE_TURN_CHARACTERS = 60_000
-_WINDOW_BYTE_CEILING = 2_000_000
-_HISTORY_MERGE_TURNS = 8
-_HISTORY_MERGE_CHARACTERS = 8_000
+#: The write rule (`turn_batches`): every 5 complete turns, or fewer when
+#: the next turn would take the batch past 8000 HY-read characters.
+TURN_BATCH_TURNS = 5
+TURN_BATCH_CHARACTERS = 8_000
+#: A live conversation's last turn counts as complete, and a short batch is
+#: written, after this long without a new message in that conversation.
+TURN_IDLE_FLUSH_SECONDS = 600
 
 
 def memory_task_timeout_seconds(model_timeout_seconds: float) -> float:
@@ -57,180 +61,83 @@ def _coding_tool_result_max_bytes() -> int:
 _HY_TOOL_CHARACTERS = _coding_tool_result_max_bytes()
 
 
-def _conversation_chunks(
-    records: list[EvidenceRecord],
-    *,
-    max_characters: int | None = None,
-    max_turns: int | None = None,
-    completed_only: bool = False,
-) -> list[tuple[EvidenceRecord, ...]]:
-    """Build bounded, complete user/assistant/tool windows."""
+@dataclass(frozen=True)
+class TurnBatches:
+    """What a conversation's records become: batches to write, records to keep."""
+
+    #: Each batch is whole turns; every batch holds at least one user message.
+    batches: tuple[tuple[EvidenceRecord, ...], ...]
+    #: Not written yet: the open last turn, complete turns short of a batch,
+    #: or records with no user message before them. Fed back in, in order,
+    #: with the conversation's next records.
+    carry: tuple[EvidenceRecord, ...]
+
+
+def conversation_turns(records: Sequence[EvidenceRecord]) -> list[tuple[EvidenceRecord, ...]]:
+    """One user message and every record after it up to the next user message.
+
+    Records before the first user message are not a turn of their own: they
+    join the first turn (an assistant proposal and the user turn adopting it
+    reach HY together). With no user message at all the result is one group
+    that does not open with a user, which `turn_batches` never writes.
+    """
     turns: list[list[EvidenceRecord]] = []
     for record in records:
         if not _is_hy_turn_record(record):
             continue
-        pending_proposal = bool(turns) and not _opens_turn(turns[-1][0])
-        if not turns or (_opens_turn(record) and not pending_proposal):
+        # Until the first user message, the group is leading records only.
+        leading = bool(turns) and not any(_opens_turn(item) for item in turns[-1])
+        if not turns or (_opens_turn(record) and not leading):
             turns.append([record])
         else:
             turns[-1].append(record)
-
-    if completed_only and turns and _trailing_user_turn_is_open(turns[-1]):
-        turns.pop()
-
-    whole = _whole_turns() and max_characters is None
-    if whole:
-        size = _hy_read_length
-        max_characters = _WHOLE_TURN_CHARACTERS
-        max_turns = max_turns or 1_000_000
-    else:
-        size = lambda record: len(record.verbatim)  # noqa: E731
-        max_characters = max_characters or 12_000
-        max_turns = max_turns or 24
-
-    chunks: list[tuple[EvidenceRecord, ...]] = []
-    for turn in turns:
-        original_question = turn[0] if _opens_turn(turn[0]) else None
-        expanded: list[EvidenceRecord] = []
-        for record in turn:
-            reserve = (
-                size(original_question)
-                if original_question is not None
-                and record is not original_question
-                and size(original_question) < max_characters
-                else 0
-            )
-            segment_limit = max(1, max_characters - reserve)
-            if size(record) <= segment_limit:
-                expanded.append(record)
-                continue
-            spans = _semantic_segment_spans(record.verbatim, segment_limit)
-            expanded.extend(
-                segment_evidence_record(record, index=index, start=start, end=end)
-                if len(spans) > 1
-                else record
-                for index, (start, end) in enumerate(spans)
-            )
-        question = expanded[0] if expanded and _opens_turn(expanded[0]) else None
-        current: list[EvidenceRecord] = []
-        for record in expanded:
-            if current and (
-                len(current) >= max_turns
-                or sum(size(item) for item in current) + size(record) > max_characters
-                or sum(_window_bytes(item) for item in current) + _window_bytes(record)
-                > _WINDOW_BYTE_CEILING
-            ):
-                chunks.append(tuple(current))
-                current = (
-                    [question]
-                    if question is not None
-                    and max_turns > 1
-                    and size(question) + size(record) <= max_characters
-                    else []
-                )
-            current.append(record)
-        chunks.append(tuple(current))
-    return _merge_tool_only_chunks(chunks)
+    return [tuple(turn) for turn in turns]
 
 
-def _semantic_segment_spans(text: str, limit: int) -> list[tuple[int, int]]:
-    if len(text) <= limit:
-        return [(0, len(text))]
-    spans: list[tuple[int, int]] = []
-    start = 0
-    separators = ("\n\n", "\n", "。", "！", "？", ". ", "! ", "? ")
-    while start < len(text):
-        hard_end = min(len(text), start + limit)
-        if hard_end == len(text):
-            end = hard_end
-        else:
-            floor = start + max(1, limit // 2)
-            candidates = [text.rfind(separator, floor, hard_end) for separator in separators]
-            boundary = max(candidates)
-            if boundary < floor:
-                end = hard_end
-            else:
-                matched = next(
-                    separator for separator in separators if text.startswith(separator, boundary)
-                )
-                end = boundary + len(matched)
-        spans.append((start, end))
-        start = end
-    return spans
+def turn_batches(records: Sequence[EvidenceRecord], *, flush: bool) -> TurnBatches:
+    """Group a conversation's records into write batches by the turn rule.
 
-
-def _merge_tool_only_chunks(
-    chunks: list[tuple[EvidenceRecord, ...]],
-) -> list[tuple[EvidenceRecord, ...]]:
-    if not any(chunk and all(record.role == "tool" for record in chunk) for chunk in chunks):
-        return chunks
-    merged: list[tuple[EvidenceRecord, ...]] = []
-    leading_tools: list[EvidenceRecord] = []
-    for chunk in chunks:
-        if not chunk:
-            continue
-        if all(record.role == "tool" for record in chunk):
-            if merged:
-                merged[-1] = (*merged[-1], *chunk)
-            else:
-                leading_tools.extend(chunk)
-            continue
-        if leading_tools:
-            chunk = (*leading_tools, *chunk)
-            leading_tools = []
-        merged.append(chunk)
-    return [] if leading_tools else merged
-
-
-def _merge_historical_chunks(
-    chunks: list[tuple[EvidenceRecord, ...]],
-    *,
-    max_turns: int = _HISTORY_MERGE_TURNS,
-    max_characters: int = _HISTORY_MERGE_CHARACTERS,
-) -> list[tuple[EvidenceRecord, ...]]:
-    if not chunks:
-        return []
-    merged: list[tuple[EvidenceRecord, ...]] = []
-    current: list[EvidenceRecord] = []
-    current_turns = 0
+    - A turn is complete once the next user message follows it; `flush`
+      completes the last one too (history at end of file, or 10 idle minutes).
+    - Every 5 complete turns are one batch. Before 5, a turn that would take
+      the batch past 8000 characters starts the next batch; one turn over
+      8000 is a batch by itself and is never split.
+    - Characters are what HY reads: user/assistant text in full, a tool
+      record up to `_HY_TOOL_CHARACTERS`.
+    - Fewer than 5 complete turns within 8000 are written only on `flush`.
+    - No batch is written without a user message; records with none before
+      them are carried, and dropped on `flush`.
+    """
+    turns = conversation_turns(records)
+    if turns and not any(_opens_turn(record) for record in turns[0]):
+        # Records with no user message yet: nothing to write until one comes.
+        return TurnBatches(batches=(), carry=() if flush else turns[0])
+    complete = turns if flush else turns[:-1]
+    open_tail = () if flush or not turns else turns[-1]
+    batches: list[tuple[EvidenceRecord, ...]] = []
+    current: list[tuple[EvidenceRecord, ...]] = []
     current_characters = 0
-    for chunk in chunks:
-        chunk_characters = sum(len(record.verbatim) for record in chunk)
-        if current and (
-            current_turns >= max_turns or current_characters + chunk_characters > max_characters
-        ):
-            merged.append(tuple(current))
-            current = []
-            current_turns = 0
-            current_characters = 0
-        current.extend(chunk)
-        current_turns += 1
-        current_characters += chunk_characters
-    if current:
-        merged.append(tuple(current))
-    return merged
-
-
-def _chunk_input_ref(chunk: tuple[EvidenceRecord, ...]) -> str:
-    user_records = tuple(item for item in chunk if item.role == "user")
-    return user_records[-1].evidence_id if user_records else chunk[-1].evidence_id
+    for turn in complete:
+        characters = sum(_hy_read_length(record) for record in turn)
+        if current and current_characters + characters > TURN_BATCH_CHARACTERS:
+            batches.append(tuple(record for item in current for record in item))
+            current, current_characters = [], 0
+        current.append(turn)
+        current_characters += characters
+        if len(current) >= TURN_BATCH_TURNS or current_characters > TURN_BATCH_CHARACTERS:
+            batches.append(tuple(record for item in current for record in item))
+            current, current_characters = [], 0
+    if current and flush:
+        batches.append(tuple(record for item in current for record in item))
+        current = []
+    carry = tuple(record for item in current for record in item) + tuple(open_tail)
+    return TurnBatches(batches=tuple(batches), carry=carry)
 
 
 def _opens_turn(record: EvidenceRecord) -> bool:
     return record.role == "user"
 
 
-def _trailing_user_turn_is_open(turn: list[EvidenceRecord]) -> bool:
-    if len(turn) != 1:
-        return False
-    opener = turn[0]
-    return (
-        _opens_turn(opener)
-        and opener.content_kind is EvidenceContentKind.RAW_CONVERSATION
-        and not _is_hand_added_memory(opener)
-    )
-
-
 def _is_hy_turn_record(record: EvidenceRecord) -> bool:
     if record.role == "tool":
         return record.content_kind is EvidenceContentKind.RAW_CONVERSATION and bool(
@@ -254,23 +161,6 @@ def _is_hy_input_turn(record: EvidenceRecord) -> bool:
     return _HOST_ENVELOPE.search(source[:2_000]) is None
 
 
-def _is_hand_added_memory(record: EvidenceRecord) -> bool:
-    meta = record.origin_meta.get(EvidenceOrigin.MANUAL_IMPORT)
-    return (
-        record.source_id == "one-memory-manual"
-        and meta is not None
-        and meta.raw_meta.get("input_mode") == "direct_memory_note"
-    )
-
-
-def _window_bytes(record: EvidenceRecord) -> int:
-    return len(record.verbatim.encode("utf-8"))
-
-
-def _whole_turns() -> bool:
-    return os.environ.get("ONE_MEMORY_SPLIT_TURNS", "").strip() != "1"
-
-
 def _hy_read_length(record: EvidenceRecord) -> int:
     if record.role == "tool":
         return min(len(record.verbatim), _HY_TOOL_CHARACTERS)

memhub/apps/core/src/one_memory/hy_runtime_config.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from typing import Literal, cast
 
 HY_MEMORY_DISTRIBUTION = "hy-memory"
-HY_MEMORY_VERSION = "2.0.6"
+HY_MEMORY_VERSION = "2.0.7"
 HY_MEMORY_NAMESPACE = "governed-v2"
 DiagnosticLogLevel = Literal["error", "warning", "info", "debug"]
 

memhub/apps/core/src/one_memory/hy_sdk_migration.py

@@ -82,7 +82,10 @@ def repair_sdk_metadata_from_bundle(
     database_path: Path,
     now: datetime | None = None,
 ) -> MetadataRepairResult:
-    """Restore metadata lost by historical cutovers from their verified bundles."""
+    """Restore metadata lost by historical cutovers from their verified bundles.
+
+    Offline only: no SDK client may be open on this home (see `_repair_vector`).
+    """
     migration_id = _sdk_migration_id(database_path)
     if migration_id is None:
         return MetadataRepairResult(None, 0, 0, 0)
@@ -329,8 +332,11 @@ def _repair_vector(
     from one_memory.sdk_vector_reattach import open_offline_chroma
 
     client = open_offline_chroma(directory)
-    # The SDK client is not open yet (startup has not created its handle), so
-    # this is the only writer; release the store before the SDK takes it.
+    # This must be the only Chroma client on the directory: Chroma shares one
+    # System per path, and this client's close() drops it from under any
+    # other (KeyError on its next lookup). `sdk-migrate` runs it after the
+    # cutover with Runtime stopped; Core's startup fallback runs it while
+    # `hold_stores_offline` keeps every SDK client waiting.
     try:
         repaired = 0
         ids = tuple(mapping)
@@ -420,6 +426,7 @@ def _repair_coding(
         path = root / name
         if not path.is_file():
             continue
+        _ensure_coding_schema(path)
         with closing(sqlite3.connect(path)) as database:
             tables = {
                 str(row[0])
@@ -431,10 +438,14 @@ def _repair_coding(
             ):
                 if table not in tables:
                     continue
+                # _ensure_coding_schema added it; an SDK without the column
+                # (before 2.0.7) leaves the time to the vector repair alone.
+                has_memory_at = "memory_at" in _columns(database, table)
                 rows = database.execute(
-                    f"SELECT {identifier},source,created_at FROM {table}"
+                    f"SELECT {identifier},source,created_at,"
+                    f"{'memory_at' if has_memory_at else 'NULL'} FROM {table}"
                 ).fetchall()
-                for memory_id, source, created_at in rows:
+                for memory_id, source, created_at, memory_at in rows:
                     desired = mapping.get(str(memory_id))
                     if desired is None:
                         continue
@@ -443,6 +454,13 @@ def _repair_coding(
                     if desired.source and source in {None, "", "default"}:
                         updates.append("source=?")
                         values.append(desired.source)
+                    if (
+                        has_memory_at
+                        and desired.memory_at is not None
+                        and memory_at != desired.memory_at
+                    ):
+                        updates.append("memory_at=?")
+                        values.append(desired.memory_at)
                     if desired.created_at is not None:
                         desired_created_at = (
                             datetime.fromtimestamp(desired.created_at, tz=UTC)
@@ -464,6 +482,32 @@ def _repair_coding(
     return repaired
 
 
+def _ensure_coding_schema(path: Path) -> None:
+    """Bring a Coding/Working store to the installed SDK's schema.
+
+    A migrated home's stores were written by an older SDK and may lack columns
+    the repair writes (``memory_at`` since hy-memory 2.0.7). The repair runs
+    before any SDK client opens them, so the SDK's own schema step is run here
+    rather than repeating its DDL. ``CodingMemoryStore`` has no public
+    schema-only entry point: ``initialize()`` also opens its Chroma
+    collections, which must stay closed during the offline repair. So only its
+    SQLite step runs, on a store object that carries nothing but that path.
+    """
+    from hy_memory.coding.store import (  # type: ignore[import-untyped]
+        CodingMemoryStore,
+    )
+
+    # Relies on the pinned SDK's internals (hy-memory==2.0.7 is pinned exactly).
+    store = CodingMemoryStore.__new__(CodingMemoryStore)
+    store._db_path = str(path)
+    store._conn = None
+    try:
+        store._init_sqlite_sync()
+    finally:
+        if store._conn is not None:
+            store._conn.close()
+
+
 def _batches[T](values: list[T] | tuple[T, ...]) -> Iterator[tuple[T, ...]]:
     for start in range(0, len(values), _BATCH):
         yield tuple(values[start : start + _BATCH])

memhub/apps/core/src/one_memory/hy_sdk_runtime.py

@@ -36,7 +36,11 @@ from one_memory.hy_graph_admin import (
 )
 from one_memory.hy_sdk_types import HySdkRuntimeError
 from one_memory.local_embedding import LocalEmbedder
-from one_memory.model_profiles import ModelProfileSnapshot, ProviderKind
+from one_memory.model_profiles import (
+    ModelProfileSnapshot,
+    ProviderKind,
+    disable_thinking_params,
+)
 
 _RUNTIME_LOCK = threading.RLock()
 BUILTIN_RETRIEVAL_PROFILE_ID = "builtin-retrieval"
@@ -52,6 +56,16 @@ class _RuntimeEntry:
 
 _RUNTIMES: dict[Path, _RuntimeEntry] = {}
 
+# Homes whose SDK-owned local stores are held offline for maintenance, and
+# the client constructions in flight per home. Chroma keeps one process-wide
+# System per path with no lock around create, lookup and release, so a
+# second client that opens or closes beside the SDK's pulls the System out
+# from under it (KeyError on the next lookup). Maintenance therefore runs
+# only while no SDK client exists for that home, and new clients wait.
+_STORE_GATE = threading.Condition()
+_OFFLINE_STORES: set[Path] = set()
+_OPENING: dict[Path, int] = {}
+
 
 class LocalOnnxEmbeddingService:
     """Public ``EmbeddingService`` implementation backed by MemHub's ONNX model."""
@@ -244,14 +258,22 @@ def build_sdk_config(
         embedding_provider, embedding_endpoint, embedding_model, dimensions
     )
 
+    llm: dict[str, Any] = {
+        "provider": "openai",
+        "model": profile.model,
+        "api_key": generation_key,
+        "base_url": generation_endpoint,
+    }
+    # The SDK sends `extra_body` on every chat call, chat and coding/working
+    # alike. Without it DeepSeek spent the whole agent_max_tokens on thinking
+    # and answered finish_reason=length with no content.
+    thinking_off = disable_thinking_params(profile)
+    if thinking_off:
+        llm["extra_body"] = thinking_off
+
     return MemoryConfig.from_dict(
         {
-            "llm": {
-                "provider": "openai",
-                "model": profile.model,
-                "api_key": generation_key,
-                "base_url": generation_endpoint,
-            },
+            "llm": llm,
             "embedder": {
                 "provider": "openai",
                 "model": (
@@ -295,7 +317,60 @@ def acquire_client(
     data_dir: Path,
     http_transport_factory: Callable[[], Any] | None = None,
 ) -> ClientLease:
-    """Construct a public SDK client and return its explicit runtime lease."""
+    """Construct a public SDK client and return its explicit runtime lease.
+
+    Waits while `hold_stores_offline` holds this home's stores.
+    """
+    key = data_dir.resolve()
+    with _STORE_GATE:
+        while key in _OFFLINE_STORES:
+            _STORE_GATE.wait()
+        _OPENING[key] = _OPENING.get(key, 0) + 1
+    try:
+        return _acquire_client(
+            profile,
+            api_key=api_key,
+            data_dir=data_dir,
+            http_transport_factory=http_transport_factory,
+        )
+    finally:
+        with _STORE_GATE:
+            _OPENING[key] -= 1
+            if not _OPENING[key]:
+                del _OPENING[key]
+            _STORE_GATE.notify_all()
+
+
+def hold_stores_offline(data_dir: Path) -> bool:
+    """Hold this home's SDK stores for offline maintenance; False if a client is open.
+
+    While held, `acquire_client` for the home waits. Every SDK client in Core
+    is built there, so nothing reads or writes Chroma beside the maintenance.
+    `release_stores` ends the hold; it is idempotent.
+    """
+    key = data_dir.resolve()
+    with _STORE_GATE:
+        with _RUNTIME_LOCK:
+            open_runtime = key in _RUNTIMES
+        if key in _OFFLINE_STORES or _OPENING.get(key) or open_runtime:
+            return False
+        _OFFLINE_STORES.add(key)
+        return True
+
+
+def release_stores(data_dir: Path) -> None:
+    with _STORE_GATE:
+        _OFFLINE_STORES.discard(data_dir.resolve())
+        _STORE_GATE.notify_all()
+
+
+def _acquire_client(
+    profile: ModelProfileSnapshot,
+    *,
+    api_key: str | None,
+    data_dir: Path,
+    http_transport_factory: Callable[[], Any] | None,
+) -> ClientLease:
     config = build_sdk_config(profile, api_key=api_key, data_dir=data_dir)
     config.http_transport_factory = http_transport_factory
     mismatch = inspect_graph_width_mismatch(config)

memhub/apps/core/src/one_memory/hy_sdk_types.py

@@ -151,6 +151,7 @@ class HySdkMemoryRecord(BaseModel):
         if not content.strip():
             raise HySdkReadError("hy_memory_record_shape_unknown")
         status = str(raw.get("status") or "active")
+        memory_at = raw.get("memory_at")
         created_at = _epoch(raw.get("created_at"))
         updated_at = _epoch(raw.get("updated_at"))
         try:
@@ -167,6 +168,9 @@ class HySdkMemoryRecord(BaseModel):
                     str(raw["session_id"]) if isinstance(raw.get("session_id"), str) else None
                 ),
                 source=str(raw["source"]) if isinstance(raw.get("source"), str) else None,
+                # Conversation time; hy-memory 2.0.7 stores it for experiences
+                # and facts. Older stores leave it unknown.
+                memory_at=memory_at if isinstance(memory_at, int) else None,
                 gmt_created=created_at,
                 scene=cast(MemoryScene, variant),
                 kind=cast(MemoryKindName, kind),

memhub/apps/core/src/one_memory/imports/runner.py

@@ -662,11 +662,15 @@ class ImportRunner:
                         and self.sdk_queue is not None
                         and event.cursor.read_to_end
                     ):
+                        # History ends at this end of file, so its last turn
+                        # and a short batch are written. Live polling keeps
+                        # them for the next increment or the idle flush.
                         self.sdk_queue.finalize_sdk_file(
                             user_id=current.user_id,
                             source_id=current.source_id,
                             file_key=event.cursor.file_key,
                             connection=connection,
+                            flush=not self._live_import,
                         )
                         continue
                     if self.on_stable_evidence is not None and event.cursor.read_to_end:

memhub/apps/core/src/one_memory/live_capture.py

@@ -256,24 +256,19 @@ def record_capture_rejection(
 
 
 def _submit_live_memory(store: EvidenceStore, record: EvidenceRecord, *, changed: bool) -> None:
+    """Add the record to its conversation's live assembling row.
+
+    The turn rule decides what is written (`turn_batches`): a turn is
+    complete when the next user message arrives, batches are 5 turns or
+    8000 characters, and Core writes the rest after 10 quiet minutes.
+    """
     if not changed or not _sync_open(store.database_path, record.source_id):
         return
-    from one_memory.conversation_processing import _conversation_chunks
     from one_memory.memory_server.durable_queue import DurableWriteQueue
-    from one_memory.memory_server.unified_entry import UnifiedWriteIntake
 
-    intake = UnifiedWriteIntake(DurableWriteQueue(store.database_path, evidence=store))
-    for chunk in _conversation_chunks(
-        store.list_conversation(record.evidence_id),
-        max_turns=12,
-        completed_only=True,
-    ):
-        if (
-            record in chunk
-            and any(item.role == "user" for item in chunk)
-            and any(item.role == "assistant" for item in chunk)
-        ):
-            intake.submit_records(chunk, priority="live")
+    DurableWriteQueue(store.database_path, evidence=store).assemble_live_records(
+        (record,), file_key=f"hook:{record.conversation_id}"
+    )
 
 
 def _conversation(payload: dict[str, object]) -> str | None:

memhub/apps/core/src/one_memory/memory_additions.py

@@ -130,10 +130,18 @@ class MemoryAdditionStore:
                 raise RuntimeError("sdk_memory_add_intake_unavailable")
             record = materialize_write_input(value, observed_at=now)
             with main_database(self.write_queue.database_path).read() as database:
+                # No grant row is written here: a missing row means allowed,
+                # only an explicit revocation refuses (same as the claim gate).
+                grant = database.execute(
+                    "SELECT revoked_at FROM source_grants WHERE user_id='local' AND source_id=?",
+                    (SOURCE_ID,),
+                ).fetchone()
                 existing = database.execute(
                     "SELECT request_id FROM memory_note_requests WHERE note_id=?",
                     (record.evidence_id,),
                 ).fetchone()
+            if grant is not None and grant[0] is not None:
+                raise ValueError("memory_add_source_revoked")
             if existing is not None:
                 current = self.require(record.evidence_id)
                 if current.verbatim != record.verbatim:

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -24,10 +24,11 @@ from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 from uuid import uuid4
 
 from one_memory.background_work import in_the_background
+from one_memory.conversation_processing import TURN_IDLE_FLUSH_SECONDS, turn_batches
 from one_memory.database_access import main_database
 from one_memory.diagnostic_log import diagnostic_event
 from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
@@ -894,16 +895,30 @@ class DurableWriteQueue:
         priority: str,
         connection: sqlite3.Connection,
     ) -> None:
-        """Append fresh-install input to an assembling queue row.
+        """Append input to its conversation's assembling queue row.
 
-        The row is not claimable until the scanner commits EOF and finalizes
-        it. No Evidence table is involved.
+        The row is not claimable. `finalize_sdk_file` turns what is complete
+        into write jobs (`turn_batches`) and keeps the rest here for the next
+        increment; `flush_idle_assemblies` writes a live row's rest once the
+        conversation has been quiet. `updated_at` moves only when a record is
+        new or revised, so a re-read does not keep a quiet conversation open.
+        No Evidence table is involved.
         """
         assembly_key = (
             f"assembly:{record.user_id}:{record.source_id}:"
             f"{hashlib.sha256(file_key.encode()).hexdigest()}:"
             f"{hashlib.sha256(record.conversation_id.encode()).hexdigest()}"
         )
+        session_key = internal_session_id_for(record.conversation_id, record.source_id)
+        written = connection.execute(
+            "SELECT 1 FROM memory_write_jobs AS job, json_each(job.evidence_ids_json) AS ref "
+            "WHERE job.session_key=? AND job.state!='assembling' AND ref.value=? LIMIT 1",
+            (session_key, record.evidence_id),
+        ).fetchone()
+        if written is not None:
+            # A re-read of a record a job already holds: staging it again
+            # would regroup it with later turns into a second job.
+            return
         row = connection.execute(
             "SELECT job_id,payload_json,priority FROM memory_write_jobs "
             "WHERE idempotency_key=? AND state='assembling'",
@@ -914,8 +929,20 @@ class DurableWriteQueue:
             payload = json.loads(str(row["payload_json"]))
             records = list(payload.get("input_records") or [])
         dumped = record.model_dump(mode="json")
-        if not any(item.get("evidence_id") == record.evidence_id for item in records):
+        index = next(
+            (
+                position
+                for position, item in enumerate(records)
+                if item.get("evidence_id") == record.evidence_id
+            ),
+            None,
+        )
+        changed = index is None or records[index] != dumped
+        if index is None:
             records.append(dumped)
+        elif changed:
+            # A revision replaces the carried copy in place.
+            records[index] = dumped
         now = datetime.now(UTC).isoformat()
         payload_json = json.dumps(
             {
@@ -926,15 +953,13 @@ class DurableWriteQueue:
             sort_keys=True,
         )
         if row is not None:
+            merged_priority = "live" if "live" in (row["priority"], priority) else "normal"
+            if not changed and merged_priority == row["priority"]:
+                return
             connection.execute(
-                "UPDATE memory_write_jobs SET payload_json=?,priority=?,updated_at=? "
-                "WHERE job_id=?",
-                (
-                    payload_json,
-                    "live" if row["priority"] == "live" or priority == "live" else "normal",
-                    now,
-                    str(row["job_id"]),
-                ),
+                "UPDATE memory_write_jobs SET payload_json=?,priority=?,"
+                "updated_at=CASE WHEN ? THEN ? ELSE updated_at END WHERE job_id=?",
+                (payload_json, merged_priority, int(changed), now, str(row["job_id"])),
             )
             return
         queue_seq = int(
@@ -952,7 +977,7 @@ class DurableWriteQueue:
             (
                 str(uuid4()),
                 queue_seq,
-                internal_session_id_for(record.conversation_id, record.source_id),
+                session_key,
                 record.source_id,
                 assembly_key,
                 assembly_key,
@@ -982,92 +1007,215 @@ class DurableWriteQueue:
         source_id: str,
         file_key: str,
         connection: sqlite3.Connection,
+        flush: bool = True,
     ) -> tuple[str, ...]:
+        """Write what one file's assembling rows hold that the turn rule completes.
+
+        `flush` (history at end of file) completes the last turn and writes a
+        short batch; without it (live polling) those stay in the row.
+        """
         rows = connection.execute(
             "SELECT * FROM memory_write_jobs WHERE state='assembling' "
             "AND source_key=? ORDER BY queue_seq",
             (source_id,),
         ).fetchall()
-        finalized = []
+        finalized: list[str] = []
         for row in rows:
             payload = json.loads(str(row["payload_json"]))
             if payload.get("file_key") != file_key:
                 continue
-            records = tuple(
-                EvidenceRecord.model_validate(item)
-                for item in payload.get("input_records") or []
-                if item.get("user_id") == user_id
+            finalized.extend(
+                self._write_assembled_turns(connection, row, user_id=user_id, flush=flush)
             )
-            if not records:
-                connection.execute(
-                    "DELETE FROM memory_write_jobs WHERE job_id=?",
-                    (str(row["job_id"]),),
+        return tuple(finalized)
+
+    def assemble_live_records(
+        self,
+        records: tuple[EvidenceRecord, ...],
+        *,
+        file_key: str,
+    ) -> tuple[str, ...]:
+        """Stage live records and write what is complete, in one transaction.
+
+        For live capture outside the import runner (host hooks): the same
+        assembling rows, turn rule and idle flush as live file polling.
+        """
+        if not records:
+            return ()
+        first = records[0]
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            for record in records:
+                self.stage_sdk_input(
+                    record, file_key=file_key, priority="live", connection=connection
                 )
-                continue
-            first = records[0]
-            session_id = internal_session_id_for(
-                first.conversation_id,
-                first.source_id,
-            )
-            material = ":".join(f"{item.evidence_id}:{item.content_hash}" for item in records)
-            request_id = f"turn:v2:{hashlib.sha256(material.encode()).hexdigest()}"
-            existing = connection.execute(
-                "SELECT job_id FROM memory_write_jobs WHERE idempotency_key=? AND job_id!=?",
-                (request_id, str(row["job_id"])),
-            ).fetchone()
-            if existing is not None:
-                connection.execute(
-                    "DELETE FROM memory_write_jobs WHERE job_id=?",
-                    (str(row["job_id"]),),
-                )
-                finalized.append(str(existing["job_id"]))
-                continue
-            request = WriteRequest(
-                internal_session_id=session_id,
+            return self.finalize_sdk_file(
+                user_id=first.user_id,
                 source_id=first.source_id,
-                batch_identity=BatchIdentity(
-                    internal_session_id=session_id,
-                    first_round=1,
-                    last_round=1,
-                ),
-                evidence_ids=[item.evidence_id for item in records],
-                input_records=records,
-                idempotency_key=request_id,
-                priority="live" if str(row["priority"]) == "live" else "normal",
-                event_at=max(
-                    (item.created_at for item in records if item.created_at),
-                    default=first.imported_at,
-                ),
-                time_quality="exact" if any(item.created_at for item in records) else "approximate",
+                file_key=file_key,
+                connection=connection,
+                flush=False,
             )
+
+    def flush_idle_assemblies(self, *, now: datetime | None = None) -> tuple[str, ...]:
+        """Write the rest of every live conversation quiet for 10 minutes.
+
+        Its last turn counts as complete and a batch short of 5 turns is
+        written. History rows are left alone: their end of file flushes them.
+        """
+        current = now or datetime.now(UTC)
+        cutoff = (current - timedelta(seconds=TURN_IDLE_FLUSH_SECONDS)).isoformat()
+        query = (
+            "SELECT * FROM memory_write_jobs WHERE state='assembling' AND priority='live' "
+            "AND updated_at<=? ORDER BY queue_seq"
+        )
+        # A read first: most ticks find nothing and must not take the writer.
+        with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
+            if connection.execute(query, (cutoff,)).fetchone() is None:
+                return ()
+        finalized: list[str] = []
+        with (
+            main_database(self.database_path).write(row_factory=sqlite3.Row) as connection,
+            connection,
+        ):
+            for row in connection.execute(query, (cutoff,)).fetchall():
+                finalized.extend(self._write_assembled_turns(connection, row, flush=True))
+        return tuple(finalized)
+
+    def _write_assembled_turns(
+        self,
+        connection: sqlite3.Connection,
+        row: sqlite3.Row,
+        *,
+        flush: bool,
+        user_id: str | None = None,
+    ) -> list[str]:
+        """Turn one assembling row's complete batches into jobs; keep the carry."""
+        payload = json.loads(str(row["payload_json"]))
+        records = tuple(
+            EvidenceRecord.model_validate(item)
+            for item in payload.get("input_records") or []
+            if user_id is None or item.get("user_id") == user_id
+        )
+        grouped = turn_batches(records, flush=flush)
+        priority: Literal["live", "normal"] = "live" if str(row["priority"]) == "live" else "normal"
+        job_ids = [
+            self._insert_turn_batch(connection, batch, priority=priority)
+            for batch in grouped.batches
+        ]
+        if not grouped.carry:
             connection.execute(
-                "UPDATE memory_write_jobs SET session_key=?,batch_id=?,"
-                "idempotency_key=?,payload_json=?,evidence_ids_json=?,state='pending',"
-                "event_at=?,time_quality=?,updated_at=? WHERE job_id=?",
+                "DELETE FROM memory_write_jobs WHERE job_id=?", (str(row["job_id"]),)
+            )
+        elif grouped.batches or len(grouped.carry) != len(records):
+            # `updated_at` stays: the idle clock counts new messages only.
+            connection.execute(
+                "UPDATE memory_write_jobs SET payload_json=? WHERE job_id=?",
                 (
-                    session_id,
-                    request.batch_identity.batch_id,
-                    request_id,
                     json.dumps(
-                        request.model_dump(
-                            mode="json",
-                            exclude={"event_at", "time_quality"},
-                        ),
+                        {
+                            **payload,
+                            "input_records": [
+                                record.model_dump(mode="json") for record in grouped.carry
+                            ],
+                        },
                         sort_keys=True,
                     ),
-                    json.dumps(list(request.evidence_ids)),
-                    (
-                        request.event_at.isoformat()
-                        if request.event_at
-                        else datetime.now(UTC).isoformat()
-                    ),
-                    request.time_quality,
-                    datetime.now(UTC).isoformat(),
                     str(row["job_id"]),
                 ),
             )
-            finalized.append(str(row["job_id"]))
-        return tuple(finalized)
+        return job_ids
+
+    def _insert_turn_batch(
+        self,
+        connection: sqlite3.Connection,
+        records: tuple[EvidenceRecord, ...],
+        *,
+        priority: Literal["live", "normal"],
+    ) -> str:
+        """One pending job for one batch of whole turns, keyed by its content.
+
+        Same records, same key: a re-read returns the job already written.
+        """
+        first = records[0]
+        session_id = internal_session_id_for(first.conversation_id, first.source_id)
+        material = ":".join(f"{item.evidence_id}:{item.content_hash}" for item in records)
+        request_id = f"turn:v2:{hashlib.sha256(material.encode()).hexdigest()}"
+        existing = connection.execute(
+            "SELECT job_id FROM memory_write_jobs WHERE idempotency_key=?",
+            (request_id,),
+        ).fetchone()
+        if existing is not None:
+            return str(existing["job_id"])
+        request = WriteRequest(
+            internal_session_id=session_id,
+            source_id=first.source_id,
+            batch_identity=BatchIdentity(
+                internal_session_id=session_id,
+                first_round=1,
+                last_round=1,
+            ),
+            evidence_ids=[item.evidence_id for item in records],
+            input_records=records,
+            idempotency_key=request_id,
+            priority=priority,
+        )
+        # The batch's time is its first user message's, as for every other job.
+        event_at, time_quality = self._event_time_for_request(request)
+        now = datetime.now(UTC)
+        job_id = str(uuid4())
+        queue_seq = int(
+            connection.execute(
+                "SELECT COALESCE(MAX(queue_seq),0)+1 FROM memory_write_jobs"
+            ).fetchone()[0]
+        )
+        event = event_at or now
+        if event.tzinfo is None:
+            event = event.replace(tzinfo=UTC)
+        connection.execute(
+            "INSERT INTO memory_write_jobs ("
+            "job_id,queue_seq,session_key,source_key,batch_id,idempotency_key,"
+            "payload_json,evidence_ids_json,state,priority,attempts,max_attempts,"
+            "lease_owner,lease_expires_at,hy_invoked,last_error,event_at,time_quality,"
+            "created_at,updated_at,sdk_raw_ids_json,sdk_memory_ids_json"
+            ") VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+            (
+                job_id,
+                queue_seq,
+                session_id,
+                first.source_id,
+                request.batch_identity.batch_id,
+                request_id,
+                json.dumps(
+                    request.model_dump(mode="json", exclude={"event_at", "time_quality"}),
+                    sort_keys=True,
+                ),
+                json.dumps(list(request.evidence_ids)),
+                PENDING,
+                priority,
+                0,
+                3,
+                None,
+                None,
+                0,
+                None,
+                event.astimezone(UTC).isoformat(),
+                time_quality,
+                now.isoformat(),
+                now.isoformat(),
+                "[]",
+                "[]",
+            ),
+        )
+        if not self.sdk_authority:
+            # Legacy catch-up reads coverage from these references.
+            connection.executemany(
+                "INSERT INTO memory_write_job_evidence (job_id, evidence_id) VALUES (?,?)",
+                ((job_id, evidence_id) for evidence_id in request.evidence_ids),
+            )
+        return job_id
 
     def _freeze_request_inputs(self, request: WriteRequest) -> WriteRequest:
         if request.input_records or self.evidence is None:
@@ -2500,6 +2648,9 @@ class DurableWriteDrain:
         #: a model outage clears on its own, and a queue that never goes idle
         #: must still re-admit the write once the wait has elapsed.
         self._last_model_readmit: datetime | None = None
+        #: A live conversation quiet for 10 minutes has its last turn and short
+        #: batch written (`flush_idle_assemblies`); checked on the same timer.
+        self._last_assembly_flush: datetime | None = None
         #: Idle polls do not need to hit the single writer every 50ms. A
         #: 1m Evidence table makes a miss expensive enough that the tail
         #: call-event write can be behind it; back off only while idle and
@@ -3032,6 +3183,15 @@ class DurableWriteDrain:
                         self.queue.readmit_model_waiters,
                         operation_name="readmit_model_waiters",
                     )
+                if (
+                    self._last_assembly_flush is None
+                    or tick_started - self._last_assembly_flush >= self._reap_interval
+                ):
+                    self._last_assembly_flush = tick_started
+                    await self._run_idle_maintenance(
+                        self.queue.flush_idle_assemblies,
+                        operation_name="flush_idle_assemblies",
+                    )
                 if self.reconcile_unknown is not None and (
                     self._last_unknown_reconcile is None
                     or tick_started - self._last_unknown_reconcile

memhub/apps/core/src/one_memory/model_profiles.py

@@ -1136,20 +1136,31 @@ class ModelProfileStore:
         )
 
 
+def disable_thinking_params(
+    profile: ModelProfile | ModelProfileSnapshot,
+) -> dict[str, object]:
+    """Provider-specific thinking-off params, empty unless the profile opts in.
+
+    The one mapping for every request Core builds itself and for the SDK's
+    `llm.extra_body`, so a model that thinks on one path thinks on none.
+    """
+    if not profile.disable_thinking:
+        return {}
+    if profile.provider_kind is ProviderKind.OPENAI_COMPATIBLE:
+        if urlparse(profile.base_url or "").hostname == "api.deepseek.com":
+            return {"thinking": {"type": "disabled"}}
+        return {"reasoning_effort": "none"}
+    if profile.provider_kind is ProviderKind.OLLAMA:
+        return {"think": False}
+    return {}
+
+
 def attach_disable_thinking(
     profile: ModelProfile | ModelProfileSnapshot,
     payload: dict[str, object],
 ) -> dict[str, object]:
     """Attach provider-specific thinking-off params when the profile opts in."""
-    if not profile.disable_thinking:
-        return payload
-    if profile.provider_kind is ProviderKind.OPENAI_COMPATIBLE:
-        if urlparse(profile.base_url or "").hostname == "api.deepseek.com":
-            payload["thinking"] = {"type": "disabled"}
-        else:
-            payload["reasoning_effort"] = "none"
-    elif profile.provider_kind is ProviderKind.OLLAMA:
-        payload["think"] = False
+    payload.update(disable_thinking_params(profile))
     return payload
 
 

memhub/apps/core/src/one_memory/sdk_migration_log.py

@@ -125,6 +125,31 @@ def describe_error(error: BaseException) -> dict[str, object]:
     return {"chain": chain}
 
 
+_PY_FRAME = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.\-]*\.py:\d{1,7}:[A-Za-z_][A-Za-z0-9_]*$")
+
+
+def error_frames(error: BaseException, *, limit: int = 16) -> tuple[str, ...]:
+    """The raised error's innermost `file.py:line:function` frames, for a stack field.
+
+    Library frames are kept (a KeyError raised inside Chroma is located by
+    them); `<module>`-style names are dropped because stack fields refuse them.
+    """
+    chain = describe_error(error)["chain"]
+    frames = chain[0].get("frames", ()) if chain else ()  # type: ignore[index,union-attr]
+    return tuple(frame for frame in frames if _PY_FRAME.fullmatch(frame))[-limit:]
+
+
+def error_chain_text(error: BaseException) -> str:
+    """`describe_error` as one log token: `Type@file:line:function>...;Cause@...`.
+
+    For a plain text log (core.log). Types and frames only, never a message.
+    """
+    return ";".join(
+        f"{entry['type']}@{'>'.join(entry.get('frames', ()))}"  # type: ignore[arg-type]
+        for entry in describe_error(error)["chain"]  # type: ignore[union-attr]
+    )
+
+
 def _build() -> dict[str, object]:
     from one_memory import __version__
 

memhub/apps/core/src/one_memory/sdk_migration_runner.py

@@ -829,6 +829,7 @@ def run_startup_migration(
                 ignore_errors=True,
             )
         discard_failed_state(data_dir)
+        _repair_metadata_after_cutover(data_dir, database_path, log)
         progress("cleanup")
         with log.step("cleanup.drop_legacy_tables") as step:
             dropped = resume_legacy_cleanup(data_dir=data_dir, database_path=database_path)
@@ -847,6 +848,25 @@ def run_startup_migration(
     )
 
 
+def _repair_metadata_after_cutover(data_dir: Path, database_path: Path, log: _StepLog) -> None:
+    """Restore lost metadata here, where Runtime is stopped and no Chroma client is open.
+
+    Writes the repair receipt, so Core's startup fallback finds it and does
+    nothing. Authority is already committed: a failure is logged and left to
+    that fallback instead of failing the migration.
+    """
+    from one_memory.hy_sdk_migration import repair_sdk_metadata_from_bundle
+
+    try:
+        with log.step("cutover.metadata_repair") as step:
+            repair = repair_sdk_metadata_from_bundle(data_dir=data_dir, database_path=database_path)
+            step["mapped"] = repair.mapped
+            step["vector_rows"] = repair.vector_rows
+            step["coding_rows"] = repair.coding_rows
+    except Exception:
+        return
+
+
 def _restore_after_failure(
     *,
     data_dir: Path,

memhub/apps/core/src/one_memory/sdk_write_processor.py

@@ -16,6 +16,7 @@ from one_memory.secrets import SecretStore, SecretStoreUnavailableError
 
 PRODUCT_AGENT_ID = "all-apps"
 _MESSAGE_ROLES = frozenset({"user", "assistant", "tool"})
+_CONVERSATION_ROLES = frozenset({"user", "assistant"})
 
 
 @dataclass(frozen=True)
@@ -41,12 +42,23 @@ class SdkWriteProcessor:
     ) -> SdkWriteReceipt:
         if not records:
             raise RuntimeError("hy_memory_task_without_inputs")
+        extracted = all(_is_extracted_content(record) for record in records)
+        if not extracted and not any(
+            record.content_kind is EvidenceContentKind.RAW_CONVERSATION
+            and record.role in _CONVERSATION_ROLES
+            for record in records
+        ):
+            # Nothing to remember: tool records alone. The SDK strips them on
+            # the chat path and answers 400 "content or messages is
+            # required", which used to settle the job as unknown. Completed
+            # with no memories, and the SDK is never called.
+            return SdkWriteReceipt()
 
         scope = _write_scope(records)
         metadata = _write_metadata(records, scope)
         client = await self._client_handle.client()
 
-        if all(_is_extracted_content(record) for record in records):
+        if extracted:
             contents = [sanitize_model_input(record.verbatim) for record in records]
             await before_hy_invocation()
             result = await asyncio.to_thread(

memhub/apps/core/tests/test_conversation_turn_grouping.py

@@ -1,90 +1,199 @@
-"""A proposal and the turn that adopts it must reach HY as one turn."""
+"""The turn rule: what a conversation's records become before they reach HY.
+
+A turn is one user message and everything up to the next one. Every 5
+complete turns are written together, sooner when the next turn would pass
+8000 characters; a short batch waits for the end of a history file or 10
+quiet minutes. No batch is written without a user message.
+"""
 
 from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from one_memory.conversation_processing import _conversation_chunks
+from one_memory.conversation_processing import (
+    TURN_BATCH_CHARACTERS,
+    TURN_BATCH_TURNS,
+    conversation_turns,
+    turn_batches,
+)
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
 
+_WRITER = {"user": "person", "assistant": "agent", "tool": "tool"}
 
-def _records(tmp_path, *speakers: tuple[str, str, str]) -> list:
+
+def _records(tmp_path, *speakers: tuple[str, str]) -> list:
     store = EvidenceStore(tmp_path / "one-memory.sqlite3")
-    records = []
-    for index, (role, written_by, text) in enumerate(speakers):
-        records.append(
-            store.put(
-                EvidenceInput(
-                    source_id="cursor",
-                    conversation_id="conversation",
-                    message_id=f"m{index}",
-                    role=role,
-                    written_by=WrittenBy(written_by),
-                    verbatim=text,
-                    created_at=datetime.now(UTC),
-                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
-                )
-            ).record
-        )
-    return records
-
-
-def test_an_unanswered_assistant_prefix_joins_the_turn_that_adopts_it(tmp_path) -> None:
-    """`assistant` proposal -> `user` explicit adoption is one turn, not two.
-
-    Splitting it sends HY an answer without what it answers, and loses the
-    authority pair the author/adoption quality gate depends on.
-    """
-    records = _records(
-        tmp_path,
-        ("assistant", "agent", "建议把超时改成 30 秒。"),
-        ("user", "person", "就这么做，超时改成 30 秒。"),
-    )
-
-    chunks = _conversation_chunks(records)
-
-    assert [[record.role for record in chunk] for chunk in chunks] == [["assistant", "user"]]
-
-
-def test_a_lone_user_message_does_not_become_a_model_input(tmp_path) -> None:
-    """A user message with no reply yet is Evidence, not a completed turn.
-
-    Concurrent-import fixtures write `user` then `assistant` pairs, but a lone
-    user message must not create an HY task: that is the complete-turn rule,
-    not a queue stall.
-    """
-    records = _records(tmp_path, ("user", "person", "把超时改成 30 秒。"))
-
-    assert _conversation_chunks(records, completed_only=True) == []
-
-
-def test_only_the_unanswered_trailing_turn_is_held_back(tmp_path) -> None:
-    """Earlier complete turns stay eligible while the open tail waits."""
-    records = _records(
-        tmp_path,
-        ("user", "person", "先把超时改成 30 秒。"),
-        ("assistant", "agent", "已经改好了。"),
-        ("user", "person", "接着把发布包传上去。"),
-    )
-
-    assert _conversation_chunks(records, completed_only=True) == [(records[0], records[1])]
-
-
-def test_the_complete_turn_contract_used_for_concurrent_import_reds(tmp_path) -> None:
-    """Node-id level answer for `[BLOCK C2 -> C1]` concurrent-import reds.
-
-    Those fixtures write `user` turns and never emit the closing
-    `assistant`/tool event, so under the complete-turn rule no HY task is
-    created inside the 5s window. That is the contract, not a queue stall:
-    Evidence is still committed and searchable, and the task appears once the
-    closing event arrives.
-    """
-    closed = _records(
-        tmp_path,
-        ("user", "person", "把超时改成 30 秒。"),
-        ("assistant", "agent", "已改为 30 秒。"),
-    )
-    assert [[record.role for record in chunk] for chunk in _conversation_chunks(closed)] == [
-        ["user", "assistant"]
+    return [
+        store.put(
+            EvidenceInput(
+                source_id="cursor",
+                conversation_id="conversation",
+                message_id=f"m{index}",
+                role=role,
+                written_by=WrittenBy(_WRITER[role]),
+                verbatim=text,
+                created_at=datetime.now(UTC),
+                origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+            )
+        ).record
+        for index, (role, text) in enumerate(speakers)
     ]
+
+
+def _turns(count: int, *, answer: str = "好的。") -> list[tuple[str, str]]:
+    return [
+        part
+        for index in range(count)
+        for part in (("user", f"第 {index} 个问题"), ("assistant", answer))
+    ]
+
+
+def _roles(batches) -> list[list[str]]:
+    return [[record.role for record in batch] for batch in batches]
+
+
+def test_rule_constants_are_the_product_rule() -> None:
+    assert (TURN_BATCH_TURNS, TURN_BATCH_CHARACTERS) == (5, 8_000)
+
+
+def test_a_turn_runs_from_one_user_message_to_the_next(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("user", "跑一下测试"),
+        ("assistant", "好"),
+        ("tool", "3 passed"),
+        ("assistant", "都通过了"),
+        ("user", "提交吧"),
+        ("assistant", "已提交"),
+    )
+
+    assert [len(turn) for turn in conversation_turns(records)] == [4, 2]
+
+
+def test_records_before_the_first_user_message_join_the_first_turn(tmp_path) -> None:
+    """An assistant proposal and the user turn adopting it reach HY together."""
+    records = _records(
+        tmp_path,
+        ("assistant", "建议把超时改成 30 秒。"),
+        ("user", "就这么做，超时改成 30 秒。"),
+        ("assistant", "改好了。"),
+        ("user", "再把重试改成 3 次。"),
+        ("assistant", "也改好了。"),
+    )
+
+    assert _roles(conversation_turns(records)) == [
+        ["assistant", "user", "assistant"],
+        ["user", "assistant"],
+    ]
+
+
+def test_five_complete_turns_are_one_batch_and_the_open_turn_waits(tmp_path) -> None:
+    records = _records(tmp_path, *_turns(6))
+
+    grouped = turn_batches(records, flush=False)
+
+    assert [len(batch) for batch in grouped.batches] == [10]
+    assert grouped.batches[0] == tuple(records[:10])
+    assert grouped.carry == tuple(records[10:])
+
+
+def test_complete_turns_short_of_five_wait_for_flush(tmp_path) -> None:
+    records = _records(tmp_path, *_turns(3))
+
+    waiting = turn_batches(records, flush=False)
+    assert waiting.batches == ()
+    assert waiting.carry == tuple(records)
+
+    flushed = turn_batches(records, flush=True)
+    assert flushed.batches == (tuple(records),)
+    assert flushed.carry == ()
+
+
+def test_a_lone_user_message_is_not_written_until_flush(tmp_path) -> None:
+    records = _records(tmp_path, ("user", "把超时改成 30 秒。"))
+
+    assert turn_batches(records, flush=False).batches == ()
+    assert turn_batches(records, flush=True).batches == (tuple(records),)
+
+
+def test_the_turn_that_would_pass_8000_characters_starts_the_next_batch(tmp_path) -> None:
+    answer = "句" * 3_000
+    records = _records(tmp_path, *_turns(4, answer=answer))
+
+    grouped = turn_batches(records, flush=False)
+
+    # Turns 1+2 fit; turn 3 would pass 8000, so 1+2 are written now. Turn 3
+    # is complete but alone, and turn 4 is still open.
+    assert grouped.batches == (tuple(records[:4]),)
+    assert grouped.carry == tuple(records[4:])
+    assert sum(len(record.verbatim) for record in grouped.batches[0]) <= TURN_BATCH_CHARACTERS
+
+
+def test_one_turn_over_8000_characters_is_a_batch_of_its_own_and_not_split(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("user", "先问个小问题"),
+        ("assistant", "好"),
+        ("user", "写一份方案"),
+        ("assistant", "段落。" * 7_000),
+        ("user", "谢谢"),
+    )
+
+    grouped = turn_batches(records, flush=False)
+
+    assert grouped.batches == (tuple(records[:2]), tuple(records[2:4]))
+    assert grouped.carry == (records[4],)
+    assert all(":segment:" not in record.evidence_id for record in grouped.batches[1])
+
+
+def test_tool_output_counts_only_what_hy_reads(tmp_path) -> None:
+    """A tool record counts up to `_HY_TOOL_CHARACTERS`, not its full text."""
+    from one_memory.conversation_processing import _HY_TOOL_CHARACTERS
+
+    parts = [
+        part
+        for index in range(5)
+        for part in (("user", f"第 {index} 步"), ("tool", "x" * 50_000), ("assistant", "完成"))
+    ]
+    records = _records(tmp_path, *parts, ("user", "下一步"))
+
+    grouped = turn_batches(records, flush=False)
+
+    assert _HY_TOOL_CHARACTERS == 512
+    assert [len(batch) for batch in grouped.batches] == [15]
+
+
+def test_records_with_no_user_message_are_never_a_batch(tmp_path) -> None:
+    """The SDK strips tool records on the chat path and refuses what is left.
+
+    A tool-only batch came back 400 "content or messages is required" and
+    was recorded unknown, 16 times. Carried until a user message arrives;
+    dropped when the conversation is flushed without one.
+    """
+    records = _records(tmp_path, ("tool", "ls output"), ("assistant", "看完了"))
+
+    waiting = turn_batches(records, flush=False)
+    assert waiting.batches == ()
+    assert waiting.carry == tuple(records)
+    assert turn_batches(records, flush=True) == turn_batches((), flush=True)
+
+    later = _records(tmp_path, ("user", "那就开始吧"), ("assistant", "开始"))
+    joined = turn_batches((*waiting.carry, *later), flush=True)
+    assert _roles(joined.batches) == [["tool", "assistant", "user", "assistant"]]
+
+
+def test_every_batch_holds_a_user_message(tmp_path) -> None:
+    records = _records(
+        tmp_path,
+        ("tool", "前置输出"),
+        *_turns(7, answer="答" * 2_500),
+        ("tool", "尾部输出"),
+    )
+
+    for flush in (False, True):
+        grouped = turn_batches(records, flush=flush)
+        assert grouped.batches
+        assert all(any(record.role == "user" for record in batch) for batch in grouped.batches)
+        written = [record for batch in grouped.batches for record in batch]
+        assert written + list(grouped.carry) == records

memhub/apps/core/tests/test_core_ready_before_recovery.py

@@ -142,3 +142,112 @@ def test_shutdown_during_recovery_stops_it_without_starting_more(tmp_path, monke
             release.set()
 
     asyncio.run(scenario())
+
+
+def test_metadata_repair_never_overlaps_an_sdk_client(tmp_path, monkeypatch):
+    """The repair's own Chroma client dropped the SDK's shared System (KeyError).
+
+    The first library request opened the SDK client while the repair ran
+    after readiness. Now every SDK client of the home waits for the repair.
+    """
+    from one_memory import hy_sdk_runtime
+    from one_memory.hy_sdk_runtime import ClientLease
+
+    release = threading.Event()
+    entered = threading.Event()
+    events: list[str] = []
+
+    def slow_repair(**_):
+        entered.set()
+        events.append("repair_started")
+        release.wait(10)
+        events.append("repair_finished")
+        return MetadataRepairResult(None, 0, 0, 0)
+
+    def open_client(profile, *, api_key, data_dir, http_transport_factory):
+        events.append("client_opened")
+        return ClientLease(object(), data_dir.resolve())
+
+    monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", slow_repair)
+    monkeypatch.setattr(hy_sdk_runtime, "_acquire_client", open_client)
+    monkeypatch.setattr(ClientLease, "close", lambda self: None)
+
+    async def scenario():
+        app = application(tmp_path)
+        try:
+            async with app.router.lifespan_context(app), client_for(app) as client:
+                assert await asyncio.to_thread(entered.wait, 5)
+                handle = hy_sdk_runtime.SdkClientHandle(
+                    profile=hy_sdk_runtime.builtin_retrieval_snapshot(),
+                    api_key=None,
+                    data_dir=app.state.settings.data_dir,
+                )
+                opening = asyncio.create_task(handle.client())
+                await asyncio.sleep(0.2)
+                # Readiness and reads answer; the SDK client waits.
+                assert (await health(client))["status"] == "ok"
+                assert not opening.done()
+                release.set()
+                await asyncio.wait_for(opening, 5)
+                await wait_for_recovery(client)
+        finally:
+            release.set()
+
+    asyncio.run(scenario())
+    assert events == ["repair_started", "repair_finished", "client_opened"]
+
+
+def test_stores_are_not_held_beside_an_open_sdk_client(tmp_path):
+    from one_memory import hy_sdk_runtime
+
+    key = tmp_path.resolve()
+    with hy_sdk_runtime._STORE_GATE:
+        hy_sdk_runtime._OPENING[key] = 1
+    try:
+        assert hy_sdk_runtime.hold_stores_offline(tmp_path) is False
+    finally:
+        with hy_sdk_runtime._STORE_GATE:
+            del hy_sdk_runtime._OPENING[key]
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is True
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is False
+    hy_sdk_runtime.release_stores(tmp_path)
+    hy_sdk_runtime.release_stores(tmp_path)
+    assert hy_sdk_runtime.hold_stores_offline(tmp_path) is True
+    hy_sdk_runtime.release_stores(tmp_path)
+
+
+def test_a_failed_recovery_step_records_where_it_failed(tmp_path, monkeypatch, caplog):
+    """`error_type=KeyError` alone named no location; frames, never the message."""
+    from one_memory import api
+
+    def lookup(systems):
+        return systems["private-identifier"]
+
+    def broken_repair(**_):
+        return lookup({})
+
+    events: list[dict] = []
+    monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", broken_repair)
+    monkeypatch.setattr(api, "diagnostic_event", lambda **values: events.append(values))
+
+    async def scenario():
+        app = application(tmp_path)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            await wait_for_recovery(client)
+
+    with caplog.at_level("WARNING"):
+        asyncio.run(scenario())
+
+    [line] = [
+        record.getMessage()
+        for record in caplog.records
+        if "core_recovery_step_failed" in record.getMessage()
+    ]
+    assert "step=sdk_metadata_repair error_type=KeyError chain=KeyError@" in line
+    assert "test_core_ready_before_recovery.py:" in line and ":lookup" in line
+    assert "private-identifier" not in line
+    [failed] = [event for event in events if event.get("event") == "core.recovery_step_failed"]
+    assert failed["error_type"] == "KeyError"
+    assert failed["safe_stack"][-1].startswith("test_core_ready_before_recovery.py:")
+    assert failed["safe_stack"][-1].endswith(":lookup")
+    assert "private-identifier" not in repr(failed)

memhub/apps/core/tests/test_disable_thinking.py

@@ -8,6 +8,7 @@ import json
 import httpx
 from pydantic import SecretStr
 
+from one_memory.hy_sdk_runtime import build_sdk_config
 from one_memory.imports.models import (
     EvidenceInput,
     EvidenceOrigin,
@@ -17,6 +18,7 @@ from one_memory.imports.models import (
 from one_memory.imports.storage import EvidenceStore
 from one_memory.memory_batch_extraction import MemoryBatchExtractor
 from one_memory.memory_qa import _answer_request
+from one_memory.model_probe import _structured_request
 from one_memory.model_profiles import (
     ModelProfileInput,
     ModelProfileStore,
@@ -243,3 +245,94 @@ def test_ollama_batch_sends_think_false_only_when_disabled(tmp_path) -> None:
     )
     assert payloads[0]["think"] is False
     assert "reasoning_effort" not in payloads[0]
+
+
+def _deepseek(store, secrets, *, disable_thinking: bool):
+    return store.create(
+        ModelProfileInput(
+            name=f"deepseek-{disable_thinking}",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://api.deepseek.com/v1",
+            model="deepseek-v4-flash",
+            api_key=SecretStr("sk-test"),
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+            disable_thinking=disable_thinking,
+        ),
+        secrets,
+    )
+
+
+def test_sdk_config_turns_deepseek_thinking_off_like_the_probe(tmp_path) -> None:
+    """DeepSeek Flash spent the SDK's 2000 agent tokens thinking: EMPTY_RESPONSE.
+
+    The SDK sends `llm.extra_body` on every chat call (chat and the shared
+    coding/working provider), and the probe must ask the model the same way.
+    """
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    profile = _deepseek(store, secrets, disable_thinking=True)
+    snapshot = store.snapshot(profile.profile_id, profile.revision)
+
+    config = build_sdk_config(snapshot, api_key="sk-test", data_dir=tmp_path / "data")
+
+    assert config.llm.extra_body == {"thinking": {"type": "disabled"}}
+    _m, _u, _h, probe = _structured_request(profile, "sk-test")
+    assert probe["thinking"] == config.llm.extra_body["thinking"]
+    assert "reasoning_effort" not in probe
+
+
+def test_sdk_config_sends_no_thinking_fields_unless_the_profile_opts_in(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    profile = _deepseek(store, secrets, disable_thinking=False)
+    snapshot = store.snapshot(profile.profile_id, profile.revision)
+
+    config = build_sdk_config(snapshot, api_key="sk-test", data_dir=tmp_path / "data")
+
+    assert not config.llm.extra_body
+
+
+def test_sdk_config_maps_other_providers_through_the_same_table(tmp_path) -> None:
+    database = tmp_path / "one-memory.sqlite3"
+    secrets = InMemorySecretStore()
+    store = ModelProfileStore(database)
+    openai = store.create(
+        ModelProfileInput(
+            name="openai",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://api.example.com/v1",
+            model="gpt-test",
+            api_key=SecretStr("sk-test"),
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+        ),
+        secrets,
+    )
+    ollama = store.create(
+        ModelProfileInput(
+            name="ollama",
+            provider_kind=ProviderKind.OLLAMA,
+            base_url="http://127.0.0.1:11434",
+            model="qwen",
+            embedding_model="embedding-model",
+            embedding_provider_kind=ProviderKind.LOCAL_ONNX,
+        ),
+        secrets,
+    )
+
+    openai_config = build_sdk_config(
+        store.snapshot(openai.profile_id, openai.revision),
+        api_key="sk-test",
+        data_dir=tmp_path / "data",
+    )
+    ollama_config = build_sdk_config(
+        store.snapshot(ollama.profile_id, ollama.revision),
+        api_key=None,
+        data_dir=tmp_path / "data",
+    )
+
+    assert openai_config.llm.extra_body == {"reasoning_effort": "none"}
+    assert ollama_config.llm.extra_body == {"think": False}

memhub/apps/core/tests/test_import_runner.py

@@ -773,3 +773,131 @@ def test_cancel_during_callback_keeps_a_replayable_pending_marker(tmp_path) -> N
 
     assert evidence_store.count() == 1
     assert len(_pending_markers(evidence_store.database_path)) == 1
+
+
+def said(message_id: str, role: str, text: str) -> EvidenceInput:
+    writer = {"user": WrittenBy.PERSON, "assistant": WrittenBy.AGENT, "tool": WrittenBy.TOOL}
+    return evidence(message_id).model_copy(
+        update={"role": role, "written_by": writer[role], "verbatim": text}
+    )
+
+
+def _sdk_runner(tmp_path):
+    database = tmp_path / "one-memory.sqlite3"
+    bootstrap_fresh_management_schema(database)
+    job_store = ImportJobStore(database)
+    queue = DurableWriteQueue(database, sdk_authority=True)
+    runner = ImportRunner(
+        evidence_store=RetiredEvidenceStore(database),  # type: ignore[arg-type]
+        job_store=job_store,
+        sdk_queue=queue,
+        sdk_authority=lambda: True,
+    )
+    return database, job_store, queue, runner
+
+
+def _read(runner, job_store, parts, *, operation: str, first_line: int = 1):
+    events: list[ScanEvent] = [
+        ScannedEvidence(evidence=part, cursor=cursor(f"line:{first_line + index}"))
+        for index, part in enumerate(parts)
+    ]
+    last = f"line:{first_line + len(parts) - 1}"
+    events.append(ScanCursorTouch(cursor=cursor(last).model_copy(update={"read_to_end": True})))
+    job = job_store.create(user_id="local", source_id="fixture", operation=operation)
+    asyncio.run(runner.run(adapter=FixtureAdapter(events), job_id=job.job_id))
+
+
+def _jobs(database) -> list[tuple[str, list[str]]]:
+    import json
+
+    with sqlite3.connect(database) as connection:
+        rows = connection.execute(
+            "SELECT state,payload_json FROM memory_write_jobs ORDER BY queue_seq"
+        ).fetchall()
+    return [
+        (state, [item["role"] for item in json.loads(payload)["input_records"]])
+        for state, payload in rows
+    ]
+
+
+def _turns(count: int, *, start: int = 0) -> list[EvidenceInput]:
+    return [
+        said(f"{role[0]}{index}", role, f"{role} {index}")
+        for index in range(start, start + count)
+        for role in ("user", "assistant")
+    ]
+
+
+def test_history_import_writes_five_turn_batches_and_flushes_the_rest_at_eof(tmp_path):
+    """A first import used to be one giant job per conversation."""
+    database, job_store, _queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(7), operation="history")
+
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant"] * 5),
+        ("pending", ["user", "assistant"] * 2),
+    ]
+
+
+def test_live_polling_carries_an_open_turn_and_a_tool_only_increment(tmp_path):
+    """A transcript read mid-turn used to give tool-only jobs (16 unknown)."""
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(2), operation="sync")
+    assert _jobs(database) == [("assembling", ["user", "assistant"] * 2)]
+
+    tools = [said("t1", "tool", "ls"), said("t2", "tool", "3 passed")]
+    _read(runner, job_store, tools, operation="sync", first_line=5)
+    assert _jobs(database) == [
+        ("assembling", ["user", "assistant", "user", "assistant", "tool", "tool"])
+    ]
+
+    # Ten quiet minutes: the open turn is complete and the short batch goes.
+    assert queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=9)) == ()
+    flushed = queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+    assert len(flushed) == 1
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant", "user", "assistant", "tool", "tool"])
+    ]
+
+
+def test_live_polling_writes_five_complete_turns_and_keeps_the_open_one(tmp_path):
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(5), operation="sync")
+    assert [state for state, _ in _jobs(database)] == ["assembling"]
+
+    _read(runner, job_store, [said("u5", "user", "user 5")], operation="sync", first_line=11)
+    assert _jobs(database) == [
+        ("assembling", ["user"]),
+        ("pending", ["user", "assistant"] * 5),
+    ]
+
+
+def test_a_reread_increment_creates_no_second_job(tmp_path):
+    database, job_store, queue, runner = _sdk_runner(tmp_path)
+
+    _read(runner, job_store, _turns(6), operation="sync")
+    _read(runner, job_store, _turns(6), operation="sync")
+    queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+    _read(runner, job_store, _turns(6), operation="sync")
+    queue.flush_idle_assemblies(now=datetime.now(UTC) + timedelta(minutes=11))
+
+    assert _jobs(database) == [
+        ("pending", ["user", "assistant"] * 5),
+        ("pending", ["user", "assistant"]),
+    ]
+
+
+def test_history_records_with_no_user_message_are_dropped_at_eof(tmp_path):
+    database, job_store, _queue, runner = _sdk_runner(tmp_path)
+
+    _read(
+        runner,
+        job_store,
+        [said("t1", "tool", "ls"), said("a1", "assistant", "ok")],
+        operation="history",
+    )
+
+    assert _jobs(database) == []

memhub/apps/core/tests/test_live_capture.py

@@ -5,6 +5,7 @@ import json
 import os
 import sqlite3
 from contextlib import closing
+from datetime import UTC, datetime, timedelta
 from io import StringIO
 from pathlib import Path
 
@@ -107,7 +108,15 @@ def test_same_message_id_from_the_file_does_not_duplicate(tmp_path) -> None:
     assert store.count() == 1
 
 
+def _quiet_for_ten_minutes(database: Path, store: EvidenceStore) -> tuple[str, ...]:
+    """What Core's drain does once a live conversation has been quiet."""
+    return DurableWriteQueue(database, evidence=store).flush_idle_assemblies(
+        now=datetime.now(UTC) + timedelta(minutes=11)
+    )
+
+
 def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None:
+    """The turn rule holds the turn until the next user message or 10 quiet minutes."""
     store = _grant(tmp_path, "workbuddy", sync=True)
     database = tmp_path / "one-memory.sqlite3"
 
@@ -132,10 +141,13 @@ def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None:
         database,
     )
 
+    with closing(sqlite3.connect(database)) as db:
+        assert db.execute("SELECT state FROM memory_write_jobs").fetchall() == [("assembling",)]
+    assert len(_quiet_for_ten_minutes(database, store)) == 1
     with closing(sqlite3.connect(database)) as db:
         row = db.execute(
             "SELECT priority,evidence_ids_json FROM memory_write_jobs "
-            "WHERE idempotency_key LIKE 'turn:v1:%'"
+            "WHERE idempotency_key LIKE 'turn:v2:%'"
         ).fetchone()
     assert row[0] == "live"
     assert len(json.loads(row[1])) == 2
@@ -161,6 +173,41 @@ def test_hook_submits_complete_turn_to_the_live_durable_lane(tmp_path) -> None:
         assert db.execute("SELECT COUNT(*) FROM evidence").fetchone() == (2,)
 
 
+def test_hook_writes_five_turns_once_the_next_user_message_arrives(tmp_path) -> None:
+    store = _grant(tmp_path, "workbuddy", sync=True)
+    database = tmp_path / "one-memory.sqlite3"
+
+    def say(event: str, field: str, message_id: str, text: str) -> None:
+        assert capture_hook(
+            "workbuddy",
+            {
+                "hook_event_name": event,
+                "session_id": "sess-five",
+                "message_id": message_id,
+                field: text,
+            },
+            database,
+        )
+
+    for index in range(5):
+        say("UserPromptSubmit", "prompt", f"u{index}", f"第 {index} 个偏好")
+        say("Stop", "last_assistant_message", f"a{index}", f"记住第 {index} 个")
+    with closing(sqlite3.connect(database)) as db:
+        assert db.execute("SELECT state FROM memory_write_jobs").fetchall() == [("assembling",)]
+
+    say("UserPromptSubmit", "prompt", "u5", "第 5 个偏好")
+
+    with closing(sqlite3.connect(database)) as db:
+        rows = db.execute(
+            "SELECT state,evidence_ids_json FROM memory_write_jobs ORDER BY queue_seq"
+        ).fetchall()
+    assert [(state, len(json.loads(ids))) for state, ids in rows] == [
+        ("assembling", 0),
+        ("pending", 10),
+    ]
+    assert store.count() == 11
+
+
 def test_file_adopts_the_hook_row_when_the_text_is_unique(tmp_path) -> None:
     store = _grant(tmp_path, "workbuddy", sync=True)
     database = tmp_path / "one-memory.sqlite3"
@@ -211,6 +258,7 @@ def test_hook_job_completes_after_file_identity_adoption(tmp_path) -> None:
         },
         database,
     )
+    assert len(_quiet_for_ten_minutes(database, store)) == 1
     for message_id, role, author, text in (
         ("file-user", "user", WrittenBy.PERSON, "我喜欢蓝莓"),
         ("file-assistant", "assistant", WrittenBy.AGENT, "我会记住你喜欢蓝莓"),

memhub/apps/core/tests/test_memory_additions.py

@@ -166,6 +166,57 @@ def test_revoked_manual_source_is_not_silently_reauthorized(tmp_path):
     asyncio.run(scenario())
 
 
+@pytest.mark.fresh_sdk
+def test_sdk_manual_add_without_a_grant_row_is_claimed(tmp_path, monkeypatch):
+    """The SDK add path writes no grant row; the claim gate used to require one.
+
+    The intake admitted the job, the drain skipped it on every tick, and the
+    note stayed Pending with attempts=0 while later jobs ran. A missing grant
+    row is allowed; only an explicit revocation holds the job.
+    """
+
+    async def scenario():
+        app = application(tmp_path)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            addition = {"request_id": str(uuid4()), "text": "Remember this.", "scope": "all-apps"}
+            assert (await client.post("/api/v2/memories", json=addition)).status_code == 201
+            database = app.state.settings.data_dir / "one-memory.sqlite3"
+            with sqlite3.connect(database) as db:
+                assert (
+                    db.execute(
+                        "SELECT COUNT(*) FROM source_grants WHERE source_id='one-memory-manual'"
+                    ).fetchone()[0]
+                    == 0
+                )
+                request_id = db.execute("SELECT request_id FROM memory_note_requests").fetchone()[0]
+            job = app.state.memory_write_queue.get(request_id)
+            assert job is not None and job.state == "pending"
+            monkeypatch.setattr(
+                app.state.model_profile_store, "active_identity", lambda: ("profile", 1)
+            )
+
+            assert app.state.memory_write_drain.claim(limit=1) == (job.job_id,)
+
+            # An explicit revocation still holds the job and refuses new adds.
+            app.state.memory_write_queue.settle(job.job_id, state="pending", owner=None)
+            with sqlite3.connect(database) as db:
+                db.execute(
+                    "INSERT INTO source_grants (user_id,source_id,root_path,requested_path,"
+                    "root_identity,candidate_kind,granted_at,revoked_at,revision) VALUES "
+                    "('local','one-memory-manual','x','x','manual','default',"
+                    "'2026-10-01T00:00:00+00:00','2026-10-27T00:00:00+00:00',1)"
+                )
+            assert app.state.memory_write_drain.claim(limit=1) == ()
+            refused = await client.post(
+                "/api/v2/memories",
+                json={**addition, "request_id": str(uuid4())},
+            )
+            assert refused.status_code == 409
+            assert refused.json()["detail"] == "memory_add_source_revoked"
+
+    asyncio.run(scenario())
+
+
 @pytest.mark.fresh_sdk
 def test_sdk_memory_write_accepts_openai_messages_without_evidence_tables(tmp_path):
     async def scenario():

memhub/apps/core/tests/test_memory_library_all_scenes.py

@@ -55,6 +55,9 @@ def chat_row(memory_id, content, *, source="cursor"):
     }
 
 
+EXPERIENCE_MEMORY_AT = 1_789_990_000
+
+
 def experience_row(memory_id, task, solution, *, variant="working", source="codebuddy"):
     return {
         "experience_id": memory_id,
@@ -63,6 +66,8 @@ def experience_row(memory_id, task, solution, *, variant="working", source="code
         "task": task,
         "solution": solution,
         "source": source,
+        # Conversation time, epoch seconds like chat (hy-memory 2.0.7).
+        "memory_at": EXPERIENCE_MEMORY_AT,
         "created_at": "2026-10-01T08:00:00",
         "updated_at": "2026-10-02T08:00:00",
         "_variant": variant,
@@ -90,6 +95,7 @@ class FakeSdk:
         self.productivity = {}
         self.search_calls = []
         self.list_calls = []
+        self.stats_calls = []
         self.get_calls = []
         self.update_calls = []
         self.delete_calls = []
@@ -227,6 +233,22 @@ class FakeSdk:
             "elapsed_ms": 1.0,
         }
 
+    def memory_stats(self, *, user_id, statuses=None):
+        self.stats_calls.append({"user_id": user_id, "statuses": statuses})
+        experiences = [
+            variant for (kind, variant, _id) in self.productivity if kind == "experience"
+        ]
+        facts = [variant for (kind, variant, _id) in self.productivity if kind == "fact"]
+        scenes = {"chat": {"count": len(self.chat)}}
+        for variant in ("coding", "working"):
+            scenes[variant] = {"count": experiences.count(variant), "facts": facts.count(variant)}
+        return {
+            "total": len(self.chat) + len(experiences),
+            "scenes": scenes,
+            "request_id": "stats",
+            "elapsed_ms": 1.0,
+        }
+
     def get(self, memory_id, kind=None, variant=None):
         self.get_calls.append((memory_id, kind, variant))
         match = self._matches(memory_id, kind, variant)
@@ -429,6 +451,7 @@ def test_keyword_filter_reads_experience_and_fact_text(tmp_path, monkeypatch):
         assert experience["content"] == "Fix flaky tests\nPin the pytest seed"
         assert experience["status"] == "active"
         assert experience["updated_at"] > experience["created_at"]
+        assert experience["memory_at"] == EXPERIENCE_MEMORY_AT
         fact = by_id["fact-1"]
         assert (fact["scene"], fact["kind"], fact["variant"]) == ("coding", "fact", "coding")
         assert fact["content"] == "The pytest suite runs in CI"
@@ -471,6 +494,10 @@ def test_empty_query_lists_all_scenes_with_source_filter_and_totals(tmp_path, mo
             "chat",
             "chat",
         ]
+        assert [item["memory_at"] for item in body["memories"][:2]] == [
+            EXPERIENCE_MEMORY_AT,
+            EXPERIENCE_MEMORY_AT,
+        ]
         call = sdk.list_calls[0]
         assert call["scene"] == "all"
         assert call["include_sources"] is None
@@ -510,6 +537,7 @@ def test_detail_edit_delete_pass_kind_and_variant(tmp_path, monkeypatch):
         assert record["task"] == "Old task"
         assert record["solution"] == "Old solution"
         assert record["content"] == "Old task\nOld solution"
+        assert record["memory_at"] == EXPERIENCE_MEMORY_AT
         assert sdk.get_calls[-1] == ("exp-1", "experience", "working")
 
         # Chat detail with no namespace keeps the plain SDK call.
@@ -647,3 +675,34 @@ def test_scene_filter_narrows_search_and_binds_the_window(tmp_path, monkeypatch)
         assert len(sdk.search_calls) == 2
 
     run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_home_summary_counts_what_the_memory_page_lists(tmp_path, monkeypatch):
+    """Home showed 55 (chat only) while the memory page showed 94."""
+
+    async def scenario(app, sdk, client):
+        sdk.add_chat(chat_row("chat-1", "cursor chat"))
+        sdk.add_chat(chat_row("chat-2", "codebuddy chat", source="codebuddy"))
+        sdk.add_productivity("experience", experience_row("exp-w", "task w", "solution w"))
+        sdk.add_productivity(
+            "experience", experience_row("exp-c", "task c", "solution c", variant="coding")
+        )
+        # Facts are not listed on the memory page, so they are not counted.
+        sdk.add_productivity("fact", fact_row("fact-1", "The pytest suite runs in CI"))
+
+        response = await client.get("/api/v2/home-summary")
+        assert response.status_code == 200
+        body = response.json()
+        library = (await client.get("/api/v2/memory-library")).json()
+        assert body["memory_count"] == library["total"] == 4
+        assert sdk.stats_calls == [{"user_id": "local", "statuses": ["active"]}]
+        # The newest memory across scenes, not the newest chat memory.
+        home_list = sdk.list_calls[0]
+        assert home_list["scene"] == "all"
+        assert (home_list["offset"], home_list["limit"], home_list["order"]) == (0, 1, "desc")
+        assert body["latest"]["memory_id"] == "exp-w"
+        assert body["latest"]["scene"] == "working"
+        assert body["latest"]["memory_at"] == EXPERIENCE_MEMORY_AT
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -1118,6 +1118,75 @@ def test_sdk_write_processor_uses_add_extracted_for_memory_content_and_additions
     assert len(client.calls) == 1
 
 
+def test_a_batch_with_no_user_or_assistant_message_is_nothing_to_remember(tmp_path) -> None:
+    """Tool-only input came back 400 "content or messages is required" (16 unknown).
+
+    The SDK strips tool records on the chat path. Such a batch completes with
+    no memories; the SDK is never called and nothing is recorded unknown.
+    """
+    from one_memory.imports.models import WrittenBy
+    from one_memory.sdk_write_processor import SdkWriteProcessor
+
+    client = _SdkPublicClient()
+    processor = SdkWriteProcessor(_SdkPublicHandle(client))  # type: ignore[arg-type]
+    tools = tuple(
+        _sdk_evidence(f"tool-{index}", text="3 passed").model_copy(
+            update={"role": "tool", "written_by": WrittenBy.TOOL}
+        )
+        for index in range(2)
+    )
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    record = queue.submit(
+        WriteRequest(
+            internal_session_id="sess-tools",
+            source_id="cursor",
+            batch_identity=BatchIdentity(
+                internal_session_id="sess-tools", first_round=1, last_round=1
+            ),
+            evidence_ids=[item.evidence_id for item in tools],
+            input_records=tools,
+            idempotency_key="tools-only",
+        )
+    )
+    _claim_through_drain(queue, record.job_id)
+
+    asyncio.run(
+        execute_job(
+            record.job_id,
+            queue=queue,
+            supervisor=processor,
+            evidence=_EvidenceStore(set()),
+        )
+    )
+
+    assert client.calls == []
+    assert client.events == []
+    [row] = _rows(path)
+    assert (row["state"], row["hy_invoked"], row["sdk_memory_ids_json"]) == (COMPLETED, 0, "[]")
+
+
+def test_the_drain_tick_writes_a_quiet_live_conversation(tmp_path, monkeypatch) -> None:
+    """The 10-minute idle flush runs on the drain's maintenance timer."""
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    flushed: list[str] = []
+    monkeypatch.setattr(queue, "flush_idle_assemblies", lambda: flushed.append("flush") or ())
+    drain = DurableWriteDrain(queue, execute=_never_execute, reap_interval=timedelta(0))
+
+    async def tick() -> None:
+        task = asyncio.create_task(drain.run_forever())
+        for _ in range(100):
+            await asyncio.sleep(0.01)
+            if flushed:
+                break
+        drain.stop()
+        await task
+
+    asyncio.run(tick())
+    assert flushed
+
+
 def test_sdk_unknown_outcome_inquiry_uses_only_public_receipt() -> None:
     from one_memory.sdk_write_processor import SdkWriteProcessor, SdkWriteReceipt
 

memhub/apps/core/tests/test_policy_api_wiring.py

@@ -127,6 +127,28 @@ def test_the_manual_source_is_admitted_by_grant_without_a_toggle(tmp_path) -> No
     assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
 
 
+def test_the_manual_source_is_allowed_without_a_grant_row_until_revoked(tmp_path) -> None:
+    """The SDK add path writes no grant row, and intake admits the record.
+
+    Claim must answer the same: a missing row is allowed, an explicit
+    revocation is not.
+    """
+    receipt: list[object] = [None]
+    app = SimpleNamespace(
+        state=SimpleNamespace(
+            config_store=ConfigStore(tmp_path),
+            model_profile_store=SimpleNamespace(active_identity=lambda: ("profile", 1)),
+            source_grant_store=SimpleNamespace(receipt=lambda **_kwargs: receipt[0]),
+        )
+    )
+
+    assert _source_writes_on(app, "one-memory-manual") is True
+    assert _sedimentation_claim_enabled(app, "one-memory-manual") is True
+    receipt[0] = SimpleNamespace(revoked_at="2026-10-27")
+    assert _source_writes_on(app, "one-memory-manual") is False
+    assert _sedimentation_claim_enabled(app, "one-memory-manual") is False
+
+
 def test_claim_gate_waits_without_polling_when_no_model_is_active(tmp_path) -> None:
     config = ConfigStore(tmp_path)
     app = SimpleNamespace(

memhub/apps/core/tests/test_sdk_data_plane_migration.py

@@ -137,19 +137,38 @@ def test_post_cutover_metadata_repair_restores_source_and_original_times(store):
     )
     coding = data_dir / "hy-memory" / "governed-v2" / "coding_cv.sqlite3"
     with sqlite3.connect(coding) as database:
-        database.execute(
-            "CREATE TABLE coding_memory_meta ("
-            "experience_id TEXT,user_id TEXT,agent_id TEXT,solution TEXT,"
-            "source TEXT,created_at TEXT,updated_at TEXT)"
+        # A store written by hy-memory 2.0.6, before the conversation time
+        # column: the repair must add it through the SDK and then fill it.
+        database.executescript(
+            """
+            CREATE TABLE coding_memory_meta (
+                experience_id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
+                agent_id TEXT NOT NULL DEFAULT 'default_agent', task TEXT NOT NULL,
+                search_keys TEXT NOT NULL DEFAULT '[]', solution TEXT NOT NULL,
+                boundary_envs TEXT NOT NULL DEFAULT '', workspace_id TEXT,
+                branch TEXT, session_id TEXT, files TEXT NOT NULL DEFAULT '[]',
+                source_fact_ids TEXT NOT NULL DEFAULT '[]',
+                confidence REAL NOT NULL DEFAULT 0.7,
+                source TEXT NOT NULL DEFAULT 'default',
+                extraction_source TEXT NOT NULL DEFAULT 'auto_extract',
+                created_at TEXT NOT NULL, updated_at TEXT NOT NULL
+            );
+            CREATE TABLE working_fact (
+                fact_id TEXT PRIMARY KEY, user_id TEXT NOT NULL,
+                agent_id TEXT NOT NULL DEFAULT 'default_agent', session_id TEXT,
+                workspace_id TEXT, branch TEXT, source TEXT NOT NULL DEFAULT 'default',
+                statement TEXT NOT NULL, files TEXT NOT NULL DEFAULT '[]',
+                subject TEXT NOT NULL DEFAULT '', details TEXT NOT NULL DEFAULT '',
+                fact_link TEXT, derived_experience_ids TEXT NOT NULL DEFAULT '[]',
+                status TEXT NOT NULL DEFAULT 'active',
+                fact_chain_summary TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL
+            );
+            """
         )
         database.execute(
-            "CREATE TABLE working_fact ("
-            "fact_id TEXT,user_id TEXT,agent_id TEXT,statement TEXT,"
-            "source TEXT,created_at TEXT)"
-        )
-        database.execute(
-            "INSERT INTO coding_memory_meta VALUES "
-            "('sdk-1','local','scope','remember','default','now','now')"
+            "INSERT INTO coding_memory_meta "
+            "(experience_id,user_id,agent_id,task,solution,source,created_at,updated_at) "
+            "VALUES ('sdk-1','local','scope','remember','remember','default','now','now')"
         )
 
     result = repair_sdk_metadata_from_bundle(
@@ -168,10 +187,11 @@ def test_post_cutover_metadata_repair_restores_source_and_original_times(store):
         2026, 10, 18, tzinfo=UTC
     )
     with sqlite3.connect(coding) as database:
-        source, created_at = database.execute(
-            "SELECT source,created_at FROM coding_memory_meta WHERE experience_id='sdk-1'"
+        source, memory_at, created_at = database.execute(
+            "SELECT source,memory_at,created_at FROM coding_memory_meta WHERE experience_id='sdk-1'"
         ).fetchone()
     assert source == "workbuddy"
+    assert datetime.fromtimestamp(memory_at, tz=UTC) == datetime(2026, 10, 17, tzinfo=UTC)
     assert created_at.startswith("2026-10-18T00:00:00")
 
     repeated = repair_sdk_metadata_from_bundle(
@@ -705,6 +725,29 @@ def test_cutover_waits_until_the_new_runtime_reads_every_chat_memory(store, acti
     assert store.authority() is MemoryAuthority.SDK
 
 
+def test_sdk_migrate_repairs_metadata_after_cutover_and_writes_the_receipt(store, active_profile):
+    """76/76 memories showed source=default after an upgrade whose repair failed.
+
+    The repair now runs here, offline, right after the cutover commits; Core's
+    startup finds the receipt and does nothing.
+    """
+    _confirmed(store, "m1")
+    home = store.database_path.parent
+
+    result = run_startup_migration(
+        data_dir=home,
+        runtime_build="21",
+        sdk_version="2.0.5",
+        namespace="governed-v2",
+        open_client=_reading({"m1"}),
+    )
+
+    assert result.state == "cutover"
+    receipt = home / f".sdk-metadata-repair-v2-{result.migration_id}.json"
+    assert receipt.is_file()
+    assert json.loads(receipt.read_text())["migration_id"] == result.migration_id
+
+
 def test_retired_coding_experiences_do_not_block_but_an_empty_store_does(store, active_profile):
     _confirmed(store, "chat-1")
     _confirmed(store, "exp-kept", "exp-retired", kind="coding")

memhub/apps/core/tests/test_sdk_migration_bundle.py

@@ -91,7 +91,7 @@ def test_bundle_contains_management_sdk_config_and_verified_manifest(tmp_path):
     assert manifest["bundle_id"] == "migration-1"
     assert manifest["sdk_namespace"] == "governed-v2"
     assert manifest["runtime_build"] == "build-13"
-    assert manifest["snapshot_sdk_version"] == "2.0.6"
+    assert manifest["snapshot_sdk_version"] == "2.0.7"
     paths = {item["path"]: item for item in manifest["files"]}
     assert {
         "management/one-memory.sqlite3",

memhub/apps/core/tests/test_unified_write_entry_wiring.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import json
 import sqlite3
 from uuid import uuid4
 
@@ -197,17 +198,20 @@ def test_live_import_callback_submits_realtime_priority(tmp_path) -> None:
             )
             app.state.config_store.set_sync(source_id, True)
             user = _committed_record(app, source_id=source_id, message_id="live-user")
-            record = app.state.evidence_store.put(
-                EvidenceInput(
-                    source_id=source_id,
-                    conversation_id=user.conversation_id,
-                    message_id="live-assistant",
-                    role="assistant",
-                    written_by=WrittenBy.AGENT,
-                    verbatim="Live reply",
-                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
-                )
-            ).record
+            # Live writes by the turn rule: 5 complete turns, the fifth
+            # completed by the next user message.
+            for index in range(10):
+                record = app.state.evidence_store.put(
+                    EvidenceInput(
+                        source_id=source_id,
+                        conversation_id=user.conversation_id,
+                        message_id=f"live-{index}",
+                        role="assistant" if index % 2 == 0 else "user",
+                        written_by=WrittenBy.AGENT if index % 2 == 0 else WrittenBy.PERSON,
+                        verbatim=f"Live line {index}",
+                        origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+                    )
+                ).record
             callback = app.state.import_supervisor.on_stable_evidence
             assert callback is not None
             token = live_sync_arrival.set(True)
@@ -216,11 +220,12 @@ def test_live_import_callback_submits_realtime_priority(tmp_path) -> None:
             finally:
                 live_sync_arrival.reset(token)
             with sqlite3.connect(app.state.evidence_store.database_path) as db:
-                priority = db.execute(
-                    "SELECT priority FROM memory_write_jobs WHERE idempotency_key LIKE ?",
+                rows = db.execute(
+                    "SELECT priority,evidence_ids_json FROM memory_write_jobs "
+                    "WHERE idempotency_key LIKE ?",
                     ("turn:v1:%",),
-                ).fetchone()[0]
-            assert priority == "live"
+                ).fetchall()
+            assert [(priority, len(json.loads(ids))) for priority, ids in rows] == [("live", 10)]
 
     asyncio.run(scenario())
 

memhub/apps/core/tests/test_whole_turn_hy_input.py

@@ -1,8 +1,8 @@
-"""D74: a turn reaches HY whole; only what HY will read counts toward the ceiling."""
+"""D74: a turn reaches HY whole; only what HY will read counts toward the batch size."""
 
 from datetime import UTC, datetime
 
-from one_memory.conversation_processing import _conversation_chunks
+from one_memory.conversation_processing import turn_batches
 from one_memory.imports.models import EvidenceInput, EvidenceOrigin, OriginMetadata, WrittenBy
 from one_memory.imports.storage import EvidenceStore
 
@@ -36,66 +36,33 @@ def test_tool_heavy_turn_is_one_window(tmp_path):
         *[("assistant", "定位中") for _ in range(30)],
         ("assistant", "原因是 fixture 没清理" * 100),
     )
-    chunks = _conversation_chunks(records)
-    assert len(chunks) == 1
-    assert chunks[0] == tuple(records)
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
 
 
 def test_long_answer_is_not_cut(tmp_path):
     records = _turn(tmp_path, ("user", "写一份方案"), ("assistant", "段落。" * 7_000))
-    chunks = _conversation_chunks(records)
-    assert chunks == [tuple(records)]
-    assert all(":segment:" not in record.evidence_id for record in chunks[0])
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
+    assert all(":segment:" not in record.evidence_id for record in batches[0])
 
 
-def test_text_over_the_ceiling_still_splits(tmp_path):
-    records = _turn(tmp_path, ("user", "整理"), ("assistant", "句子。" * 25_000))
-    chunks = _conversation_chunks(records)
-    assert len(chunks) > 1
-    assert all(chunk[0].evidence_id == records[0].evidence_id for chunk in chunks)
+def test_a_turn_far_over_the_batch_size_is_still_one_window(tmp_path):
+    """The rule never splits a turn: it is written alone, whole.
 
-
-def test_switch_restores_the_old_split(tmp_path, monkeypatch):
-    monkeypatch.setenv("ONE_MEMORY_SPLIT_TURNS", "1")
-    records = _turn(tmp_path, ("user", "写一份方案"), ("assistant", "段落。" * 7_000))
-    assert len(_conversation_chunks(records)) > 1
-
-
-def test_explicit_limits_keep_old_semantics(tmp_path):
-    records = _turn(tmp_path, ("user", "q"), *[("assistant", "a" * 100) for _ in range(30)])
-    assert len(_conversation_chunks(records, max_turns=6, max_characters=1200)) > 1
-
-
-def test_one_window_is_capped_in_raw_bytes(tmp_path):
-    """HY counts 2 KiB per tool record, but we still hold the raw text.
-
-    `client.add` is in-process, so a window's raw bytes are what the process
-    actually holds even though HY will cut each tool record down before it
-    reaches the model. Past the byte ceiling the window is cut, so no single
-    call can grow without bound.
+    The old windows cut text past 60k characters and raw bytes past 2 MB into
+    segments that repeated the user question. HY reads at most
+    `_HY_TOOL_CHARACTERS` of each tool record, so the window HY reads stays
+    bounded by the turn's own text.
     """
-    from one_memory.conversation_processing import _WINDOW_BYTE_CEILING
-
     records = _turn(
         tmp_path,
         ("user", "跑一下全部测试"),
         *[("tool", "y" * 400_000) for _ in range(20)],
+        ("assistant", "句子。" * 25_000),
     )
-    chunks = _conversation_chunks(records)
-    assert len(chunks) > 1
-    assert all(
-        sum(len(item.verbatim.encode("utf-8")) for item in chunk) <= _WINDOW_BYTE_CEILING
-        for chunk in chunks
-    )
-
-
-def test_the_byte_ceiling_counts_bytes_not_characters(tmp_path):
-    """One CJK character is three bytes; a character count would let three
-    times as much through."""
-    from one_memory.conversation_processing import _WINDOW_BYTE_CEILING, _window_bytes
-
-    assert _window_bytes(_turn(tmp_path, ("user", "中文"))[0]) > len("中文")
-    assert _WINDOW_BYTE_CEILING == 2_000_000
+    batches = turn_batches(records, flush=True).batches
+    assert batches == (tuple(records),)
 
 
 def test_the_sdk_strips_tool_messages_on_the_chat_path():

memhub/apps/core/uv.lock

@@ -991,7 +991,7 @@ wheels = [
 
 [[package]]
 name = "hy-memory"
-version = "2.0.6"
+version = "2.0.7"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "chromadb" },
@@ -1008,9 +1008,9 @@ dependencies = [
     { name = "scikit-learn" },
     { name = "tiktoken" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/18/98/312ac8bcbe41e63200a93d296300b2e6748e1a532dad01f009dad57ba408/hy_memory-2.0.6.tar.gz", hash = "sha256:d0f54d54867ee7e722a8dd1a88a00c0576904435f15ae7899fe640840bff74bc", size = 855200, upload-time = "2026-10-27T07:53:10.022Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/17/b5/87a43c8d8b77fdfbbe7974d1e93af50ddf82317c5d56e180b865fb19e181/hy_memory-2.0.7.tar.gz", hash = "sha256:85636487e5bd4b6595a8c15301499efb365b6010c236a45c51c144f6986a4222", size = 854913, upload-time = "2026-10-28T03:35:38.127Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/27/f1/d4866cffe426d77b857168cc694b15df866f3d9f45152c17c148e1595db4/hy_memory-2.0.6-py3-none-any.whl", hash = "sha256:b0e666f8e89d1763bb31c0e67502df74434134a638ff6fd1912de9e936be2483", size = 918955, upload-time = "2026-10-27T07:53:07.686Z" },
+    { url = "https://files.pythonhosted.org/packages/7c/b8/a6e0bfd17aeb432b0470f4dc707271e6476f5ac34f3b301ab3ad2e1c0d83/hy_memory-2.0.7-py3-none-any.whl", hash = "sha256:80a501e17987202ee1150a91b071313e890300620f353f9e89c967286dc1a645", size = 919841, upload-time = "2026-10-28T03:35:35.399Z" },
 ]
 
 [[package]]
@@ -1837,7 +1837,7 @@ requires-dist = [
     { name = "fastapi", specifier = ">=0.141.1" },
     { name = "filelock", specifier = ">=3.32.4" },
     { name = "httpx", specifier = ">=0.28.1" },
-    { name = "hy-memory", specifier = "==2.0.6" },
+    { name = "hy-memory", specifier = "==2.0.7" },
     { name = "keyring", specifier = ">=25.7.0" },
     { name = "numpy", specifier = "==2.5.2" },
     { name = "onnxruntime", specifier = "==1.23.2" },

memhub/apps/desktop/src/host-control-ownership.test.ts

@@ -15,6 +15,7 @@ const fake = vi.hoisted(() => ({
   missingBroker: false,
   hostCapability: true,
   open: vi.fn(),
+  refreshHosts: vi.fn(async (_layouts: unknown, _options: unknown) => ({})),
   host: vi.fn<
     (input: {
       hostId: string;
@@ -274,7 +275,9 @@ beforeEach(async () => {
   fake.open.mockResolvedValue({
     hostGrantStore: {},
     repairInstalledCommand: vi.fn(async () => []),
+    refreshInstalledHosts: fake.refreshHosts,
   });
+  fake.refreshHosts.mockClear();
   fake.host.mockResolvedValue({ state: "synthetic-remote-result" });
   Object.defineProperty(process, "resourcesPath", {
     configurable: true,
@@ -590,5 +593,11 @@ describe("packaged Desktop host-control ownership", () => {
     expect(fake.ensureRuntime).not.toHaveBeenCalled();
     expect(fake.createClient).not.toHaveBeenCalled();
     expect(fake.windows).toHaveLength(1);
+    // The embedded Runtime is this process, so Desktop refreshes the hosts
+    // it installed once the Runtime is up, as a standalone one would itself.
+    expect(fake.refreshHosts).toHaveBeenCalledTimes(1);
+    expect(fake.refreshHosts.mock.calls[0]?.[1]).toMatchObject({
+      diagnostics: expect.anything(),
+    });
   });
 });

memhub/apps/desktop/src/host-permissions.test.ts

@@ -21,6 +21,8 @@ import {
   claudeCodeLayout,
   codexLayout,
   cursorLayout,
+  MCP_TOOLSET_KEY,
+  MCP_TOOLSET_VERSION,
   mcpConfigPath,
   piLayout,
   planRecallInstall,
@@ -29,6 +31,7 @@ import {
   workbuddyLayout,
   workbuddyScatteredLayout,
 } from "@one-memory/host-bridge/installer";
+import type { DiagnosticInput } from "@one-memory/runtime-control";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import { probeRecallInstall } from "../../../packages/host-bridge/src/recall-probe.js";
 import { skillFilePath } from "../../../packages/host-bridge/src/skill-drift.js";
@@ -1384,3 +1387,197 @@ describe("upgrading installs left by an earlier release", () => {
     );
   });
 });
+
+describe("refreshing installed hosts after an upgrade", () => {
+  function recorder() {
+    const events: DiagnosticInput[] = [];
+    return {
+      events,
+      sink: {
+        log: async (event: DiagnosticInput) => {
+          events.push(event);
+        },
+      },
+    };
+  }
+
+  async function linkedWorkbuddy(context: Awaited<ReturnType<typeof fixture>>) {
+    const layout = workbuddyLayout(path.join(context.root, "workbuddy"));
+    const preview = await context.permissions.previewInstall(layout);
+    const result = await context.permissions.install(
+      layout,
+      preview.fingerprint,
+    );
+    await context.permissions.setPermission(
+      layout,
+      approval(required(result.permission.receipt)),
+    );
+    return layout;
+  }
+
+  /** Put back the skill an earlier release wrote, journaled as ours. */
+  async function ageSkill(layout: ReturnType<typeof workbuddyLayout>) {
+    const skillPath = required(skillFilePath(layout));
+    const aged = (await readFile(skillPath, "utf8"))
+      .replace(/skill-version=\d+/, "skill-version=15")
+      .replace("`MemHub_recall`", "`MemHub_ask`");
+    await writeFile(skillPath, aged);
+    const journal = JSON.parse(await readFile(layout.journalPath, "utf8"));
+    journal.installed_skill_hash = createHash("sha256")
+      .update(aged)
+      .digest("hex");
+    await writeFile(layout.journalPath, JSON.stringify(journal));
+    return aged;
+  }
+
+  it("rewrites an old skill on the first start of a build, and only once per build", async () => {
+    const context = await fixture();
+    const layout = await linkedWorkbuddy(context);
+    const current = await readFile(required(skillFilePath(layout)), "utf8");
+    await ageSkill(layout);
+    const { events, sink } = recorder();
+
+    const first = await context.permissions.refreshInstalledHosts([layout], {
+      build: "34",
+      diagnostics: sink,
+    });
+    expect(first).toMatchObject({
+      state: "refreshed",
+      synced: ["workbuddy"],
+      failed: [],
+    });
+    expect(await readFile(required(skillFilePath(layout)), "utf8")).toBe(
+      current,
+    );
+
+    // Same build again: one file read, nothing rewritten, even if stale.
+    const aged = await ageSkill(layout);
+    expect(
+      await context.permissions.refreshInstalledHosts([layout], {
+        build: "34",
+        diagnostics: sink,
+      }),
+    ).toMatchObject({ state: "already_refreshed", synced: [] });
+    expect(await readFile(required(skillFilePath(layout)), "utf8")).toBe(aged);
+
+    // The next build refreshes again.
+    expect(
+      await context.permissions.refreshInstalledHosts([layout], {
+        build: "35",
+      }),
+    ).toMatchObject({ state: "refreshed", synced: ["workbuddy"] });
+    expect(await readFile(required(skillFilePath(layout)), "utf8")).toBe(
+      current,
+    );
+
+    // Diagnostics carry host ids and counts only, never file content.
+    expect(events.map((event) => event.event)).toEqual([
+      "host_refresh.host_updated",
+      "host_refresh.completed",
+      "host_refresh.skipped",
+    ]);
+    expect(events[0].details).toEqual({ source_id: "workbuddy" });
+    expect(events[1].details).toEqual({
+      count: 1,
+      updated_count: 1,
+      error_count: 0,
+    });
+    expect(JSON.stringify(events)).not.toContain("MemHub_");
+    expect(JSON.stringify(events)).not.toContain(context.root);
+  });
+
+  it("leaves a skill the user edited untouched", async () => {
+    const context = await fixture();
+    const layout = await linkedWorkbuddy(context);
+    const edited = `${await ageSkill(layout)}\nUser addition\n`;
+    await writeFile(required(skillFilePath(layout)), edited);
+
+    expect(
+      await context.permissions.refreshInstalledHosts([layout], {
+        build: "34",
+      }),
+    ).toMatchObject({ state: "refreshed", synced: [], failed: [] });
+    expect(await readFile(required(skillFilePath(layout)), "utf8")).toBe(
+      edited,
+    );
+  });
+
+  it("rewrites an MCP entry without the tool set stamp, in the plugin manifest too", async () => {
+    const context = await fixture();
+    const layout = await linkedWorkbuddy(context);
+    const plugin = required(layout.plugin);
+    const manifests = [
+      path.join(plugin.installRoot, ".codebuddy-plugin", "plugin.json"),
+      path.join(
+        plugin.marketplaceRoot,
+        "plugins",
+        plugin.pluginName,
+        ".codebuddy-plugin",
+        "plugin.json",
+      ),
+    ];
+    // What an earlier release wrote: the same entry with no stamp, journaled.
+    const journal = JSON.parse(await readFile(layout.journalPath, "utf8"));
+    for (const manifestPath of manifests) {
+      const manifest = JSON.parse(await readFile(manifestPath, "utf8"));
+      expect(manifest.mcpServers["one-memory"].env[MCP_TOOLSET_KEY]).toBe(
+        MCP_TOOLSET_VERSION,
+      );
+      delete manifest.mcpServers["one-memory"].env[MCP_TOOLSET_KEY];
+      const text = `${JSON.stringify(manifest, null, 2)}\n`;
+      await writeFile(manifestPath, text);
+      const entry = journal.plugin_files.find(
+        (file: { path: string }) => file.path === manifestPath,
+      );
+      entry.hash = createHash("sha256").update(text).digest("hex");
+      journal.installed_server_hash = createHash("sha256")
+        .update(JSON.stringify(manifest.mcpServers["one-memory"]))
+        .digest("hex");
+    }
+    await writeFile(layout.journalPath, JSON.stringify(journal));
+    const granted = await context.permissions.getPermission(layout);
+    expect(granted.state).toBe("approved");
+    expect(
+      (await readInstalledRecallCommand(layout))?.env[MCP_TOOLSET_KEY],
+    ).toBeUndefined();
+
+    expect(
+      await context.permissions.refreshInstalledHosts([layout], {
+        build: "34",
+      }),
+    ).toMatchObject({ synced: ["workbuddy"], failed: [] });
+    for (const manifestPath of manifests) {
+      const manifest = JSON.parse(await readFile(manifestPath, "utf8"));
+      expect(manifest.mcpServers["one-memory"].env[MCP_TOOLSET_KEY]).toBe(
+        MCP_TOOLSET_VERSION,
+      );
+    }
+    // Rewriting our own entry keeps the user's authorization.
+    expect(await context.permissions.getPermission(layout)).toEqual(granted);
+  });
+
+  it("rewrites a config-file MCP entry without the stamp", async () => {
+    const context = await fixture();
+    await install(context);
+    const configPath = mcpConfigPath(context.layout);
+    const config = JSON.parse(await readFile(configPath, "utf8"));
+    delete config.mcpServers["one-memory"].env[MCP_TOOLSET_KEY];
+    await writeFile(configPath, `${JSON.stringify(config, null, 2)}\n`);
+    const journal = JSON.parse(
+      await readFile(context.layout.journalPath, "utf8"),
+    );
+    journal.installed_server_hash = createHash("sha256")
+      .update(JSON.stringify(config.mcpServers["one-memory"]))
+      .digest("hex");
+    await writeFile(context.layout.journalPath, JSON.stringify(journal));
+
+    expect(
+      await context.permissions.refreshInstalledHosts([context.layout]),
+    ).toMatchObject({ state: "refreshed", synced: ["cursor"], failed: [] });
+    const refreshed = JSON.parse(await readFile(configPath, "utf8"));
+    expect(refreshed.mcpServers["one-memory"].env[MCP_TOOLSET_KEY]).toBe(
+      MCP_TOOLSET_VERSION,
+    );
+    expect(refreshed.mcpServers.neighbor).toEqual({ command: "keep" });
+  });
+});

memhub/apps/desktop/src/main.ts

@@ -1369,6 +1369,24 @@ async function initialize(isQuitting: () => boolean): Promise<void> {
     });
     await runtimeSupervisor.start();
     if (isQuitting()) return;
+    // A standalone Runtime refreshes the hosts it installed as it starts, so
+    // an upgrade through the App or `memhub upgrade` is covered there. An
+    // embedded one is this process: do the same here, now it is up. Only a
+    // packaged build, so a source checkout never repoints real hosts at itself.
+    if (localPermissions && app.isPackaged) {
+      void localPermissions
+        .refreshInstalledHosts(
+          hosts.map((host) => host.layout),
+          {
+            build:
+              packagedBuildNumber === undefined
+                ? undefined
+                : String(packagedBuildNumber),
+            diagnostics: hostDiagnostics,
+          },
+        )
+        .catch(() => undefined);
+    }
     if (
       app.isPackaged &&
       process.platform === "darwin" &&

memhub/apps/web/src/MemoryMap.test.tsx

@@ -5,7 +5,6 @@ import type {
   MemoryGraphNode,
   MemoryGraphResponse,
   MemoryProcessingSummary,
-  SourceImportReceipt as Receipt,
   SourceResponse,
 } from "@one-memory/contracts";
 import {
@@ -40,12 +39,6 @@ import { afterEach, expect, it, vi } from "vitest";
 import { MemoryMap } from "./MemoryMap";
 import type { MemoryArrival } from "./memory-arrival-state";
 
-const { sourceReceipt } = vi.hoisted(() => ({
-  sourceReceipt: vi.fn(),
-}));
-vi.mock("./api", () => ({
-  getSourceImportReceipt: sourceReceipt,
-}));
 vi.mock("./MemoryGraph3D", () => ({
   MemoryGraph3D: ({
     graph,
@@ -517,7 +510,7 @@ it("D96: the memory page has no raw-dialogue tab and never loads the dialogue li
 it("D96: lists a memory by when its conversation happened, not when it was generated", () => {
   const today = new Date().toISOString();
   const thisYear = new Date().getFullYear();
-  const items = ["第一条", "第二条", "第三条"].map((content, i) => ({
+  const items = ["第一条", "第二条", "第三条", "第四条"].map((content, i) => ({
     ...memories[0],
     memory_id: `t${i}`,
     content,
@@ -532,10 +525,11 @@ it("D96: lists a memory by when its conversation happened, not when it was gener
         total: 519,
         offset: 100,
         limit: 100,
-        source_ids_by_memory: { t0: [], t1: [], t2: [] },
+        source_ids_by_memory: { t0: [], t1: [], t2: [], t3: [] },
         source_time_by_memory: {
           t0: { at: `${thisYear}-10-02T10:54:00`, basis: "conversation" },
           t1: { at: `${thisYear}-09-30T08:00:00`, basis: "import" },
+          t3: { at: `${thisYear}-09-29T08:00:00`, basis: "write" },
         },
       })}
     />,
@@ -547,6 +541,10 @@ it("D96: lists a memory by when its conversation happened, not when it was gener
   expect(
     within(list).getByRole("button", { name: /第二条/ }).textContent,
   ).toContain("导入于 9月30日");
+  // An experience stored without a conversation time names its write time.
+  expect(
+    within(list).getByRole("button", { name: /第四条/ }).textContent,
+  ).toContain("写入于 9月29日");
   // D99: the generated day keeps its own line, so a memory made today from an
   // old conversation still reads as new rather than looking misplaced.
   expect(
@@ -982,35 +980,6 @@ const importedSources = [
     imported_conversations: 1,
   },
 ] as SourceResponse[];
-const importedReceipt: Receipt = {
-  source_id: "codex",
-  measured_at: "2026-09-14T00:00:00Z",
-  messages: 5,
-  conversations: 2,
-  person_messages: 2,
-  agent_messages: 2,
-  tool_messages: 0,
-  unknown_author_messages: 1,
-  unknown_time_messages: 1,
-  examples: [
-    {
-      evidence_id: "raw-codex",
-      role: "user",
-      written_by: "person",
-      created_at: null,
-      excerpt: "合成来源原文，不是治理记忆。",
-      truncated: false,
-    },
-  ],
-};
-async function openSourceReceipt() {
-  await userEvent.click(
-    screen.getByRole("button", { name: "图中选择 Codex · 2 段" }),
-  );
-  await userEvent.click(screen.getByText("查看已读内容"));
-  await waitFor(() => expect(sourceReceipt).toHaveBeenCalledWith("codex"));
-}
-
 it("REV09: selecting an imported source retains the canvas and only its authorized node", async () => {
   render(<Host items={[]} sources={importedSources} />);
   const canvas = screen.getByTestId("canvas");
@@ -1025,7 +994,8 @@ it("REV09: selecting an imported source retains the canvas and only its authoriz
   expect(
     screen.getByRole("list", { name: "地图记忆列表" }).children,
   ).toHaveLength(0);
-  expect(sourceReceipt).not.toHaveBeenCalled();
+  expect(screen.getByRole("heading", { name: "记忆详情" })).toBeTruthy();
+  expect(screen.getByText("还没有已生成的记忆")).toBeTruthy();
   await userEvent.click(screen.getByRole("button", { name: /全部记忆/ }));
   expect(within(canvas).getAllByRole("button")).toHaveLength(2);
   expect(canvas.dataset.selected ?? "").toBe("");
@@ -1047,83 +1017,247 @@ it("REV09: fallback search matches source names and IDs without bypassing the se
   );
   await userEvent.type(search, "cOdEx");
   expect(screen.getByTestId("canvas").dataset.selected).toBe("source:codex");
-  expect(screen.getByText("查看已读内容")).toBeTruthy();
+  expect(screen.getByText("还没有已生成的记忆")).toBeTruthy();
   await userEvent.clear(search);
   await userEvent.type(search, "Cursor");
   expect(screen.queryByTestId("canvas")).toBeNull();
-  expect(screen.queryByText("查看已读内容")).toBeNull();
+  expect(screen.queryByText("正在读取记忆详情…")).toBeNull();
   await userEvent.clear(search);
   expect(screen.getByTestId("canvas").dataset.selected ?? "").toBe("");
-  expect(screen.queryByText("查看已读内容")).toBeNull();
+  expect(screen.queryByText("正在读取记忆详情…")).toBeNull();
 });
 
-it("REV09: source selection uses the actual permission-checked receipt API", async () => {
-  sourceReceipt.mockResolvedValue(importedReceipt);
-  render(<Host items={[]} sources={importedSources} />);
-  await openSourceReceipt();
-  expect(
-    await screen.findByText(importedReceipt.examples[0].excerpt),
-  ).toBeTruthy();
-  expect(screen.getByText("用户 · 时间未知")).toBeTruthy();
-  expect(screen.getByText("原始消息").nextElementSibling?.textContent).toBe(
-    "5",
+it("selecting a source node waits for its settled page, then opens the first memory", async () => {
+  const onQuery = vi.fn();
+  const onSource = vi.fn();
+  const resetPage = vi.fn();
+  const oldPage = {
+    items: memories,
+    total: 2,
+    offset: 0,
+    limit: 20,
+    source_ids_by_memory: { m0: ["codex"], m1: ["codex"] },
+  };
+  const view = render(
+    <Host
+      simplified
+      sources={importedSources}
+      libraryView={{
+        ...pagedView(oldPage),
+        query: "旧查询",
+        settledQuery: "旧查询",
+        onQuery,
+        onSource,
+        resetPage,
+      }}
+    />,
   );
-  // No raw conversation is read by id on the SDK data plane.
-  expect(screen.queryByRole("button", { name: "查看完整对话" })).toBeNull();
+  await act(async () => {});
+  await userEvent.click(screen.getByRole("button", { name: "图中选择 Codex" }));
+  expect(onQuery).toHaveBeenCalledWith("");
+  expect(onSource).toHaveBeenCalledWith("codex");
+  expect(resetPage).toHaveBeenCalledOnce();
+
+  view.rerender(
+    <Host
+      simplified
+      sources={importedSources}
+      libraryView={{
+        ...pagedView(oldPage),
+        source: "codex",
+        settledSource: "",
+        loading: true,
+        onQuery,
+        onSource,
+        resetPage,
+      }}
+    />,
+  );
+  expect(
+    screen
+      .getAllByRole("status")
+      .some((node) => node.textContent === "正在读取记忆详情…"),
+  ).toBe(true);
+
+  const settledItems = [memories[1], memories[0]];
+  view.rerender(
+    <Host
+      simplified
+      items={settledItems}
+      sources={importedSources}
+      libraryView={{
+        ...pagedView({
+          ...oldPage,
+          items: settledItems,
+          source_ids_by_memory: { m1: ["codex"], m0: ["codex"] },
+        }),
+        source: "codex",
+        settledSource: "codex",
+        settledQuery: "",
+        onQuery,
+        onSource,
+        resetPage,
+      }}
+    />,
+  );
+  expect(await screen.findByRole("heading", { name: "第二条" })).toBeTruthy();
+  expect(screen.queryByText("查看已读内容")).toBeNull();
+  expect(document.querySelector(".source-import-receipt")).toBeNull();
 });
 
-it.each(["revoked", "removed", "revoked-hidden"])(
-  "REV09: %s source drops fallback, preview and a late receipt without reopening on reauthorization",
-  async (state) => {
-    let finish: (value: Receipt) => void = () => {};
-    sourceReceipt.mockImplementation(
-      () =>
-        new Promise<Receipt>((resolve) => {
-          finish = resolve;
-        }),
-    );
-    const view = render(<Host items={[]} sources={[importedSources[0]]} />);
-    await openSourceReceipt();
-    view.rerender(
-      <Host
-        items={[]}
-        hidden={state === "revoked-hidden"}
-        sources={
-          state === "removed"
-            ? []
-            : [{ ...importedSources[0], authorized: false }]
-        }
-      />,
-    );
-    expect(screen.queryByTestId("canvas")).toBeNull();
-    expect(document.querySelector(".source-import-receipt")).toBeNull();
-    await act(async () => finish(importedReceipt));
-    view.rerender(<Host items={[]} sources={[importedSources[0]]} />);
-    expect(screen.getByTestId("canvas").dataset.selected ?? "").toBe("");
-    expect(screen.queryByText(importedReceipt.examples[0].excerpt)).toBeNull();
-    expect(screen.queryByText("查看已读内容")).toBeNull();
-    expect(sourceReceipt).toHaveBeenCalledTimes(1);
-  },
-);
+it("does not replace a newer memory choice when a source page settles", async () => {
+  const page = {
+    items: memories,
+    total: 2,
+    offset: 0,
+    limit: 20,
+    source_ids_by_memory: { m0: ["codex"], m1: ["codex"] },
+  };
+  const view = render(
+    <Host simplified sources={importedSources} libraryView={pagedView(page)} />,
+  );
+  await act(async () => {});
+  await userEvent.click(screen.getByRole("button", { name: "图中选择 Codex" }));
+  await userEvent.click(
+    screen.getByRole("button", { name: "图中选择 第二条" }),
+  );
+  expect(await screen.findByRole("heading", { name: "第二条" })).toBeTruthy();
 
-it("REV09: compact preview close keeps source selection for reopening", async () => {
+  view.rerender(
+    <Host
+      simplified
+      sources={importedSources}
+      libraryView={{
+        ...pagedView(page),
+        source: "codex",
+        settledSource: "codex",
+      }}
+    />,
+  );
+  expect(screen.getByRole("heading", { name: "第二条" })).toBeTruthy();
+});
+
+it("closing the compact source inspector cancels its pending auto-open", async () => {
   vi.stubGlobal("matchMedia", (query: string) => ({
     matches: query.includes("max-width"),
     addEventListener() {},
     removeEventListener() {},
   }));
-  sourceReceipt.mockResolvedValue(importedReceipt);
-  render(<Host items={[]} sources={importedSources} />);
-  await openSourceReceipt();
-  await screen.findByText(importedReceipt.examples[0].excerpt);
+  const page = {
+    items: memories,
+    total: 2,
+    offset: 0,
+    limit: 20,
+    source_ids_by_memory: { m0: ["codex"], m1: ["codex"] },
+  };
+  const view = render(
+    <Host simplified sources={importedSources} libraryView={pagedView(page)} />,
+  );
+  await act(async () => {});
+  await userEvent.click(screen.getByRole("button", { name: "图中选择 Codex" }));
   await userEvent.click(screen.getByRole("button", { name: "关闭记忆详情" }));
-  expect(document.querySelector(".source-import-receipt")).toBeNull();
+
+  view.rerender(
+    <Host
+      simplified
+      sources={importedSources}
+      libraryView={{
+        ...pagedView(page),
+        source: "codex",
+        settledSource: "codex",
+      }}
+    />,
+  );
+  expect(screen.queryByRole("dialog", { name: "记忆与原文" })).toBeNull();
+  expect(screen.queryByRole("heading", { name: "第一条" })).toBeNull();
+});
+
+it("keeps the source selection and offers retry when its page fails", async () => {
+  const onRetry = vi.fn();
+  const page = {
+    items: memories,
+    total: 2,
+    offset: 0,
+    limit: 20,
+    source_ids_by_memory: { m0: ["codex"], m1: ["codex"] },
+  };
+  const view = render(
+    <Host simplified sources={importedSources} libraryView={pagedView(page)} />,
+  );
+  await act(async () => {});
+  await userEvent.click(screen.getByRole("button", { name: "图中选择 Codex" }));
+  view.rerender(
+    <Host
+      simplified
+      sources={importedSources}
+      libraryView={{
+        ...pagedView(page),
+        source: "codex",
+        settledSource: "codex",
+        failed: true,
+        onRetry,
+      }}
+    />,
+  );
+  expect(
+    screen
+      .getAllByRole("alert")
+      .some((node) => node.textContent === "记忆暂时无法读取"),
+  ).toBe(true);
+  await userEvent.click(screen.getByRole("button", { name: "重试" }));
+  expect(onRetry).toHaveBeenCalledOnce();
+});
+
+it("drops a pending source selection when that source is revoked", async () => {
+  const page = {
+    items: memories,
+    total: 2,
+    offset: 0,
+    limit: 20,
+    source_ids_by_memory: { m0: ["codex"], m1: ["codex"] },
+  };
+  const view = render(
+    <Host
+      simplified
+      sources={[importedSources[0]]}
+      libraryView={pagedView(page)}
+    />,
+  );
+  await act(async () => {});
+  await userEvent.click(screen.getByRole("button", { name: "图中选择 Codex" }));
   expect(screen.getByTestId("canvas").dataset.selected).toBe("source:codex");
-  await userEvent.click(
-    screen.getByRole("button", { name: "查看选中来源原文" }),
+
+  view.rerender(
+    <Host
+      simplified
+      sources={[{ ...importedSources[0], authorized: false }]}
+      libraryView={{
+        ...pagedView(page),
+        source: "codex",
+        settledSource: "codex",
+      }}
+    />,
+  );
+  expect(screen.queryByRole("button", { name: "图中选择 Codex" })).toBeNull();
+  expect(screen.getByTestId("canvas").dataset.selected).not.toBe(
+    "source:codex",
+  );
+  expect(screen.queryByText("正在读取记忆详情…")).toBeNull();
+
+  view.rerender(
+    <Host
+      simplified
+      sources={[importedSources[0]]}
+      libraryView={{
+        ...pagedView(page),
+        source: "codex",
+        settledSource: "codex",
+      }}
+    />,
+  );
+  expect(screen.getByTestId("canvas").dataset.selected).not.toBe(
+    "source:codex",
   );
-  expect(screen.getByRole("dialog", { name: "记忆与原文" })).toBeTruthy();
-  expect(screen.getByText("查看已读内容")).toBeTruthy();
 });
 
 it("REV09: first governed memories replace fallback without remounting the graph or breaking memory IDs", async () => {
@@ -1162,5 +1296,4 @@ it.each([
   render(<Host items={[]} sources={sources} />);
   expect(screen.queryByTestId("canvas")).toBeNull();
   expect(document.querySelector(".source-import-receipt")).toBeNull();
-  expect(sourceReceipt).not.toHaveBeenCalled();
 });

memhub/apps/web/src/MemoryMap.tsx

@@ -48,7 +48,6 @@ import {
 import { OptionMenu } from "./OptionMenu";
 import { R6Button } from "./R6Controls";
 import { SelectedMemoryDetail } from "./SelectedMemoryDetail";
-import { SourceImportReceipt } from "./SourceImportReceipt";
 import { sourceNameOr } from "./source-label";
 import { useMediaPreference } from "./use-media-preference";
 import { Answer, NodeDetail, type SavedAnswer } from "./Workbench";
@@ -97,6 +96,8 @@ type Props = {
     loading: boolean;
     failed: boolean;
     stale: boolean;
+    settledSource?: string;
+    settledQuery?: string;
     /** The source id the SDK uses for "no source given" (shown as 未识别来源). */
     unattributedSource?: string;
     onQuery(value: string): void;
@@ -139,7 +140,7 @@ function monthDay(value: string, locale: GuiLocale): string | null {
 
 /** D96: when the cited conversation happened, not when the memory was generated. */
 function sourceStamp(
-  time: { at: string; basis: "conversation" | "import" } | undefined,
+  time: { at: string; basis: "conversation" | "import" | "write" } | undefined,
   fallback: string | null | undefined,
   locale: GuiLocale = "zh-CN",
 ): string {
@@ -147,7 +148,11 @@ function sourceStamp(
   if (time && day)
     return memoryText(
       locale,
-      time.basis === "conversation" ? "对话于 {day}" : "导入于 {day}",
+      time.basis === "conversation"
+        ? "对话于 {day}"
+        : time.basis === "write"
+          ? "写入于 {day}"
+          : "导入于 {day}",
       { day },
     );
   return dayStamp(fallback, locale);
@@ -213,6 +218,9 @@ export function MemoryMap(props: Props) {
   }, [submittedQuery]);
   const [selectedId, setSelectedId] = useState<string | null>(null);
   const [groupNode, setGroupNode] = useState<MemoryGraphNode | null>(null);
+  const [pendingSourceSelection, setPendingSourceSelection] = useState<
+    string | null
+  >(null);
   const [question, setQuestion] = useState("");
   const [inspectorOpen, setInspectorOpen] = useState(false);
   const [filterNotice, setFilterNotice] = useState("");
@@ -363,6 +371,7 @@ export function MemoryMap(props: Props) {
     // Source revocation also clears a suspended preview, without closing shared memory detail.
     if (groupNode && !visibleGroupNode) {
       setGroupNode(null);
+      setPendingSourceSelection(null);
       setInspectorOpen(false);
     }
   }, [groupNode, visibleGroupNode]);
@@ -389,6 +398,7 @@ export function MemoryMap(props: Props) {
   const pick = useCallback(
     (memoryId: string, scroll = true) => {
       if (!active.some((memory) => memory.memory_id === memoryId)) return;
+      setPendingSourceSelection(null);
       scrollSelection.current = scroll;
       if (!filtered.some((memory) => memory.memory_id === memoryId)) {
         setGroupKey("");
@@ -402,6 +412,32 @@ export function MemoryMap(props: Props) {
     },
     [active, filtered, onOpenMemory],
   );
+  useEffect(() => {
+    if (
+      !library ||
+      !pendingSourceSelection ||
+      overlaysSuspended ||
+      library.loading ||
+      library.failed ||
+      !selectedSource ||
+      selectedSource.source_id !== pendingSourceSelection ||
+      library.source !== pendingSourceSelection ||
+      (library.settledSource ?? library.source) !== pendingSourceSelection ||
+      (library.settledQuery ?? library.query).trim() ||
+      library.page?.offset !== 0
+    )
+      return;
+    setPendingSourceSelection(null);
+    const first = active[0];
+    if (first) pick(first.memory_id, false);
+  }, [
+    active,
+    library,
+    overlaysSuspended,
+    pendingSourceSelection,
+    pick,
+    selectedSource,
+  ]);
   useEffect(() => {
     if (!props.simplified || openedDefault.current || overlaysSuspended) return;
     if (library?.loading && filtered.length === 0) return;
@@ -452,6 +488,7 @@ export function MemoryMap(props: Props) {
   const clear = useCallback(() => {
     setSelectedId(null);
     setGroupNode(null);
+    setPendingSourceSelection(null);
     setInspectorOpen(false);
     onCloseDetail();
   }, [onCloseDetail]);
@@ -483,6 +520,7 @@ export function MemoryMap(props: Props) {
       clear();
       return;
     }
+    setPendingSourceSelection(null);
     setInspectorOpen(false);
   }
   function chooseGroup(key: string) {
@@ -509,8 +547,17 @@ export function MemoryMap(props: Props) {
       setAxis("source");
       setGroupKey(`source:${node.source_id}`);
       if (library) {
+        setPendingSourceSelection(node.source_id);
+        library.onQuery("");
         library.onSource(node.source_id);
         library.resetPage();
+      } else {
+        const first = sourceGroups
+          .find((entry) => entry.key === `source:${node.source_id}`)
+          ?.memoryIds.find((memoryId) =>
+            active.some((memory) => memory.memory_id === memoryId),
+          );
+        if (first) pick(first, false);
       }
     }
     if (node.node_type === "topic") {
@@ -987,11 +1034,13 @@ export function MemoryMap(props: Props) {
             <button
               type="button"
               className="quiet map-tool"
-              aria-label={text(
-                selectedSource ? "查看选中来源原文" : "查看选中记忆原文",
-              )}
+              aria-label={text("记忆详情")}
               title={text("查看原文")}
-              onClick={() => setInspectorOpen(true)}
+              onClick={() => {
+                setInspectorOpen(true);
+                if (selectedSource && library)
+                  setPendingSourceSelection(selectedSource.source_id);
+              }}
             >
               <FileText size={17} />
             </button>
@@ -1126,7 +1175,7 @@ export function MemoryMap(props: Props) {
             {inspectorDetail
               ? text("记忆详情")
               : selectedSource
-                ? text("来源原文")
+                ? text("记忆详情")
                 : visibleGroupNode
                   ? text("关联节点")
                   : text("记忆详情")}
@@ -1145,13 +1194,33 @@ export function MemoryMap(props: Props) {
         </header>
         <div className="map-inspector-body" ref={inspectorBody}>
           {selectedSource ? (
-            inspectorOpen &&
-            !overlaysSuspended && (
-              <SourceImportReceipt
-                key={selectedSource.source_id}
-                source={selectedSource}
-              />
-            )
+            <div className="map-inspector-empty map-inspector-loading">
+              {library?.failed &&
+              library.source === selectedSource.source_id ? (
+                <>
+                  <p role="alert">{text("记忆暂时无法读取")}</p>
+                  <button
+                    type="button"
+                    className="secondary"
+                    onClick={library.onRetry}
+                  >
+                    {text("重试")}
+                  </button>
+                </>
+              ) : pendingSourceSelection || library?.loading ? (
+                <>
+                  <div className="map-detail-skeleton" aria-hidden="true">
+                    <i />
+                    <i />
+                    <i />
+                    <i />
+                  </div>
+                  <p role="status">{text("正在读取记忆详情…")}</p>
+                </>
+              ) : (
+                <p>{text("还没有已生成的记忆")}</p>
+              )}
+            </div>
           ) : props.simplified && inspectorDetail ? (
             <div
               aria-busy={

memhub/apps/web/src/MemorySpace.tsx

@@ -1027,6 +1027,8 @@ export function MemorySpace({
                 loading: sdkLibrary.loading,
                 failed: sdkLibrary.failed,
                 stale: false,
+                settledSource: sdkLibrary.settledSource,
+                settledQuery: sdkLibrary.settledQuery,
                 unattributedSource: UNATTRIBUTED_SOURCE_ID,
                 onQuery: setSdkLibraryQuery,
                 onSource: setSdkLibrarySource,

memhub/apps/web/src/SelectedMemoryDetail.test.tsx

@@ -176,3 +176,41 @@ it("stamps today's source and conversation time without an evidence page", () =>
   expect(screen.getByText(/对话时间 今天 /)).toBeTruthy();
   expect(screen.queryByRole("button", { name: "查看原文" })).toBeNull();
 });
+it("shows an experience's conversation time when the SDK stored one", () => {
+  const today = new Date().toISOString();
+  render(
+    <SelectedMemoryDetail
+      detail={{
+        ...detail,
+        evidence: [],
+        sdk_status: "active",
+        source_id: "codebuddy",
+        sdk_memory_at: today,
+        sdk_created_at: today,
+        sdk_ref: { scene: "coding", kind: "experience", variant: "coding" },
+      }}
+      sources={[]}
+    />,
+  );
+  expect(screen.getByText(/对话时间 今天 /)).toBeTruthy();
+  expect(screen.queryByText(/时间未知/)).toBeNull();
+});
+it("labels an experience without a conversation time by its write time", () => {
+  render(
+    <SelectedMemoryDetail
+      detail={{
+        ...detail,
+        evidence: [],
+        sdk_status: "active",
+        source_id: "codebuddy",
+        sdk_memory_at: null,
+        sdk_created_at: new Date().toISOString(),
+        sdk_ref: { scene: "working", kind: "experience", variant: "working" },
+      }}
+      sources={[]}
+    />,
+  );
+  expect(screen.getByText(/来源 codebuddy · 写入时间 今天 /)).toBeTruthy();
+  expect(screen.queryByText(/对话时间/)).toBeNull();
+  expect(screen.queryByText(/时间未知/)).toBeNull();
+});

memhub/apps/web/src/SelectedMemoryDetail.tsx

@@ -19,7 +19,10 @@ export function SelectedMemoryDetail({
   detail: MemoryDetail;
   sources: SourceResponse[];
   /** D99: when the cited conversation happened, in the list's own terms. */
-  conversationAt?: { at: string; basis: "conversation" | "import" } | null;
+  conversationAt?: {
+    at: string;
+    basis: "conversation" | "import" | "write";
+  } | null;
 }) {
   const { locale, t } = useGuiLanguage();
   const memory = detail.memory;
@@ -42,6 +45,17 @@ export function SelectedMemoryDetail({
           ? t("selectedMemory.status.superseded")
           : t("selectedMemory.status.unknown");
   const sourceId = detail.source_id ?? evidence?.source_id;
+  const created = clockStamp(
+    detail.sdk_created_at ?? memory.updated_at,
+    locale,
+  );
+  // An SDK experience or fact stored before hy-memory 2.0.7 has no
+  // conversation time. It shows when it was written, named as such, instead
+  // of "时间未知".
+  const writtenOnly =
+    detail.sdk_ref !== undefined &&
+    detail.sdk_ref.kind !== "chat" &&
+    !detail.sdk_memory_at;
   const source =
     detail.sdk_status && sourceId === UNATTRIBUTED_SOURCE_ID
       ? t("selectedMemory.sourceUnattributed")
@@ -77,25 +91,28 @@ export function SelectedMemoryDetail({
         <p className="selected-memory-source">
           {t("selectedMemory.sourceLine", {
             source,
-            time: clockStamp(
-              detail.sdk_memory_at ?? evidence?.created_at,
-              locale,
-            ),
+            time: writtenOnly
+              ? t("selectedMemory.writeTime", { created })
+              : clockStamp(
+                  detail.sdk_memory_at ?? evidence?.created_at,
+                  locale,
+                ),
           })}
         </p>
       ) : null}
-      <p className="selected-memory-times">
-        {t("selectedMemory.timestamps", {
-          conversation: clockStamp(
-            conversationAt?.at ?? detail.sdk_memory_at ?? evidence?.created_at,
-            locale,
-          ),
-          created: clockStamp(
-            detail.sdk_created_at ?? memory.updated_at,
-            locale,
-          ),
-        })}
-      </p>
+      {writtenOnly ? null : (
+        <p className="selected-memory-times">
+          {t("selectedMemory.timestamps", {
+            conversation: clockStamp(
+              conversationAt?.at ??
+                detail.sdk_memory_at ??
+                evidence?.created_at,
+              locale,
+            ),
+            created,
+          })}
+        </p>
+      )}
     </article>
   );
 }

memhub/apps/web/src/sdk-memory-map.test.ts

@@ -119,8 +119,12 @@ it("carries each memory's scene and namespace, and edits an experience's solutio
   });
   expect(page.window_id).toBe("w1");
   expect(page.window_reset).toBe(true);
-  // An experience has no conversation time to show as "imported".
-  expect(page.source_time_by_memory?.e).toBeUndefined();
+  // Stored without a conversation time: it shows when it was written, not
+  // an "import".
+  expect(page.source_time_by_memory?.e).toEqual({
+    at: new Date((NOW - 86_400) * 1_000).toISOString(),
+    basis: "write",
+  });
   expect(page.items[1]).toMatchObject({
     kind: "working",
     updated_at: new Date((NOW - 3_600) * 1_000).toISOString(),
@@ -140,3 +144,29 @@ it("carries each memory's scene and namespace, and edits an experience's solutio
   expect(sdkEditableText(experience)).toBe("Pin seed");
   expect(sdkEditableText(record({}))).toBe("memory");
 });
+
+it("shows an experience's conversation time when the SDK stored one", () => {
+  const at = NOW - 2 * 86_400;
+  const experience = record({
+    memory_id: "e",
+    layer: "experience",
+    scene: "coding",
+    kind: "experience",
+    variant: "coding",
+    memory_at: at,
+    created_at: NOW - 3_600,
+  });
+  const page = sdkLibraryPage({
+    memories: [experience],
+    total: 1,
+    limit: 20,
+    offset: 0,
+  });
+  expect(page.source_time_by_memory?.e).toEqual({
+    at: new Date(at * 1_000).toISOString(),
+    basis: "conversation",
+  });
+  expect(sdkDetail(experience).sdk_memory_at).toBe(
+    new Date(at * 1_000).toISOString(),
+  );
+});

memhub/apps/web/src/sdk-memory-map.ts

@@ -154,22 +154,24 @@ export function sdkDetail(record: SdkMemoryRecord): MemoryDetail {
 export function sdkLibraryPage(page: SdkMemoryPage): MemoryLibraryPage {
   const sourceTime: Record<
     string,
-    { at: string; basis: "conversation" | "import" }
+    { at: string; basis: "conversation" | "import" | "write" }
   > = {};
   const sourceIds: Record<string, string[]> = {};
   const refs: Record<string, SdkMemoryRef> = {};
   for (const record of page.memories) {
     sourceIds[record.memory_id] = [sourceOf(record)];
     refs[record.memory_id] = sdkMemoryRef(record);
-    // An experience or fact has no conversation time; the list shows when it
-    // was last updated instead of calling its creation an import.
-    if (refs[record.memory_id].kind !== "chat") continue;
+    // An experience or fact stored before hy-memory 2.0.7 has no
+    // conversation time. It shows when it was written, labelled as such,
+    // rather than calling its creation an import.
+    const chat = refs[record.memory_id].kind === "chat";
     const conversation = plausible(record.memory_at);
-    const at = conversation ?? plausible(record.gmt_created);
+    const at =
+      conversation ?? plausible(chat ? record.gmt_created : record.created_at);
     if (at === null) continue;
     sourceTime[record.memory_id] = {
       at: new Date(at * 1_000).toISOString(),
-      basis: conversation !== null ? "conversation" : "import",
+      basis: conversation !== null ? "conversation" : chat ? "import" : "write",
     };
   }
   return {

memhub/docs/ARCHITECTURE.md

@@ -282,7 +282,7 @@ Support is declared per immutable GitHub Release manifest:
 - Linux x64/arm64 Runtime on native GHA runners.
 - macOS arm64/x64 App/Runtime: Developer ID, notarization, native acceptance.
 - Windows is not supported.
-- `hy-memory==2.0.6` is the current stable SDK dependency; the release
+- `hy-memory==2.0.7` is the current stable SDK dependency; the release
   lock and [`../MEMORY.md`](../MEMORY.md) are authoritative when the pin changes.
 
 See [`INSTALLATION.md`](INSTALLATION.md).

memhub/docs/contracts/one-memory-recall-install-v1.md

@@ -39,10 +39,12 @@ which talks to a broker on the same machine.
 | `ONE_MEMORY_BROKER_DISCOVERY` | absolute path to the discovery file | How the server finds the running app. Wrong value means recall returns nothing, with no error. |
 | `ONE_MEMORY_MANAGED_BY` | `one-memory` | Marks the entry as ours, so our installer will update or remove it later instead of treating it as a stranger's and leaving it alone. |
 | `ELECTRON_RUN_AS_NODE` | `1` | The MCP entry ships inside the app bundle and is started through the app's own binary. Without this the binary opens the window instead of running the script. |
+| `ONE_MEMORY_MCP_TOOLSET` | tool set version, e.g. `3` | Names the MemHub tool set the entry was written for. MemHub changes it only when tools are added, removed or renamed, so hosts that cache a server's tool list by its launch config restart the server and drop removed tools. The server itself ignores it. |
 
 `ONE_MEMORY_MANAGED_BY` is optional for a hand-made entry. Set it if you want
 MemHub to manage the entry for you afterwards; leave it out if you would
-rather own the entry yourself.
+rather own the entry yourself. `ONE_MEMORY_MCP_TOOLSET` is optional too; a
+managed entry without the current value is rewritten on the next upgrade.
 
 ## 3. MCP hosts
 
@@ -57,7 +59,8 @@ Register a stdio server named `one-memory`. The product name is MemHub; the conf
       "env": {
         "ELECTRON_RUN_AS_NODE": "1",
         "ONE_MEMORY_BROKER_DISCOVERY": "<absolute path to the discovery file>",
-        "ONE_MEMORY_MANAGED_BY": "one-memory"
+        "ONE_MEMORY_MANAGED_BY": "one-memory",
+        "ONE_MEMORY_MCP_TOOLSET": "3"
       }
     }
   }

memhub/package.json

@@ -2,7 +2,7 @@
   "name": "one-memory",
   "version": "1.0.5",
   "release": {
-    "build_number": 32,
+    "build_number": 34,
     "macos_min_version": "13.0",
     "app_update": "manual",
     "runtime_manifest_schema": 1,

memhub/packages/contracts/src/index.ts

@@ -468,10 +468,12 @@ export interface MemoryLibraryPage {
   limit: number;
   /** Active evidence sources for each returned memory, from the same authorized page snapshot. */
   source_ids_by_memory?: Record<string, string[]>;
-  /** When each returned memory's cited conversation happened (D96). `import` when only the import time is known. */
+  /** When each returned memory's cited conversation happened (D96). `import`
+   * when only the import time is known; `write` for an SDK experience or fact
+   * stored without a conversation time, which shows when it was written. */
   source_time_by_memory?: Record<
     string,
-    { at: string; basis: "conversation" | "import" }
+    { at: string; basis: "conversation" | "import" | "write" }
   >;
   /** True when the read budget ran out mid-page (typically during a history
    * import): a partial snapshot, not an authoritative view. Show "importing"

memhub/packages/host-bridge/src/install-contract.test.ts

@@ -5,6 +5,8 @@ import { describe, expect, it } from "vitest";
 import type { RecallRuntime } from "./recall-installer.js";
 import {
   DISCOVERY_KEY,
+  MCP_TOOLSET_KEY,
+  MCP_TOOLSET_VERSION,
   OWNER_KEY,
   OWNER_VALUE,
   renderConnectionMaterial,
@@ -60,6 +62,7 @@ describe("mcp install snippet", () => {
     const env = parsed.mcpServers[SERVER_NAME].env;
     expect(env[DISCOVERY_KEY]).toBe(runtime().discoveryPath);
     expect(env[OWNER_KEY]).toBe(OWNER_VALUE);
+    expect(env[MCP_TOOLSET_KEY]).toBe(MCP_TOOLSET_VERSION);
     expect(env.ELECTRON_RUN_AS_NODE).toBe("1");
   });
 
@@ -174,9 +177,10 @@ describe("public contract document", () => {
     // reports it to us. Same class of problem R-33 treated for skill text,
     // and treated the same way: one source, checked by a test.
     const doc = await fs.readFile(CONTRACT_DOC, "utf8");
-    for (const key of [DISCOVERY_KEY, OWNER_KEY]) {
+    for (const key of [DISCOVERY_KEY, OWNER_KEY, MCP_TOOLSET_KEY]) {
       expect(doc).toContain(key);
     }
+    expect(doc).toContain(`"${MCP_TOOLSET_KEY}": "${MCP_TOOLSET_VERSION}"`);
     expect(doc).toContain(OWNER_VALUE);
     expect(doc).toContain(SERVER_NAME);
   });

memhub/packages/host-bridge/src/mcp-granted-tools.test.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from "vitest";
 
 import { type GrantedTools, McpSession } from "./mcp-session.js";
+import { MCP_TOOL_NAMES, MCP_TOOLSET_VERSION } from "./mcp-toolset.js";
 
 function sessionWith(granted: GrantedTools): McpSession {
   return new McpSession(
@@ -71,3 +72,14 @@ describe("BUG-15: the listed tools follow what the grant allows", () => {
     expect(names).toHaveLength(3);
   });
 });
+
+describe("the tool set stamped on installed entries", () => {
+  it("names exactly the tools the server lists", async () => {
+    // Change a tool without moving MCP_TOOLSET_VERSION and hosts that cache
+    // tool lists by config fingerprint keep showing the old set. Update both,
+    // and this pinned version, together.
+    const names = await listedNames(sessionWith(async () => null));
+    expect(names).toEqual([...MCP_TOOL_NAMES]);
+    expect(MCP_TOOLSET_VERSION).toBe("3");
+  });
+});

memhub/packages/host-bridge/src/mcp-session.ts

@@ -127,6 +127,8 @@ export class McpSession {
   }
 
   private async listedTools() {
+    // Any change to this set moves MCP_TOOL_NAMES and MCP_TOOLSET_VERSION in
+    // `mcp-toolset.ts`, so installed entries change and hosts relist.
     const definitions = [
       recallToolDefinition(),
       writeToolDefinition(),

memhub/packages/host-bridge/src/mcp-toolset.ts

@@ -0,0 +1,29 @@
+/** The tool contract of MemHub's MCP server, and the stamp that names it.
+ *
+ * Kept apart from `mcp-session.ts` so the installer can stamp an entry without
+ * pulling in the broker client, and listed here rather than inferred so a test
+ * can hold `tools/list` to it: change a tool there and that test fails until
+ * this list, and with it the version, moves too.
+ */
+
+/** The tools `tools/list` returns, in order. */
+export const MCP_TOOL_NAMES = [
+  "MemHub_recall",
+  "MemHub_write",
+  "MemHub_diagnostic_trace",
+] as const;
+
+/** Bumped whenever a tool is added, removed or renamed.
+ *
+ * Written into every managed MCP entry under `MCP_TOOLSET_KEY`. Hosts such as
+ * WorkBuddy cache a server's tool list under a fingerprint of its launch
+ * config, and our entry is otherwise byte-identical across upgrades, since it
+ * runs through the `current` symlink. Moving this value is what changes that
+ * fingerprint, so those hosts restart the server and drop the tools it no
+ * longer has. Deliberately not the build number: that would rewrite every
+ * host's config on every upgrade, for nothing.
+ */
+export const MCP_TOOLSET_VERSION = "3";
+
+/** The env key the version is written under. */
+export const MCP_TOOLSET_KEY = "ONE_MEMORY_MCP_TOOLSET";

memhub/packages/host-bridge/src/opencode-installer.test.ts

@@ -5,6 +5,7 @@ import { describe, expect, it } from "vitest";
 
 import {
   applyRecallInstall,
+  MCP_TOOLSET_VERSION,
   mcpConfigPath,
   opencodeLayout,
   planRecallInstall,
@@ -46,6 +47,7 @@ describe("OpenCode recall installer", () => {
         ELECTRON_RUN_AS_NODE: "1",
         ONE_MEMORY_MANAGED_BY: "one-memory",
         ONE_MEMORY_BROKER_DISCOVERY: "/opt/one-memory/runtime.json",
+        ONE_MEMORY_MCP_TOOLSET: MCP_TOOLSET_VERSION,
       },
       enabled: true,
     });

memhub/packages/host-bridge/src/recall-installer.ts

@@ -58,6 +58,7 @@ import {
   readDshServer,
   removeDshServer,
 } from "./dsh-patch.js";
+import { MCP_TOOLSET_KEY, MCP_TOOLSET_VERSION } from "./mcp-toolset.js";
 import {
   RECALL_PRESENTATION_GUIDANCE,
   RECALL_PRESENTATION_GUIDANCE_ZH,
@@ -88,8 +89,12 @@ export {
   type RecallLinkState,
   type RecallProbe,
 } from "./recall-probe.js";
-
-export { planRecallRepoint, type RecallRepoint };
+export {
+  MCP_TOOLSET_KEY,
+  MCP_TOOLSET_VERSION,
+  planRecallRepoint,
+  type RecallRepoint,
+};
 
 /** Rewrite hosts whose installed command is not the one we would write now.
  *
@@ -2844,6 +2849,9 @@ function managedServer(runtime: RecallRuntime): ManagedServer {
       ELECTRON_RUN_AS_NODE: "1",
       [DISCOVERY_KEY]: runtime.discoveryPath,
       [OWNER_KEY]: OWNER_VALUE,
+      // Moves the host's fingerprint of this entry when the tool set changes;
+      // see `MCP_TOOLSET_VERSION`. An entry without it reads as stale.
+      [MCP_TOOLSET_KEY]: MCP_TOOLSET_VERSION,
       ...(runtime.credentialFile === undefined
         ? {}
         : { ONE_MEMORY_HOST_CREDENTIAL_FILE: runtime.credentialFile }),

memhub/packages/host-bridge/src/skill-drift.test.ts

@@ -5,6 +5,7 @@ import path from "node:path";
 import { describe, expect, it } from "vitest";
 
 import { applyCursorInstall, planCursorInstall } from "./cursor-installer.js";
+import { MCP_TOOLSET_KEY, MCP_TOOLSET_VERSION } from "./mcp-toolset.js";
 import {
   applyRecallInstall,
   applyRecallUninstall,
@@ -15,6 +16,7 @@ import {
   planRecallUninstall,
   SKILL_MARKER,
   SKILL_VERSION,
+  verifyRecallInstall,
 } from "./recall-installer.js";
 import {
   checkSkillDrift,
@@ -387,6 +389,59 @@ describe("skill drift (R-33)", () => {
     });
   }, 30_000);
 
+  it("marks stale an MCP entry written before the tool set was stamped, and sync stamps it", async () => {
+    // Our entry is otherwise identical across upgrades, so hosts that cache a
+    // tool list by config fingerprint keep the removed tools until it changes.
+    const cursorDir = await fixtureDir();
+    const discovery = path.join(cursorDir, "broker.json");
+    await writeFile(discovery, "{}\n");
+    await applyCursorInstall(
+      await planCursorInstall(options(cursorDir, discovery)),
+    );
+    const layout = cursorLayout(cursorDir);
+    const configPath = layout.configPath as string;
+    const config = JSON.parse(await readFile(configPath, "utf8")) as {
+      mcpServers: Record<string, { env: Record<string, string> }>;
+    };
+    expect(config.mcpServers["one-memory"].env[MCP_TOOLSET_KEY]).toBe(
+      MCP_TOOLSET_VERSION,
+    );
+    // What an earlier release wrote and journaled: the same entry, no stamp.
+    delete config.mcpServers["one-memory"].env[MCP_TOOLSET_KEY];
+    await writeFile(configPath, `${JSON.stringify(config, null, 2)}\n`);
+    const journal = JSON.parse(await readFile(layout.journalPath, "utf8"));
+    journal.installed_server_hash = createHash("sha256")
+      .update(JSON.stringify(config.mcpServers["one-memory"]))
+      .digest("hex");
+    await writeFile(layout.journalPath, `${JSON.stringify(journal)}\n`);
+
+    // Still intact as installed, so the link is not reported broken...
+    await expect(verifyRecallInstall(layout)).resolves.toMatchObject({
+      hostId: "cursor",
+    });
+    // ...but stale against this release, and safe to sync without asking.
+    await expect(
+      checkSkillDrift(layout, runtime(discovery)),
+    ).resolves.toMatchObject({
+      state: "stale",
+      detail: "the MCP entry predates the tool set this MemHub serves",
+    });
+    const sync = await planSkillSync(layout, runtime(discovery));
+    expect(sync?.requiresConfirmation).toBe(false);
+    if (sync === null) throw new Error("expected a stale entry sync");
+    await applyRecallInstall(sync.plan);
+
+    const synced = JSON.parse(await readFile(configPath, "utf8")) as {
+      mcpServers: Record<string, { env: Record<string, string> }>;
+    };
+    expect(synced.mcpServers["one-memory"].env[MCP_TOOLSET_KEY]).toBe(
+      MCP_TOOLSET_VERSION,
+    );
+    await expect(
+      checkSkillDrift(layout, runtime(discovery)),
+    ).resolves.toMatchObject({ state: "in_sync" });
+  }, 30_000);
+
   it("requires recall for dependent answers, not on every unrelated turn", async () => {
     const cursorDir = await fixtureDir();
     const discovery = path.join(cursorDir, "broker.json");

memhub/packages/host-bridge/src/skill-drift.ts

@@ -2,6 +2,7 @@ import { createHash } from "node:crypto";
 import path from "node:path";
 import { captureFilePath, namesCapture } from "./capture-hooks.js";
 import { PLUGIN_HOOKS } from "./codebuddy-plugin.js";
+import { MCP_TOOLSET_KEY, MCP_TOOLSET_VERSION } from "./mcp-toolset.js";
 import {
   asInstalled,
   expectedPluginHooks,
@@ -9,6 +10,7 @@ import {
   type RecallHostLayout,
   type RecallInstallPlan,
   type RecallRuntime,
+  readInstalledRecallCommand,
   renderSkillContent,
   SKILL_VERSION,
 } from "./recall-installer.js";
@@ -99,7 +101,23 @@ export async function checkSkillDrift(
   runtime: RecallRuntime,
 ): Promise<SkillDrift | null> {
   const skill = await checkSkillFileDrift(layout, runtime);
-  if (skill === null || skill.state === "modified") return skill;
+  if (skill === null) {
+    // A host with no skill mechanism still has an entry whose tool set can
+    // age. One whose skill is merely missing is left alone: syncing would put
+    // back a file somebody removed.
+    if (skillFilePath(layout) !== null) return null;
+    return (await serverToolsetDrift(layout)) === "stale"
+      ? {
+          hostId: layout.hostId,
+          state: "stale",
+          installedVersion: null,
+          currentVersion: SKILL_VERSION,
+          skillPath: null,
+          detail: TOOLSET_STALE_DETAIL,
+        }
+      : null;
+  }
+  if (skill.state === "modified") return skill;
   const hooks = await pluginHooksDrift(layout, runtime);
   if (hooks === "modified") {
     return {
@@ -108,15 +126,19 @@ export async function checkSkillDrift(
       detail: "the plugin hook file was edited after install",
     };
   }
-  if (hooks === "stale" && skill.state === "in_sync") {
+  if (skill.state === "stale") return skill;
+  if (hooks === "stale") {
     return {
       ...skill,
       state: "stale",
       detail: "plugin hooks are still ours, but have moved since install",
     };
   }
+  if ((await serverToolsetDrift(layout)) === "stale") {
+    return { ...skill, state: "stale", detail: TOOLSET_STALE_DETAIL };
+  }
   const capture = await captureHooksDrift(layout);
-  if (capture === "stale" && skill.state === "in_sync") {
+  if (capture === "stale") {
     return {
       ...skill,
       state: "stale",
@@ -126,6 +148,29 @@ export async function checkSkillDrift(
   return skill;
 }
 
+const TOOLSET_STALE_DETAIL =
+  "the MCP entry predates the tool set this MemHub serves";
+
+/** Whether our MCP entry names the tool set this release serves.
+ *
+ * The entry is otherwise identical across upgrades, so a host that caches a
+ * server's tool list under a fingerprint of its config (WorkBuddy does) keeps
+ * listing tools that are gone until the entry itself changes. Only an entry
+ * marked as ours is read; a CLI-channel host has no entry at all.
+ */
+async function serverToolsetDrift(
+  layout: RecallHostLayout,
+): Promise<"in_sync" | "stale" | null> {
+  if (layout.channel !== "mcp") return null;
+  // A config the host cannot parse either is not ours to judge here; the
+  // install and link checks report it.
+  const installed = await readInstalledRecallCommand(layout).catch(() => null);
+  if (installed === null) return null;
+  return installed.env[MCP_TOOLSET_KEY] === MCP_TOOLSET_VERSION
+    ? "in_sync"
+    : "stale";
+}
+
 /** Detect duplicated command hooks left by earlier installs.
  *
  * `checkSkillDrift` historically compared only the Skill and plugin hooks;

memhub/packages/i18n/src/gui-i18n-current.ts

@@ -108,6 +108,7 @@ export const currentChineseMessages = {
   "selectedMemory.version": "第 {revision} 版",
   "selectedMemory.sourceLine": "来源 {source} · {time}",
   "selectedMemory.timestamps": "对话时间 {conversation} · 沉淀时间 {created}",
+  "selectedMemory.writeTime": "写入时间 {created}",
   "selectedMemory.edit": "编辑",
   "selectedMemory.evidence": "原始依据",
   "selectedMemory.evidenceLine": "{author} · 原对话 {time}：",
@@ -258,6 +259,7 @@ export const currentEnglishMessages: Record<
   "selectedMemory.sourceLine": "Source {source} · {time}",
   "selectedMemory.timestamps":
     "Conversation time {conversation} · Created {created}",
+  "selectedMemory.writeTime": "Written {created}",
   "selectedMemory.edit": "Edit",
   "selectedMemory.evidence": "Original evidence",
   "selectedMemory.evidenceLine": "{author} · Original conversation {time}:",
@@ -418,6 +420,7 @@ export const currentSpanishMessages: Record<
   "selectedMemory.sourceLine": "Origen {source} · {time}",
   "selectedMemory.timestamps":
     "Conversación {conversation} · Guardada {created}",
+  "selectedMemory.writeTime": "Escrita {created}",
   "selectedMemory.edit": "Editar",
   "selectedMemory.evidence": "Evidencia original",
   "selectedMemory.evidenceLine": "{author} · Conversación original {time}:",

memhub/packages/i18n/src/gui-i18n-english.ts

@@ -1774,6 +1774,7 @@ const baseEnglishMessages = {
     " · Import count is still updating; refresh to check",
   "memory:对话于 {day}": "Conversation on {day}",
   "memory:导入于 {day}": "Imported on {day}",
+  "memory:写入于 {day}": "Written on {day}",
   "memory:已开始处理所选对话": "Processing selected conversations",
   "memory:处理请求已受理": "Processing request received",
   "memory:模型连接已通过。首批记忆生成后，会逐步出现在记忆列表里。":

memhub/packages/i18n/src/gui-i18n-memory.ts

@@ -336,6 +336,7 @@ export const memoryChineseMessages = {
   "memory: · 导入数量仍在更新，可刷新核对": " · 导入数量仍在更新，可刷新核对",
   "memory:对话于 {day}": "对话于 {day}",
   "memory:导入于 {day}": "导入于 {day}",
+  "memory:写入于 {day}": "写入于 {day}",
   "memory:已开始处理所选对话": "已开始处理所选对话",
   "memory:处理请求已受理": "处理请求已受理",
   "memory:模型连接已通过。首批记忆生成后，会逐步出现在记忆列表里。":
@@ -750,6 +751,7 @@ export const memorySpanishMessages: Record<
     " · La importación continúa; actualiza para verificar",
   "memory:对话于 {day}": "Conversación del {day}",
   "memory:导入于 {day}": "Importada el {day}",
+  "memory:写入于 {day}": "Escrita el {day}",
   "memory:已开始处理所选对话": "Procesamiento de conversaciones iniciado",
   "memory:处理请求已受理": "Solicitud de procesamiento aceptada",
   "memory:模型连接已通过。首批记忆生成后，会逐步出现在记忆列表里。":

memhub/packages/runtime-control/src/host-permissions.ts

@@ -41,6 +41,7 @@ import {
   repairInstalledRecallCommands,
   verifyRecallInstall,
 } from "@one-memory/host-bridge/installer";
+import type { DiagnosticInput, DiagnosticSink } from "./diagnostic-log.js";
 import {
   hasSyncHookArtifacts,
   loadSyncHookAssets,
@@ -96,8 +97,22 @@ interface InstallationBinding {
 }
 
 const CREDENTIAL_ENV = "ONE_MEMORY_HOST_CREDENTIAL_FILE";
+/** The build whose host refresh last completed, beside the grants. */
+const HOST_REFRESH_FILE = "host-refresh.json";
 const locks = new Map<string, Promise<void>>();
 
+/** What one pass of `refreshInstalledHosts` did, by host id. */
+export interface HostRefreshReport {
+  /** `already_refreshed` when this build had already completed a pass. */
+  state: "refreshed" | "already_refreshed";
+  checked: number;
+  /** Hosts whose installed command or discovery path was moved. */
+  repointed: string[];
+  /** Hosts whose stale skill, hooks, plugin files or MCP entry were rewritten. */
+  synced: string[];
+  failed: string[];
+}
+
 export const LOCAL_EXPERIENCE_UNAVAILABLE = "体验版暂不支持";
 
 /** Trusted local Runtime control plane. Never exposed through MCP or the Core API.
@@ -201,11 +216,20 @@ export class HostPermissions {
    * touching authorization. A skill the user edited is left for the detail page.
    */
   async syncStaleInstalls(layouts: RecallHostLayout[]): Promise<string[]> {
+    return (await this.syncStale(layouts)).synced;
+  }
+
+  private async syncStale(
+    layouts: RecallHostLayout[],
+  ): Promise<{ synced: string[]; failed: string[] }> {
     const synced: string[] = [];
+    const failed: string[] = [];
     for (const layout of layouts) {
       try {
         await this.withHostLock(layout, async () => {
-          await this.requireLinked(layout);
+          // Not linked from this data directory is not ours to sync, and not
+          // a failure either.
+          if (!(await this.isLinked(layout))) return;
           const sync = await planSkillSync(layout, this.runtimeFor(layout));
           if (sync === null || sync.requiresConfirmation) return;
           const preview = await previewRecallInstall(sync.plan);
@@ -215,9 +239,128 @@ export class HostPermissions {
         });
       } catch {
         // One host that cannot be synced must not stop the runtime starting.
+        failed.push(layout.hostId);
       }
     }
-    return synced;
+    return { synced, failed };
+  }
+
+  /** Refresh every host this directory installed, once per build.
+   *
+   * An upgrade changes the call rules, hooks and MCP entry we would write, but
+   * nothing rewrote the copies hosts already had, so they kept telling the
+   * model to call tools that were gone. This runs the same repoint and stale
+   * sync the detail page offers, without asking, for files that are still
+   * exactly ours; an edited skill stays for the user to confirm.
+   *
+   * The build it last completed for is recorded beside the grants, so later
+   * launches of the same build cost one file read. A run in which any host
+   * failed is not recorded, so the next launch tries again. With no build to
+   * name (a source checkout) it runs every time and records nothing.
+   */
+  async refreshInstalledHosts(
+    layouts: RecallHostLayout[],
+    options: { build?: string; diagnostics?: DiagnosticSink } = {},
+  ): Promise<HostRefreshReport> {
+    const log = (input: DiagnosticInput) =>
+      options.diagnostics?.log(input).catch(() => undefined);
+    const build = options.build?.trim() || undefined;
+    if (build !== undefined && (await this.readRefreshedBuild()) === build) {
+      await log({
+        level: "info",
+        component: "host.refresh",
+        event: "host_refresh.skipped",
+        outcome: "skipped",
+        details: { reason: "already_refreshed", count: layouts.length },
+      });
+      return {
+        state: "already_refreshed",
+        checked: layouts.length,
+        repointed: [],
+        synced: [],
+        failed: [],
+      };
+    }
+    const started = Date.now();
+    const repointed: string[] = [];
+    const failed: string[] = [];
+    for (const layout of layouts) {
+      try {
+        repointed.push(...(await this.repairInstalledCommands([layout])));
+      } catch {
+        failed.push(layout.hostId);
+      }
+    }
+    const stale = await this.syncStale(layouts);
+    failed.push(...stale.failed.filter((hostId) => !failed.includes(hostId)));
+    for (const hostId of new Set([...repointed, ...stale.synced])) {
+      await log({
+        level: "info",
+        component: "host.refresh",
+        event: "host_refresh.host_updated",
+        outcome: "succeeded",
+        details: { source_id: hostId },
+      });
+    }
+    for (const hostId of failed) {
+      await log({
+        level: "warn",
+        component: "host.refresh",
+        event: "host_refresh.host_failed",
+        outcome: "failed",
+        error_code: "host_refresh_failed",
+        details: { source_id: hostId },
+      });
+    }
+    if (build !== undefined && failed.length === 0) {
+      await this.writeRefreshedBuild(build).catch(() => undefined);
+    }
+    await log({
+      level: failed.length === 0 ? "info" : "warn",
+      component: "host.refresh",
+      event: "host_refresh.completed",
+      outcome: failed.length === 0 ? "succeeded" : "degraded",
+      duration_ms: Date.now() - started,
+      details: {
+        count: layouts.length,
+        updated_count: new Set([...repointed, ...stale.synced]).size,
+        error_count: failed.length,
+      },
+    });
+    return {
+      state: "refreshed",
+      checked: layouts.length,
+      repointed,
+      synced: stale.synced,
+      failed,
+    };
+  }
+
+  private async readRefreshedBuild(): Promise<string | null> {
+    try {
+      return await storageGuard(async () => {
+        const file = path.join(this.directory, HOST_REFRESH_FILE);
+        if ((await optionalStat(file)) === null) return null;
+        const value: unknown = JSON.parse(await privateRead(file));
+        return isObject(value) &&
+          value.schemaVersion === 1 &&
+          typeof value.build === "string"
+          ? value.build
+          : null;
+      });
+    } catch {
+      // Unreadable reads as never refreshed: the refresh is idempotent.
+      return null;
+    }
+  }
+
+  private async writeRefreshedBuild(build: string): Promise<void> {
+    await storageGuard(() =>
+      privateWrite(
+        path.join(this.directory, HOST_REFRESH_FILE),
+        `${JSON.stringify({ schemaVersion: 1, build })}\n`,
+      ),
+    );
   }
 
   /** Same repair for one host. Does not revoke or reissue its authorization. */
@@ -942,6 +1085,16 @@ export class HostPermissions {
       throw new Error("host_permission_installation_incomplete");
   }
 
+  /** `requireLinked` as a question, for sweeps that skip hosts not ours. */
+  private async isLinked(layout: RecallHostLayout): Promise<boolean> {
+    if (!supported(layout)) return false;
+    const current = await this.inspect(layout, false);
+    return (
+      (current.state === "approved" || current.state === "install-only") &&
+      current.reason === null
+    );
+  }
+
   /** Turning off a host with nothing of ours on it is an empty removal, not a
    * failure; one whose state cannot be settled is an empty removal carrying
    * what to look at, which uninstall refuses. Only what is ours for certain

memhub/packages/runtime-control/src/index.ts

@@ -79,6 +79,7 @@ export {
 } from "./host-control-client.js";
 export {
   HostPermissions,
+  type HostRefreshReport,
   LOCAL_EXPERIENCE_UNAVAILABLE,
   type RecallPermissionStatus,
   type RevokeRecallPermissionInput,

memhub/packages/runtime-control/src/runtime-cli.test.ts

@@ -226,6 +226,7 @@ describe("host runtime attestation", () => {
     "repairInstalledCommand",
     "repairInstalledCommands",
     "syncStaleInstalls",
+    "refreshInstalledHosts",
     "previewInstall",
     "install",
     "setPermission",

memhub/packages/runtime-control/src/runtime-cli.ts

@@ -600,6 +600,7 @@ export function guardedPermissions(
     "previewSyncHook",
     "applySyncHook",
     "syncStaleInstalls",
+    "refreshInstalledHosts",
   ]);
   return new Proxy(owner, {
     get(target, property, receiver) {
@@ -739,10 +740,17 @@ export async function createRuntimeService(
       ).filter((host) => host.hostId !== "openclaw");
       await verifyHostRuntime().then(
         async () => {
-          await permissions.repairInstalledCommands(
+          // Every Runtime start after an upgrade, the App's or `memhub
+          // upgrade`'s, comes through here first; once per build after that.
+          await permissions.refreshInstalledHosts(
             hosts.map((host) => host.layout),
+            {
+              build:
+                config.buildIdentity?.buildNumber ??
+                config.buildIdentity?.version,
+              diagnostics: hostDiagnostics,
+            },
           );
-          await permissions.syncStaleInstalls(hosts.map((host) => host.layout));
         },
         () => undefined,
       );

memhub/packages/runtime-control/src/runtime-service.test.ts

@@ -2,9 +2,11 @@ import {
   chmod,
   mkdir,
   mkdtemp,
+  readdir,
   readFile,
   rm,
   stat,
+  symlink,
   writeFile,
 } from "node:fs/promises";
 import { tmpdir } from "node:os";
@@ -38,6 +40,38 @@ function supervisor() {
   };
 }
 
+const PREVIOUS_BOOT = "00000000-0000-4000-8000-00000000000a";
+const CURRENT_BOOT = "00000000-0000-4000-8000-00000000000b";
+
+/** Seed a lock as an existing owner would leave it; no owner = legacy debris. */
+async function seedLock(dataDir: string, owner?: object): Promise<string> {
+  const lock = path.join(dataDir, "runtime-service.lock");
+  await mkdir(lock, { mode: 0o700 });
+  if (owner) {
+    await writeFile(path.join(lock, "owner.json"), JSON.stringify(owner), {
+      mode: 0o600,
+    });
+  }
+  await chmod(lock, 0o700);
+  return lock;
+}
+
+async function ownerOf(dataDir: string): Promise<Record<string, unknown>> {
+  return JSON.parse(
+    await readFile(
+      path.join(dataDir, "runtime-service.lock", "owner.json"),
+      "utf8",
+    ),
+  );
+}
+
+/** Every lock, staging and cleared-lease entry in the data dir. */
+async function lockEntries(dataDir: string): Promise<string[]> {
+  return (await readdir(dataDir))
+    .filter((name) => name.startsWith("runtime-service.lock"))
+    .sort();
+}
+
 describe("RuntimeService", () => {
   it("publishes and removes an outer Runtime receipt without secrets", async () => {
     const dataDir = await directory("runtime-service-");
@@ -71,18 +105,7 @@ describe("RuntimeService", () => {
 
   it("refuses a live owner of the same Runtime home", async () => {
     const dataDir = await directory("runtime-service-live-");
-    const lock = path.join(dataDir, "runtime-service.lock");
-    await import("node:fs/promises").then(({ mkdir }) =>
-      mkdir(lock, { mode: 0o700 }),
-    );
-    await writeFile(
-      path.join(lock, "owner.json"),
-      JSON.stringify({ pid: 777 }),
-      {
-        mode: 0o600,
-      },
-    );
-    await chmod(lock, 0o700);
+    await seedLock(dataDir, { pid: 777 });
     const service = new RuntimeService({
       dataDir,
       createSupervisor: supervisor,
@@ -94,27 +117,228 @@ describe("RuntimeService", () => {
     );
   });
 
-  it("reclaims a stale validated lease", async () => {
-    const dataDir = await directory("runtime-service-stale-");
-    const lock = path.join(dataDir, "runtime-service.lock");
-    await import("node:fs/promises").then(({ mkdir }) =>
-      mkdir(lock, { mode: 0o700 }),
-    );
-    await writeFile(
-      path.join(lock, "owner.json"),
-      JSON.stringify({ pid: 777 }),
-      {
-        mode: 0o600,
-      },
-    );
-    await chmod(lock, 0o700);
+  it("reclaims a previous-boot lease even when a recycled PID answers the liveness probe", async () => {
+    // Issue #184: a forced power-off leaves runtime-service.lock behind;
+    // after the reboot the owner's PID belongs to another process, so a
+    // liveness-only check reports "already running" forever and launchd's
+    // KeepAlive loops the wrapper. A lease from another boot can never have
+    // a living owner, whatever the PID table says.
+    const dataDir = await directory("runtime-service-prevboot-");
+    await seedLock(dataDir, { pid: 777, boot_id: PREVIOUS_BOOT });
     const service = new RuntimeService({
       dataDir,
       createSupervisor: supervisor,
+      pid: 4321,
+      processIsAlive: () => true,
+      bootId: () => CURRENT_BOOT,
+    });
+
+    await service.start();
+    expect(await ownerOf(dataDir)).toMatchObject({
+      pid: 4321,
+      boot_id: CURRENT_BOOT,
+    });
+    await service.stop();
+    expect(await lockEntries(dataDir)).toEqual([]);
+  });
+
+  it("refuses a live owner from the current boot", async () => {
+    const dataDir = await directory("runtime-service-sameboot-live-");
+    await seedLock(dataDir, { pid: 777, boot_id: CURRENT_BOOT });
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      processIsAlive: (pid) => pid === 777,
+      bootId: () => CURRENT_BOOT,
+    });
+
+    await expect(service.start()).rejects.toThrow(
+      "runtime_service_already_running",
+    );
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 777 });
+  });
+
+  it("reclaims a dead owner from the current boot", async () => {
+    const dataDir = await directory("runtime-service-sameboot-dead-");
+    await seedLock(dataDir, { pid: 777, boot_id: CURRENT_BOOT });
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      processIsAlive: () => false,
+      bootId: () => CURRENT_BOOT,
+    });
+
+    await service.start();
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
+    await service.stop();
+  });
+
+  it.each([
+    ["the lease predates boot ids", { pid: 777 }, CURRENT_BOOT],
+    ["the lease has an empty boot id", { pid: 777, boot_id: "" }, CURRENT_BOOT],
+    [
+      "the current boot id is unknown",
+      { pid: 777, boot_id: PREVIOUS_BOOT },
+      null,
+    ],
+  ])("falls back to the PID probe when %s", async (_case, owner, current) => {
+    const dataDir = await directory("runtime-service-unknown-boot-");
+    await seedLock(dataDir, owner);
+    const live = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      processIsAlive: (pid) => pid === 777,
+      bootId: () => current,
+    });
+    await expect(live.start()).rejects.toThrow(
+      "runtime_service_already_running",
+    );
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 777 });
+
+    const dead = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      processIsAlive: () => false,
+      bootId: () => current,
+    });
+    await dead.start();
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
+    await dead.stop();
+  });
+
+  it("records the platform boot id in owner.json", async () => {
+    const dataDir = await directory("runtime-service-boot-id-");
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+    });
+
+    await service.start();
+    const owner = await ownerOf(dataDir);
+    if (process.platform === "darwin" || process.platform === "linux") {
+      expect(owner.boot_id).toMatch(/^[0-9A-Za-z-]{8,64}$/);
+    } else {
+      expect(owner.boot_id).toBeNull();
+    }
+    expect(owner.pid).toBe(process.pid);
+    await service.stop();
+  });
+
+  it("leaves no lock behind when publishing fails before the rename", async () => {
+    // The lock only appears through a rename of a fully written staging
+    // directory, so a failure (or power cut) while staging cannot leave a
+    // lock without owner.json.
+    const dataDir = await directory("runtime-service-publish-failure-");
+    const broken = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      bootId: () => {
+        throw new Error("simulated crash while staging");
+      },
+    });
+
+    await expect(broken.start()).rejects.toThrow(
+      "runtime_service_lock_unavailable",
+    );
+    expect(await lockEntries(dataDir)).toEqual([]);
+
+    const next = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      bootId: () => CURRENT_BOOT,
+    });
+    await next.start();
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
+    await next.stop();
+  });
+
+  it("clears an owner-less legacy lock created before the current boot", async () => {
+    // Older versions created the lock with mkdir and wrote owner.json
+    // afterwards; a power cut in between left an empty lock that used to
+    // fail as runtime_service_lock_invalid forever.
+    const dataDir = await directory("runtime-service-legacy-prevboot-");
+    const lock = await seedLock(dataDir);
+    const createdAt = (await stat(lock)).mtimeMs;
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      bootId: () => CURRENT_BOOT,
+      legacyBootTime: () => createdAt + 60_000,
+    });
+
+    await service.start();
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
+    await service.stop();
+  });
+
+  it.each([
+    ["it may be from the current boot", (createdAt: number) => createdAt - 1],
+    ["the boot time is unknown", () => null],
+  ])(
+    "keeps refusing an owner-less legacy lock when %s",
+    async (_case, boot) => {
+      const dataDir = await directory("runtime-service-legacy-sameboot-");
+      const lock = await seedLock(dataDir);
+      const createdAt = (await stat(lock)).mtimeMs;
+      const service = new RuntimeService({
+        dataDir,
+        createSupervisor: supervisor,
+        bootId: () => CURRENT_BOOT,
+        legacyBootTime: () => boot(createdAt),
+      });
+
+      await expect(service.start()).rejects.toThrow(
+        "runtime_service_lock_invalid",
+      );
+      // rename() may replace an empty directory; the publish step must not.
+      expect(await readdir(lock)).toEqual([]);
+      expect(await lockEntries(dataDir)).toEqual(["runtime-service.lock"]);
+    },
+  );
+
+  it("removes staging and cleared-lease debris after acquiring", async () => {
+    const dataDir = await directory("runtime-service-debris-");
+    const staging = path.join(dataDir, "runtime-service.lock.tmp-crashed");
+    await mkdir(staging, { mode: 0o700 });
+    await writeFile(
+      path.join(staging, "owner.json"),
+      JSON.stringify({ pid: 9 }),
+    );
+    await mkdir(path.join(dataDir, "runtime-service.lock.stale.crashed"), {
+      mode: 0o700,
+    });
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
+      bootId: () => CURRENT_BOOT,
+    });
+
+    await service.start();
+    expect(await lockEntries(dataDir)).toEqual(["runtime-service.lock"]);
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
+    await service.stop();
+    expect(await lockEntries(dataDir)).toEqual([]);
+  });
+
+  it("reclaims a stale validated lease", async () => {
+    const dataDir = await directory("runtime-service-stale-");
+    await seedLock(dataDir, { pid: 777 });
+    const service = new RuntimeService({
+      dataDir,
+      createSupervisor: supervisor,
+      pid: 4321,
       processIsAlive: () => false,
     });
 
     await service.start();
+    expect(await ownerOf(dataDir)).toMatchObject({ pid: 4321 });
     await service.stop();
   });
 
@@ -386,7 +610,7 @@ describe("RuntimeService", () => {
     const real = path.join(parent, "real");
     const link = path.join(parent, "data");
     await mkdir(real, { mode: 0o755 });
-    await import("node:fs/promises").then(({ symlink }) => symlink(real, link));
+    await symlink(real, link);
     const service = new RuntimeService({
       dataDir: link,
       createSupervisor: () => supervisor(),

memhub/packages/runtime-control/src/runtime-service.ts

@@ -1,12 +1,14 @@
+import { execFileSync } from "node:child_process";
 import { randomUUID } from "node:crypto";
+import { readFileSync, type Stats } from "node:fs";
 import {
   chmod,
   lstat,
   mkdir,
+  readdir,
   readFile,
   rename,
   rm,
-  rmdir,
   writeFile,
 } from "node:fs/promises";
 import path from "node:path";
@@ -35,6 +37,14 @@ export interface RuntimeServiceOptions {
   ): ManagedRuntime | Promise<ManagedRuntime>;
   pid?: number;
   processIsAlive?: (pid: number) => boolean;
+  /** Identity of the current boot (`kern.bootsessionuuid` on macOS,
+   * `/proc/sys/kernel/random/boot_id` on Linux, `null` when unknown).
+   * Injectable so tests can simulate a reboot without one. */
+  bootId?: () => string | null;
+  /** Wall-clock epoch (ms) the kernel reports for the current boot, or `null`
+   * when unknown. Consulted only for a legacy lock directory without
+   * owner.json; see removeUnownedLease. Injectable for tests. */
+  legacyBootTime?: () => number | null;
   diagnostics?: DiagnosticSink;
 }
 
@@ -186,29 +196,81 @@ export class RuntimeService {
     const lock = this.lockPath();
     for (let attempt = 0; attempt < 2; attempt += 1) {
       try {
-        await mkdir(lock, { mode: 0o700 });
-        const identity = await lstat(lock);
-        this.lease = { dev: identity.dev, ino: identity.ino };
-        try {
-          await privateWrite(
-            path.join(lock, "owner.json"),
-            JSON.stringify({
-              pid: this.pid(),
-              created_at: new Date().toISOString(),
-            }),
-          );
-        } catch (error) {
-          await this.releaseLease();
-          throw error;
-        }
-        return;
+        await this.publishLease(lock);
+        break;
       } catch (error) {
-        if ((error as NodeJS.ErrnoException).code !== "EEXIST" || attempt > 0) {
+        if (!leaseTaken(error) || attempt > 0) {
           throw new Error("runtime_service_lock_unavailable", { cause: error });
         }
         await this.removeStaleLease(lock);
       }
     }
+    await this.sweepLeaseDebris();
+  }
+
+  /** Publish the lease in one step: the lock directory becomes visible with
+   * owner.json (pid, created_at, boot_id) already inside it, so a power cut
+   * can leave a staging directory behind but never a lock without an owner.
+   * The old mkdir -> write sequence had exactly that window (issue #184). */
+  private async publishLease(lock: string): Promise<void> {
+    const staging = `${lock}${STAGING_INFIX}${randomUUID()}`;
+    await mkdir(staging, { mode: 0o700 });
+    try {
+      await privateWrite(
+        path.join(staging, "owner.json"),
+        JSON.stringify({
+          pid: this.pid(),
+          created_at: new Date().toISOString(),
+          boot_id: this.currentBootId(),
+        }),
+      );
+      // POSIX rename() silently REPLACES an existing *empty* directory, so on
+      // its own it is not an exclusive create. A non-empty lock (every lease
+      // this version publishes) makes it fail with EEXIST/ENOTEMPTY. An empty
+      // lock can only come from an older version's two-step mkdir -> write:
+      // either power-cut debris, or a live acquire in flight. The latter must
+      // not be overwritten, and the former has to go through
+      // removeStaleLease's legacy rule rather than be replaced blindly, so an
+      // existing lock of any kind is reported as taken here. The residual race
+      // (an old binary mkdir-ing between this check and the rename) needs an
+      // old and a new Runtime starting on the same home within microseconds;
+      // this version never creates an empty lock, since releaseLease also
+      // removes it with a single rename.
+      if (await pathExists(lock)) throw leaseTakenError();
+      await rename(staging, lock);
+    } catch (error) {
+      await rm(staging, { recursive: true, force: true }).catch(
+        () => undefined,
+      );
+      throw error;
+    }
+    // rename() keeps the inode, so this is the published directory's identity;
+    // releaseLease compares it before touching the lock.
+    const identity = await lstat(lock);
+    this.lease = { dev: identity.dev, ino: identity.ino };
+  }
+
+  /** Remove staging and cleared-lease directories left behind by a crash.
+   * Runs only while this process holds the lease, so a staging directory is
+   * either debris or belongs to a competitor that has already lost (its
+   * rename cannot succeed over our non-empty lock). Best-effort. */
+  private async sweepLeaseDebris(): Promise<void> {
+    const base = path.basename(this.lockPath());
+    const names = await readdir(this.options.dataDir).catch((): string[] => []);
+    await Promise.all(
+      names
+        .filter(
+          (name) =>
+            name.startsWith(`${base}${STAGING_INFIX}`) ||
+            name.startsWith(`${base}${STALE_INFIX}`),
+        )
+        .map((name) =>
+          rm(path.join(this.options.dataDir, name), {
+            recursive: true,
+            force: true,
+          }).catch(() => undefined),
+        ),
+    );
   }
 
   private async removeStaleLease(lock: string): Promise<void> {
@@ -223,25 +285,81 @@ export class RuntimeService {
     ) {
       throw new Error("runtime_service_lock_unsafe");
     }
+    let raw: string;
+    try {
+      raw = await readFile(path.join(lock, "owner.json"), "utf8");
+    } catch (error) {
+      if ((error as NodeJS.ErrnoException).code === "ENOENT") {
+        await this.removeUnownedLease(lock, info);
+        return;
+      }
+      throw new Error("runtime_service_lock_invalid", { cause: error });
+    }
     let owner: unknown;
     try {
-      owner = JSON.parse(await readFile(path.join(lock, "owner.json"), "utf8"));
+      owner = JSON.parse(raw);
     } catch {
       throw new Error("runtime_service_lock_invalid");
     }
-    const pid =
-      owner && typeof owner === "object"
-        ? (owner as { pid?: unknown }).pid
-        : undefined;
+    if (!owner || typeof owner !== "object" || Array.isArray(owner)) {
+      throw new Error("runtime_service_lock_invalid");
+    }
+    // A lease written during another boot cannot have a living owner: every
+    // process of that boot is gone, and a PID that answers the liveness probe
+    // now merely reused the number. Without this, a forced power-off left the
+    // lock behind, the reboot recycled the owner's PID, and launchd's
+    // KeepAlive restarted the wrapper into runtime_service_already_running
+    // without end (issue #184). The kernel's boot identity is compared, not
+    // the wall clock: the clock can jump forward while a lease is held (no
+    // RTC, VM resume, NTP or manual correction) and make a live lease look
+    // older than the boot. When either side is unknown (a lease from an older
+    // version, an unsupported platform) the PID probe below decides, as before.
+    const ownerBoot = (owner as { boot_id?: unknown }).boot_id;
+    const currentBoot = this.currentBootId();
+    if (
+      typeof ownerBoot === "string" &&
+      ownerBoot !== "" &&
+      currentBoot !== null &&
+      ownerBoot !== currentBoot
+    ) {
+      await this.clearLeaseDir(lock);
+      return;
+    }
+    const pid = (owner as { pid?: unknown }).pid;
     if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 1) {
       throw new Error("runtime_service_lock_invalid");
     }
     if ((this.options.processIsAlive ?? processIsAlive)(pid)) {
       throw new Error("runtime_service_already_running");
     }
-    const stale = `${lock}.stale.${randomUUID()}`;
+    await this.clearLeaseDir(lock);
+  }
+
+  /** A lock directory without owner.json. This version publishes owner.json
+   * together with the lock and releases it with one rename, so such a
+   * directory was written by an older version (mkdir -> write, where a power
+   * cut in between leaves it empty) or by hand. It names no PID and no boot,
+   * so the only usable evidence is "created before the current boot": clear
+   * it then, and otherwise keep the previous refusal
+   * (runtime_service_lock_invalid), including when the boot time is unknown.
+   *
+   * The kernel boot time and the directory mtime are both wall-clock values,
+   * which is why lease ownership itself no longer depends on them. Here a
+   * forward clock jump can only hurt if an older binary is live inside its
+   * mkdir -> write window at that very moment; a backward jump only keeps the
+   * old refusal. */
+  private async removeUnownedLease(lock: string, info: Stats): Promise<void> {
+    const bootTime = (this.options.legacyBootTime ?? currentBootTimeMs)();
+    if (bootTime === null || !(info.mtimeMs < bootTime)) {
+      throw new Error("runtime_service_lock_invalid");
+    }
+    await this.clearLeaseDir(lock);
+  }
+
+  private async clearLeaseDir(lock: string): Promise<void> {
+    const stale = `${lock}${STALE_INFIX}${randomUUID()}`;
     await rename(lock, stale);
-    await rm(stale, { recursive: true });
+    await rm(stale, { recursive: true, force: true });
   }
 
   private async releaseLease(): Promise<void> {
@@ -252,8 +370,19 @@ export class RuntimeService {
     const current = await lstat(lock).catch(() => null);
     if (!current || current.dev !== lease.dev || current.ino !== lease.ino)
       return;
-    await rm(path.join(lock, "owner.json"), { force: true });
-    await rmdir(lock).catch(() => undefined);
+    // Take the lock away with one rename so no observer ever sees it without
+    // owner.json: an owner-less lock is left to mean legacy debris.
+    const released = `${lock}${STALE_INFIX}${randomUUID()}`;
+    try {
+      await rename(lock, released);
+    } catch {
+      return;
+    }
+    await rm(released, { recursive: true, force: true }).catch(() => undefined);
+  }
+
+  private currentBootId(): string | null {
+    return (this.options.bootId ?? currentBootId)();
   }
 
   private async writeReceipt(
@@ -302,6 +431,91 @@ export class RuntimeService {
   }
 }
 
+const STAGING_INFIX = ".tmp-";
+const STALE_INFIX = ".stale.";
+
+function leaseTaken(error: unknown): boolean {
+  const code = (error as NodeJS.ErrnoException | undefined)?.code;
+  return code === "EEXIST" || code === "ENOTEMPTY";
+}
+
+function leaseTakenError(): NodeJS.ErrnoException {
+  return Object.assign(new Error("runtime_service_lock_exists"), {
+    code: "EEXIST",
+  });
+}
+
+async function pathExists(target: string): Promise<boolean> {
+  try {
+    await lstat(target);
+    return true;
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return false;
+    throw error;
+  }
+}
+
+function sysctl(name: string): string {
+  return execFileSync("/usr/sbin/sysctl", ["-n", name], {
+    encoding: "utf8",
+    timeout: 1000,
+    maxBuffer: 4096,
+    killSignal: "SIGKILL",
+    shell: false,
+    stdio: ["ignore", "pipe", "ignore"],
+    env: { PATH: "/usr/bin:/bin:/usr/sbin:/sbin", LANG: "C", LC_ALL: "C" },
+  });
+}
+
+let cachedBootId: string | null | undefined;
+
+/** The kernel's identity for the current boot, cached per process (it cannot
+ * change while the process lives). `null` on unsupported platforms or when
+ * the source cannot be read; the lease logic then falls back to the PID
+ * probe. */
+function currentBootId(): string | null {
+  if (cachedBootId === undefined) cachedBootId = readBootId();
+  return cachedBootId;
+}
+
+function readBootId(): string | null {
+  try {
+    let raw: string;
+    if (process.platform === "darwin") raw = sysctl("kern.bootsessionuuid");
+    else if (process.platform === "linux")
+      raw = readFileSync("/proc/sys/kernel/random/boot_id", "utf8");
+    else return null;
+    const id = raw.trim();
+    return /^[0-9A-Za-z-]{8,64}$/.test(id) ? id : null;
+  } catch {
+    return null;
+  }
+}
+
+/** Wall-clock epoch (ms) the kernel reports for the current boot, or `null`.
+ * Used only for owner-less legacy lock directories. */
+function currentBootTimeMs(): number | null {
+  try {
+    if (process.platform === "darwin") {
+      const match = /\bsec\s*=\s*(\d+)\s*,\s*usec\s*=\s*(\d+)/.exec(
+        sysctl("kern.boottime"),
+      );
+      return match
+        ? Number(match[1]) * 1000 + Math.floor(Number(match[2]) / 1000)
+        : null;
+    }
+    if (process.platform === "linux") {
+      const match = /^btime\s+(\d+)\s*$/m.exec(
+        readFileSync("/proc/stat", "utf8"),
+      );
+      return match ? Number(match[1]) * 1000 : null;
+    }
+  } catch {
+    // Unknown boot time: the caller keeps refusing.
+  }
+  return null;
+}
+
 async function privateWrite(target: string, value: string): Promise<void> {
   const temporary = `${target}.${randomUUID()}.tmp`;
   try {

memhub/packages/runtime-control/src/sdk-migration-log-order.test.ts

@@ -0,0 +1,101 @@
+import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+import { describe, expect, it } from "vitest";
+import {
+  sdkMigrationLogPath,
+  writeSdkMigrationLog,
+} from "./sdk-migration-log.js";
+
+async function events(dataDir: string): Promise<string[]> {
+  return (await readFile(sdkMigrationLogPath(dataDir), "utf8"))
+    .trim()
+    .split("\n")
+    .map((line) => JSON.parse(line).event);
+}
+
+describe("SDK migration log ordering", () => {
+  it("preserves invocation order for concurrent progress and completion writes", async () => {
+    const dataDir = await mkdtemp(
+      path.join(os.tmpdir(), "migration-log-order-"),
+    );
+    try {
+      const expected = Array.from(
+        { length: 100 },
+        (_, i) => `migration.progress.${i}`,
+      );
+      expected.push("migration.finished");
+      await Promise.all(
+        expected.map((event) =>
+          writeSdkMigrationLog(dataDir, {
+            writer: "desktop",
+            event,
+            outcome: "started",
+          }),
+        ),
+      );
+      expect(await events(dataDir)).toEqual(expected);
+    } finally {
+      await rm(dataDir, { recursive: true, force: true });
+    }
+  });
+
+  it("awaiting the terminal event also drains preceding progress writes", async () => {
+    const dataDir = await mkdtemp(
+      path.join(os.tmpdir(), "migration-log-drain-"),
+    );
+    const pending: Promise<void>[] = [];
+    try {
+      const expected = Array.from(
+        { length: 50 },
+        (_, i) => `migration.progress.${i}`,
+      );
+      for (const event of expected) {
+        pending.push(
+          writeSdkMigrationLog(dataDir, {
+            writer: "desktop",
+            event,
+            outcome: "started",
+          }),
+        );
+      }
+      await writeSdkMigrationLog(dataDir, {
+        writer: "desktop",
+        event: "migration.finished",
+        outcome: "succeeded",
+      });
+      expect(await events(dataDir)).toEqual([
+        ...expected,
+        "migration.finished",
+      ]);
+    } finally {
+      await Promise.all(pending);
+      await rm(dataDir, { recursive: true, force: true });
+    }
+  });
+
+  it("does not let a failed append block subsequent writes to the same home", async () => {
+    const dataDir = await mkdtemp(
+      path.join(os.tmpdir(), "migration-log-recovery-"),
+    );
+    try {
+      await writeFile(path.join(dataDir, "logs"), "not a directory");
+      await expect(
+        writeSdkMigrationLog(dataDir, {
+          writer: "desktop",
+          event: "migration.progress",
+          outcome: "started",
+        }),
+      ).resolves.toBeUndefined();
+      await rm(path.join(dataDir, "logs"));
+      await writeSdkMigrationLog(dataDir, {
+        writer: "desktop",
+        event: "migration.finished",
+        outcome: "succeeded",
+      });
+      expect(await events(dataDir)).toEqual(["migration.finished"]);
+    } finally {
+      await rm(dataDir, { recursive: true, force: true });
+    }
+  });
+});

memhub/packages/runtime-control/src/sdk-migration-log.ts

@@ -33,6 +33,9 @@ export const SDK_MIGRATION_LOG_NAME = "sdk-migration.log";
 const MAX_BYTES = 4 * 1024 * 1024;
 const SAFE_TEXT = /^[A-Za-z0-9_.:/@+-]{0,160}$/;
 const MAX_ITEMS = 32;
+// Preserve invocation order within this process, including fire-and-forget
+// progress followed by an awaited terminal event. Homes remain independent.
+const pendingWrites = new Map<string, Promise<void>>();
 
 export function sdkMigrationLogPath(dataDir: string): string {
   return path.join(dataDir, "logs", SDK_MIGRATION_LOG_NAME);
@@ -103,17 +106,28 @@ export async function writeSdkMigrationLog(
       platform: process.platform,
       ...(entry.details === undefined ? {} : { details: entry.details }),
     });
-    const target = sdkMigrationLogPath(dataDir);
-    await mkdir(path.dirname(target), { recursive: true, mode: 0o700 });
-    const size = await stat(target).then(
-      (info) => info.size,
-      () => 0,
-    );
-    if (size > MAX_BYTES) await rename(target, `${target}.1`);
-    await appendFile(target, `${JSON.stringify(line)}\n`, {
-      encoding: "utf8",
-      mode: 0o600,
-    });
+    const target = path.resolve(sdkMigrationLogPath(dataDir));
+    const previous = pendingWrites.get(target) ?? Promise.resolve();
+    const pending = previous
+      .catch(() => undefined)
+      .then(async () => {
+        await mkdir(path.dirname(target), { recursive: true, mode: 0o700 });
+        const size = await stat(target).then(
+          (info) => info.size,
+          () => 0,
+        );
+        if (size > MAX_BYTES) await rename(target, `${target}.1`);
+        await appendFile(target, `${JSON.stringify(line)}\n`, {
+          encoding: "utf8",
+          mode: 0o600,
+        });
+      });
+    pendingWrites.set(target, pending);
+    try {
+      await pending;
+    } finally {
+      if (pendingWrites.get(target) === pending) pendingWrites.delete(target);
+    }
   } catch {
     // Never let the log decide whether a migration or a startup happens.
   }

memhub/scripts/generate_sbom.py

@@ -36,7 +36,7 @@ MEMMY_LICENSE = (
 MEMMY_PROVENANCE = MEMMY_LICENSE.with_name("PROVENANCE.md")
 MEMMY_PIN = "25acd5f29fc7df73b943713abe61f1b2f9839887"
 HY_MEMORY_NAME = "hy-memory"
-HY_MEMORY_VERSION = "2.0.6"
+HY_MEMORY_VERSION = "2.0.7"
 
 
 class SbomError(RuntimeError):

memhub/scripts/install-docs.test.mjs

@@ -12,7 +12,7 @@ test("installation docs disclose publication and lifecycle boundaries", async ()
     "utf8",
   );
 
-  assert.match(readme, /`hy-memory==2\.0\.6`/);
+  assert.match(readme, /`hy-memory==2\.0\.7`/);
   assert.match(readme, /`v1\.0\.5` 的发行目标是 Linux x64\/arm64 Runtime/);
   assert.match(readme, /Linux 制品由 GitHub Actions/);
   assert.match(readme, /macOS.*Developer ID 签名、公证/s);

memhub/scripts/test_generate_sbom.py

@@ -13,7 +13,7 @@ spec.loader.exec_module(module)
 
 
 def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) -> None:
-    dist_info = tmp_path / "hy_memory-2.0.6.dist-info"
+    dist_info = tmp_path / "hy_memory-2.0.7.dist-info"
     licenses = dist_info / "licenses"
     licenses.mkdir(parents=True)
     (dist_info / "METADATA").write_text(
@@ -21,7 +21,7 @@ def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) ->
             (
                 "Metadata-Version: 2.4",
                 "Name: hy-memory",
-                "Version: 2.0.6",
+                "Version: 2.0.7",
                 "License-Expression: Apache-2.0",
                 "",
             )
@@ -38,7 +38,7 @@ def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) ->
     notice = module.hy_memory_notice(dist_info)
 
     assert notice["name"] == "hy-memory"
-    assert notice["version"] == "2.0.6"
+    assert notice["version"] == "2.0.7"
     assert notice["license"] == "Apache-2.0"
     assert notice["copyright"] == "Copyright (C) 2026 Tencent. All rights reserved."
     assert "Licensed under the Apache License" in notice["text"]

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "hy-memory"
-version = "2.0.6"
+version = "2.0.8"
 description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
 readme = "README.md"
 requires-python = ">=3.9"

tests/coding/test_content_log_levels.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+"""对话、记忆正文和模型输出只进 DEBUG，不进 INFO / WARNING。
+
+MemHub 等调用方默认以 INFO 运行；这些文本一旦进 INFO，就会写进常驻的
+hy_memory.log。这里逐条锁住曾经把正文打在 INFO / WARNING 的日志。
+"""
+
+import logging
+
+import pytest
+
+from hy_memory.agent.reconciler import ReconcileOp as AgentReconcileOp
+from hy_memory.agent.reconciler import _log_ops_detail
+from hy_memory.coding.extractor import CodingMemoryExtractor
+from hy_memory.coding.fact_reconciler import WorkingFactReconciler
+from hy_memory.coding.judge import classify_and_rewrite_queries
+from hy_memory.coding.reconciler import CodingMemoryReconciler
+from hy_memory.coding.types import CodingMemory, WorkingFact
+
+from tests.coding._helpers import MockLLMProvider
+
+SECRET = "SECRET-CONVERSATION-TEXT-42"
+
+
+def _above_debug(caplog):
+    return [
+        r.getMessage() for r in caplog.records
+        if r.name.startswith("hy_memory") and r.levelno > logging.DEBUG
+    ]
+
+
+def _debug(caplog):
+    return [
+        r.getMessage() for r in caplog.records
+        if r.name.startswith("hy_memory") and r.levelno == logging.DEBUG
+    ]
+
+
+def _assert_secret_only_at_debug(caplog):
+    assert not [m for m in _above_debug(caplog) if SECRET in m]
+    assert [m for m in _debug(caplog) if SECRET in m]
+
+
+@pytest.mark.asyncio
+class TestStoreInsertLogs:
+    async def test_experience_task_not_at_info(self, coding_store, caplog):
+        caplog.set_level(logging.DEBUG, logger="hy_memory")
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u1", task=SECRET, solution="S",
+        ))
+        info = [m for m in _above_debug(caplog) if "inserted experience_id=e1" in m]
+        assert info
+        _assert_secret_only_at_debug(caplog)
+
+    async def test_fact_statement_not_at_info(self, coding_store, caplog):
+        caplog.set_level(logging.DEBUG, logger="hy_memory")
+        await coding_store.insert_fact(WorkingFact(
+            fact_id="f1", user_id="u1", agent_id="a1", session_id="s1",
+            workspace_id="ws/a", statement=SECRET,
+        ))
+        info = [m for m in _above_debug(caplog) if "inserted fact_id=f1" in m]
+        assert info
+        _assert_secret_only_at_debug(caplog)
+
+
+class TestParseFailureWarnings:
+    """解析失败仍发 WARNING，但只带长度，模型输出进 DEBUG。"""
+
+    @pytest.mark.parametrize("parse", [
+        CodingMemoryExtractor._parse_output,
+        CodingMemoryReconciler._parse_ops,
+        WorkingFactReconciler._parse_ops,
+    ])
+    def test_warning_keeps_length_drops_text(self, parse, caplog):
+        caplog.set_level(logging.DEBUG, logger="hy_memory")
+        bad = f"not json {SECRET}"
+        parse(bad)
+        warnings = [
+            r.getMessage() for r in caplog.records
+            if r.levelno == logging.WARNING and "failed to parse" in r.getMessage()
+        ]
+        assert warnings and f"({len(bad)} chars)" in warnings[0]
+        _assert_secret_only_at_debug(caplog)
+
+
+@pytest.mark.asyncio
+class TestJudgeLogs:
+    async def test_rewrite_not_at_info(self, caplog):
+        caplog.set_level(logging.DEBUG, logger="hy_memory")
+        llm = MockLLMProvider(f"1\n{SECRET}")
+        result = await classify_and_rewrite_queries(["q"], llm)
+        assert result["rewrite_query"] == SECRET
+        _assert_secret_only_at_debug(caplog)
+
+    async def test_unrecognized_head_not_at_warning(self, caplog):
+        caplog.set_level(logging.DEBUG, logger="hy_memory")
+        llm = MockLLMProvider(f"{SECRET}\nrewrite")
+        result = await classify_and_rewrite_queries(["q"], llm)
+        assert result["ok"] is False
+        _assert_secret_only_at_debug(caplog)
+
+
+def test_reconciler_ops_detail_is_debug(caplog):
+    caplog.set_level(logging.DEBUG, logger="hy_memory")
+    _log_ops_detail([AgentReconcileOp(op="ADD", content=SECRET)])
+    _assert_secret_only_at_debug(caplog)
+
+
+class _StopAfterLogging(Exception):
+    pass
+
+
+def _add_client():
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    from hy_memory.client import HyMemoryClient
+
+    client = HyMemoryClient.__new__(HyMemoryClient)
+    client._mode = "lite"
+    client._config = SimpleNamespace(coding=None)
+    writer = SimpleNamespace(
+        VERSION="test", write=AsyncMock(side_effect=_StopAfterLogging),
+    )
+    client._registry = SimpleNamespace(get_writer=AsyncMock(return_value=writer))
+    return client
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("data", [
+    SECRET,
+    [{"role": "user", "content": SECRET}],
+])
+async def test_add_does_not_log_input_at_info(data, caplog):
+    caplog.set_level(logging.DEBUG, logger="hy_memory")
+    client = _add_client()
+    with pytest.raises(_StopAfterLogging):
+        await client._async_add_impl(data, user_id="u1")
+    assert [m for m in _above_debug(caplog) if m.startswith("[add] writing")]
+    _assert_secret_only_at_debug(caplog)

tests/coding/test_memory_at.py

@@ -0,0 +1,229 @@
+# -*- coding: utf-8 -*-
+"""经验与事实记下对话时间 memory_at，并以秒级 epoch 对外返回（与 chat 一致）。
+
+之前生产力链路丢了 add() 传进来的 memory_at，coding_memory_meta / working_fact
+也没有这一列，MemHub 里所有经验都显示"时间未知"。
+"""
+
+import sqlite3
+import time
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from hy_memory.client import HyMemoryClient
+from hy_memory.coding.extractor import CodingMemoryExtractor
+from hy_memory.coding.reconciler import CodingMemoryReconciler
+from hy_memory.coding.store import CodingMemoryStore, _CREATE_TABLE_SQL, _CREATE_FACT_TABLE_SQL
+from hy_memory.coding.types import CodingMemory, WorkingFact
+from hy_memory.coding.writer import CodingWriter
+
+from tests.coding._helpers import MockLLMProvider
+from tests.coding.test_list_and_stats_by_scene import _client, _unwrapped
+from tests.coding.test_write_scene_routing import _messages
+from tests.coding.test_writer_e2e import _coding_messages, _extract_resp, _reconcile_resp
+
+T1 = datetime(2026, 10, 1, 9, 30, tzinfo=timezone.utc)
+T2 = T1 + timedelta(days=3)
+
+
+def _epoch(value: datetime) -> int:
+    return int(value.timestamp())
+
+
+def _writer(store, embed, llm):
+    return CodingWriter(
+        store=store,
+        extractor=CodingMemoryExtractor(llm),
+        reconciler=CodingMemoryReconciler(store, embed, llm),
+        llm_provider=llm, embed_service=embed,
+    )
+
+
+@pytest.mark.asyncio
+class TestStore:
+    async def test_round_trip_as_epoch(self, coding_store):
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="T", solution="S", memory_at=T1,
+        ))
+        await coding_store.insert_fact(WorkingFact(
+            fact_id="f1", user_id="u", statement="x", memory_at=T1,
+        ))
+        exp = await coding_store.get_by_id("e1", user_id="u")
+        fact = await coding_store.get_fact("f1", user_id="u")
+        assert _epoch(exp.memory_at) == _epoch(T1)
+        assert exp.to_dict()["memory_at"] == _epoch(T1)
+        assert fact.to_dict()["memory_at"] == _epoch(T1)
+
+    async def test_missing_time_stays_unknown(self, coding_store):
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="T", solution="S",
+        ))
+        exp = await coding_store.get_by_id("e1", user_id="u")
+        assert exp.memory_at is None and exp.to_dict()["memory_at"] is None
+
+    @pytest.mark.parametrize("old, new", [(T1, T2), (T2, T1), (T2, None)])
+    async def test_update_keeps_the_later_time(self, coding_store, old, new):
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="T", solution="S",
+            session_id="s", memory_at=old,
+        ))
+        await coding_store.update(CodingMemory(
+            experience_id="e1", user_id="u", task="T", solution="S2",
+            session_id="s", memory_at=new,
+        ))
+        exp = await coding_store.get_by_id("e1", user_id="u")
+        assert exp.solution == "S2"
+        assert _epoch(exp.memory_at) == _epoch(T2)
+
+    async def test_edit_keeps_the_time(self, coding_store):
+        await coding_store.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="T", solution="S", memory_at=T1,
+        ))
+        await coding_store.edit_experience("e1", user_id="u", solution="S2")
+        exp = await coding_store.get_by_id("e1", user_id="u")
+        assert _epoch(exp.memory_at) == _epoch(T1)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("variant", ["working", "coding"])
+async def test_existing_database_gains_the_column_on_open(coding_config, mock_embed, variant):
+    store = CodingMemoryStore(coding_config, embed_service=mock_embed, variant=variant)
+    # 先按老 schema（没有 memory_at）建库并放一行
+    conn = sqlite3.connect(store._db_path)
+    conn.executescript(_CREATE_TABLE_SQL.replace("    memory_at        INTEGER,\n", ""))
+    conn.executescript(_CREATE_FACT_TABLE_SQL.replace("    memory_at              INTEGER,\n", ""))
+    conn.execute(
+        "INSERT INTO coding_memory_meta (experience_id, user_id, task, solution, "
+        "created_at, updated_at) VALUES ('old', 'u', 'T', 'S', '2026-01-01T00:00:00', "
+        "'2026-01-01T00:00:00')"
+    )
+    conn.commit()
+    conn.close()
+
+    await store.initialize()
+    try:
+        old = await store.get_by_id("old", user_id="u")
+        assert old is not None and old.memory_at is None
+        await store.insert_fact(WorkingFact(
+            fact_id="f1", user_id="u", statement="x", memory_at=T1,
+        ))
+        assert _epoch((await store.get_fact("f1")).memory_at) == _epoch(T1)
+    finally:
+        await store.close()
+    # 再开一次不出错（ADD COLUMN 是幂等的）
+    again = CodingMemoryStore(coding_config, embed_service=mock_embed, variant=variant)
+    await again.initialize()
+    await again.close()
+
+
+@pytest.mark.asyncio
+class TestWriter:
+    def _llm(self):
+        return MockLLMProvider([
+            _extract_resp(
+                [{"task": "打包发布", "solution": "S", "confidence": 0.9}],
+                [{"statement": "发布脚本在 scripts/publish_package.sh"}],
+            ),
+            _reconcile_resp([{"draft_idx": 0, "action": "ADD"}]),
+        ])
+
+    async def test_conversation_time_reaches_both_layers(self, coding_store, mock_embed):
+        resp = await _writer(coding_store, mock_embed, self._llm()).write(
+            _coding_messages(), user_id="u", session_id="s", memory_at=T1,
+        )
+        exp = await coding_store.get_by_id(resp["experience_ids"][0], user_id="u")
+        fact = await coding_store.get_fact(resp["fact_ids"][0], user_id="u")
+        assert _epoch(exp.memory_at) == _epoch(T1)
+        assert _epoch(fact.memory_at) == _epoch(T1)
+
+    async def test_defaults_to_request_time_like_chat(self, coding_store, mock_embed):
+        before = int(time.time())
+        resp = await _writer(coding_store, mock_embed, self._llm()).write(
+            _coding_messages(), user_id="u", session_id="s",
+        )
+        exp = await coding_store.get_by_id(resp["experience_ids"][0], user_id="u")
+        fact = await coding_store.get_fact(resp["fact_ids"][0], user_id="u")
+        for stamp in (_epoch(exp.memory_at), _epoch(fact.memory_at)):
+            assert before <= stamp <= int(time.time()) + 1
+
+
+@pytest.mark.asyncio
+async def test_add_passes_the_time_to_the_productivity_writer(monkeypatch):
+    seen = {}
+
+    async def _judge(*args, **kwargs):
+        return "working"
+
+    class _Writer:
+        async def write(self, *args, **kwargs):
+            seen.update(kwargs)
+            return {"success": True, "experience_ids": [], "fact_ids": []}
+
+    async def _get_writer(*, variant):
+        return _Writer()
+
+    monkeypatch.setattr("hy_memory.client.classify_messages_scene", _judge)
+    ns = SimpleNamespace(
+        _config=SimpleNamespace(coding=SimpleNamespace(enable=True)),
+        _coding_llm_provider=object(),
+        _cache=None,
+        _get_coding_writer=_get_writer,
+        _normalize_scene=HyMemoryClient._normalize_scene,
+        _scene_variant=HyMemoryClient._scene_variant,
+        _parse_input=HyMemoryClient._parse_input,
+    )
+    await HyMemoryClient._async_add_impl(
+        ns, _messages(), user_id="u1", agent_id="a1", memory_at=T1,
+    )
+    assert seen["memory_at"] == T1
+
+
+@pytest.mark.asyncio
+async def test_list_get_and_search_return_epoch_seconds(coding_config, mock_embed):
+    coding = CodingMemoryStore(coding_config, embed_service=mock_embed, variant="coding")
+    await coding.initialize()
+    try:
+        await coding.insert(CodingMemory(
+            experience_id="e1", user_id="u", task="打包发布", solution="S", memory_at=T1,
+        ))
+        await coding.insert_fact(WorkingFact(
+            fact_id="f1", user_id="u", statement="打包发布", memory_at=T2,
+            derived_experience_ids=["e1"],
+        ))
+        client = _client([], {"coding": coding})
+        list_memories = _unwrapped("async_list_memories")
+
+        page = await list_memories(client, user_id="u", scene="coding")
+        assert page["experiences"][0]["memory_at"] == _epoch(T1)
+
+        merged = await list_memories(client, user_id="u", scene="all")
+        assert merged["memories"][0]["memory_at"] == _epoch(T1)
+
+        exp = await coding.get_by_id("e1", user_id="u")
+        got = HyMemoryClient._format_productivity_get(
+            exp, kind="experience", variant="coding", request_id="r", elapsed_ms=0,
+        )
+        assert got["memory_at"] == _epoch(T1)
+
+        client._embed_service = mock_embed
+        client._coding_store = coding
+
+        async def _ready(variant="working"):
+            return None
+
+        client._get_coding_writer = _ready
+        client._cache = None
+        resp = await HyMemoryClient._search_coding(
+            client, target_query="打包发布", user_ids=["u"], workspace_id=None,
+            branch=None, limit=5, working_fact_min_score=0.0,
+            working_experience_min_score=0.0, request_id="r",
+            t0=time.perf_counter(), variant="coding",
+        )
+        assert resp["facts"][0]["memory_at"] == _epoch(T2)
+        assert resp["experiences"][0]["memory_at"] == _epoch(T1)
+        unified = HyMemoryClient._search_lane_items("coding", resp)
+        assert {item["memory_at"] for item in unified} == {_epoch(T1), _epoch(T2)}
+    finally:
+        await coding.close()

tests/test_async_executor_context.py

@@ -0,0 +1,51 @@
+"""The guarded to_thread adapter preserves asyncio.to_thread context semantics."""
+import asyncio
+import contextvars
+import importlib.util
+from pathlib import Path
+import unittest
+
+
+# This helper is dependency-free; exercise it without initializing the SDK.
+_spec = importlib.util.spec_from_file_location(
+    "async_executor_under_test",
+    Path(__file__).resolve().parents[1] / "hy_memory/utils/async_executor.py",
+)
+_executor = importlib.util.module_from_spec(_spec)
+_spec.loader.exec_module(_executor)
+
+
+class ThreadContextTests(unittest.IsolatedAsyncioTestCase):
+    async def test_copies_context_and_does_not_propagate_worker_changes(self):
+        request_id = contextvars.ContextVar("request_id", default="missing")
+        request_id.set("caller")
+
+        def work(prefix, *, suffix):
+            value = request_id.get()
+            request_id.set("worker")
+            return prefix + value + suffix
+
+        result = await _executor.to_thread_guarded(work, "[", suffix="]")
+        self.assertEqual(result, "[caller]")
+        self.assertEqual(request_id.get(), "caller")
+
+    async def test_concurrent_callers_keep_separate_contexts(self):
+        request_id = contextvars.ContextVar("request_id", default="missing")
+
+        async def call(value):
+            request_id.set(value)
+            await asyncio.sleep(0)
+            return await _executor.to_thread_guarded(request_id.get)
+
+        self.assertEqual(await asyncio.gather(call("a"), call("b")), ["a", "b"])
+
+    async def test_context_is_copied_when_coroutine_runs(self):
+        request_id = contextvars.ContextVar("request_id", default="missing")
+        request_id.set("created")
+        pending = _executor.to_thread_guarded(request_id.get)
+        request_id.set("awaited")
+        self.assertEqual(await pending, "awaited")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_embed_warning_log.py

@@ -90,7 +90,10 @@ class TestEmbedInfoLog:
             assert payload["provider"] == "openai"
             assert payload["model"] == "text-embedding-test"
             assert payload["api"] == "https://example.com/v1/embeddings"
-            assert payload["request"]["input"] == ["hello world"]
+            # 待向量化的原文不进 INFO，只记每条字符数
+            assert "input" not in payload["request"]
+            assert payload["request"]["input_chars"] == [11]
+            assert "hello world" not in info_msgs[0]
             assert payload["request"]["dimensions"] == 8
             assert payload["response"]["vector_dims"] == [8]
             assert payload["response"]["vector_count"] == 1

tests/test_literal_special_token_counting.py

@@ -0,0 +1,29 @@
+"""User text may quote tokenizer control spellings without using control tokens."""
+
+import pytest
+import tiktoken
+
+from hy_memory.core.embed_service import _count_tokens, _truncate_for_embedding
+from hy_memory.utils.token_count import count_tokens
+
+
+@pytest.mark.parametrize('counter', [count_tokens, _count_tokens])
+@pytest.mark.parametrize('marker', ['<|endoftext|>', '<|fim_prefix|>'])
+def test_literal_special_tokens_are_counted_as_ordinary_text(marker, counter):
+    text = marker + '你好世界' * 20
+    expected = len(tiktoken.get_encoding('cl100k_base').encode_ordinary(text))
+    assert counter(text) == expected
+
+
+def test_literal_special_token_does_not_bypass_embedding_limit():
+    text = '<|endoftext|>' + '你好世界' * 3000
+    encoder = tiktoken.get_encoding('cl100k_base')
+    truncated = _truncate_for_embedding(text)
+    assert truncated != text
+    assert len(encoder.encode_ordinary(truncated)) <= 8192 - 64
+    assert truncated.startswith('<|endoftext|>')
+
+
+def test_short_literal_special_token_text_is_preserved():
+    text = 'Explain <|endoftext|> as a literal string.'
+    assert _truncate_for_embedding(text) == text

tests/test_pipeline_observability.py

@@ -1,4 +1,4 @@
-"""Pipeline log（始终）与 trace（可关）分离。"""
+"""Pipeline log（默认开，可关）与 trace（可关）分离。"""
 
 import asyncio
 import json
@@ -10,6 +10,7 @@ from hy_memory.config import MemoryConfig
 from hy_memory.data.cache_sqlite import SqliteCache
 from hy_memory.utils.pipeline_observability import (
     is_llm_content_logging_enabled,
+    is_pipeline_log_enabled,
     is_pipeline_trace_enabled,
     resolve_pipeline_log_dir,
 )
@@ -39,6 +40,14 @@ class TestPipelineObservabilityConfig:
         monkeypatch.setenv("MEMORY_DATA_DIR", str(tmp_path / "data"))
         assert resolve_pipeline_log_dir() == str(tmp_path / "data" / "logs" / "pipeline")
 
+    def test_log_enabled_default(self, monkeypatch):
+        monkeypatch.delenv("MEMORY_PIPELINE_LOG_ENABLED", raising=False)
+        assert is_pipeline_log_enabled() is True
+
+    def test_log_disabled(self, monkeypatch):
+        monkeypatch.setenv("MEMORY_PIPELINE_LOG_ENABLED", "false")
+        assert is_pipeline_log_enabled() is False
+
     def test_trace_enabled_default(self, monkeypatch):
         monkeypatch.delenv("MEMORY_PIPELINE_TRACE_ENABLED", raising=False)
         assert is_pipeline_trace_enabled() is True
@@ -91,6 +100,98 @@ class TestObservabilityHookBehavior:
 
         run(_test())
 
+    def test_log_flag_on_writes_jsonl(self, env_setup, monkeypatch):
+        """显式开启与默认一致：照常写 JSONL。"""
+        monkeypatch.setenv("MEMORY_PIPELINE_LOG_ENABLED", "true")
+        monkeypatch.setenv("MEMORY_PIPELINE_TRACE_ENABLED", "false")
+
+        async def _test():
+            from datetime import date
+            from hy_memory.utils.pipeline_observability import (
+                drain_pipeline_log_tasks,
+                install_pipeline_observability,
+            )
+
+            config = MemoryConfig.from_dict({"cache": {"backend": "sqlite"}})
+            cache = SqliteCache(config)
+            await cache.initialize()
+            install_pipeline_observability(cache, log_label="test")
+            await cache.store_pipeline_log(
+                request_id="req-on", user_id="u1", agent_id="a1", step="S1_EMBED",
+            )
+            await drain_pipeline_log_tasks()
+
+            log_file = Path(env_setup) / "default" / f"{date.today().isoformat()}.log"
+            assert json.loads(log_file.read_text(encoding="utf-8").strip())["step"] == "S1_EMBED"
+            await cache.close()
+
+        run(_test())
+
+    def test_log_off_trace_off_is_noop(self, env_setup, monkeypatch):
+        """两者都关：不包装、不写文件、不写库，store_pipeline_log 直接返回 True。"""
+        monkeypatch.setenv("MEMORY_PIPELINE_LOG_ENABLED", "false")
+        monkeypatch.setenv("MEMORY_PIPELINE_TRACE_ENABLED", "false")
+
+        async def _test():
+            from hy_memory.utils.pipeline_observability import (
+                _noop_store_pipeline_log,
+                drain_pipeline_log_tasks,
+                install_pipeline_observability,
+            )
+
+            config = MemoryConfig.from_dict({"cache": {"backend": "sqlite"}})
+            cache = SqliteCache(config)
+            await cache.initialize()
+            original_store = cache.store_pipeline_log
+            install_pipeline_observability(cache, log_label="test")
+            assert cache.store_pipeline_log is _noop_store_pipeline_log
+
+            ok = await cache.store_pipeline_log(
+                request_id="req-off", user_id="ns__u1", agent_id="a1",
+                step="S1_EMBED", prompt="p", response="r",
+            )
+            await drain_pipeline_log_tasks()
+
+            assert ok is True
+            assert not Path(env_setup).exists()
+            # 原始 store 仍能写库，说明表本身可用；空操作确实没写
+            assert await cache.get_pipeline_logs(request_id="req-off") == []
+            await original_store(
+                request_id="req-probe", user_id="u1", agent_id="a1",
+                step="PROBE", prompt="", response="",
+            )
+            assert len(await cache.get_pipeline_logs(request_id="req-probe")) == 1
+            await cache.close()
+
+        run(_test())
+
+    def test_log_off_trace_on_writes_db_only(self, env_setup, monkeypatch):
+        """只关 Log：不写 JSONL，Trace 照常落库。"""
+        monkeypatch.setenv("MEMORY_PIPELINE_LOG_ENABLED", "false")
+        monkeypatch.setenv("MEMORY_PIPELINE_TRACE_ENABLED", "true")
+
+        async def _test():
+            from hy_memory.utils.pipeline_observability import (
+                drain_pipeline_log_tasks,
+                install_pipeline_observability,
+            )
+
+            config = MemoryConfig.from_dict({"cache": {"backend": "sqlite"}})
+            cache = SqliteCache(config)
+            await cache.initialize()
+            install_pipeline_observability(cache, log_label="test")
+            await cache.store_pipeline_log(
+                request_id="req-db", user_id="u1", agent_id="a1", step="S1_EMBED",
+            )
+            await drain_pipeline_log_tasks()
+
+            assert not Path(env_setup).exists()
+            logs = await cache.get_pipeline_logs(request_id="req-db")
+            assert [row["step"] for row in logs] == ["S1_EMBED"]
+            await cache.close()
+
+        run(_test())
+
     def test_store_returns_before_slow_persist(self, env_setup, monkeypatch):
         """慢 DB / 慢磁盘不得堵住 await store_pipeline_log。"""
         monkeypatch.setenv("MEMORY_PIPELINE_TRACE_ENABLED", "true")