fix(memhub): export complete logs from settings

Co-authored-by: Cursor <cursoragent@cursor.com>

CHANGELOG.md

@@ -4,13 +4,14 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
-- MemHub App now exposes the complete memory-mode lifecycle. New installs
-  explicitly choose lite or pro during onboarding; pro completes only after a
-  generation model is active, while ultra remains an advanced Settings choice.
-  Settings shows mode effects, graph backfill, the default 03:00 evolve
-  schedule and manual evolve. The memory page shows original conversations in
-  lite and advanced memories in pro/ultra. Existing installs remain pro unless
-  the user changes them, and embedding maintenance blocks mode/evolve changes.
+## Hy Memory 2.0.14 — 2026-11-04
+
+- `search()` / `async_search()` take an optional `statuses` for explicit
+  L1_RAW search (`layers=["l1_raw"]`, or any search in lite mode), e.g.
+  `statuses=["active", "shadow"]`, so raw turns a pro extraction already
+  shadowed can still be searched. Default stays active-only. Raw hits now
+  report their real `status`. Passing `statuses` to a non-raw search or to
+  `scene="all"` raises `ValueError` instead of being ignored.
 
 ## Hy Memory 2.0.13 — 2026-11-02
 
@@ -81,11 +82,63 @@ Hy Memory and MemHub are versioned and released independently.
   like the other feature switches. The API answer gains `parameters`, and
   `pipeline.features` can now carry the threshold as a number. Refs #241.
 
-## MemHub 1.0.6 — 2026-11-02
+## MemHub 1.0.6 — 2026-11-04
 
-Build 40, on `hy-memory==2.0.13`. Linux x64/arm64 Runtime and macOS arm64
+Build 41, on `hy-memory==2.0.14`. Linux x64/arm64 Runtime and macOS arm64
 App and Runtime.
 
+- MemHub App now exposes the complete memory-mode lifecycle. New installs
+  explicitly choose lite or pro during onboarding; pro completes only after a
+  generation model is active, while ultra remains an advanced Settings choice.
+  Settings shows mode effects, graph backfill, the default 03:00 evolve
+  schedule and manual evolve. The memory page shows original conversations in
+  lite and advanced memories in pro/ultra. Existing installs remain pro unless
+  the user changes them, and embedding maintenance blocks mode/evolve changes.
+- MemHub pins `hy-memory==2.0.14`. Lite search is plain RAG over raw: the
+  App's lite search, CLI context search, `/api/v1/recall` and the Host Bridge
+  (`MemHub_recall`, context search) return every vector hit above the floor,
+  ranked by score, over active and shadow raws, so history pro already
+  processed is still found after switching to lite. The turn view no longer
+  drops a hit whose turns do not literally contain the query; it shows the
+  hit's closest turn instead. Pro and ultra raw search stays active-only.
+  Fixes every raw search failing in packaged builds since #316, which passed
+  `statuses` to an SDK (2.0.13) that did not accept it. The lite raw-turn
+  page also stops re-listing the whole raw layer on each poll when a live
+  session only adds new raws.
+- MemHub Core no longer stalls writes or freezes the App for tens of seconds
+  on a machine with a long live-sync history. Finished live-sync import jobs
+  are pruned to the newest 20 per source (at startup, in bounded batches, and
+  every 10 minutes); history imports and every unfinished job are kept. Job
+  lookups are filtered in SQL with new indexes (schema migration 63) instead
+  of reading the whole table. The durable write queue runs its claims,
+  settles and lease renewals on its own threads, so import work can no longer
+  starve a lease. Import abandon, resume and enqueue, and the file watcher's
+  path attribution, no longer touch the database or the disk on the event
+  loop, and the WAL cadence checkpoint runs on its own thread instead of on
+  the writer's. The write drain now restarts itself after an unexpected exit,
+  reports the failure with its stack, and shows `write_drain` /
+  `write_drain_error` in memory-processing and `memhub status`.
+- Model Center (#293, #295): Settings splits into 界面语言 / 记忆模式 /
+  模型配置 pages, with separate 记忆整理 (generation) and 记忆搜索
+  (embedding) lanes. Changing one lane sends only that lane's patch. An
+  embedding change shows the switch plan, confirmation, progress, cancel and
+  rollback, and the search model only changes after the rebuilt index is
+  verified. Model panels open inside the lane that launched them and close
+  only after Core confirms the profile is active (#310).
+- Lite keeps raw conversations (#297): lite raw writes are no longer stopped
+  by the "pause model sedimentation" switch, the memory page lists every
+  imported raw conversation by role (by turn since #316), and lite adds a
+  manual raw input (source `one-memory-manual`) that never calls a
+  generation model.
+- The home page follows the memory mode (#298): lite counts raw
+  conversations and previews the latest one; pro and ultra read memories.
+  Pro and ultra now require an active generation model, in both the mode
+  preview and the switch.
+- Onboarding (#304) asks for lite or pro and resumes after a restart (schema
+  migration 62); pro finishes only once a generation model is active.
+- Smaller App fixes: compact home status copy (#307), clearer mode-switch
+  effects (#312), Ultra topic nodes are hover-only (#315), and settings
+  controls use the content width (#317).
 - Interfaces in 11 languages: English, Simplified and Traditional Chinese,
   Japanese, Spanish, German, Portuguese, French, Korean, Russian and
   Indonesian. Every interface language, Simplified Chinese included, sets the

README.md

@@ -33,7 +33,7 @@ English | [中文](README.zh.md)
 ## Latest News
 
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 🚀 **Open-sourced** quantized **FP8** (**[Hy-Memory-30B-A3B-FP8](https://huggingface.co/tencent/Hy-Memory-30B-A3B-FP8)**) and **GGUF** (**[Hy-Memory-30B-A3B-GGUF](https://huggingface.co/tencent/Hy-Memory-30B-A3B-GGUF)**) versions of **Hy-Memory-30B-A3B** on Hugging Face!
-* *<mark>[new]</mark>* 🔥🔥🔥 **[2026-11]**: 📦 **Released [MemHub v1.0.6](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.6)** with interfaces in 11 languages, memory modes (lite / pro / ultra with nightly evolve), safe embedding-model switching, multi-user support for the CLI and API, memory writes that retry themselves and pause while the model provider is down, and more reliable upgrades, on **`hy-memory==2.0.13`**. See the release notes for details.
+* *<mark>[new]</mark>* 🔥🔥🔥 **[2026-11]**: 📦 **Released [MemHub v1.0.6](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.6)** with interfaces in 11 languages, memory modes (lite / pro / ultra with nightly evolve), safe embedding-model switching, multi-user support for the CLI and API, memory writes that retry themselves and pause while the model provider is down, and more reliable upgrades, on **`hy-memory==2.0.14`**. See the release notes for details.
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 📦 **Released [MemHub v1.0.5](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.5)** with Chinese, English and Spanish interfaces, one-time pairing codes for more custom Harnesses, and smoother browsing, search, startup and recovery, on **`hy-memory==2.0.8`**. See the release notes for details.
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 📦 **Released [MemHub v1.0.4](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.4)** and **[memhub-cli 1.0.5](https://www.npmjs.com/package/memhub-cli/v/1.0.5)**, with a signed macOS desktop app and verified macOS/Linux Runtimes!
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 🚀 **Released** **Hy Memory 2.0 (`hy-memory==2.0.4`)** on [PyPI](https://pypi.org/project/hy-memory/2.0.4/) and open-sourced the **Hy-Memory** monorepo together with **[MemHub](memhub/README.md)**!

README.zh.md

@@ -33,7 +33,7 @@
 ## 最新动态
 
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 🚀 **模型开源** 在 Hugging Face 发布 **Hy-Memory-30B-A3B** 的 **FP8**（**[Hy-Memory-30B-A3B-FP8](https://huggingface.co/tencent/Hy-Memory-30B-A3B-FP8)**）与 **GGUF**（**[Hy-Memory-30B-A3B-GGUF](https://huggingface.co/tencent/Hy-Memory-30B-A3B-GGUF)**）量化版本模型！
-* *<mark>[new]</mark>* 🔥🔥🔥 **[2026-11]**: 📦 **正式发布 [MemHub v1.0.6](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.6)**：界面支持 11 种语言，新增记忆模式（lite / pro / ultra，ultra 每晚自动整理），可安全切换嵌入模型，CLI 和 API 支持多用户，记忆写入自动重试、模型服务异常时暂停排队，升级更可靠；底层使用 **`hy-memory==2.0.13`**。详见发布说明。
+* *<mark>[new]</mark>* 🔥🔥🔥 **[2026-11]**: 📦 **正式发布 [MemHub v1.0.6](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.6)**：界面支持 11 种语言，新增记忆模式（lite / pro / ultra，ultra 每晚自动整理），可安全切换嵌入模型，CLI 和 API 支持多用户，记忆写入自动重试、模型服务异常时暂停排队，升级更可靠；底层使用 **`hy-memory==2.0.14`**。详见发布说明。
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 📦 **正式发布 [MemHub v1.0.5](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.5)**：支持中文、英文和西班牙语界面，可通过一次性配对码连接更多自定义 Harness，并优化了记忆浏览、搜索、启动和异常恢复；底层使用 **`hy-memory==2.0.8`**。详见发布说明。
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 📦 **正式发布 [MemHub v1.0.4](https://github.com/Tencent-Hunyuan/Hy-Memory/releases/tag/v1.0.4)** 与 **[memhub-cli 1.0.5](https://www.npmjs.com/package/memhub-cli/v/1.0.5)**，提供已签名的 macOS 桌面应用及经过验证的 macOS/Linux Runtime！
 * *<mark>[new]</mark>* 🔥🔥🔥 **[2026-10]**: 🚀 **正式发布** **Hy Memory 2.0 (`hy-memory==2.0.4`)**（已上线 [PyPI](https://pypi.org/project/hy-memory/2.0.4/)），并开源 **Hy-Memory** monorepo 与 **[MemHub](memhub/README.md)**！

hy_memory/client.py

@@ -1864,6 +1864,7 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         request_id: Optional[str] = None,
         layers: Optional[List[Any]] = None,
+        statuses: Optional[List[Any]] = None,
         embed_override: Optional[Dict[str, Any]] = None,
         query_embedding: Optional[List[float]] = None,
         rerank: bool = True,
@@ -1956,6 +1957,8 @@ class HyMemoryClient:
             layers:             可选。pro/ultra 下只召回指定层级（字符串或
                                 MemoryLayer），在召回阶段生效。lite 下忽略该参数并
                                 强制只召回 L1_RAW。
+            statuses:           仅用于 L1_RAW 定向搜索，可显式包含 active / shadow。
+                                非 raw 搜索或 scene="all" 传入时拒绝，避免静默忽略。
             rerank:             默认 True（原行为）。False 时本次调用跳过全部重排阶段：
                                 chat 路的本地 cross-encoder 重排（recall.local_rerank_enabled）
                                 与记忆关系图重排（graph_rerank / episode 扩展），working/
@@ -1999,7 +2002,7 @@ class HyMemoryClient:
             intention_limit=intention_limit, created_after=created_after,
             reader=reader,
             workspace_id=workspace_id, branch=branch,
-            request_id=request_id, layers=layers,
+            request_id=request_id, layers=layers, statuses=statuses,
             embed_override=embed_override,
             query_embedding=query_embedding,
             rerank=rerank,
@@ -3387,12 +3390,14 @@ class HyMemoryClient:
         request_id: str,
         t0: float,
         query_embedding: Optional[List[float]],
+        statuses: Optional[List[Any]],
     ) -> Dict[str, Any]:
         """Implement `search(..., layers=['l1_raw'])` without a second API."""
         from .models.memory import MemoryLayer, MemoryStatus
 
         embedding = query_embedding or await self._embed_service.embed(query)
         fetch = min(max(limit * (10 if session_ids else 2), limit), 10_000)
+        status_filter = self._normalize_statuses(statuses) or [MemoryStatus.ACTIVE]
         hits = await self._vector_store.search(
             query_embedding=embedding,
             user_ids=user_ids,
@@ -3402,7 +3407,7 @@ class HyMemoryClient:
             layers=[MemoryLayer.L1_RAW],
             limit=fetch,
             score_threshold=min_score,
-            status_filter=[MemoryStatus.ACTIVE],
+            status_filter=status_filter,
             only_latest=False,
             created_after=created_after,
         )
@@ -3420,7 +3425,7 @@ class HyMemoryClient:
                     "memory_id": row.pop("raw_id"),
                     **row,
                     "layer": MemoryLayer.L1_RAW.value,
-                    "status": MemoryStatus.ACTIVE.value,
+                    "status": node.status.value,
                     "score": round(float(hit.get("score", 0.0)), 4),
                 }
             )
@@ -3864,6 +3869,7 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         request_id: Optional[str] = None,
         layers: Optional[List[Any]] = None,
+        statuses: Optional[List[Any]] = None,
         embed_override: Optional[Dict[str, Any]] = None,
         query_embedding: Optional[List[float]] = None,
         rerank: bool = True,
@@ -3893,7 +3899,7 @@ class HyMemoryClient:
                     schema_min_score=schema_min_score, schema_limit=schema_limit,
                     intention_limit=intention_limit, created_after=created_after,
                     reader=reader, workspace_id=workspace_id, branch=branch,
-                    request_id=request_id, layers=layers,
+                    request_id=request_id, layers=layers, statuses=statuses,
                     query_embedding=query_embedding,
                     rerank=rerank,
                 )
@@ -3908,7 +3914,7 @@ class HyMemoryClient:
             schema_min_score=schema_min_score, schema_limit=schema_limit,
             intention_limit=intention_limit, created_after=created_after,
             reader=reader, workspace_id=workspace_id, branch=branch,
-            request_id=request_id, layers=layers,
+            request_id=request_id, layers=layers, statuses=statuses,
             query_embedding=query_embedding,
             rerank=rerank,
         )
@@ -3938,6 +3944,7 @@ class HyMemoryClient:
         branch: Optional[str] = None,
         request_id: Optional[str] = None,
         layers: Optional[List[Any]] = None,
+        statuses: Optional[List[Any]] = None,
         query_embedding: Optional[List[float]] = None,
         rerank: bool = True,
     ) -> Dict[str, Any]:
@@ -4023,6 +4030,17 @@ class HyMemoryClient:
                 "Use single agent_id with multiple session_ids, or multiple agent_ids without session_ids."
             )
 
+        if statuses is not None:
+            from .models.memory import MemoryLayer
+
+            requested_layers = self._normalize_layers(layers)
+            if scene_all or (
+                not lite_mode and requested_layers != [MemoryLayer.L1_RAW]
+            ):
+                raise ValueError(
+                    "statuses is supported only for explicit L1_RAW search"
+                )
+
         if scene_all:
             return await self._search_all_scenes(
                 query, queries=queries, search_mode=search_mode,
@@ -4358,6 +4376,7 @@ class HyMemoryClient:
                 request_id=request_id,
                 t0=t0,
                 query_embedding=query_embedding,
+                statuses=statuses,
             )
 
         read_req = ReadRequest(

memhub/README.md

@@ -18,8 +18,8 @@ If you run into a problem, please open an Issue with the version, platform,repro
 
 ## 发行与平台合同
 
-源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.13`。MemHub 源码版本为 `1.0.6`
-（build 40），GitHub 仓库为
+源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.14`。MemHub 源码版本为 `1.0.6`
+（build 41），GitHub 仓库为
 [`Tencent-Hunyuan/Hy-Memory`](https://github.com/Tencent-Hunyuan/Hy-Memory)，npm 包名为
 `memhub-cli`。当前代码架构见
 [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)。
@@ -118,7 +118,7 @@ MemHub 的 SQLite 数据默认位于 `~/.one-memory`，不在 npm 包或 Runtime
 
 ## Hy Memory
 
-SDK 依赖固定为公开 PyPI 正式版 `hy-memory==2.0.13`。SDK 配置、权限边界及 search
+SDK 依赖固定为公开 PyPI 正式版 `hy-memory==2.0.14`。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.13",
+      "versionInfo": "2.0.14",
       "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.13
+## hy-memory 2.0.14
 
 License: Apache-2.0
 

memhub/apps/cli/src/status-report.test.ts

@@ -109,6 +109,39 @@ describe("status on a paused model", () => {
     });
     expect(renderStatusText(report, "MemHub")).not.toContain("模型暂停");
   });
+
+  it("reports a restarting write drain and its exit code", async () => {
+    const report = await compose({
+      ...processing({ llm_backoff_until: null, llm_backoff_error: null }),
+      write_drain: "restarting",
+      write_drain_error: "operational_error",
+    });
+    expect(report.sections.processing).toMatchObject({
+      write_drain: "restarting",
+      write_drain_error: "operational_error",
+    });
+    const text = renderStatusText(report, "MemHub");
+    expect(text).toContain(
+      "写入队列：异常退出（operational_error），正在自动重启",
+    );
+    expect(text).not.toContain("write_drain");
+  });
+
+  it("says nothing about a running drain and nulls what an older Core omits", async () => {
+    const running = await compose({
+      ...processing({ llm_backoff_until: null, llm_backoff_error: null }),
+      write_drain: "running",
+      write_drain_error: null,
+    });
+    expect(renderStatusText(running, "MemHub")).not.toContain("写入队列");
+    const older = await compose(
+      processing({ llm_backoff_until: null, llm_backoff_error: null }),
+    );
+    expect(older.sections.processing).toMatchObject({
+      write_drain: null,
+      write_drain_error: null,
+    });
+  });
 });
 
 describe("status and doctor on an unfinished upgrade migration", () => {

memhub/apps/cli/src/status-report.ts

@@ -323,6 +323,8 @@ export function renderStatusText(
   if (embeddingSwitch !== null) lines.push(`向量模型：${embeddingSwitch}`);
   const backoff = modelBackoffLine(report.sections.processing);
   if (backoff !== null) lines.push(backoff);
+  const drain = writeDrainLine(report.sections.processing);
+  if (drain !== null) lines.push(drain);
   if (report.summary.state === "partial") {
     const notes = [
       report.summary.degraded.length > 0
@@ -353,6 +355,20 @@ function modelBackoffLine(section: Section | undefined): string | null {
   }`;
 }
 
+/** The durable write drain, when it is not simply running: it exited on an
+ * error and is coming back (`restarting`), or it is not running at all. */
+function writeDrainLine(section: Section | undefined): string | null {
+  if (section?.state !== "ok") return null;
+  const state = section.write_drain;
+  if (state !== "restarting" && state !== "stopped") return null;
+  const error = section.write_drain_error;
+  const reason =
+    typeof error === "string" && error.length > 0 ? `（${error}）` : "";
+  return state === "restarting"
+    ? `写入队列：异常退出${reason}，正在自动重启`
+    : `写入队列：已停止${reason}`;
+}
+
 export function pipelineLine(pipeline: PipelineSummary): string {
   const parts: string[] = [pipeline.mode];
   if (pipeline.forced_by_env)
@@ -410,6 +426,8 @@ const SECTION_LINE_OMITTED = new Set([
   "embedding_switch",
   "llm_backoff_until",
   "llm_backoff_error",
+  "write_drain",
+  "write_drain_error",
   "content_plane",
   "visible_count",
   "effective_mode",
@@ -549,6 +567,7 @@ function describeProcessing(
     })),
     suspended: typeof summary.suspended === "number" ? summary.suspended : 0,
     ...llmBackoffFields(summary),
+    ...writeDrainFields(summary),
     ...embeddingSwitchField(summary),
   };
 }
@@ -569,6 +588,24 @@ function llmBackoffFields(summary: MemoryProcessingSummary): {
   };
 }
 
+/** Durable write drain liveness. Null when Core does not report it (older
+ * Core, or recovery has not started the drain yet). */
+function writeDrainFields(summary: MemoryProcessingSummary): {
+  write_drain: "running" | "restarting" | "stopped" | null;
+  write_drain_error: string | null;
+} {
+  const state = summary.write_drain;
+  const error = summary.write_drain_error;
+  return {
+    write_drain:
+      state === "running" || state === "restarting" || state === "stopped"
+        ? state
+        : null,
+    write_drain_error:
+      typeof error === "string" && /^[a-z_]{1,64}$/.test(error) ? error : null,
+  };
+}
+
 /** The running switch or merge, projected; absent when none runs. */
 function embeddingSwitchField(
   summary: MemoryProcessingSummary,

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.13",
+    "hy-memory==2.0.14",
     # 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

@@ -49,6 +49,7 @@ from one_memory.auth import local_api_guard, renderer_origins
 from one_memory.background_work import (
     in_sdk_read,
     in_the_background,
+    in_the_write_queue,
     shutdown_bulk_pool,
     warm_bulk_pool,
 )
@@ -164,6 +165,7 @@ from one_memory.imports.grants import (
 )
 from one_memory.imports.host_packages import CapabilityState, HostPackageCapabilities
 from one_memory.imports.jobs import (
+    PRUNE_BATCH_ROWS,
     ImportIssueRecord,
     ImportJob,
     ImportJobStore,
@@ -270,6 +272,7 @@ from one_memory.onboarding import (
     OnboardingState,
     OnboardingStore,
 )
+from one_memory.raw_turns import RawDisplayTurn, raw_display_turns, raw_prefix_added
 from one_memory.recall import (
     GovernedRecallTarget,
     RecallReceipt,
@@ -604,24 +607,30 @@ class SdkRawItem(HySdkRawRecord):
     role: str | None = None
     content_hash: str
     created: datetime | None = None
+    turn_id: str | None = Field(default=None, min_length=1, max_length=96)
+    turn_index: int | None = Field(default=None, ge=0)
+    turn_count: int | None = Field(default=None, ge=1)
 
     @classmethod
-    def of(cls, record: HySdkRawRecord) -> "SdkRawItem":
-        messages = tuple(
-            SdkRawMessage(role=role, content=text)
-            for role, text in raw_turn_messages(record.content)
-        )
+    def of(cls, record: HySdkRawRecord, turn: RawDisplayTurn | None = None) -> "SdkRawItem":
+        content = turn.content if turn is not None else record.content
+        raw_messages = turn.messages if turn is not None else raw_turn_messages(record.content)
+        messages = tuple(SdkRawMessage(role=role, content=text) for role, text in raw_messages)
         roles = {message.role for message in messages}
         return cls(
-            **record.model_dump(),
+            **record.model_dump(exclude={"content"}),
+            content=content,
             messages=messages,
             role=next(iter(roles)) if len(roles) == 1 and "" not in roles else None,
-            content_hash="sha256:" + hashlib.sha256(record.content.encode("utf-8")).hexdigest(),
+            content_hash="sha256:" + hashlib.sha256(content.encode("utf-8")).hexdigest(),
             created=(
                 datetime.fromtimestamp(record.gmt_created, tz=UTC)
                 if record.gmt_created is not None
                 else None
             ),
+            turn_id=turn.turn_id if turn is not None else None,
+            turn_index=turn.turn_index if turn is not None else None,
+            turn_count=turn.turn_count if turn is not None else None,
         )
 
 
@@ -634,9 +643,9 @@ class SdkRawPage(BaseModel):
     total: int = Field(ge=0)
     offset: int = Field(ge=0)
     limit: int = Field(ge=1, le=500)
-    #: Set when a time filter was applied to this page in Core (the SDK list
-    #: has none): `items` is the page's rows inside the range, `total` stays
-    #: the SDK's count before that filter.
+    unit: Literal["raw", "turn"] = "raw"
+    #: Set when a time filter was applied in Core. For storage `raw`, `total`
+    #: remains the SDK count; for product `turn`, it is the filtered turn count.
     time_filtered: bool = False
 
 
@@ -736,6 +745,12 @@ class MemoryProcessingSummary(BaseModel):
     #: The last provider failure behind the pause, as a stable code
     #: (`model_auth_failed`, `model_not_found`, `model_unreachable`, ...).
     llm_backoff_error: str | None = None
+    #: The durable write drain, the queue's only consumer: `running`,
+    #: `restarting` (it exited on an error and comes back after a pause) or
+    #: `stopped`; None before startup recovery has started it.
+    write_drain: Literal["running", "restarting", "stopped"] | None = None
+    #: Why the drain last exited unexpectedly, as a stable code; None if never.
+    write_drain_error: str | None = None
 
 
 class MemoryModeState(BaseModel):
@@ -1495,7 +1510,7 @@ def create_app(
                 # under a new SDK request id; the re-run first asks the SDK
                 # whether the old one landed (`execute_job`). Nothing stays
                 # `unknown` waiting for a person.
-                await in_the_background(
+                await in_the_write_queue(
                     memory_write_queue.recover_unknown,
                     in_flight=sdk_write_in_flight,
                 )
@@ -1624,6 +1639,7 @@ def create_app(
             # Started by `recover_after_ready`: a running drain claims jobs and
             # holds the single writer, which the rest of startup still needs.
             app.state.memory_write_drain_task = None
+            app.state.import_job_pruning = None
             # D52: every entry that already holds committed Evidence goes
             # through the one durable queue. The intake turns one record into
             # one complete-turn request; the drain stays the only consumer, so
@@ -1866,12 +1882,21 @@ def create_app(
                         app.state.embedding_switch.start_retention()
 
                 async def start_write_drain() -> None:
+                    # Supervised: an unexpected exit is reported and the drain
+                    # restarts, instead of stranding every queued write.
                     app.state.memory_write_drain_task = watched_task(
                         asyncio.get_running_loop(),
-                        lambda: memory_write_drain.run_forever(),
+                        lambda: memory_write_drain.run_supervised(),
                         name="one-memory-durable-write-drain",
                     )
 
+                async def start_import_job_pruning() -> None:
+                    app.state.import_job_pruning = watched_task(
+                        asyncio.get_running_loop(),
+                        lambda: _prune_finished_import_jobs(app.state.import_job_store),
+                        name="one-memory-import-job-pruning",
+                    )
+
                 async def resume_graph_backfill() -> None:
                     saved = app.state.evolve_state_store.graph_build()
                     if saved.state != "running":
@@ -1942,7 +1967,7 @@ def create_app(
                             )
 
                 async def resume_imports() -> None:
-                    app.state.import_supervisor.resume_interrupted()
+                    await app.state.import_supervisor.resume_interrupted_off_loop()
 
                 async def start_live_sync() -> None:
                     if runtime_settings.live_sync_enabled:
@@ -1982,6 +2007,7 @@ def create_app(
                         await step("evolve_schedule", start_evolve_schedule)
                     else:
                         await step("pending_dispatch_replay", replay_pending_dispatch)
+                    await step("import_job_pruning", start_import_job_pruning)
                     await step("import_resume", resume_imports)
                     await step("live_sync", start_live_sync)
                     await step("source_reconcile", _reconcile_source_sync_from_config)
@@ -2071,6 +2097,10 @@ def create_app(
                     # Recovery never reached the metadata repair step.
                     release_stores(runtime_settings.data_dir)
                     app.state.sdk_stores_held = False
+                if app.state.import_job_pruning is not None:
+                    app.state.import_job_pruning.cancel()
+                    with suppress(asyncio.CancelledError):
+                        await app.state.import_job_pruning
                 await app.state.import_supervisor.shutdown()
                 app.state.memory_write_drain.stop()
                 if app.state.memory_write_drain_task is not None:
@@ -2809,17 +2839,16 @@ def create_app(
         await _refuse_overlapping_root(source_id, registration.candidate.path, DEFAULT_USER_ID)
         root_changed = prior is not None and prior.root_id != request.expected_root_id
         if root_changed:
-            source_jobs = [job for job in _job_store(app).list_jobs() if job.source_id == source_id]
+            source_jobs = await asyncio.to_thread(
+                partial(
+                    _job_store(app).find_jobs,
+                    source_id=source_id,
+                    statuses=_UNFINISHED_IMPORT_STATUSES,
+                )
+            )
             if any(job.status is ImportStatus.RUNNING for job in source_jobs):
                 raise HTTPException(status_code=409, detail="source_import_busy")
             for job in source_jobs:
-                if job.status not in {
-                    ImportStatus.PENDING,
-                    ImportStatus.PAUSED,
-                    ImportStatus.INTERRUPTED,
-                    ImportStatus.FAILED,
-                }:
-                    continue
                 await _supervisor(app).cancel(job.job_id)
         try:
             receipt = await asyncio.to_thread(
@@ -2998,21 +3027,18 @@ def create_app(
         user's own cursors and history job.
         """
         await _live_sync(app).initialize_source(source_id, baseline_missing_files=True)
-        active = {
-            ImportStatus.PENDING,
-            ImportStatus.PAUSED,
-            ImportStatus.INTERRUPTED,
-            ImportStatus.RUNNING,
-            ImportStatus.FAILED,
-        }
-        if any(
-            lane_key(job.user_id, job.source_id) == source_id
-            and job.operation == "history"
-            and job.status in active
-            for job in _job_store(app).list_jobs()
-        ):
-            return
         user_id, bare_source_id = split_lane(source_id)
+        active_history = await asyncio.to_thread(
+            partial(
+                _job_store(app).latest_job,
+                user_id=user_id,
+                source_id=bare_source_id,
+                operation="history",
+                statuses=_UNFINISHED_IMPORT_STATUSES,
+            )
+        )
+        if active_history is not None:
+            return
         await asyncio.to_thread(
             _job_store(app).create,
             user_id=user_id,
@@ -3022,18 +3048,25 @@ def create_app(
 
     async def _activate_source_resume(source_id: str) -> None:
         """Run work prepared for a source after config.json has committed."""
+        user_id, bare_source_id = split_lane(source_id)
         try:
-            for job in _job_store(app).list_jobs():
-                if lane_key(job.user_id, job.source_id) != source_id:
-                    continue
+            for job in await asyncio.to_thread(
+                partial(
+                    _job_store(app).find_jobs,
+                    user_id=user_id,
+                    source_id=bare_source_id,
+                    statuses=(
+                        ImportStatus.PENDING,
+                        ImportStatus.PAUSED,
+                        ImportStatus.INTERRUPTED,
+                        ImportStatus.FAILED,
+                    ),
+                )
+            ):
                 if job.status is ImportStatus.PENDING:
-                    _supervisor(app).enqueue(job.job_id)
-                elif job.status in {
-                    ImportStatus.PAUSED,
-                    ImportStatus.INTERRUPTED,
-                    ImportStatus.FAILED,
-                }:
-                    _supervisor(app).resume(job.job_id)
+                    await _supervisor(app).enqueue_off_loop(job.job_id, job=job)
+                else:
+                    await _supervisor(app).resume_off_loop(job.job_id)
         except SourceAccessError as error:
             raise HTTPException(status_code=409, detail=error.code) from None
         if app.state.settings.automatic_memory_processing:
@@ -3060,22 +3093,19 @@ def create_app(
         _require_registration(app, source_id)
         if user_id != DEFAULT_USER_ID:
             return await _revoke_other_user_grant(source_id, user_id)
-        receipt = _grant_store(app).receipt(
-            user_id="local",
-            source_id=source_id,
+        receipt = await asyncio.to_thread(
+            partial(_grant_store(app).receipt, user_id="local", source_id=source_id)
         )
         if receipt is None:
             raise HTTPException(status_code=404, detail="grant_not_found")
-        active_statuses = {
-            ImportStatus.PENDING,
-            ImportStatus.RUNNING,
-            ImportStatus.PAUSED,
-            ImportStatus.INTERRUPTED,
-            ImportStatus.FAILED,
-        }
-        for job in _job_store(app).list_jobs():
-            if job.source_id == source_id and job.status in active_statuses:
-                await _supervisor(app).cancel(job.job_id)
+        for job in await asyncio.to_thread(
+            partial(
+                _job_store(app).find_jobs,
+                source_id=source_id,
+                statuses=_UNFINISHED_IMPORT_STATUSES,
+            )
+        ):
+            await _supervisor(app).cancel(job.job_id)
         # Revocation is the user's node for "stop using this source" (D73):
         # park unfinished extraction instead of leaving it to fail later, so
         # re-authorizing resumes the same work.
@@ -3087,9 +3117,8 @@ def create_app(
         await _live_sync(app).stop_source(source_id)
         _supervisor(app).unregister_adapter(source_id)
         _detection(app).pop(source_id, None)
-        receipt = _grant_store(app).revoke(
-            user_id="local",
-            source_id=source_id,
+        receipt = await asyncio.to_thread(
+            partial(_grant_store(app).revoke, user_id="local", source_id=source_id)
         )
         await asyncio.to_thread(_config_store(app).set_sync, source_id, False)
         await _refresh_live_watch(app)
@@ -3105,16 +3134,15 @@ def create_app(
         )
         if receipt is None:
             raise HTTPException(status_code=404, detail="grant_not_found")
-        active_statuses = {
-            ImportStatus.PENDING,
-            ImportStatus.RUNNING,
-            ImportStatus.PAUSED,
-            ImportStatus.INTERRUPTED,
-            ImportStatus.FAILED,
-        }
-        for job in _job_store(app).list_jobs():
-            if lane_key(job.user_id, job.source_id) == lane and job.status in active_statuses:
-                await _supervisor(app).cancel(job.job_id)
+        for job in await asyncio.to_thread(
+            partial(
+                _job_store(app).find_jobs,
+                user_id=user_id,
+                source_id=source_id,
+                statuses=_UNFINISHED_IMPORT_STATUSES,
+            )
+        ):
+            await _supervisor(app).cancel(job.job_id)
         await _live_sync(app).stop_source(lane)
         _supervisor(app).unregister_adapter(lane)
         receipt = await asyncio.to_thread(
@@ -3179,7 +3207,7 @@ def create_app(
             )
         except SourceAccessError as error:
             raise HTTPException(status_code=409, detail=error.code) from None
-        _supervisor(app).enqueue(job.job_id)
+        await _supervisor(app).enqueue_off_loop(job.job_id, job=job)
         return job
 
     @app.post(
@@ -3474,8 +3502,10 @@ def create_app(
         user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN),
     ) -> list[ImportJob]:
         # One user's jobs: the App reads `local`'s and must not show another
-        # user's Hermes import as its own.
-        return [job for job in _job_store(app).list_jobs() if job.user_id == user_id]
+        # user's Hermes import as its own. Every one of them, oldest first, as
+        # before; finished live syncs past the newest twenty per source are
+        # pruned (`prune_finished_sync_jobs`), so this list stays short.
+        return _job_store(app).find_jobs(user_id=user_id)
 
     @app.get(
         "/api/v1/live-sync",
@@ -3572,7 +3602,7 @@ def create_app(
         dependencies=protected,
     )
     async def pause_import_job(job_id: str) -> ImportJob:
-        _require_job(app, job_id)
+        await asyncio.to_thread(_require_job, app, job_id)
         return await _translate_transition(_supervisor(app).pause(job_id))
 
     @app.post(
@@ -3581,11 +3611,11 @@ def create_app(
         dependencies=protected,
     )
     async def resume_import_job(job_id: str) -> ImportJob:
-        job = _require_job(app, job_id)
+        job = await asyncio.to_thread(_require_job, app, job_id)
         if job.status in {ImportStatus.COMPLETED, ImportStatus.CANCELLED}:
             raise HTTPException(status_code=409, detail="job_not_resumable")
         try:
-            return _supervisor(app).resume(job_id)
+            return await _supervisor(app).resume_off_loop(job_id)
         except InvalidJobTransitionError:
             raise HTTPException(409, "invalid_job_transition") from None
 
@@ -3595,7 +3625,7 @@ def create_app(
         dependencies=protected,
     )
     async def cancel_import_job(job_id: str) -> ImportJob:
-        _require_job(app, job_id)
+        await asyncio.to_thread(_require_job, app, job_id)
         return await _translate_transition(_supervisor(app).cancel(job_id))
 
     @app.get(
@@ -4441,6 +4471,16 @@ def create_app(
             raise _sdk_http_error(error, "hy_memory_delete_failed") from None
         return Response(status_code=204)
 
+    raw_turn_cache: dict[
+        tuple[object, ...],
+        tuple[
+            int,
+            tuple[tuple[str, str | None], ...],
+            tuple[tuple[HySdkRawRecord, RawDisplayTurn], ...],
+        ],
+    ] = {}
+    raw_turn_cache_locks: dict[tuple[object, ...], asyncio.Lock] = {}
+
     @app.get(
         "/api/v2/raw",
         response_model=SdkRawPage,
@@ -4450,6 +4490,7 @@ def create_app(
         source: list[str] | None = Query(default=None),  # noqa: B008
         offset: int = Query(default=0, ge=0),
         limit: int = Query(default=100, ge=1, le=500),
+        unit: Literal["raw", "turn"] = Query(default="raw"),
         user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN),
         exclude_sources: list[str] | None = Query(default=None),  # noqa: B008
         session_id: list[str] | None = Query(default=None),  # noqa: B008
@@ -4463,7 +4504,10 @@ def create_app(
         The SDK lists raw only when asked for `layers=["l1_raw"]`, and only
         active rows unless told otherwise. Pro shadows a raw once extraction
         succeeds, so a raw page reads active and shadow (`statuses` narrows
-        it). `source` and `session_id` (a conversation) may repeat or be comma
+        it). `unit=turn` projects storage batches into stable product turns;
+        the first read builds an append-only in-memory index and later polls
+        only probe the latest SDK page unless it changed. `source` and
+        `session_id` (a conversation) may repeat or be comma
         separated. `after`/`before` (unix seconds, against the conversation
         time, else the stored time) have no SDK list filter: they cut this
         page in Core and mark it `time_filtered`.
@@ -4477,24 +4521,158 @@ def create_app(
         sessions = split_csv(session_id)
         try:
             client = await _sdk_client_handle(app).client()
-            value = await asyncio.to_thread(
-                client.list_memories,
-                user_id=user_id,
+            query = {
+                "user_id": user_id,
                 **raw_list_filters(chosen_statuses),
-                include_sources=sources or None,
-                exclude_sources=excluded or None,
-                session_ids=sessions or None,
-                offset=offset,
-                limit=limit,
-                order=order,
-            )
-            page = HySdkRawPage.from_memory_list(value, limit=limit, offset=offset)
+                "include_sources": sources or None,
+                "exclude_sources": excluded or None,
+                "session_ids": sessions or None,
+                "order": order if unit == "raw" else "desc",
+            }
+            if unit == "raw":
+                value = await asyncio.to_thread(
+                    client.list_memories,
+                    **query,
+                    offset=offset,
+                    limit=limit,
+                )
+                page = HySdkRawPage.from_memory_list(value, limit=limit, offset=offset)
+                records = page.raw
+            else:
+                cache_key = (
+                    user_id,
+                    tuple(sorted(chosen_statuses)),
+                    tuple(sorted(sources)),
+                    tuple(sorted(excluded)),
+                    tuple(sorted(sessions)),
+                )
+                if len(raw_turn_cache_locks) > 64:
+                    for stale_key, stale_lock in tuple(raw_turn_cache_locks.items()):
+                        if stale_key not in raw_turn_cache and not stale_lock.locked():
+                            raw_turn_cache_locks.pop(stale_key, None)
+                lock = raw_turn_cache_locks.setdefault(cache_key, asyncio.Lock())
+                async with lock:
+                    for _attempt in range(3):
+                        value = await asyncio.to_thread(
+                            client.list_memories,
+                            **query,
+                            offset=0,
+                            limit=500,
+                        )
+                        first_page = HySdkRawPage.from_memory_list(value, limit=500, offset=0)
+                        signature = tuple(
+                            (record.raw_id, record.status) for record in first_page.raw
+                        )
+                        cached = raw_turn_cache.get(cache_key)
+                        if (
+                            cached is not None
+                            and cached[0] == first_page.total
+                            and cached[1] == signature
+                        ):
+                            projected_turns = cached[2]
+                            break
+                        added = (
+                            None
+                            if cached is None
+                            else raw_prefix_added(cached[1], signature, cached[0], first_page.total)
+                        )
+                        if cached is not None and added is not None:
+                            # A live session only puts new raws on top: project
+                            # those and keep the rest, instead of re-listing the
+                            # whole raw layer on every poll.
+                            projected_turns = (
+                                tuple(
+                                    (record, turn)
+                                    for record in first_page.raw[:added]
+                                    for turn in reversed(raw_display_turns(record))
+                                )
+                                + cached[2]
+                            )
+                            raw_turn_cache[cache_key] = (
+                                first_page.total,
+                                signature,
+                                projected_turns,
+                            )
+                            break
+
+                        records_list = list(first_page.raw)
+                        raw_offset = len(first_page.raw)
+                        raw_total = first_page.total
+                        while first_page.raw and raw_offset < raw_total:
+                            value = await asyncio.to_thread(
+                                client.list_memories,
+                                **query,
+                                offset=raw_offset,
+                                limit=500,
+                            )
+                            raw_page = HySdkRawPage.from_memory_list(
+                                value, limit=500, offset=raw_offset
+                            )
+                            records_list.extend(raw_page.raw)
+                            raw_offset += len(raw_page.raw)
+                            if not raw_page.raw:
+                                break
+
+                        verify_value = await asyncio.to_thread(
+                            client.list_memories,
+                            **query,
+                            offset=0,
+                            limit=500,
+                        )
+                        verify = HySdkRawPage.from_memory_list(verify_value, limit=500, offset=0)
+                        verify_signature = tuple(
+                            (record.raw_id, record.status) for record in verify.raw
+                        )
+                        if verify.total != first_page.total or verify_signature != signature:
+                            continue
+                        projected_turns = tuple(
+                            (record, turn)
+                            for record in records_list
+                            for turn in reversed(raw_display_turns(record))
+                        )
+                        raw_turn_cache[cache_key] = (
+                            first_page.total,
+                            signature,
+                            projected_turns,
+                        )
+                        while len(raw_turn_cache) > 32:
+                            oldest = next(iter(raw_turn_cache))
+                            raw_turn_cache.pop(oldest)
+                            old_lock = raw_turn_cache_locks.get(oldest)
+                            if old_lock is not None and not old_lock.locked():
+                                raw_turn_cache_locks.pop(oldest, None)
+                        break
+                    else:
+                        raise RuntimeError("hy_memory_raw_snapshot_busy")
+                if cache_key not in raw_turn_cache and not lock.locked():
+                    raw_turn_cache_locks.pop(cache_key, None)
         except HTTPException:
             raise
         except Exception as error:
             raise _sdk_http_error(error, "hy_memory_raw_list_failed") from None
-        records = page.raw
         time_filtered = after is not None or before is not None
+        if unit == "turn":
+            turns = projected_turns
+            if time_filtered:
+                turns = tuple(
+                    (record, turn)
+                    for record, turn in turns
+                    if (at := record.memory_at or record.gmt_created) is not None
+                    and (after is None or at >= after)
+                    and (before is None or at < before)
+                )
+            if order == "asc":
+                turns = tuple(reversed(turns))
+            selected = turns[offset : offset + limit]
+            return SdkRawPage(
+                user_id=user_id,
+                items=tuple(SdkRawItem.of(record, turn) for record, turn in selected),
+                total=len(turns),
+                offset=offset,
+                limit=limit,
+                unit="turn",
+                time_filtered=time_filtered,
+            )
         if time_filtered:
             records = tuple(
                 record
@@ -4509,6 +4687,7 @@ def create_app(
             total=page.total,
             offset=offset,
             limit=limit,
+            unit="raw",
             time_filtered=time_filtered,
         )
 
@@ -4705,6 +4884,7 @@ def create_app(
         running_rows = counts.get(MemoryTaskStatus.RUNNING, 0)
         backoff = cast(ModelBackoff | None, getattr(app.state, "model_backoff", None))
         backoff_state = backoff.state() if backoff is not None else None
+        drain = getattr(app.state, "memory_write_drain", None)
         return MemoryProcessingSummary(
             pending=(
                 sdk_counts.get("pending", 0)
@@ -4752,6 +4932,8 @@ def create_app(
                 if backoff_state is not None and backoff_state.until is not None
                 else None
             ),
+            write_drain=getattr(drain, "liveness", None),
+            write_drain_error=getattr(drain, "last_error_code", None),
         )
 
     @app.get(
@@ -5549,6 +5731,16 @@ def _diagnostic_request_details(request: Request, event: str | None) -> dict[str
     return details
 
 
+#: Every state a job can still leave: the ones a revoke or re-root cancels.
+_UNFINISHED_IMPORT_STATUSES = (
+    ImportStatus.PENDING,
+    ImportStatus.RUNNING,
+    ImportStatus.PAUSED,
+    ImportStatus.INTERRUPTED,
+    ImportStatus.FAILED,
+)
+
+
 def _require_job(app: FastAPI, job_id: str) -> ImportJob:
     try:
         return _job_store(app).require(job_id)
@@ -5823,6 +6015,54 @@ def _backfill_state(
     return "running"
 
 
+#: How often finished live syncs are pruned while the Core runs, and how many
+#: bounded batches one such pass may take (`prune_finished_sync_jobs`).
+IMPORT_JOB_PRUNE_SECONDS = 600.0
+IMPORT_JOB_PRUNE_PASS_BATCHES = 10
+
+
+async def _prune_finished_import_jobs(
+    store: ImportJobStore,
+    *,
+    interval_seconds: float = IMPORT_JOB_PRUNE_SECONDS,
+    pass_batches: int = IMPORT_JOB_PRUNE_PASS_BATCHES,
+) -> None:
+    """Keep `import_jobs` bounded: at startup down to the keep count, then on a timer.
+
+    Every batch is its own short write transaction, made off the loop, with a
+    pause between batches so the writer's line keeps moving. The first pass
+    runs until nothing is left, which on a home that grew for weeks is the
+    one long pass; after it each pass is at most `pass_batches` batches.
+    A busy database skips the rest of a pass rather than failing it.
+    """
+    first = True
+    while True:
+        removed_total = 0
+        batches = 0
+        while first or batches < pass_batches:
+            try:
+                removed = await in_the_background(store.prune_finished_sync_jobs)
+            except sqlite3.OperationalError as error:
+                if "locked" not in str(error).lower() and "busy" not in str(error).lower():
+                    raise
+                logger.info("import_jobs_prune_deferred error_type=%s", type(error).__name__)
+                break
+            batches += 1
+            removed_total += removed
+            if removed < PRUNE_BATCH_ROWS:
+                break
+            await asyncio.sleep(0.05)
+        if removed_total:
+            logger.info(
+                "import_jobs_pruned removed=%d batches=%d startup=%s",
+                removed_total,
+                batches,
+                first,
+            )
+        first = False
+        await asyncio.sleep(interval_seconds)
+
+
 async def _keep_loop_heartbeat(watchdog: LoopWatchdog) -> None:
     """The loop saying it is still able to run something."""
     # The watchdog already logs each stall with a stack, but a stack is prose:
@@ -6222,7 +6462,7 @@ async def _search_sdk_raw(
     `filters` are a caller's pass-through options: its floor and row filters
     apply, and its `include_sources` only narrow `sources`, never widen them.
     """
-    from one_memory.context_routes import raw_lane_overrides
+    from one_memory.context_routes import raw_lane_overrides, raw_lane_statuses
 
     include = sorted(sources) if sources else None
     if filters is not None and filters.include_sources is not None:
@@ -6249,6 +6489,8 @@ async def _search_sdk_raw(
             "scene": "chat",
             "reader": HY_MEMORY_READER,
             "layers": ["l1_raw"],
+            # Lite recall is plain RAG over every raw, shadow included.
+            **raw_lane_statuses(lite=app.state.memory_mode() == "lite"),
             **(raw_lane_overrides(filters) if filters is not None else {}),
         },
     )
@@ -6863,12 +7105,12 @@ def _history_import_gate_released(app: FastAPI) -> bool:
         app.state.history_import_gate_released = True
         return True
     selected = set(state.selected_source_ids)
+    # Asked inside the drain's claim, per candidate: history jobs of `local`
+    # only, in SQL, never the whole table.
     matching = [
         job
-        for job in jobs.list_jobs()
-        if job.operation == "history"
-        and job.user_id == DEFAULT_USER_ID
-        and job.source_id in selected
+        for job in jobs.find_jobs(user_id=DEFAULT_USER_ID, operation="history")
+        if job.source_id in selected
         and (state.started_at is None or job.created_at >= state.started_at)
     ]
     if {job.source_id for job in matching} != selected:

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

@@ -20,6 +20,15 @@ SOURCE_READER_WORKERS = 4
 SDK_WRITE_WORKERS = 12
 # Recall's SDK reads: one search and the `get`s behind it (`_SDK_GET_CONCURRENCY`).
 SDK_READ_WORKERS = 8
+# The durable write queue's own bookkeeping: claim, settle, the per-job reads
+# around an SDK write, and the drain's maintenance passes. Each is one short
+# statement on the single writer or a reader; the drain runs at most
+# `max_concurrent_sources` jobs, and they wait for the writer, not for CPU.
+DURABLE_QUEUE_WORKERS = 4
+# Lease renewal alone. Renewal is the drain's ownership fence: a lease that
+# lapses while its SDK write is still running hands the job to recovery, so it
+# must never wait for a thread behind any other work, its own queue's included.
+LEASE_RENEWAL_WORKERS = 2
 
 _pool: ThreadPoolExecutor | None = None
 _snapshot_pool: ThreadPoolExecutor | None = None
@@ -28,6 +37,8 @@ _raw_read_pool: ThreadPoolExecutor | None = None
 _memory_read_pool: ThreadPoolExecutor | None = None
 _source_reader_pool: ThreadPoolExecutor | None = None
 _sdk_write_pool: ThreadPoolExecutor | None = None
+_durable_queue_pool: ThreadPoolExecutor | None = None
+_lease_renewal_pool: ThreadPoolExecutor | None = None
 _warmed_readers: weakref.WeakSet[ThreadPoolExecutor] = weakref.WeakSet()
 _reader_warm_lock = threading.Lock()
 _interactive_reads = 0
@@ -109,6 +120,35 @@ def sdk_write_pool() -> ThreadPoolExecutor:
     return _sdk_write_pool
 
 
+def durable_queue_pool() -> ThreadPoolExecutor:
+    """The durable write queue's database work, apart from bulk work.
+
+    The queue used to hand its claims, settles and maintenance to the bulk
+    pool, whose three threads also carry import heartbeats and job listings.
+    Under an import peak all three sat in those for tens of seconds, and the
+    queue's next statement waited for a thread before it could even join the
+    writer's line, and leases lapsed while their SDK writes were still running.
+    """
+    global _durable_queue_pool
+    if _durable_queue_pool is None:
+        _durable_queue_pool = ThreadPoolExecutor(
+            max_workers=DURABLE_QUEUE_WORKERS,
+            thread_name_prefix="one-memory-write-queue",
+        )
+    return _durable_queue_pool
+
+
+def lease_renewal_pool() -> ThreadPoolExecutor:
+    """Threads that only renew durable write leases (`DurableWriteDrain.renew`)."""
+    global _lease_renewal_pool
+    if _lease_renewal_pool is None:
+        _lease_renewal_pool = ThreadPoolExecutor(
+            max_workers=LEASE_RENEWAL_WORKERS,
+            thread_name_prefix="one-memory-lease-renewal",
+        )
+    return _lease_renewal_pool
+
+
 def source_reader_pool() -> ThreadPoolExecutor:
     """Bounded process-wide pool for source iteration and parsing."""
     global _source_reader_pool
@@ -128,9 +168,21 @@ def warm_bulk_pool() -> None:
     Starting a thread is nothing on an idle machine and 2.5 seconds on a busy
     one, which is precisely when bulk work is being handed over. So it is
     done here instead, at startup, before there is anything to be late for.
+
+    The durable write queue's pools are warmed here too, for the same reason:
+    their first renewal must not pay for starting a thread.
     """
-    pool = bulk_pool()
-    started = tuple(threading.Event() for _ in range(BULK_WORKERS))
+    _start_every_worker(bulk_pool(), BULK_WORKERS, "bulk_pool_warmup_timeout")
+    _start_every_worker(
+        durable_queue_pool(), DURABLE_QUEUE_WORKERS, "durable_queue_pool_warmup_timeout"
+    )
+    _start_every_worker(
+        lease_renewal_pool(), LEASE_RENEWAL_WORKERS, "lease_renewal_pool_warmup_timeout"
+    )
+
+
+def _start_every_worker(pool: ThreadPoolExecutor, workers: int, timeout_code: str) -> None:
+    started = tuple(threading.Event() for _ in range(workers))
     release = threading.Event()
 
     def occupy(marker: threading.Event) -> None:
@@ -146,7 +198,7 @@ def warm_bulk_pool() -> None:
     try:
         for marker in started:
             if not marker.wait(10):
-                raise RuntimeError("bulk_pool_warmup_timeout")
+                raise RuntimeError(timeout_code)
     finally:
         release.set()
     for future in futures:
@@ -228,6 +280,38 @@ async def in_sdk_write[T](
     )
 
 
+async def in_the_write_queue[T](
+    work: Callable[..., T],
+    /,
+    *args: object,
+    **kwargs: object,
+) -> T:
+    """One durable write queue statement, in the queue's pool (`durable_queue_pool`).
+
+    It still goes through the one writer and its line, like every write; what
+    it no longer does is wait for a bulk thread first.
+    """
+    return await asyncio.get_running_loop().run_in_executor(
+        durable_queue_pool(),
+        copy_context().run,
+        partial(work, *args, **kwargs),
+    )
+
+
+async def in_lease_renewal[T](
+    work: Callable[..., T],
+    /,
+    *args: object,
+    **kwargs: object,
+) -> T:
+    """One lease renewal, in the threads kept for nothing else (`lease_renewal_pool`)."""
+    return await asyncio.get_running_loop().run_in_executor(
+        lease_renewal_pool(),
+        copy_context().run,
+        partial(work, *args, **kwargs),
+    )
+
+
 def _retrieve[T](future: "asyncio.Future[T]") -> None:
     """Mark a shielded read's outcome as seen.
 
@@ -314,7 +398,13 @@ def _warm_reader_pool(reader: ThreadPoolExecutor) -> None:
 
 def shutdown_bulk_pool() -> None:
     global _pool, _snapshot_pool, _interactive_pool, _raw_read_pool, _memory_read_pool
-    global _source_reader_pool, _sdk_write_pool
+    global _source_reader_pool, _sdk_write_pool, _durable_queue_pool, _lease_renewal_pool
+    if _durable_queue_pool is not None:
+        _durable_queue_pool.shutdown(wait=False, cancel_futures=True)
+        _durable_queue_pool = None
+    if _lease_renewal_pool is not None:
+        _lease_renewal_pool.shutdown(wait=False, cancel_futures=True)
+        _lease_renewal_pool = None
     if _sdk_write_pool is not None:
         _sdk_write_pool.shutdown(wait=False, cancel_futures=True)
         _sdk_write_pool = None

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

@@ -66,6 +66,8 @@ class SdkContextSearchRequest(ContextSearchRequest, SdkSearchOptions):
     limit: int = Field(default=10, ge=1, le=20, strict=True)
     schema_limit: int = Field(default=3, ge=0, le=20, strict=True)
     intention_limit: int = Field(default=1, ge=0, le=20, strict=True)
+    #: Product callers can request one item per turn instead of SDK raw batches.
+    raw_unit: Literal["raw", "turn"] = "raw"
     #: Whose memories to search. The App never sends it and reads `local`.
     user_id: UserId = DEFAULT_USER_ID
 
@@ -123,6 +125,9 @@ class SdkRawContextItem(BaseModel):
     memory_at: datetime | None = None
     score: float = Field(ge=0, le=1)
     truncated: bool = False
+    turn_id: str | None = Field(default=None, min_length=1, max_length=96)
+    turn_index: int | None = Field(default=None, ge=0)
+    turn_count: int | None = Field(default=None, ge=1)
 
 
 class SdkMemoryContextItem(BaseModel):

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

@@ -30,6 +30,7 @@ from one_memory.hy_sdk_types import (
     parse_sdk_search,
 )
 from one_memory.imports.storage import EvidenceStore
+from one_memory.raw_turns import raw_display_turns, turns_for_hit
 from one_memory.sdk_search_options import SdkSearchFilters, SearchLane
 from one_memory.semantic_deadline import SemanticMemoryHit
 from one_memory.user_scope import owns
@@ -70,6 +71,18 @@ def raw_lane_overrides(value: SdkSearchFilters) -> dict[str, object]:
     return found
 
 
+def raw_lane_statuses(*, lite: bool) -> dict[str, list[str]]:
+    """Which raw statuses the raw lane searches.
+
+    Lite extracts nothing, so its search is plain RAG over every raw,
+    including the shadow raws pro already processed before the switch to
+    lite. Pro and ultra keep the SDK default (active only): a shadow raw is
+    already represented by the memories extracted from it, and returning it
+    again would repeat that content in the caller's context budget.
+    """
+    return {"statuses": ["active", "shadow"]} if lite else {}
+
+
 def hit_namespace(hit: SemanticMemoryHit) -> dict[str, str]:
     """The `get` namespace of a hit: nothing for chat, kind and variant otherwise."""
     if hit.kind in {"experience", "fact"} and hit.variant is not None:
@@ -120,6 +133,9 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
             sources = raw_lane_sources(authority.raw_sources, value.include_sources)
             if sources is None:
                 return ()
+            # Plain RAG: every vector hit over the floor is an answer, ranked
+            # by its score. Each hit yields at least one item, so the SDK
+            # needs to return no more hits than the caller's limit.
             raw = await asyncio.to_thread(
                 client.search,
                 value.query,
@@ -135,27 +151,60 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
                     "scene": "chat",
                     "reader": HY_MEMORY_READER,
                     "layers": ["l1_raw"],
+                    **raw_lane_statuses(lite=lite),
                     **raw_lane_overrides(value),
                 },
             )
             hits = tuple(
                 hit for hit in parse_sdk_raw_search(raw) if value.passes_floor("raw", hit.score)
             )
-            return tuple(
-                SdkRawContextItem(
-                    raw_id=hit.raw_id,
-                    content=hit.content,
-                    source=hit.source,
-                    session_id=hit.session_id,
-                    memory_at=(
-                        datetime.fromtimestamp(hit.memory_at, tz=UTC)
-                        if hit.memory_at is not None
-                        else None
-                    ),
-                    score=hit.score,
+            # One item per hit first, so a hit is never crowded out by another
+            # hit's sibling turns; the siblings fill whatever room is left.
+            primary: list[SdkRawContextItem] = []
+            siblings: list[SdkRawContextItem] = []
+            for hit in hits:
+                memory_at = (
+                    datetime.fromtimestamp(hit.memory_at, tz=UTC)
+                    if hit.memory_at is not None
+                    else None
                 )
-                for hit in hits
-            )
+                turns = raw_display_turns(hit) if value.raw_unit == "turn" else ()
+                if not turns:
+                    primary.append(
+                        SdkRawContextItem(
+                            raw_id=hit.raw_id,
+                            content=hit.content,
+                            source=hit.source,
+                            session_id=hit.session_id,
+                            memory_at=memory_at,
+                            score=hit.score,
+                        )
+                    )
+                    continue
+                # The embedding is the batch's: the hit is shown by the turns
+                # that contain the query, or else by its closest turn, each
+                # with the hit's score.
+                shown = [
+                    SdkRawContextItem(
+                        raw_id=hit.raw_id,
+                        turn_id=turn.turn_id,
+                        turn_index=turn.turn_index,
+                        turn_count=turn.turn_count,
+                        content=turn.content,
+                        source=hit.source,
+                        session_id=hit.session_id,
+                        memory_at=memory_at,
+                        score=hit.score,
+                    )
+                    for turn in turns_for_hit(turns, value.query)
+                ]
+                primary.append(shown[0])
+                siblings.extend(shown[1:])
+            chosen = primary[: value.limit]
+            chosen.extend(siblings[: value.limit - len(chosen)])
+            # Stable: a hit's turns stay together, in hit-score order.
+            chosen.sort(key=lambda item: item.score, reverse=True)
+            return tuple(chosen)
 
         async def memory_lane() -> tuple[SdkMemoryContextItem, ...]:
             # lite extracts nothing: its search answers raw turns, which the

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

@@ -242,6 +242,75 @@ class _InTurn:
             self._line.remove(place)
 
 
+class _Checkpointer:
+    """The one thread that runs cadence checkpoints, for every main database.
+
+    A database asks for at most one checkpoint at a time: a request made
+    while one is already waiting is the same request. The checkpoint goes
+    through the database's own writer, in its line, like any other write.
+    A waiting request holds the database weakly: it must not keep alive a
+    database nobody else holds any more.
+    """
+
+    def __init__(self) -> None:
+        self._state = threading.Condition()
+        self._due: OrderedDict[int, weakref.ref[MainDatabase]] = OrderedDict()
+        self._running = 0
+        self._thread: threading.Thread | None = None
+
+    def request(self, database: "MainDatabase") -> None:
+        with self._state:
+            self._due.setdefault(id(database), weakref.ref(database))
+            if self._thread is None or not self._thread.is_alive():
+                self._thread = threading.Thread(
+                    target=self._run,
+                    name="one-memory-checkpoint",
+                    daemon=True,
+                )
+                self._thread.start()
+            self._state.notify_all()
+
+    def forget(self, database: "MainDatabase") -> None:
+        with self._state:
+            self._due.pop(id(database), None)
+            self._state.notify_all()
+
+    def wait_idle(self, timeout: float) -> bool:
+        """Whether every requested checkpoint has run within `timeout` seconds."""
+        with self._state:
+            return self._state.wait_for(lambda: not self._due and not self._running, timeout)
+
+    def _run(self) -> None:
+        while True:
+            with self._state:
+                self._state.wait_for(lambda: bool(self._due))
+                _key, reference = self._due.popitem(last=False)
+                database = reference()
+                if database is None:
+                    self._state.notify_all()
+                    continue
+                self._running += 1
+            try:
+                database._cadence_checkpoint()
+            except Exception:  # noqa: BLE001 - maintenance must not end the thread
+                logger.warning(
+                    "main_database_checkpoint_failed file=%s", database.path.name, exc_info=True
+                )
+            finally:
+                del database
+                with self._state:
+                    self._running -= 1
+                    self._state.notify_all()
+
+
+_checkpointer = _Checkpointer()
+
+
+def wait_for_cadence_checkpoints(timeout: float = 5.0) -> bool:
+    """Wait for the cadence checkpoints already requested; for tests and shutdown."""
+    return _checkpointer.wait_idle(timeout)
+
+
 def _locked() -> sqlite3.OperationalError:
     # Callers tell a busy database from a broken one by these, as they would
     # for SQLite's own refusal.
@@ -614,13 +683,23 @@ class MainDatabase:
             return None
 
     def _maybe_checkpoint_after_write(self) -> None:
-        """Checkpoint on a write cadence without opening another connection.
+        """Count a write, and hand a due cadence checkpoint to the checkpointer.
 
         SQLite's default autocheckpoint only runs when the writer itself
         crosses the page threshold. Production hit a state where the WAL grew
         to 13GB, so the main database needs its own bounded, observable
-        maintenance path. This runs after the caller's checkout has been reset
-        and only re-enters the same serialized writer queue.
+        maintenance path: every `checkpoint_write_interval()` writes or
+        `MAIN_DATABASE_CHECKPOINT_SECONDS`, whichever comes first.
+
+        The checkpoint itself used to run right here, on the thread that had
+        just written, as one more turn in the writer's line. That thread was
+        an import heartbeat or a durable lease renewal as often as not, and
+        it then waited behind the whole line a second time before it could
+        return. The idle TRUNCATE (#278) cannot replace the cadence: under a
+        sustained import the database is never idle, and that is exactly when
+        the WAL grows. So the cadence stays, and one checkpointer thread runs
+        it (`_Checkpointer`); the writer stays the only connection that
+        writes, and no caller waits for maintenance it did not ask for.
         """
         now = time.monotonic()
         with self._state:
@@ -632,8 +711,12 @@ class MainDatabase:
             if due:
                 self._writes_since_checkpoint = 0
                 self._last_checkpoint_at = now
-        if not due:
-            return
+        if due:
+            _checkpointer.request(self)
+
+    def _cadence_checkpoint(self) -> None:
+        """One PASSIVE checkpoint through the writer, and the WAL alert after it."""
+        now = time.monotonic()
         try:
             busy, log_frames, checkpointed = self._checkpoint_from_maintenance("PASSIVE")
         except sqlite3.Error:
@@ -707,6 +790,8 @@ class MainDatabase:
 
     def close(self) -> None:
         """Close every idle connection now; ones in use close when handed back."""
+        # A checkpoint still waiting would open the writer again after this.
+        _checkpointer.forget(self)
         with self._state:
             self._generation += 1
             self._identity = None

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.13"
+HY_MEMORY_VERSION = "2.0.14"
 HY_MEMORY_NAMESPACE = "governed-v2"
 DiagnosticLogLevel = Literal["error", "warning", "info", "debug"]
 

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

@@ -503,7 +503,7 @@ def _ensure_coding_schema(path: Path) -> None:
         CodingMemoryStore,
     )
 
-    # Relies on the pinned SDK's internals (hy-memory==2.0.13 is pinned exactly).
+    # Relies on the pinned SDK's internals (hy-memory==2.0.14 is pinned exactly).
     store = CodingMemoryStore.__new__(CodingMemoryStore)
     store._db_path = str(path)
     store._conn = None

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

@@ -1,6 +1,7 @@
 import logging
 import re
 import sqlite3
+from collections.abc import Iterable
 from contextlib import AbstractContextManager, nullcontext
 from datetime import UTC, datetime
 from enum import StrEnum
@@ -18,13 +19,27 @@ from one_memory.imports.adapters import ScanPhase, SourceCursor
 #: the drift would only surface on a fresh database.
 #:
 #: This runs on SQLite today, so the SQLite rendering is used.
-from one_memory.memory_server.migrations import import_jobs_ddl
+from one_memory.memory_server.migrations import (
+    import_job_lookup_indexes_ddl,
+    import_jobs_ddl,
+)
 
 logger = logging.getLogger("one_memory.imports.jobs")
 
 JOB_SCHEMA = import_jobs_ddl("sqlite")
+#: Migration 63, applied by the store as well so a running home gets it.
+JOB_LOOKUP_INDEXES = import_job_lookup_indexes_ddl("sqlite")
 MAX_LIVE_BACKLOG_BYTES = 256 * 1024
 
+#: Finished live syncs kept per (user, source). Live sync opens a job for every
+#: change it imports, and nothing removed them: hundreds a day on an active
+#: machine. The App shows the latest job per source and the active ones; twenty is
+#: that plus enough recent history to read a failure back.
+SYNC_JOBS_KEPT_PER_SOURCE = 20
+#: Rows one pruning transaction may delete, so it holds the single writer for
+#: a few milliseconds at most however large the table grew.
+PRUNE_BATCH_ROWS = 200
+
 
 class ImportStatus(StrEnum):
     PENDING = "pending"
@@ -36,6 +51,12 @@ class ImportStatus(StrEnum):
     CANCELLED = "cancelled"
 
 
+#: States no job ever leaves (`ImportJobStore._TRANSITIONS`). Failed and
+#: interrupted are not among them: both can be resumed, and both still hold the
+#: source's one active slot.
+FINISHED_STATUSES = (ImportStatus.COMPLETED, ImportStatus.CANCELLED)
+
+
 class ImportJob(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid")
 
@@ -272,6 +293,9 @@ class ImportJobStore:
                 WHERE status IN ('pending', 'running', 'paused', 'interrupted', 'failed')
                 """
             )
+        for statement in JOB_LOOKUP_INDEXES.split(";"):
+            if statement.strip():
+                connection.execute(statement)
 
     def create(
         self,
@@ -334,8 +358,8 @@ class ImportJobStore:
         read again.
         """
         opened: list[ImportJob] = []
-        for job in self.list_jobs(status=ImportStatus.FAILED):
-            if job.operation != "history" or job.stored != 0:
+        for job in self.find_jobs(operation="history", statuses=(ImportStatus.FAILED,)):
+            if job.stored != 0:
                 continue
             reasons = {issue.reason for issue in self.list_issues(job.job_id)}
             if not reasons or not reasons <= self._UNPUBLISHED_SNAPSHOT_FAILURES:
@@ -799,16 +823,172 @@ class ImportJobStore:
         return self.require(job_id)
 
     def list_jobs(self, *, status: ImportStatus | None = None) -> list[ImportJob]:
+        """Every job, oldest first. Hot paths ask `find_jobs` a narrower question."""
+        return self.find_jobs(statuses=(status,) if status is not None else None)
+
+    def find_jobs(
+        self,
+        *,
+        user_id: str | None = None,
+        source_id: str | None = None,
+        operation: Literal["history", "sync"] | None = None,
+        statuses: Iterable[ImportStatus] | None = None,
+        newest_first: bool = False,
+        limit: int | None = None,
+    ) -> list[ImportJob]:
+        """The jobs matching every given filter, oldest first unless asked otherwise.
+
+        The filters are SQL, served by `import_jobs_lane` / `import_jobs_status`
+        (migration 63). Reading the whole table and filtering here was what
+        three bulk threads and three request threads were all inside, for tens
+        of seconds, while the loop waited on them.
+        """
+        clauses: list[str] = []
+        parameters: list[str | int] = []
+        if user_id is not None:
+            clauses.append("user_id = ?")
+            parameters.append(user_id)
+        if source_id is not None:
+            clauses.append("source_id = ?")
+            parameters.append(source_id)
+        if operation is not None:
+            clauses.append("operation = ?")
+            parameters.append(operation)
+        if statuses is not None:
+            wanted = tuple(dict.fromkeys(ImportStatus(status) for status in statuses))
+            if not wanted:
+                return []
+            clauses.append(f"status IN ({', '.join('?' for _ in wanted)})")
+            parameters.extend(wanted)
         query = "SELECT * FROM import_jobs"
-        parameters: tuple[str, ...] = ()
-        if status is not None:
-            query += " WHERE status = ?"
-            parameters = (status,)
-        query += " ORDER BY created_at, job_id"
+        if clauses:
+            query += " WHERE " + " AND ".join(clauses)
+        query += (
+            " ORDER BY created_at DESC, job_id DESC"
+            if newest_first
+            else " ORDER BY created_at, job_id"
+        )
+        if limit is not None:
+            if limit < 1:
+                raise ValueError("limit must be positive")
+            query += " LIMIT ?"
+            parameters.append(limit)
         with self._connect() as connection:
             rows = connection.execute(query, parameters).fetchall()
         return [self._to_job(row) for row in rows]
 
+    def latest_job(
+        self,
+        *,
+        user_id: str,
+        source_id: str,
+        operation: Literal["history", "sync"] | None = None,
+        statuses: Iterable[ImportStatus] | None = None,
+    ) -> ImportJob | None:
+        jobs = self.find_jobs(
+            user_id=user_id,
+            source_id=source_id,
+            operation=operation,
+            statuses=statuses,
+            newest_first=True,
+            limit=1,
+        )
+        return jobs[0] if jobs else None
+
+    def prune_finished_sync_jobs(
+        self,
+        *,
+        keep: int = SYNC_JOBS_KEPT_PER_SOURCE,
+        max_rows: int = PRUNE_BATCH_ROWS,
+    ) -> int:
+        """Delete finished live-sync jobs beyond the newest `keep` of each source.
+
+        One bounded transaction: at most `max_rows` jobs, with their issues and
+        skip counts. Returns how many jobs went; fewer than `max_rows` means
+        nothing is left to prune, so calling again until then finishes the job.
+
+        Only `sync` jobs in a state nothing leaves (completed, cancelled) are
+        ever deleted. Failed and interrupted jobs hold the source's active slot
+        and can be resumed; history jobs are named by connection receipts, by
+        the onboarding gate and by the first-import review, and there are a
+        handful per source, so they are all kept. Cursors live in their own
+        tables keyed by file, not by job, so a pruned job takes no position
+        with it. The newest job of every source is always among those kept,
+        which is the one the App and live sync health name.
+        """
+        if keep < 1:
+            raise ValueError("keep must be positive")
+        if max_rows < 1:
+            raise ValueError("max_rows must be positive")
+        finished = tuple(FINISHED_STATUSES)
+        placeholders = ", ".join("?" for _ in finished)
+        deleted = 0
+        with self._connect(writing=True) as connection:
+            lanes = connection.execute(
+                f"""
+                SELECT user_id, source_id, COUNT(*) AS finished
+                FROM import_jobs
+                WHERE operation = 'sync' AND status IN ({placeholders})
+                GROUP BY user_id, source_id
+                HAVING COUNT(*) > ?
+                """,
+                (*finished, keep),
+            ).fetchall()
+            for lane in lanes:
+                if deleted >= max_rows:
+                    break
+                # The oldest job still kept: everything strictly older goes.
+                boundary = connection.execute(
+                    f"""
+                    SELECT created_at, job_id FROM import_jobs
+                    WHERE user_id = ? AND source_id = ? AND operation = 'sync'
+                      AND status IN ({placeholders})
+                    ORDER BY created_at DESC, job_id DESC
+                    LIMIT 1 OFFSET ?
+                    """,
+                    (lane["user_id"], lane["source_id"], *finished, keep - 1),
+                ).fetchone()
+                if boundary is None:
+                    continue
+                victims = [
+                    str(row["job_id"])
+                    for row in connection.execute(
+                        f"""
+                        SELECT job_id FROM import_jobs
+                        WHERE user_id = ? AND source_id = ? AND operation = 'sync'
+                          AND status IN ({placeholders})
+                          AND (created_at < ? OR (created_at = ? AND job_id < ?))
+                        ORDER BY created_at, job_id
+                        LIMIT ?
+                        """,
+                        (
+                            lane["user_id"],
+                            lane["source_id"],
+                            *finished,
+                            boundary["created_at"],
+                            boundary["created_at"],
+                            boundary["job_id"],
+                            max_rows - deleted,
+                        ),
+                    )
+                ]
+                if not victims:
+                    continue
+                marks = ", ".join("?" for _ in victims)
+                # Foreign keys cascade these too; deleting them by name keeps
+                # the outcome the same on a connection that has them off.
+                connection.execute(f"DELETE FROM import_issues WHERE job_id IN ({marks})", victims)
+                connection.execute(
+                    f"DELETE FROM import_skip_reason_counts WHERE job_id IN ({marks})", victims
+                )
+                removed = connection.execute(
+                    f"DELETE FROM import_jobs WHERE job_id IN ({marks}) "
+                    f"AND operation = 'sync' AND status IN ({placeholders})",
+                    (*victims, *finished),
+                )
+                deleted += removed.rowcount
+        return deleted
+
     def get_source_cursors(
         self,
         *,

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

@@ -28,7 +28,7 @@ from one_memory.imports.adapters import (
 )
 from one_memory.imports.jobs import ImportJob, ImportJobStore, ImportStatus
 from one_memory.imports.supervisor import ImportSupervisor
-from one_memory.user_scope import DEFAULT_USER_ID, lane_key, split_lane
+from one_memory.user_scope import DEFAULT_USER_ID, split_lane
 
 #: How long an automatic retry will wait for a wedged worker to notice cancel.
 ABANDON_TIMEOUT_SECONDS = 5.0
@@ -249,13 +249,15 @@ class PollingLiveSyncSupervisor:
                     ),
                 )
                 return None
-            source_jobs = await in_the_background(self.job_store.list_jobs)
+            # This lane's unfinished jobs only: the newest unfinished sync, and
+            # whatever history shares its cursor. Asked in SQL; listing every
+            # job ever run and filtering here was the poll's biggest read.
+            source_jobs = await in_the_background(self._lane_jobs, source_id)
             active_sync = next(
                 (
                     job
                     for job in reversed(source_jobs)
-                    if lane_key(job.user_id, job.source_id) == source_id
-                    and job.operation == "sync"
+                    if job.operation == "sync"
                     and job.status
                     in {
                         ImportStatus.PENDING,
@@ -275,7 +277,9 @@ class PollingLiveSyncSupervisor:
                     active_sync.job_id
                 ):
                     attempted_at = datetime.now(UTC)
-                    self.import_supervisor.enqueue(active_sync.job_id)
+                    await self.import_supervisor.enqueue_off_loop(
+                        active_sync.job_id, job=active_sync
+                    )
                     if not wait_for_completion:
                         self._observe_sync(source_id, attempted_at, active_sync.job_id)
                         return active_sync
@@ -295,22 +299,20 @@ class PollingLiveSyncSupervisor:
                 # by a process restart between yield and claim.  Enqueue is
                 # the operation that actually creates the task.
                 attempted_at = datetime.now(UTC)
-                self.import_supervisor.resume(active_sync.job_id)
-                self.import_supervisor.enqueue(active_sync.job_id)
+                resumed = await self.import_supervisor.resume_off_loop(active_sync.job_id)
+                await self.import_supervisor.enqueue_off_loop(active_sync.job_id, job=resumed)
                 completed = await self.import_supervisor.wait(active_sync.job_id)
                 self._record_outcome(source_id, attempted_at, completed)
                 return completed
             competing = [
                 job
                 for job in source_jobs
-                if lane_key(job.user_id, job.source_id) == source_id
-                and job.operation != "sync"
+                if job.operation != "sync"
                 and job.status
                 in {ImportStatus.PENDING, ImportStatus.RUNNING, ImportStatus.INTERRUPTED}
             ]
             bounded_history_incomplete = any(
-                lane_key(job.user_id, job.source_id) == source_id
-                and job.operation == "history"
+                job.operation == "history"
                 and job.history_after is not None
                 and job.status is not ImportStatus.COMPLETED
                 for job in source_jobs
@@ -354,6 +356,38 @@ class PollingLiveSyncSupervisor:
             # automatically after the realtime slice completes.
             return await self._run_sync(source_id, wait_for_completion=wait_for_completion)
 
+    def _lane_jobs(self, source_id: str) -> list[ImportJob]:
+        """The lane's jobs `_sync` decides on, oldest first.
+
+        Everything not completed, of either operation: an unfinished sync to
+        pick up, a history that competes for the cursor, and a bounded history
+        that has not completed yet (which includes a cancelled one). Finished
+        syncs, the rows that grow without end, are never read.
+        """
+        user_id, bare_source_id = split_lane(source_id)
+        unfinished = tuple(
+            status for status in ImportStatus if status is not ImportStatus.COMPLETED
+        )
+        return [
+            *self.job_store.find_jobs(
+                user_id=user_id,
+                source_id=bare_source_id,
+                operation="history",
+                statuses=unfinished,
+            ),
+            *self.job_store.find_jobs(
+                user_id=user_id,
+                source_id=bare_source_id,
+                operation="sync",
+                statuses=(
+                    ImportStatus.PENDING,
+                    ImportStatus.RUNNING,
+                    ImportStatus.PAUSED,
+                    ImportStatus.INTERRUPTED,
+                ),
+            ),
+        ]
+
     async def _has_work(self, source_id: str) -> bool:
         """Ask the source whether a scan would write anything, before taking a job out.
 
@@ -452,7 +486,7 @@ class PollingLiveSyncSupervisor:
             ),
         )
         self._active_jobs[source_id] = job.job_id
-        self.import_supervisor.enqueue(job.job_id)
+        await self.import_supervisor.enqueue_off_loop(job.job_id, job=job)
         if not wait_for_completion:
             self._observe_sync(source_id, attempted_at, job.job_id)
             return job

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

@@ -23,6 +23,7 @@ from pathlib import Path
 
 from watchfiles import awatch
 
+from one_memory.background_work import in_the_background
 from one_memory.diagnostic_log import diagnostic_event
 from one_memory.imports.jobs import ImportStatus
 from one_memory.imports.live_sync import PollingLiveSyncSupervisor
@@ -90,7 +91,8 @@ class FileWatchLiveSync:
                 # turn that caused setup. The watcher cannot observe a change
                 # from before it existed, so immediately run the same bounded
                 # trigger once instead of waiting for the 60-second backstop.
-                self._queue_changes(set(self.roots.values()))
+                # The roots are this watcher's own sources: nothing to resolve.
+                self._queue_sources(tuple(self.roots), count=len(self.roots))
 
     async def shutdown(self) -> None:
         task = self._task
@@ -139,28 +141,54 @@ class FileWatchLiveSync:
         observed_at: datetime | None = None,
     ) -> tuple[str, ...]:
         """Queue affected sources and wait for their coalesced work to settle."""
-        workers = self._queue_changes(changed, observed_at=observed_at)
+        workers = await self._queue_changes(changed, observed_at=observed_at)
         await asyncio.gather(*(asyncio.shield(worker) for worker in workers.values()))
         return tuple(workers)
 
-    def _queue_changes(
+    def sources_for(self, changed: set[Path]) -> tuple[str, ...]:
+        """The sources owning `changed`, in path order. Resolves every path."""
+        return tuple(
+            dict.fromkeys(
+                source
+                for source in (self.source_for(path) for path in sorted(changed))
+                if source is not None
+            )
+        )
+
+    async def _queue_changes(
         self, changed: set[Path], *, observed_at: datetime | None = None
+    ) -> dict[str, asyncio.Task[None]]:
+        """Attribute a batch of changed paths to sources and queue them.
+
+        Attributing resolves each path, a `stat` per component, and a busy
+        host reports hundreds of paths at once. On the loop that was the main
+        thread inside `source_for` while nothing else was answered, so the
+        resolving happens off it and only the queueing here.
+        """
+        if self._stop.is_set():
+            return {}
+        observed = observed_at or datetime.now(UTC)
+        sources = await in_the_background(self.sources_for, changed)
+        return self._queue_sources(sources, count=len(changed), observed_at=observed)
+
+    def _queue_sources(
+        self,
+        sources: tuple[str, ...],
+        *,
+        count: int,
+        observed_at: datetime | None = None,
     ) -> dict[str, asyncio.Task[None]]:
         if self._stop.is_set():
             return {}
         observed = observed_at or datetime.now(UTC)
         workers: dict[str, asyncio.Task[None]] = {}
-        for source_id in dict.fromkeys(
-            source
-            for source in (self.source_for(path) for path in sorted(changed))
-            if source is not None
-        ):
+        for source_id in sources:
             diagnostic_event(
                 level="debug",
                 component="import.live_watch",
                 event="live_sync.hook_received",
                 outcome="queued",
-                details={"source_id": source_id, "count": len(changed)},
+                details={"source_id": source_id, "count": count},
             )
             self._pending[source_id] = max(self._pending.get(source_id, observed), observed)
             worker = self._workers.get(source_id)
@@ -239,7 +267,7 @@ class FileWatchLiveSync:
                 stop_event=self._stop,
                 ignore_permission_denied=True,
             ):
-                self._queue_changes({Path(path) for _change, path in changes})
+                await self._queue_changes({Path(path) for _change, path in changes})
         except asyncio.CancelledError:
             raise
         except Exception:

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

@@ -1,6 +1,7 @@
 import asyncio
 from collections.abc import Awaitable, Callable, Mapping
 from contextlib import suppress
+from functools import partial
 from time import monotonic
 from typing import Literal
 
@@ -82,6 +83,10 @@ class ImportSupervisor:
         )
         self._tasks: dict[str, asyncio.Task[None]] = {}
         self._drains: dict[str, asyncio.Task[None]] = {}
+        #: Pause and cancel settle a stopped job with a read and a transition.
+        #: Made on the loop, two of them could not interleave; made off it,
+        #: they are serialized here so they still land in the order they woke.
+        self._settling = asyncio.Lock()
 
     def register_adapter(self, adapter: SourceAdapter, *, user_id: str = "local") -> None:
         """Adapters are keyed by lane: the source id for `local`, `source@user`
@@ -116,11 +121,44 @@ class ImportSupervisor:
             raise UnknownSourceError(lane)
         if self.source_enabled is not None and not self.source_enabled(lane):
             return job
-        existing = self._tasks.get(job_id)
+        self._start(job)
+        return job
+
+    async def enqueue_off_loop(self, job_id: str, *, job: ImportJob | None = None) -> ImportJob:
+        """`enqueue`, with its job read and its grant check made off the loop.
+
+        Both are database reads, and the grant check resolves the source root
+        on disk. Live sync enqueues from its poll, so on the loop they were
+        the main thread waiting for a reader behind an import peak. `job` is
+        the row the caller has just read, when it has one.
+        """
+        if job_id in self._drains:
+            raise InvalidJobTransitionError("import_cleanup_in_progress")
+
+        def admit() -> tuple[ImportJob, bool]:
+            current = job if job is not None and job.job_id == job_id else None
+            if current is None:
+                current = self.job_store.require(job_id)
+            lane = self.lane_of(current)
+            return current, self.source_enabled is None or self.source_enabled(lane)
+
+        current, enabled = await in_the_background(admit)
+        lane = self.lane_of(current)
+        if lane not in self.adapters:
+            raise UnknownSourceError(lane)
+        if job_id in self._drains:
+            raise InvalidJobTransitionError("import_cleanup_in_progress")
+        if enabled:
+            self._start(current)
+        return current
+
+    def _start(self, job: ImportJob) -> None:
+        lane = self.lane_of(job)
+        existing = self._tasks.get(job.job_id)
         if existing is None or existing.done():
-            self._tasks[job_id] = asyncio.create_task(
+            self._tasks[job.job_id] = asyncio.create_task(
                 self._execute(
-                    job_id,
+                    job.job_id,
                     source_id=lane,
                     operation=job.operation,
                 )
@@ -130,10 +168,24 @@ class ImportSupervisor:
                 component="import.supervisor",
                 event="import.queued",
                 outcome="queued",
-                job_id=job_id,
+                job_id=job.job_id,
                 details={"source_id": job.source_id, "mode": job.operation},
             )
-        return job
+
+    async def resume_interrupted_off_loop(self) -> None:
+        """`resume_interrupted` for the running Core: every read and write off the loop."""
+        for status in (ImportStatus.INTERRUPTED, ImportStatus.PENDING):
+            jobs = await in_the_background(partial(self.job_store.find_jobs, statuses=(status,)))
+            for job in jobs:
+                if self.lane_of(job) not in self.adapters:
+                    continue
+                if job.status is not ImportStatus.PENDING:
+                    if self.source_enabled is not None and not await in_the_background(
+                        self.source_enabled, self.lane_of(job)
+                    ):
+                        continue
+                    job = await in_the_background(self.job_store.queue_for_resume, job.job_id)
+                await self.enqueue_off_loop(job.job_id, job=job)
 
     def resume_interrupted(self) -> None:
         # A process restart must not place recoverable work behind every newly
@@ -166,40 +218,64 @@ class ImportSupervisor:
         self.enqueue(job_id)
         return queued
 
+    async def resume_off_loop(self, job_id: str) -> ImportJob:
+        """`resume`, with its reads and its write made off the event loop."""
+        if job_id in self._drains:
+            raise InvalidJobTransitionError("import_cleanup_in_progress")
+        queued = await in_the_background(self._queue_resumable, job_id)
+        if queued.status is not ImportStatus.PENDING:
+            return queued
+        return await self.enqueue_off_loop(job_id, job=queued)
+
+    def _queue_resumable(self, job_id: str) -> ImportJob:
+        job = self.job_store.require(job_id)
+        if job.status not in {
+            ImportStatus.PENDING,
+            ImportStatus.PAUSED,
+            ImportStatus.INTERRUPTED,
+            ImportStatus.FAILED,
+        }:
+            return job
+        return self.job_store.queue_for_resume(job_id)
+
     async def wait(self, job_id: str) -> ImportJob:
         task = self._tasks.get(job_id)
         if task is not None:
             await asyncio.shield(task)
-        return self.job_store.require(job_id)
+        return await in_the_background(self.job_store.require, job_id)
 
     async def pause(self, job_id: str) -> ImportJob:
-        job = self.job_store.require(job_id)
+        job = await in_the_background(self.job_store.require, job_id)
         if job.status in {ImportStatus.PAUSED, ImportStatus.INTERRUPTED}:
             return job
         await self._cancel_task(job_id)
-        current = self.job_store.require(job_id)
-        if current.status is ImportStatus.COMPLETED:
-            return current
-        return self.job_store.transition(job_id, ImportStatus.PAUSED)
+        async with self._settling:
+            return await in_the_background(self._settle_stopped, job_id, ImportStatus.PAUSED)
 
     async def cancel(self, job_id: str) -> ImportJob:
-        job = self.job_store.require(job_id)
+        job = await in_the_background(self.job_store.require, job_id)
         if job.status is ImportStatus.CANCELLED:
             return job
         await self._cancel_task(job_id)
+        async with self._settling:
+            return await in_the_background(self._settle_stopped, job_id, ImportStatus.CANCELLED)
+
+    def _settle_stopped(self, job_id: str, target: ImportStatus) -> ImportJob:
+        """A stopped runner's job moves to `target`, unless it completed first."""
         current = self.job_store.require(job_id)
         if current.status is ImportStatus.COMPLETED:
             return current
-        return self.job_store.transition(job_id, ImportStatus.CANCELLED)
+        return self.job_store.transition(job_id, target)
 
     async def shutdown(self) -> None:
-        running = [
-            job
-            for job in self.job_store.list_jobs(status=ImportStatus.RUNNING)
-            if job.job_id in self._tasks
-        ]
-        for job in running:
-            self.job_store.interrupt_running(job.job_id)
+        mine = frozenset(self._tasks)
+
+        def interrupt_mine() -> None:
+            for job in self.job_store.find_jobs(statuses=(ImportStatus.RUNNING,)):
+                if job.job_id in mine:
+                    self.job_store.interrupt_running(job.job_id)
+
+        await in_the_background(interrupt_mine)
         await asyncio.gather(
             *(self._cancel_task(job_id) for job_id in tuple(self._tasks)),
         )
@@ -355,9 +431,7 @@ class ImportSupervisor:
             try:
                 await asyncio.wait_for(asyncio.shield(task), timeout)
             except TimeoutError:
-                current = self.job_store.require(job_id)
-                if current.status is ImportStatus.RUNNING:
-                    self.job_store.interrupt_running(job_id)
+                await in_the_background(self._interrupt_if_running, job_id)
                 return False
             except asyncio.CancelledError:
                 # The worker noticed cancel and exited. A cancel of this
@@ -365,10 +439,16 @@ class ImportSupervisor:
                 caller = asyncio.current_task()
                 if caller is not None and caller.cancelling():
                     raise
+        # A read and a write on the single writer: off the loop. Live sync
+        # calls this from its poll, and on the loop it was the main thread
+        # waiting its turn for the writer behind an import peak.
+        await in_the_background(self._interrupt_if_running, job_id)
+        return True
+
+    def _interrupt_if_running(self, job_id: str) -> None:
         current = self.job_store.require(job_id)
         if current.status is ImportStatus.RUNNING:
             self.job_store.interrupt_running(job_id)
-        return True
 
     async def _cancel_task(self, job_id: str) -> None:
         drain = self._drains.get(job_id)

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

@@ -28,10 +28,11 @@ from pathlib import Path
 from typing import Any, Literal
 from uuid import uuid4
 
-from one_memory.background_work import in_the_background
+from one_memory.background_work import in_lease_renewal, in_the_write_queue
 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.exception_reporting import report_exception
 from one_memory.imports.models import EvidenceOrigin, EvidenceRecord
 from one_memory.imports.storage import EvidenceStore
 from one_memory.memory_server.contracts import (
@@ -451,6 +452,14 @@ class _LeaseRenewalFatalError(RuntimeError):
     """A non-transient renewal failure that must stop the worker."""
 
 
+def _drain_exit_code(error: BaseException) -> str:
+    """A stable, lowercase code for why the drain exited; never message text."""
+    if isinstance(error, _LeaseRenewalFatalError):
+        return "lease_renewal_failed"
+    name = re.sub(r"(?<!^)(?=[A-Z])", "_", type(error).__name__).lower()
+    return name if re.fullmatch(r"[a-z_]{1,64}", name) else "drain_error"
+
+
 @dataclass(frozen=True)
 class UnknownWriteJob:
     """An `unknown` row that still needs an external inquiry.
@@ -2562,14 +2571,14 @@ async def execute_job(
     # Every store call here is synchronous SQLite work on the single writer or
     # a reader; made from the loop it would block health checks and the API for
     # as long as the write lock is held.
-    request = await in_the_background(queue.payload_for, job_id)
+    request = await in_the_write_queue(queue.payload_for, job_id)
     if request is None:
         # A claimed row whose payload is missing or unreadable (reaped while
         # still claimable, corrupted, or a foreign row) must not be left
         # `running`: nothing was sent, so it is a terminal failure rather than
         # an unknown, and the lease must be released so it is not held until it
         # lapses.
-        await in_the_background(
+        await in_the_write_queue(
             queue.settle,
             job_id,
             state=FAILED,
@@ -2579,9 +2588,9 @@ async def execute_job(
         )
         return
     try:
-        records = await in_the_background(evidence_for_request, evidence, request)
+        records = await in_the_write_queue(evidence_for_request, evidence, request)
     except LookupError as error:
-        await in_the_background(
+        await in_the_write_queue(
             queue.settle,
             job_id,
             state=FAILED,
@@ -2618,7 +2627,7 @@ async def execute_job(
         source_id = lane_key(next(iter(users), DEFAULT_USER_ID), source_id)
 
         async def mark() -> bool:
-            return await in_the_background(queue.mark_invoked, job_id, owner=owner)
+            return await in_the_write_queue(queue.mark_invoked, job_id, owner=owner)
 
         if admit_hy_invocation is not None:
             admitted = await admit_hy_invocation(source_id, requires_generation, mark)
@@ -2633,7 +2642,7 @@ async def execute_job(
             raise RuntimeError("durable_write_lease_lost_before_invocation")
         hy_invoked = True
 
-    attempt = await in_the_background(queue.sdk_attempt, job_id)
+    attempt = await in_the_write_queue(queue.sdk_attempt, job_id)
     window = model_failures() if model_failures is not None else nullcontext(None)
     provider: ModelAttempt | None = None
     try:
@@ -2675,7 +2684,7 @@ async def execute_job(
             }
             and not hy_invoked
         ):
-            await in_the_background(queue.release_unstarted, job_id, owner=owner)
+            await in_the_write_queue(queue.release_unstarted, job_id, owner=owner)
             return
         provider_failure = provider.failure if provider is not None else None
         if hy_invoked and provider_failure is not None:
@@ -2683,7 +2692,7 @@ async def execute_job(
             # gone, a provider that is down): the job is fine and waits for
             # the provider, in its place, without spending a retry. The
             # breaker paces when it is tried again.
-            await in_the_background(
+            await in_the_write_queue(
                 queue.requeue_for_model, job_id, error_code=provider_failure, owner=owner
             )
             logger.info(
@@ -2712,7 +2721,7 @@ async def execute_job(
         elif not hy_invoked and code in MODEL_NOT_READY_CODES:
             state = WAITING_FOR_MODEL
         if state is not None:
-            await in_the_background(
+            await in_the_write_queue(
                 queue.settle,
                 job_id,
                 state=state,
@@ -2724,7 +2733,7 @@ async def execute_job(
             # Anything else is this job's own trouble: content the model or
             # the SDK refused, a reply that did not parse, a check that failed
             # after the write. Retried a few times, then `failed`.
-            state = await in_the_background(
+            state = await in_the_write_queue(
                 queue.retry_or_fail, job_id, error_code=code, owner=owner
             )
         # Never `logger.exception` here: it prints the exception's message,
@@ -2751,7 +2760,7 @@ async def execute_job(
                 error_chain_text(error),
             )
         return
-    await in_the_background(
+    await in_the_write_queue(
         queue.settle,
         job_id,
         state=COMPLETED,
@@ -2760,13 +2769,13 @@ async def execute_job(
         sdk_memory_ids=memory_ids,
     )
     release = getattr(evidence, "release_sdk_staging", None)
-    durable_receipt = await in_the_background(queue.sdk_receipt, job_id)
+    durable_receipt = await in_the_write_queue(queue.sdk_receipt, job_id)
     if (
         callable(release)
         and durable_receipt is not None
         and (durable_receipt.raw_ids or durable_receipt.memory_ids)
     ):
-        await in_the_background(
+        await in_the_write_queue(
             release,
             tuple(record.evidence_id for record in records),
         )
@@ -2891,6 +2900,16 @@ class DurableWriteDrain:
         #: a minute against a provider that refused every one (#245).
         self.model_claim_allowance = model_claim_allowance
         self._last_unknown_reconcile: datetime | None = None
+        #: What `run_supervised` says about the drain: None before it was
+        #: first started, then `running`, `restarting` (it exited on an error
+        #: and waits out its backoff) or `stopped`. Shown by memory-processing.
+        self.liveness: Literal["running", "restarting", "stopped"] | None = None
+        #: The stable code of the last unexpected exit, kept after a restart.
+        self.last_error_code: str | None = None
+        self.restarts = 0
+        self._stop_requested = False
+        #: Wakes a restart backoff, so `stop()` does not wait it out.
+        self._stopping = asyncio.Event()
 
     def claim(
         self,
@@ -3218,7 +3237,7 @@ class DurableWriteDrain:
             # Claim is a blocking SQLite write on the single writer connection.
             # Run it off the loop, like every other DB call in this chain; a
             # busy writer would otherwise stall the whole event loop.
-            ids = await in_the_background(
+            ids = await in_the_write_queue(
                 self.claim,
                 limit=limit,
                 budget_ms=self.claim_budget_ms,
@@ -3231,10 +3250,10 @@ class DurableWriteDrain:
         await asyncio.gather(*(self._run_claimed_job(job_id) for job_id in ids))
 
     async def _run_claimed_job(self, job_id: str) -> None:
-        if not await in_the_background(self.still_owns, job_id):
+        if not await in_the_write_queue(self.still_owns, job_id):
             logger.warning("durable_write_drain_lease_lost_before_execution job_id=%s", job_id)
             return
-        deadline = await in_the_background(self.lease_deadline, job_id)
+        deadline = await in_the_write_queue(self.lease_deadline, job_id)
         if deadline is None:
             return
         renewal = asyncio.create_task(self._renew_loop(job_id, deadline))
@@ -3261,11 +3280,9 @@ class DurableWriteDrain:
                 "durable_write_drain_job_ceiling_exceeded job_id=%s",
                 job_id,
             )
-            await in_the_background(
-                self.queue.settle_worker_error,
+            await self._settle_worker_error(
                 job_id,
                 error_code="durable_work_ceiling_exceeded",
-                owner=self.owner,
             )
         except Exception as error:
             logger.error(
@@ -3273,12 +3290,15 @@ class DurableWriteDrain:
                 job_id,
                 type(error).__name__,
             )
+            report_exception(
+                "memory.write_queue",
+                "memory.write_queue.executor_failed",
+                error,
+            )
             self._record_database_error(error)
-            await in_the_background(
-                self.queue.settle_worker_error,
+            await self._settle_worker_error(
                 job_id,
                 error_code="durable_executor_error",
-                owner=self.owner,
             )
         finally:
             execution.cancel()
@@ -3287,9 +3307,39 @@ class DurableWriteDrain:
                 with contextlib.suppress(asyncio.CancelledError, Exception):
                     await task
 
+    async def _settle_worker_error(self, job_id: str, *, error_code: str) -> None:
+        """Best-effort settlement must not kill the long-lived drain.
+
+        The row still has a durable lease when this fallback runs. If the
+        single SQLite writer is busy, waiting and retrying here only extends
+        the same contention that made the executor fail. Leave the row where
+        it is instead: lease recovery later moves a never-invoked job back to
+        pending and a possibly-invoked job to unknown, preserving the
+        no-blind-retry boundary.
+
+        Only transient lock/busy refusals are deferred. A malformed query or
+        other programming error still escapes to the outer per-job fence.
+        """
+        try:
+            await in_the_write_queue(
+                self.queue.settle_worker_error,
+                job_id,
+                error_code=error_code,
+                owner=self.owner,
+            )
+        except sqlite3.OperationalError as error:
+            if "locked" not in str(error).lower() and "busy" not in str(error).lower():
+                raise
+            self._sample_database_failure("settle_worker_error", error)
+            logger.warning(
+                "durable_write_worker_settlement_deferred job_id=%s error_type=%s",
+                job_id,
+                type(error).__name__,
+            )
+
     async def _execute_with_ceiling(self, job_id: str) -> None:
         ceiling = (
-            await in_the_background(self.work_ceiling)
+            await in_the_write_queue(self.work_ceiling)
             if callable(self.work_ceiling)
             else self.work_ceiling
         )
@@ -3324,7 +3374,33 @@ class DurableWriteDrain:
                 finished = {task for task in active if task.done()}
                 active.difference_update(finished)
                 for task in finished:
-                    await task
+                    try:
+                        await task
+                    except asyncio.CancelledError:
+                        raise
+                    except _LeaseRenewalFatalError:
+                        # Renewal is the ownership fence. If it is broken,
+                        # continuing could let another worker reclaim a job
+                        # while this process is still executing it.
+                        raise
+                    except Exception as error:
+                        # A job task is a fault-containment boundary. Its row
+                        # retains the durable lease and is recovered according
+                        # to hy_invoked; killing this only consumer would strand
+                        # every unrelated write behind it.
+                        self._record_database_error(error)
+                        logger.error(
+                            "durable_write_drain_job_task_failed task_name=%s error_type=%s",
+                            task.get_name(),
+                            type(error).__name__,
+                        )
+                        # With its stack: a type name alone cannot tell one
+                        # stray failure from a bug that fails every job.
+                        report_exception(
+                            "memory.write_queue",
+                            "memory.write_queue.job_task_failed",
+                            error,
+                        )
                 if self.budget is not None and work_deadline is not None:
                     now = _time.monotonic()
                     # Always admit the first unit of a cycle even when the
@@ -3340,7 +3416,7 @@ class DurableWriteDrain:
                 try:
                     available = self.max_concurrent_sources - len(active)
                     ids = (
-                        await in_the_background(
+                        await in_the_write_queue(
                             self.claim,
                             limit=available,
                             budget_ms=self.claim_budget_ms,
@@ -3349,7 +3425,12 @@ class DurableWriteDrain:
                         else ()
                     )
                     for job_id in ids:
-                        active.add(asyncio.create_task(self._run_claimed_job(job_id)))
+                        active.add(
+                            asyncio.create_task(
+                                self._run_claimed_job(job_id),
+                                name=f"durable-write:{job_id}",
+                            )
+                        )
                     executed = len(ids)
                     self._database_failure_streaks.pop("claim", None)
                 except asyncio.CancelledError:
@@ -3447,7 +3528,79 @@ class DurableWriteDrain:
                 await asyncio.gather(*active, return_exceptions=True)
 
     def stop(self) -> None:
+        self._stop_requested = True
         self.running = False
+        self._stopping.set()
+
+    async def run_supervised(
+        self,
+        *,
+        idle_pause: float = 0.05,
+        restart_backoff: float = 1.0,
+        max_restart_backoff: float = 60.0,
+    ) -> None:
+        """`run_forever`, started again whenever it exits on an error.
+
+        The drain is the queue's only consumer, so a drain that died stranded
+        every write behind it until the next start, and nothing said so: an
+        unexpected SQLite error on the claim path (disk I/O, a vanished file)
+        or an exception from a claim gate ended it for good. Each exit is now
+        reported with its stack, and the drain comes back after a pause that
+        doubles up to `max_restart_backoff`, back to `restart_backoff` once it
+        has run for that long. `stop()` and cancellation end it as before.
+
+        Ownership across a restart: `run_forever` only returns once every job
+        task it started has finished (its `finally`), and a task whose lease
+        renewal broke (`_LeaseRenewalFatalError`) waits for its execution to
+        return first, so nothing of the old drain is still running a job when
+        the new one claims. The new one also takes a new lease owner, so a
+        statement of the old one still in a pool thread can neither renew nor
+        settle a row under the new drain's name, and the old drain's leases
+        are only ever recovered by expiry (`reclaim_expired`), which decides
+        by `hy_invoked` exactly as for a drain that died with its process.
+        """
+        backoff = restart_backoff
+        while not self._stop_requested:
+            started = _time.monotonic()
+            self.liveness = "running"
+            try:
+                await self.run_forever(idle_pause=idle_pause)
+            except asyncio.CancelledError:
+                self.liveness = "stopped"
+                raise
+            except Exception as error:
+                self.last_error_code = _drain_exit_code(error)
+                self.restarts += 1
+                report_exception(
+                    "memory.write_queue",
+                    "memory.write_queue.drain_exited",
+                    error,
+                    context={"restarts": self.restarts},
+                )
+                logger.error(
+                    "durable_write_drain_exited error_code=%s restarts=%d chain=%s",
+                    self.last_error_code,
+                    self.restarts,
+                    error_chain_text(error),
+                )
+            else:
+                if self._stop_requested:
+                    break
+            if self._stop_requested:
+                break
+            self.owner = f"durable-drain:{uuid4()}"
+            if _time.monotonic() - started >= max_restart_backoff:
+                backoff = restart_backoff
+            self.liveness = "restarting"
+            try:
+                await asyncio.wait_for(self._stopping.wait(), timeout=backoff)
+            except TimeoutError:
+                pass
+            except asyncio.CancelledError:
+                self.liveness = "stopped"
+                raise
+            backoff = min(backoff * 2, max_restart_backoff)
+        self.liveness = "stopped"
 
     async def _run_idle_maintenance(
         self,
@@ -3465,7 +3618,7 @@ class DurableWriteDrain:
         persisting provider or user content.
         """
         try:
-            await in_the_background(operation)
+            await in_the_write_queue(operation)
             self._database_failure_streaks.pop(f"maintenance:{operation_name}", None)
         except asyncio.CancelledError:
             raise
@@ -3493,7 +3646,7 @@ class DurableWriteDrain:
                 renewed_at = datetime.now(UTC)
                 remaining = (deadline - renewed_at).total_seconds()
                 if remaining <= 0 or not await asyncio.wait_for(
-                    in_the_background(self.renew, job_id), timeout=remaining
+                    in_lease_renewal(self.renew, job_id), timeout=remaining
                 ):
                     return False
                 deadline = renewed_at + self.lease

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

@@ -3235,6 +3235,65 @@ async def _062_onboarding_progress(cursor: Any) -> None:
             raise
 
 
+IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION = 63
+
+
+def import_job_lookup_indexes_ddl(dialect: str = "mysql") -> str:
+    """Index the per-source and per-status import job lookups.
+
+    Migration 7 gave SQLite `import_jobs` no index beyond its primary key (the
+    one-active-job index is partial), so every "this source's jobs" or "the
+    jobs in this state" question read and sorted the whole table. Live sync
+    adds a row per change it imports and nothing removed them, so on a machine
+    a few weeks old that table was thousands of rows read several times a poll.
+
+    `import_jobs_lane` serves the lane lookups (one user, one source, one
+    operation, a status, newest first) and the pruning of finished syncs;
+    `import_jobs_status` serves restart recovery and the status filters.
+    MySQL's migration 7 already has a status index and no `operation` column,
+    so it only gains the ordered source index.
+    """
+    if dialect == "sqlite":
+        return """
+CREATE INDEX IF NOT EXISTS import_jobs_lane
+ON import_jobs(user_id, source_id, operation, status, created_at);
+CREATE INDEX IF NOT EXISTS import_jobs_status
+ON import_jobs(status, created_at);
+"""
+    return """
+CREATE INDEX idx_import_jobs_source_created
+ON import_jobs(user_id, source_id, created_at);
+"""
+
+
+def _is_duplicate_index(error: Exception) -> bool:
+    """MySQL's "index name already there" (1061): the index is in place."""
+    args: tuple[object, ...] = getattr(error, "args", ())
+    return bool(args) and args[0] == 1061
+
+
+async def _063_import_job_lookup_indexes(cursor: Any) -> None:
+    dialect = _dialect(cursor)
+    if dialect == "sqlite":
+        # `operation` was added by the job store, not by migration 7, so a
+        # database built from the catalog alone may not have it yet. Same
+        # declaration as the store's own addition.
+        if "operation" not in await cursor.columns("import_jobs"):
+            await cursor.execute(
+                "ALTER TABLE import_jobs ADD COLUMN operation TEXT NOT NULL DEFAULT 'history'"
+            )
+        for statement in _split_statements(import_job_lookup_indexes_ddl(dialect)):
+            await cursor.execute(statement)
+        return
+    for statement in _split_statements(import_job_lookup_indexes_ddl(dialect)):
+        try:
+            await cursor.execute(statement)
+        except Exception as error:  # noqa: BLE001 - MySQL driver error type varies
+            if _is_duplicate_index(error):
+                continue
+            raise
+
+
 def _split_statements(sql: str) -> list[str]:
     """Split a multi-statement script, dropping blanks, PRAGMA and comments.
 
@@ -3541,4 +3600,9 @@ MIGRATIONS: tuple[Migration, ...] = (
         _062_onboarding_progress,
         ("onboarding_state",),
     ),
+    Migration(
+        IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION,
+        "import_job_lookup_indexes",
+        _063_import_job_lookup_indexes,
+    ),
 )

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

@@ -0,0 +1,123 @@
+"""Turn-level read projection for SDK raw batches.
+
+HY keeps several complete conversation turns in one L1 raw node so model
+writes stay batched. Product readers must not expose that storage batch as
+one conversation item: this module gives every contained turn a stable
+identity while retaining the parent raw id for traceability.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import re
+from dataclasses import dataclass
+
+from one_memory.hy_sdk_types import HySdkRawRecord, raw_turn_messages
+
+
+@dataclass(frozen=True)
+class RawDisplayTurn:
+    turn_id: str
+    turn_index: int
+    turn_count: int
+    content: str
+    messages: tuple[tuple[str, str], ...]
+
+
+def raw_display_turns(record: HySdkRawRecord) -> tuple[RawDisplayTurn, ...]:
+    """Split one stored raw batch into user-opened turns.
+
+    A user message opens a new turn; assistant/tool/system messages stay with
+    that user message until the next user message. Legacy content with no user
+    marker remains one turn so it never disappears from the product.
+    """
+    messages = raw_turn_messages(record.content)
+    grouped: list[list[tuple[str, str]]] = []
+    for message in messages:
+        role, _ = message
+        has_user = bool(grouped) and any(item[0] == "user" for item in grouped[-1])
+        if not grouped or (role == "user" and has_user):
+            grouped.append([message])
+        else:
+            grouped[-1].append(message)
+    if not grouped:
+        grouped = [[("", record.content)]]
+
+    digest = hashlib.sha256(record.raw_id.encode("utf-8")).hexdigest()
+    count = len(grouped)
+    return tuple(
+        RawDisplayTurn(
+            turn_id=f"raw-turn:{digest}:{index}",
+            turn_index=index,
+            turn_count=count,
+            content=_turn_content(turn),
+            messages=tuple(turn),
+        )
+        for index, turn in enumerate(grouped)
+    )
+
+
+def _turn_content(messages: list[tuple[str, str]]) -> str:
+    return "\n".join(f"[{role}]: {content}" if role else content for role, content in messages)
+
+
+def raw_prefix_added(
+    cached: tuple[tuple[str, str | None], ...],
+    current: tuple[tuple[str, str | None], ...],
+    cached_total: int,
+    current_total: int,
+) -> int | None:
+    """How many raws were only put on top of a cached newest-first listing.
+
+    `cached` and `current` are the (raw id, status) of the first page of each
+    listing. The answer is a count when the current page is the cached one
+    with new raws before it and nothing else changed; None when anything else
+    happened (a delete, a status change, a raw dated into the middle), which
+    needs a full re-listing.
+    """
+    added = current_total - cached_total
+    if added <= 0 or added > len(current):
+        return None
+    kept = current[added:]
+    if not kept and cached_total:
+        return None
+    if kept != cached[: len(kept)]:
+        return None
+    return added
+
+
+_WORD = re.compile(r"\w+")
+
+
+def _query_terms(query: str) -> frozenset[str]:
+    """Words of the query, plus character bigrams of unspaced (CJK) words."""
+    terms: set[str] = set()
+    for word in _WORD.findall(query.casefold()):
+        terms.add(word)
+        if not word.isascii() and len(word) > 1:
+            terms.update(word[index : index + 2] for index in range(len(word) - 1))
+    return frozenset(terms)
+
+
+def turns_for_hit(turns: tuple[RawDisplayTurn, ...], query: str) -> tuple[RawDisplayTurn, ...]:
+    """The turns one semantic raw hit shows, never none while it has turns.
+
+    The SDK embeds a whole raw batch, so a vector hit cannot say which of its
+    turns matched. Every turn containing the query text is shown; when none
+    does, the turn sharing the most query terms (the earliest on a tie) stands
+    for the hit. A semantic hit is never dropped for lacking the literal query.
+    """
+    if not turns:
+        return ()
+    needle = query.casefold().strip()
+    if needle:
+        lexical = tuple(turn for turn in turns if needle in turn.content.casefold())
+        if lexical:
+            return lexical
+    terms = _query_terms(query)
+
+    def overlap(turn: RawDisplayTurn) -> tuple[int, int]:
+        content = turn.content.casefold()
+        return (sum(1 for term in terms if term in content), -turn.turn_index)
+
+    return (max(turns, key=overlap),)

memhub/apps/core/tests/TEST_INDEX.md

@@ -123,6 +123,7 @@
 | `test_import_api.py` | `integration` | 覆盖 resume during cleanup returns conflict without internal error、authorize detect import and revoke api flow、protected api rejects missing token and foreign origin 等 5 个场景。 |
 | `test_import_control_races_review.py` | `contract` | Read-only review repro: real runner/store, synthetic delayed adapter cleanup. |
 | `test_import_evidence_atomicity.py` | `contract` | 覆盖 counter failure rolls back evidence versions and cursor、same identity cursor rewind is blocked and logged、replaced identity may reset checkpoint 等 20 个场景。 |
+| `test_import_job_retention.py` | `integration` | `import_jobs` stays bounded, and the questions asked of it are SQL, not scans. |
 | `test_import_oversized_verbatim.py` | `contract` | 覆盖 normal verbatim is unchanged、oversized verbatim is truncated instead of rejected 场景。 |
 | `test_import_runner.py` | `contract` | 覆盖 job transitions are explicit、runner keeps per record errors and continues、sync slices reuse the same source listing 等 22 个场景。 |
 | `test_import_skip_batching.py` | `contract` | 覆盖 batch limit and tail use fewer commits、batched skips keep their order against the next write、flushes before every nonmatching event 等 14 个场景。 |
@@ -213,6 +214,7 @@
 | `test_onboarding_profile.py` | `integration` | R-09：首次安装引导把画像录下来，不问用户一句话。 |
 | `test_oversized_file_guard.py` | `contract` | One pass over a source file is bounded by a byte budget, not skipped. |
 | `test_parent_watchdog.py` | `integration` | A Core that outlives its launcher makes the next launch impossible. |
+| `test_pinned_sdk_raw_search.py` | `integration` | MemHub's raw search against the hy-memory wheel it pins, not the repo source. |
 | `test_policy_api_wiring.py` | `integration` | 覆盖 global model permission off does not backfill、backfill puts history on write jobs without scheduling memory tasks、source pause skips history until the source is reenabled 场景。 |
 | `test_policy_production_entry.py` | `contract` | 覆盖 production serve passes only supported app options、maintenance serve does not own the runtime 场景。 |
 | `test_precipitation_concurrency_setting.py` | `contract` | Rule 2/9: precipitation concurrency is configurable, not hardcoded to 1. |
@@ -321,4 +323,4 @@
 | `test_workspace_recall.py` | `contract` | Workspace memories were unreachable from every host; these pin the way in. |
 | `test_write_queue_lease_renewal.py` | `contract` | Rule 8: a lease must outlive the work it covers, and be renewed while it runs. |
 
-共 316 个测试文件。
+共 318 个测试文件。

memhub/apps/core/tests/core_test_classification.py

@@ -14,8 +14,8 @@ from pathlib import Path
 import pytest
 
 PRIMARY_MARKERS = ("unit", "contract", "integration", "performance", "live")
-EXPECTED_TEST_FILE_COUNT = 316
-EXPECTED_TEST_FILES_SHA256 = "011482974a26e9d43ba03bb88cbd168e645175beacd96c02ff642bc17dac49da"
+EXPECTED_TEST_FILE_COUNT = 318
+EXPECTED_TEST_FILES_SHA256 = "df8752566a0bf9e540f68b4948f3f5dbc409bd6836707b1330206e2349d70940"
 
 # Pure transformations, validation models, parity checks and state machines.
 # This list is intentionally small: uncertain files default to contract.
@@ -70,6 +70,7 @@ INTEGRATION_FILES = frozenset(
         "test_file_import_operations.py",
         "test_graph_width_endpoint.py",
         "test_import_api.py",
+        "test_import_job_retention.py",
         "test_live_capture.py",
         "test_live_watch_grants.py",
         "test_log_privacy.py",
@@ -93,6 +94,7 @@ INTEGRATION_FILES = frozenset(
         "test_model_profile_context_reset.py",
         "test_onboarding_profile.py",
         "test_parent_watchdog.py",
+        "test_pinned_sdk_raw_search.py",
         "test_policy_api_wiring.py",
         "test_recall_scope_control.py",
         "test_retrieval_lifespan.py",

memhub/apps/core/tests/test_health.py

@@ -150,7 +150,18 @@ def test_lifespan_content_backfill_failure_is_visible(tmp_path, monkeypatch) ->
 
     async def scenario() -> None:
         async with app.router.lifespan_context(app):
-            health = await _get_json(app, "/api/v1/health")
+            # The rebuild is released when post-ready recovery finishes, and
+            # recovery now resumes imports off the loop, so it can still be
+            # running at the first request.
+            for _ in range(500):
+                if app.state.recovery in {"done", "failed"}:
+                    break
+                await asyncio.sleep(0.01)
+            for _ in range(500):
+                health = await _get_json(app, "/api/v1/health")
+                if health["content_candidate_backfill"] != "running":
+                    break
+                await asyncio.sleep(0.01)
             assert health["content_candidate_backfill"] == "failed"
             assert health["content_candidate_backfill_failure"] == "database_locked"
 

memhub/apps/core/tests/test_import_api.py

@@ -34,10 +34,10 @@ def test_resume_during_cleanup_returns_conflict_without_internal_error(tmp_path,
             job = supervisor.job_store.create(user_id="local", source_id="fixture")
             supervisor.job_store.transition(job.job_id, ImportStatus.PAUSED)
 
-            def draining(_job_id):
+            async def draining(_job_id):
                 raise InvalidJobTransitionError("import_cleanup_in_progress")
 
-            monkeypatch.setattr(supervisor, "resume", draining)
+            monkeypatch.setattr(supervisor, "resume_off_loop", draining)
             response = await client.post(f"/api/v1/import-jobs/{job.job_id}/resume")
             assert response.status_code == 409
             assert response.json() == {"detail": "invalid_job_transition"}

memhub/apps/core/tests/test_import_job_retention.py

@@ -0,0 +1,383 @@
+"""`import_jobs` stays bounded, and the questions asked of it are SQL, not scans.
+
+Live sync opened a job per change it imported and nothing ever removed one, so
+the table only grew; every caller then read all of it and filtered in Python.
+These tests hold the three halves of the fix: what pruning keeps, that the
+filtered queries answer exactly what the old filters did, and that the indexes
+behind them arrive by migration.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import random
+import sqlite3
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from pydantic import SecretStr
+
+from one_memory.api import _prune_finished_import_jobs, create_app
+from one_memory.config import Settings
+from one_memory.imports.jobs import (
+    FINISHED_STATUSES,
+    SYNC_JOBS_KEPT_PER_SOURCE,
+    ImportJob,
+    ImportJobStore,
+    ImportStatus,
+)
+from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate
+from one_memory.memory_server.migrations import (
+    IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION,
+    MIGRATIONS,
+)
+
+_BASE = datetime(2026, 1, 1, tzinfo=UTC)
+
+
+def _store(tmp_path) -> ImportJobStore:
+    return ImportJobStore(tmp_path / "one-memory.sqlite3")
+
+
+def _job(
+    store: ImportJobStore,
+    *,
+    user_id: str = "local",
+    source_id: str = "fixture",
+    operation: str = "sync",
+    status: ImportStatus = ImportStatus.COMPLETED,
+    minute: int,
+    issue: bool = False,
+) -> str:
+    """Insert one row as its terminal shape, at a chosen creation time."""
+    created = (_BASE + timedelta(minutes=minute)).isoformat()
+    job_id = f"{source_id}-{user_id}-{operation}-{minute:05d}"
+    with store._connect(writing=True) as connection:
+        connection.execute(
+            "INSERT INTO import_jobs (job_id,user_id,source_id,status,created_at,updated_at,"
+            "operation) VALUES (?,?,?,?,?,?,?)",
+            (job_id, user_id, source_id, status, created, created, operation),
+        )
+        if issue:
+            connection.execute(
+                "INSERT INTO import_issues (job_id,kind,conversation_id,reason,created_at) "
+                "VALUES (?,'error',NULL,'synthetic_reason',?)",
+                (job_id, created),
+            )
+            connection.execute(
+                "INSERT INTO import_skip_reason_counts (job_id,reason,count) VALUES (?,'r',1)",
+                (job_id,),
+            )
+    return job_id
+
+
+def _ids(store: ImportJobStore, **filters) -> list[str]:
+    return [job.job_id for job in store.find_jobs(**filters)]
+
+
+def _count(store: ImportJobStore, table: str, job_id: str) -> int:
+    with store._connect() as connection:
+        return int(
+            connection.execute(
+                f"SELECT COUNT(*) FROM {table} WHERE job_id=?", (job_id,)
+            ).fetchone()[0]
+        )
+
+
+def test_pruning_keeps_the_newest_finished_syncs_of_every_source(tmp_path) -> None:
+    store = _store(tmp_path)
+    keep = 5
+    for source in ("alpha", "beta"):
+        for minute in range(12):
+            status = ImportStatus.CANCELLED if minute % 4 == 0 else ImportStatus.COMPLETED
+            _job(store, source_id=source, minute=minute, status=status, issue=True)
+    # A source with fewer than `keep` keeps all of them.
+    for minute in range(3):
+        _job(store, source_id="gamma", minute=minute)
+    # Never pruned: the active sync, history of any state, another user's lane.
+    running = _job(store, source_id="alpha", minute=1000, status=ImportStatus.RUNNING)
+    failed_history = _job(
+        store, source_id="alpha", operation="history", minute=2, status=ImportStatus.FAILED
+    )
+    histories = [
+        _job(store, source_id="alpha", operation="history", minute=100 + minute)
+        for minute in range(10)
+    ]
+    other_user = [_job(store, user_id="other", source_id="alpha", minute=m) for m in range(8)]
+
+    removed = store.prune_finished_sync_jobs(keep=keep)
+
+    assert removed == (12 - keep) * 2 + (8 - keep)
+    for source in ("alpha", "beta"):
+        finished = _ids(
+            store, user_id="local", source_id=source, operation="sync", statuses=FINISHED_STATUSES
+        )
+        assert finished == [f"{source}-local-sync-{m:05d}" for m in range(12 - keep, 12)]
+        # Their issues and skip counts went with them; the kept ones still have theirs.
+        assert _count(store, "import_issues", f"{source}-local-sync-00000") == 0
+        assert _count(store, "import_skip_reason_counts", f"{source}-local-sync-00000") == 0
+        assert _count(store, "import_issues", f"{source}-local-sync-00011") == 1
+    assert len(_ids(store, source_id="gamma")) == 3
+    assert store.require(running).status is ImportStatus.RUNNING
+    assert store.require(failed_history).status is ImportStatus.FAILED
+    assert all(store.get(job_id) is not None for job_id in histories)
+    assert _ids(store, user_id="other") == other_user[-keep:]
+    # The newest job of every lane, which the App and live sync health show,
+    # is always kept.
+    assert store.latest_job(user_id="local", source_id="beta").job_id == "beta-local-sync-00011"
+
+    before = _ids(store)
+    assert store.prune_finished_sync_jobs(keep=keep) == 0
+    assert _ids(store) == before
+
+
+def test_pruning_is_bounded_per_call_and_finishes_when_repeated(tmp_path) -> None:
+    store = _store(tmp_path)
+    for minute in range(60):
+        _job(store, minute=minute)
+
+    passes = []
+    while True:
+        removed = store.prune_finished_sync_jobs(keep=10, max_rows=15)
+        passes.append(removed)
+        if removed < 15:
+            break
+
+    assert passes == [15, 15, 15, 5]
+    assert len(_ids(store)) == 10
+
+
+def test_the_default_keep_count_is_twenty(tmp_path) -> None:
+    store = _store(tmp_path)
+    for minute in range(SYNC_JOBS_KEPT_PER_SOURCE + 7):
+        _job(store, minute=minute)
+    assert store.prune_finished_sync_jobs() == 7
+    assert len(_ids(store)) == SYNC_JOBS_KEPT_PER_SOURCE == 20
+
+
+def test_startup_pruning_runs_to_completion_off_the_loop(tmp_path) -> None:
+    store = _store(tmp_path)
+    for minute in range(500):
+        _job(store, minute=minute)
+
+    async def scenario() -> None:
+        task = asyncio.create_task(_prune_finished_import_jobs(store, interval_seconds=3600))
+        for _ in range(500):
+            await asyncio.sleep(0.01)
+            if len(await asyncio.to_thread(_ids, store)) == SYNC_JOBS_KEPT_PER_SOURCE:
+                break
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    asyncio.run(scenario())
+    assert len(_ids(store)) == SYNC_JOBS_KEPT_PER_SOURCE
+
+
+def test_filtered_queries_answer_what_the_python_filters_did(tmp_path) -> None:
+    store = _store(tmp_path)
+    rng = random.Random(7)
+    lanes = [(user, source) for user in ("local", "other") for source in ("a", "b", "c")]
+    for minute in range(240):
+        user, source = rng.choice(lanes)
+        _job(
+            store,
+            user_id=user,
+            source_id=source,
+            operation=rng.choice(("sync", "history")),
+            # Only terminal shapes: the active-slot index allows one active job per lane.
+            status=rng.choice(FINISHED_STATUSES),
+            minute=minute,
+        )
+    everything: list[ImportJob] = store.list_jobs()
+    assert len(everything) == 240
+
+    status_sets = [
+        None,
+        (ImportStatus.COMPLETED,),
+        (ImportStatus.CANCELLED, ImportStatus.COMPLETED),
+        (ImportStatus.RUNNING,),
+    ]
+    for user_id in (None, "local", "other"):
+        for source_id in (None, "a", "c"):
+            for operation in (None, "sync", "history"):
+                for statuses in status_sets:
+                    expected = [
+                        job
+                        for job in everything
+                        if (user_id is None or job.user_id == user_id)
+                        and (source_id is None or job.source_id == source_id)
+                        and (operation is None or job.operation == operation)
+                        and (statuses is None or job.status in statuses)
+                    ]
+                    found = store.find_jobs(
+                        user_id=user_id,
+                        source_id=source_id,
+                        operation=operation,
+                        statuses=statuses,
+                    )
+                    assert found == expected
+                    newest = store.find_jobs(
+                        user_id=user_id,
+                        source_id=source_id,
+                        operation=operation,
+                        statuses=statuses,
+                        newest_first=True,
+                        limit=1,
+                    )
+                    assert newest == (
+                        [max(expected, key=lambda job: (job.created_at, job.job_id))]
+                        if expected
+                        else []
+                    )
+    for status in ImportStatus:
+        assert store.list_jobs(status=status) == [job for job in everything if job.status is status]
+    assert store.find_jobs(statuses=()) == []
+
+
+def test_the_lookup_indexes_arrive_by_migration_and_serve_the_lane_query(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    older = tuple(
+        item for item in MIGRATIONS if item.version < IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION
+    )
+    asyncio.run(migrate(SqliteMigrationStore(path), older))
+    with sqlite3.connect(path) as database:
+        before = {
+            row[0] for row in database.execute("SELECT name FROM sqlite_master WHERE type='index'")
+        }
+    assert "import_jobs_lane" not in before
+
+    result = asyncio.run(migrate(SqliteMigrationStore(path), MIGRATIONS))
+
+    assert result.applied == (IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION,)
+    with sqlite3.connect(path) as database:
+        indexes = {
+            row[0] for row in database.execute("SELECT name FROM sqlite_master WHERE type='index'")
+        }
+        plan = " ".join(
+            str(row[-1])
+            for row in database.execute(
+                "EXPLAIN QUERY PLAN SELECT * FROM import_jobs "
+                "WHERE user_id=? AND source_id=? AND operation=? AND status IN (?,?) "
+                "ORDER BY created_at DESC LIMIT 1",
+                ("local", "fixture", "sync", "pending", "running"),
+            )
+        )
+    assert {"import_jobs_lane", "import_jobs_status"} <= indexes
+    assert "import_jobs_lane" in plan
+    # Re-running it is a no-op, and a store opened on the database agrees.
+    asyncio.run(migrate(SqliteMigrationStore(path), MIGRATIONS))
+    ImportJobStore(path)
+
+
+def test_a_running_home_gets_the_indexes_from_the_store(tmp_path) -> None:
+    path = tmp_path / "one-memory.sqlite3"
+    ImportJobStore(path)
+    with sqlite3.connect(path) as database:
+        database.execute("DROP INDEX import_jobs_lane")
+        database.execute("DROP INDEX import_jobs_status")
+    ImportJobStore(path)
+    with sqlite3.connect(path) as database:
+        indexes = {
+            row[0] for row in database.execute("SELECT name FROM sqlite_master WHERE type='index'")
+        }
+    assert {"import_jobs_lane", "import_jobs_status"} <= indexes
+
+
+def test_the_mysql_rendering_tolerates_an_index_already_there() -> None:
+    class Cursor:
+        dialect = "mysql"
+
+        def __init__(self) -> None:
+            self.attempted: list[str] = []
+
+        async def execute(self, statement: str) -> None:
+            self.attempted.append(statement)
+            raise RuntimeError(1061, "duplicate key name")
+
+    cursor = Cursor()
+    migration = next(
+        item for item in MIGRATIONS if item.version == IMPORT_JOB_LOOKUP_INDEXES_MIGRATION_VERSION
+    )
+    asyncio.run(migration.apply(cursor))
+    assert cursor.attempted and "import_jobs" in cursor.attempted[0]
+
+
+def test_the_import_jobs_route_keeps_its_shape_and_order(tmp_path) -> None:
+    async def scenario() -> None:
+        app = create_app(
+            Settings(
+                data_dir=tmp_path / "data",
+                source_home=tmp_path / "home",
+                api_token=SecretStr("synthetic"),
+                automatic_memory_processing=False,
+                live_sync_enabled=False,
+            )
+        )
+        async with (
+            app.router.lifespan_context(app),
+            AsyncClient(
+                transport=ASGITransport(app=app),
+                base_url="http://test",
+                headers={"Authorization": "Bearer synthetic"},
+            ) as client,
+        ):
+            store: ImportJobStore = app.state.import_job_store
+            for minute in (3, 1, 2):
+                _job(store, source_id="fixture", minute=minute)
+            _job(store, source_id="fixture", operation="history", minute=0)
+            _job(store, user_id="other", source_id="hermes", minute=4)
+
+            response = await client.get("/api/v1/import-jobs")
+            other = await client.get("/api/v1/import-jobs", params={"user_id": "other"})
+
+            assert response.status_code == 200
+            body = response.json()
+            expected = [job for job in store.list_jobs() if job.user_id == "local"]
+            assert [item["job_id"] for item in body] == [job.job_id for job in expected]
+            assert [item["created_at"] for item in body] == sorted(
+                item["created_at"] for item in body
+            )
+            assert set(body[0]) == set(ImportJob.model_fields)
+            assert [item["user_id"] for item in other.json()] == ["other"]
+
+    asyncio.run(scenario())
+
+
+def test_memory_processing_reports_the_write_drain(tmp_path) -> None:
+    async def scenario() -> None:
+        app = create_app(
+            Settings(
+                data_dir=tmp_path / "data",
+                source_home=tmp_path / "home",
+                api_token=SecretStr("synthetic"),
+                automatic_memory_processing=False,
+                live_sync_enabled=False,
+            )
+        )
+        async with (
+            app.router.lifespan_context(app),
+            AsyncClient(
+                transport=ASGITransport(app=app),
+                base_url="http://test",
+                headers={"Authorization": "Bearer synthetic"},
+            ) as client,
+        ):
+            for _ in range(500):
+                if app.state.recovery in {"done", "failed"}:
+                    break
+                await asyncio.sleep(0.01)
+            body = (await client.get("/api/v1/memory-processing")).json()
+            assert body["write_drain"] == "running"
+            assert body["write_drain_error"] is None
+
+            drain = app.state.memory_write_drain
+            drain.liveness = "restarting"
+            drain.last_error_code = "operational_error"
+            body = (await client.get("/api/v1/memory-processing")).json()
+            assert body["write_drain"] == "restarting"
+            assert body["write_drain_error"] == "operational_error"
+            drain.liveness = "running"
+
+    asyncio.run(scenario())

memhub/apps/core/tests/test_import_supervisor.py

@@ -415,15 +415,15 @@ def test_shutdown_preserves_a_job_that_finished_after_the_running_snapshot(
         task = asyncio.create_task(waiting())
         supervisor._tasks[job.job_id] = task
         await asyncio.sleep(0)
-        original = jobs.list_jobs
+        original = jobs.find_jobs
 
         def finish_after_snapshot(*args, **kwargs):
             found = original(*args, **kwargs)
-            if kwargs.get("status") is ImportStatus.RUNNING:
+            if tuple(kwargs.get("statuses") or ()) == (ImportStatus.RUNNING,):
                 jobs.finish_owned(job.job_id, runner_token=token, target=terminal)
             return found
 
-        monkeypatch.setattr(jobs, "list_jobs", finish_after_snapshot)
+        monkeypatch.setattr(jobs, "find_jobs", finish_after_snapshot)
         try:
             await supervisor.shutdown()
             assert jobs.require(job.job_id).status is terminal
@@ -433,3 +433,45 @@ def test_shutdown_preserves_a_job_that_finished_after_the_running_snapshot(
             await asyncio.gather(task, return_exceptions=True)
 
     asyncio.run(scenario())
+
+
+def test_abandon_makes_its_database_calls_off_the_event_loop(tmp_path, monkeypatch) -> None:
+    """Live sync abandons from its poll; a writer wait there froze the whole Core."""
+    import threading
+
+    async def scenario() -> None:
+        evidence, jobs = stores(tmp_path)
+        supervisor = ImportSupervisor(adapters={}, evidence_store=evidence, job_store=jobs)
+        job = jobs.create(user_id="local", source_id="fixture", operation="sync")
+        job, _token = jobs.claim(job.job_id)
+        loop_thread = threading.current_thread()
+        calls: list[tuple[str, threading.Thread]] = []
+        for name in ("require", "interrupt_running"):
+            original = getattr(jobs, name)
+
+            def recorded(*args, _name=name, _original=original, **kwargs):
+                calls.append((_name, threading.current_thread()))
+                return _original(*args, **kwargs)
+
+            monkeypatch.setattr(jobs, name, recorded)
+
+        async def wedged() -> None:
+            # Ignores the first cancel, like a runner stuck in a thread.
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                await asyncio.sleep(0.2)
+
+        task = asyncio.create_task(wedged())
+        supervisor._tasks[job.job_id] = task
+        await asyncio.sleep(0)
+        try:
+            assert await supervisor.abandon(job.job_id, timeout=0.01) is False
+        finally:
+            await asyncio.gather(task, return_exceptions=True)
+
+        assert {name for name, _thread in calls} == {"require", "interrupt_running"}
+        assert all(thread is not loop_thread for _name, thread in calls)
+        assert jobs.require(job.job_id).status is ImportStatus.INTERRUPTED
+
+    asyncio.run(scenario())

memhub/apps/core/tests/test_live_watch.py

@@ -641,3 +641,32 @@ def _job(*, created_at: datetime) -> ImportJob:
         created_at=created_at,
         updated_at=datetime.now(UTC),
     )
+
+
+def test_attributing_changed_paths_happens_off_the_event_loop(tmp_path: Path, monkeypatch) -> None:
+    """Resolving a host's batch of changed paths blocked the loop for seconds."""
+    import threading
+
+    async def scenario() -> None:
+        root, (_evidence, live, imports) = _source(tmp_path)
+        watch = FileWatchLiveSync(
+            roots={PI_DESCRIPTOR.source_id: root},
+            supervisor=live,
+            debounce_seconds=0.1,
+            min_interval_seconds=0.1,
+        )
+        loop_thread = threading.current_thread()
+        resolved_on: list[threading.Thread] = []
+        original = watch.source_for
+
+        def recorded(path: Path) -> str | None:
+            resolved_on.append(threading.current_thread())
+            return original(path)
+
+        monkeypatch.setattr(watch, "source_for", recorded)
+
+        assert await watch.handle_changes({tmp_path / "elsewhere/session.jsonl"}) == ()
+        assert resolved_on and loop_thread not in resolved_on
+        await imports.shutdown()
+
+    asyncio.run(scenario())

memhub/apps/core/tests/test_main_database_access.py

@@ -318,9 +318,31 @@ def test_main_database_checkpoints_on_a_write_cadence(tmp_path, monkeypatch):
     with database.write() as db, db:
         db.execute("UPDATE counter SET value = 2 WHERE id = 1")
 
+    assert _access.wait_for_cadence_checkpoints()
     assert checkpoints and checkpoints[0][0] == "PASSIVE"
 
 
+def test_the_cadence_checkpoint_does_not_run_on_the_writing_thread(tmp_path, monkeypatch):
+    """The thread that wrote returns at once; the checkpointer takes the next turn."""
+    path = _database(tmp_path)
+    database = main_database(path)
+    ran_on = []
+    original = database._checkpoint_from_maintenance
+
+    def observed(mode):
+        ran_on.append(threading.current_thread().name)
+        return original(mode)
+
+    monkeypatch.setattr(database, "_checkpoint_from_maintenance", observed)
+    monkeypatch.setattr(_access, "MAIN_DATABASE_CHECKPOINT_WRITE_INTERVAL", 1)
+
+    with database.write() as db, db:
+        db.execute("UPDATE counter SET value = 1 WHERE id = 1")
+
+    assert _access.wait_for_cadence_checkpoints()
+    assert ran_on == ["one-memory-checkpoint"]
+
+
 def test_main_database_warns_when_wal_stays_above_the_high_water_mark(
     tmp_path, monkeypatch, caplog
 ):
@@ -332,6 +354,7 @@ def test_main_database_warns_when_wal_stays_above_the_high_water_mark(
 
     with database.write() as db, db:
         db.execute("UPDATE counter SET value = 1 WHERE id = 1")
+    assert _access.wait_for_cadence_checkpoints()
 
     assert any("main_database_wal_high" in record.message for record in caplog.records)
 
@@ -349,6 +372,7 @@ def test_a_wal_whose_frames_are_all_checkpointed_is_not_reported_high(
 
     with database.write() as db, db:
         db.execute("UPDATE counter SET value = 1 WHERE id = 1")
+    assert _access.wait_for_cadence_checkpoints()
 
     assert database._wal_bytes() > 0
     assert not any("main_database_wal_high" in record.message for record in caplog.records)

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -4138,6 +4138,71 @@ def test_a_failing_renewal_does_not_kill_the_drain(tmp_path) -> None:
     assert queue.get("idem-1").state == COMPLETED
 
 
+def test_a_locked_fallback_settlement_does_not_kill_the_drain(tmp_path, monkeypatch) -> None:
+    """The incident shape: executor and its fallback both lose the writer.
+
+    The first row keeps its lease for normal reclaim; an unrelated row must
+    still run instead of being stranded behind a dead drain task.
+    """
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    first = queue.submit(_request(key="first", source="cursor"))
+    second = queue.submit(_request(key="second", source="codex"))
+    executed: list[str] = []
+
+    async def execute(job_id: str) -> None:
+        executed.append(job_id)
+        if job_id == first.job_id:
+            raise RuntimeError("executor failed")
+        queue.settle(job_id, state=COMPLETED)
+        drain.stop()
+
+    drain = DurableWriteDrain(queue, execute=execute, max_concurrent_sources=1)
+
+    def locked_settlement(*_args, **_kwargs):
+        raise sqlite3.OperationalError("database is locked")
+
+    monkeypatch.setattr(queue, "settle_worker_error", locked_settlement)
+
+    asyncio.run(asyncio.wait_for(drain.run_forever(idle_pause=0.005), 1))
+
+    assert executed == [first.job_id, second.job_id]
+    assert queue.get("first").state == RUNNING
+    assert queue.get("second").state == COMPLETED
+
+
+def test_an_unexpected_job_task_failure_is_contained_by_the_drain(
+    tmp_path, monkeypatch, caplog
+) -> None:
+    """A programming error stays visible without taking down the consumer."""
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    first = queue.submit(_request(key="first", source="cursor"))
+    second = queue.submit(_request(key="second", source="codex"))
+    executed: list[str] = []
+
+    async def run_job(job_id: str) -> None:
+        executed.append(job_id)
+        if job_id == first.job_id:
+            raise RuntimeError("unexpected job task failure")
+        drain.stop()
+
+    drain = DurableWriteDrain(
+        queue,
+        execute=_never_execute,
+        max_concurrent_sources=1,
+    )
+    monkeypatch.setattr(drain, "_run_claimed_job", run_job)
+
+    with caplog.at_level(logging.ERROR, logger="one_memory.memory_server.durable_queue"):
+        asyncio.run(asyncio.wait_for(drain.run_forever(idle_pause=0.005), 1))
+
+    assert executed == [first.job_id, second.job_id]
+    assert any(
+        "durable_write_drain_job_task_failed" in message.getMessage()
+        and "error_type=RuntimeError" in message.getMessage()
+        for message in caplog.records
+    )
+
+
 @pytest.mark.parametrize("terminal", [COMPLETED, FAILED, DEAD])
 def test_a_terminal_row_is_sticky_against_a_late_settle(tmp_path, terminal: str) -> None:
     """A late settle must not overwrite an outcome that already happened.
@@ -4980,6 +5045,27 @@ def test_permanent_renewal_error_stops_the_worker(tmp_path, monkeypatch) -> None
         asyncio.run(drain.run_one())
 
 
+def test_permanent_renewal_error_stops_the_long_lived_drain(tmp_path, monkeypatch) -> None:
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    queue.submit(_request())
+    drain = DurableWriteDrain(
+        queue,
+        execute=lambda _job_id: asyncio.sleep(0.03),
+        lease=timedelta(seconds=1),
+        renew_every=timedelta(milliseconds=10),
+    )
+    monkeypatch.setattr(
+        drain,
+        "renew",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            sqlite3.OperationalError("database schema is corrupt")
+        ),
+    )
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(asyncio.wait_for(drain.run_forever(idle_pause=0.005), 1))
+
+
 def test_lost_renewal_waits_for_execution_before_releasing_source(tmp_path, monkeypatch) -> None:
     path = tmp_path / "one-memory.sqlite3"
     queue = DurableWriteQueue(path)
@@ -5442,3 +5528,190 @@ def test_a_disabled_source_suspends_its_jobs_and_enabling_it_resumes_them(tmp_pa
     assert runtime.status_counts() == {PENDING: 3, COMPLETED: 1}
     assert runtime.queue_depths().fresh == 3
     assert len(drain.claim(limit=10)) == 1
+
+
+def test_a_claim_path_error_restarts_the_drain_and_later_jobs_still_run(
+    tmp_path, monkeypatch
+) -> None:
+    """An unexpected SQLite error on the claim path used to end the drain for good."""
+    from one_memory.memory_server import durable_queue as _queue_module
+
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    queue.submit(_request(key="later", source="cursor"))
+    reported: list[tuple[str, BaseException]] = []
+    monkeypatch.setattr(
+        _queue_module,
+        "report_exception",
+        lambda component, event, error, **_kwargs: reported.append((event, error)),
+    )
+
+    async def execute(job_id: str) -> None:
+        queue.settle(job_id, state=COMPLETED)
+        drain.stop()
+
+    drain = DurableWriteDrain(queue, execute=execute, max_concurrent_sources=1)
+    real_claim = drain.claim
+    calls = {"claim": 0}
+
+    def failing_once(**kwargs):
+        calls["claim"] += 1
+        if calls["claim"] == 1:
+            raise sqlite3.OperationalError("disk I/O error")
+        return real_claim(**kwargs)
+
+    monkeypatch.setattr(drain, "claim", failing_once)
+    first_owner = drain.owner
+
+    asyncio.run(
+        asyncio.wait_for(
+            drain.run_supervised(idle_pause=0.005, restart_backoff=0.01, max_restart_backoff=0.05),
+            2,
+        )
+    )
+
+    assert queue.get("later").state == COMPLETED
+    assert drain.restarts == 1
+    assert drain.last_error_code == "operational_error"
+    assert drain.liveness == "stopped"
+    # A restarted drain leases under a new name: nothing the old one left in a
+    # pool thread can renew or settle as it.
+    assert drain.owner != first_owner
+    assert [event for event, _error in reported] == ["memory.write_queue.drain_exited"]
+    assert reported[0][1].__traceback__ is not None
+
+
+def test_a_claim_gate_exception_restarts_the_drain(tmp_path, monkeypatch) -> None:
+    from one_memory.memory_server import durable_queue as _queue_module
+
+    monkeypatch.setattr(_queue_module, "report_exception", lambda *_args, **_kwargs: None)
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    queue.submit(_request(key="gated", source="cursor"))
+    asked = {"count": 0}
+
+    def gate(_source_key: str) -> bool:
+        asked["count"] += 1
+        if asked["count"] == 1:
+            raise KeyError("gate bug")
+        return True
+
+    async def execute(job_id: str) -> None:
+        queue.settle(job_id, state=COMPLETED)
+        drain.stop()
+
+    drain = DurableWriteDrain(queue, execute=execute, source_claim_enabled=gate)
+
+    asyncio.run(asyncio.wait_for(drain.run_supervised(idle_pause=0.005, restart_backoff=0.01), 2))
+
+    assert queue.get("gated").state == COMPLETED
+    assert drain.last_error_code == "key_error"
+
+
+def test_stop_ends_a_restart_backoff_at_once(tmp_path, monkeypatch) -> None:
+    from one_memory.memory_server import durable_queue as _queue_module
+
+    monkeypatch.setattr(_queue_module, "report_exception", lambda *_args, **_kwargs: None)
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    drain = DurableWriteDrain(queue, execute=_never_execute)
+
+    def broken(**_kwargs):
+        raise sqlite3.DatabaseError("database disk image is malformed")
+
+    monkeypatch.setattr(drain, "claim", broken)
+
+    async def scenario() -> None:
+        task = asyncio.create_task(drain.run_supervised(restart_backoff=30))
+        for _ in range(200):
+            if drain.liveness == "restarting":
+                break
+            await asyncio.sleep(0.005)
+        assert drain.liveness == "restarting"
+        drain.stop()
+        await asyncio.wait_for(task, 1)
+
+    asyncio.run(scenario())
+    assert drain.liveness == "stopped"
+    assert drain.last_error_code == "database_error"
+
+
+def test_a_cancelled_supervised_drain_stays_cancelled(tmp_path) -> None:
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    drain = DurableWriteDrain(queue, execute=_never_execute)
+
+    async def scenario() -> None:
+        task = asyncio.create_task(drain.run_supervised(idle_pause=0.005))
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    asyncio.run(scenario())
+    assert drain.liveness == "stopped"
+    assert drain.restarts == 0
+
+
+def test_a_contained_job_task_failure_is_reported_with_its_stack(tmp_path, monkeypatch) -> None:
+    from one_memory.memory_server import durable_queue as _queue_module
+
+    reported: list[tuple[str, BaseException]] = []
+    monkeypatch.setattr(
+        _queue_module,
+        "report_exception",
+        lambda component, event, error, **_kwargs: reported.append((event, error)),
+    )
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    first = queue.submit(_request(key="first", source="cursor"))
+    queue.submit(_request(key="second", source="codex"))
+
+    async def run_job(job_id: str) -> None:
+        if job_id == first.job_id:
+            raise RuntimeError("systematic job task bug")
+        drain.stop()
+
+    drain = DurableWriteDrain(queue, execute=_never_execute, max_concurrent_sources=1)
+    monkeypatch.setattr(drain, "_run_claimed_job", run_job)
+
+    asyncio.run(asyncio.wait_for(drain.run_forever(idle_pause=0.005), 1))
+
+    assert [event for event, _error in reported] == ["memory.write_queue.job_task_failed"]
+    assert reported[0][1].__traceback__ is not None
+
+
+def test_a_lease_is_renewed_while_every_bulk_thread_is_blocked(tmp_path) -> None:
+    """The incident: bulk threads stuck in job listings starved lease renewal."""
+    from one_memory.background_work import BULK_WORKERS, bulk_pool, shutdown_bulk_pool
+
+    queue = DurableWriteQueue(tmp_path / "one-memory.sqlite3")
+    queue.submit(_request(key="slow", source="cursor"))
+    release = threading.Event()
+    blocked = [bulk_pool().submit(release.wait, 10) for _ in range(BULK_WORKERS * 2)]
+    renewed: list[bool] = []
+
+    async def execute(job_id: str) -> None:
+        # Longer than the lease: only renewals keep it.
+        await asyncio.sleep(0.8)
+        queue.settle(job_id, state=COMPLETED, owner=drain.owner)
+
+    drain = DurableWriteDrain(
+        queue,
+        execute=execute,
+        lease=timedelta(milliseconds=400),
+        renew_every=timedelta(milliseconds=50),
+    )
+    real_renew = drain.renew
+
+    def counted(job_id: str, **kwargs) -> bool:
+        result = real_renew(job_id, **kwargs)
+        renewed.append(result)
+        return result
+
+    drain.renew = counted  # type: ignore[method-assign]
+    try:
+        assert asyncio.run(asyncio.wait_for(drain.run_one(), 3)) == 1
+    finally:
+        release.set()
+        for future in blocked:
+            future.result(timeout=5)
+        shutdown_bulk_pool()
+
+    assert renewed and all(renewed)
+    assert queue.get("slow").state == COMPLETED

memhub/apps/core/tests/test_pinned_sdk_raw_search.py

@@ -0,0 +1,237 @@
+"""MemHub's raw search against the hy-memory wheel it pins, not the repo source.
+
+The suite's pythonpath puts the repository root first, so `import hy_memory`
+in a test reads the repo's `hy_memory/`, which can be ahead of the pinned
+release. Search kwargs the release does not accept then pass every test and
+fail every raw search in a packaged build (a `statuses` argument did exactly
+that against 2.0.13). These tests run in a child interpreter started with
+`-I`, so `hy_memory` resolves to the wheel installed in the Core venv, and
+drive the real routes over a real lite SDK client with a deterministic local
+embedder (no model, no network, no ONNX assets).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+from tests.test_sdk_api_surface import SDK_SEARCH_PARAMETERS
+
+pytestmark = pytest.mark.integration
+
+_CHILD = textwrap.dedent(
+    r'''
+    import asyncio
+    import hashlib
+    import inspect
+    import json
+    import math
+    import re
+    import sys
+    from importlib.metadata import version
+    from pathlib import Path
+
+    import hy_memory
+    from hy_memory import HyMemoryClient
+
+    from one_memory.hy_runtime_config import HY_MEMORY_VERSION
+
+    home = Path(sys.argv[1])
+    wheel = Path(hy_memory.__file__).resolve()
+    assert "site-packages" in wheel.parts, f"not the installed wheel: {wheel}"
+    assert version("hy-memory") == HY_MEMORY_VERSION, version("hy-memory")
+    parameters = sorted(
+        name
+        for name, parameter in inspect.signature(HyMemoryClient.search).parameters.items()
+        if name not in {"self", "query"} and parameter.kind is not inspect.Parameter.VAR_KEYWORD
+    )
+    var_keyword = any(
+        parameter.kind is inspect.Parameter.VAR_KEYWORD
+        for parameter in inspect.signature(HyMemoryClient.search).parameters.values()
+    )
+
+    from httpx import ASGITransport, AsyncClient
+    from pydantic import SecretStr
+
+    import one_memory.api as api
+    from one_memory.api import create_app
+    from one_memory.config import Settings
+    from one_memory.embedding_assets import BUILTIN_EMBEDDING_DIMENSIONS
+    from one_memory.hy_sdk_runtime import (
+        LocalOnnxEmbeddingService,
+        SdkClientHandle,
+        builtin_retrieval_snapshot,
+    )
+    from one_memory.secrets import InMemorySecretStore
+
+    WORD = re.compile(r"\w+")
+
+    class HashedEmbedder:
+        """Bag of words and CJK bigrams, hashed: texts sharing words are near."""
+
+        dimensions = BUILTIN_EMBEDDING_DIMENSIONS
+
+        def encode(self, texts):
+            vectors = []
+            for text in texts:
+                vector = [0.0] * self.dimensions
+                for word in WORD.findall(text.casefold()):
+                    grams = [word]
+                    if not word.isascii():
+                        grams += [word[i : i + 2] for i in range(len(word) - 1)]
+                    for gram in grams:
+                        digest = hashlib.sha256(gram.encode()).digest()
+                        vector[int.from_bytes(digest[:4], "big") % self.dimensions] += 1.0
+                norm = math.sqrt(sum(value * value for value in vector)) or 1.0
+                vectors.append([value / norm for value in vector])
+            return vectors
+
+    LocalOnnxEmbeddingService._embedder = HashedEmbedder()
+
+    async def lite_has_no_model(_app):
+        return False
+
+    api._recall_model_ready = lite_has_no_model
+
+    async def main():
+        settings = Settings(
+            data_dir=home / "data",
+            source_home=home,
+            api_token=SecretStr("pinned-sdk"),
+            imports_enabled=False,
+            automatic_memory_processing=False,
+            live_sync_enabled=False,
+        )
+        app = create_app(settings, secret_store=InMemorySecretStore())
+        app.state.memory_mode = lambda: "lite"
+        handle = SdkClientHandle(
+            builtin_retrieval_snapshot(),
+            api_key=None,
+            data_dir=home / "data",
+            mode="lite",
+        )
+        result = {"parameters": parameters, "var_keyword": var_keyword}
+        async with app.router.lifespan_context(app):
+            app.state.sdk_client_handle = handle
+            client = await handle.client()
+            for session, messages in (
+                (
+                    "s-hike",
+                    [
+                        ("user", "我叫小林"),
+                        ("assistant", "你好小林"),
+                        ("user", "I love hiking mountain trails at dawn"),
+                        ("assistant", "Noted, early hikes."),
+                    ],
+                ),
+                (
+                    "s-tea",
+                    [
+                        ("user", "my favourite drink is green tea"),
+                        ("assistant", "Green tea, noted."),
+                    ],
+                ),
+            ):
+                written = client.add(
+                    [{"role": role, "content": content} for role, content in messages],
+                    user_id="local",
+                    source="hermes",
+                    session_id=session,
+                )
+                assert written.get("success", True), written
+            listed = client.list_memories(user_id="local", layers=["l1_raw"], limit=10)
+            rows = listed.get("vdb", listed)["memories"]
+            tea = next(row for row in rows if "green tea" in row["content"])
+            # What pro leaves behind: the raw it already processed is shadow.
+            client._loop_thread.run(
+                client._vector_store.update_payload(tea["memory_id"], {"status": "shadow"})
+            )
+            async with AsyncClient(
+                transport=ASGITransport(app=app),
+                base_url="http://test",
+                headers={"Authorization": "Bearer pinned-sdk"},
+            ) as http:
+                for name, body in (
+                    ("hike_turn", {"query": "hiking trails mountain", "raw_unit": "turn"}),
+                    ("tea_turn", {"query": "favourite green tea drink", "raw_unit": "turn"}),
+                    ("tea_raw", {"query": "favourite green tea drink"}),
+                ):
+                    response = await http.post(
+                        "/api/v2/context/search", json={"kinds": ["raw"], "limit": 5, **body}
+                    )
+                    result[name] = {"status": response.status_code, "body": response.json()}
+                response = await http.post(
+                    "/api/v1/recall", json={"query": "favourite green tea drink"}
+                )
+                result["recall"] = {"status": response.status_code, "body": response.json()}
+        await handle.close()
+        print("RESULT " + json.dumps(result, ensure_ascii=False))
+
+    asyncio.run(main())
+    '''
+)
+
+
+@pytest.fixture(scope="module")
+def pinned(tmp_path_factory) -> dict:
+    root = tmp_path_factory.mktemp("pinned-sdk")
+    script = root / "scripts" / "child.py"
+    script.parent.mkdir()
+    script.write_text(_CHILD, encoding="utf-8")
+    home = root / "home"
+    home.mkdir()
+    environment = {
+        **os.environ,
+        "ANONYMIZED_TELEMETRY": "False",
+        "ONE_MEMORY_DATA_DIR": str(home / "data"),
+    }
+    completed = subprocess.run(
+        [sys.executable, "-I", str(script), str(home)],
+        cwd=home,
+        env=environment,
+        capture_output=True,
+        text=True,
+        timeout=240,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr[-4000:]
+    [line] = [line for line in completed.stdout.splitlines() if line.startswith("RESULT ")]
+    return json.loads(line.removeprefix("RESULT "))
+
+
+def test_the_fake_sdk_accepts_exactly_the_pinned_search_parameters(pinned) -> None:
+    # The route tests' stand-in rejects what the pinned SDK rejects.
+    assert pinned["var_keyword"] is False
+    assert sorted(SDK_SEARCH_PARAMETERS) == pinned["parameters"]
+
+
+def test_lite_turn_search_returns_semantic_hits_without_the_literal_query(pinned) -> None:
+    hike = pinned["hike_turn"]
+    assert hike["status"] == 200, hike
+    items = hike["body"]["items"]
+    assert items, hike["body"]
+    top = items[0]
+    # No turn contains "hiking trails mountain"; the hit still comes back,
+    # shown by its closest turn and carrying the hit's score.
+    assert "hiking mountain trails" in top["content"]
+    assert "小林" not in top["content"]
+    assert top["turn_index"] == 1 and top["turn_count"] == 2
+    assert [item["score"] for item in items] == sorted(
+        (item["score"] for item in items), reverse=True
+    )
+
+
+def test_lite_search_and_recall_include_shadow_raws(pinned) -> None:
+    for name in ("tea_turn", "tea_raw"):
+        result = pinned[name]
+        assert result["status"] == 200, result
+        assert any("green tea" in item["content"] for item in result["body"]["items"]), result
+    recall = pinned["recall"]
+    assert recall["status"] == 200, recall
+    assert recall["body"]["mode"] == "raw"
+    assert any("green tea" in hit["content"] for hit in recall["body"]["hits"]), recall

memhub/apps/core/tests/test_policy_api_wiring.py

@@ -189,7 +189,11 @@ def test_first_import_gate_waits_for_every_selected_history_job() -> None:
     app = SimpleNamespace(
         state=SimpleNamespace(
             onboarding_store=SimpleNamespace(get=lambda: state),
-            import_job_store=SimpleNamespace(list_jobs=lambda: jobs),
+            import_job_store=SimpleNamespace(
+                find_jobs=lambda *, user_id, operation: [
+                    job for job in jobs if job.user_id == user_id and job.operation == operation
+                ]
+            ),
         )
     )
 

memhub/apps/core/tests/test_sdk_api_surface.py

@@ -42,7 +42,10 @@ RAW_ROWS = {
     "local": [
         {
             "memory_id": "raw-1",
-            "content": "[user]: 我叫费天翔\n[assistant]: 好的，记住了。",
+            "content": (
+                "[user]: 我叫费天翔\n[assistant]: 好的，记住了。"
+                "\n[user]: 我喜欢绿茶\n[assistant]: 也记住了。"
+            ),
             "status": "shadow",
             "session_id": "s-1",
             "source": "hermes",
@@ -88,6 +91,40 @@ def _chat(user_id: str, memory_id: str, content: str, layer: str, source: str) -
     }
 
 
+#: The keyword parameters of the pinned `HyMemoryClient.search`, which takes
+#: no **kwargs. `test_pinned_sdk_raw_search` checks this set against the
+#: installed wheel, so the stand-in below rejects what the release rejects.
+SDK_SEARCH_PARAMETERS = frozenset(
+    {
+        "queries",
+        "scene",
+        "search_mode",
+        "user_ids",
+        "agent_ids",
+        "session_ids",
+        "include_sources",
+        "exclude_sources",
+        "limit",
+        "min_score",
+        "working_fact_min_score",
+        "working_experience_min_score",
+        "schema_min_score",
+        "schema_limit",
+        "intention_limit",
+        "created_after",
+        "reader",
+        "workspace_id",
+        "branch",
+        "request_id",
+        "layers",
+        "statuses",
+        "embed_override",
+        "query_embedding",
+        "rerank",
+    }
+)
+
+
 class RecordingSdk:
     """The public client surface, recording every call's keyword arguments."""
 
@@ -106,6 +143,10 @@ class RecordingSdk:
     # The SDK answers rows below a caller's floor in some lanes (a schema
     # floor, a fused rank score), so Core must filter what it returns.
     def search(self, query, **kwargs):
+        unknown = set(kwargs) - SDK_SEARCH_PARAMETERS
+        if unknown:
+            # The pinned `HyMemoryClient.search` takes no **kwargs.
+            raise TypeError(f"search() got unexpected keyword arguments {sorted(unknown)}")
         self._record("search", {"query": query, **kwargs})
         [user_id] = kwargs["user_ids"]
         if kwargs.get("layers") == ["l1_raw"]:
@@ -271,6 +312,10 @@ async def _model_ready(_app) -> bool:
     return True
 
 
+async def _model_not_ready(_app) -> bool:
+    return False
+
+
 def run_with_sdk(tmp_path, monkeypatch, scenario) -> None:
     async def main() -> None:
         app = application(tmp_path)
@@ -347,6 +392,75 @@ def test_every_context_search_option_reaches_the_sdk_and_absent_ones_keep_the_ol
     run_with_sdk(tmp_path, monkeypatch, scenario)
 
 
+@pytest.mark.fresh_sdk
+def test_raw_context_search_can_return_turns_instead_of_storage_batches(
+    tmp_path, monkeypatch
+) -> None:
+    async def scenario(app, sdk, client):
+        response = await client.post(
+            "/api/v2/context/search",
+            json={
+                "query": "绿茶",
+                "kinds": ["raw"],
+                "raw_unit": "turn",
+                "limit": 20,
+            },
+        )
+        assert response.status_code == 200, response.text
+        raw_one = [item for item in response.json()["items"] if item["raw_id"] == "raw-1"]
+        assert len(raw_one) == 1
+        assert raw_one[0]["turn_index"] == 1
+        assert "我喜欢绿茶" in raw_one[0]["content"]
+        assert "我叫费天翔" not in raw_one[0]["content"]
+        # pro/ultra: a shadow raw is already in its extracted memories.
+        assert "statuses" not in sdk.last("search")
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_lite_raw_search_is_plain_rag_over_active_and_shadow_raws(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        app.state.memory_mode = lambda: "lite"
+        response = await client.post(
+            "/api/v2/context/search",
+            json={"query": "喜欢喝绿茶的人", "kinds": ["raw"], "raw_unit": "turn", "limit": 5},
+        )
+        assert response.status_code == 200, response.text
+        sent = sdk.last("search")
+        assert sent["statuses"] == ["active", "shadow"] and sent["limit"] == 5
+        items = response.json()["items"]
+        # No turn contains the query, yet every vector hit is returned, in
+        # score order: raw-1 by its closest turn, raw-2 by its only one.
+        assert [(item["raw_id"], item["turn_index"]) for item in items] == [
+            ("raw-1", 1),
+            ("raw-2", 0),
+        ]
+        assert [item["score"] for item in items] == [0.97, 0.3]
+        assert "我喜欢绿茶" in items[0]["content"]
+
+        floored = await client.post(
+            "/api/v2/context/search",
+            json={
+                "query": "喜欢喝绿茶的人",
+                "kinds": ["raw"],
+                "raw_unit": "turn",
+                "min_score": 0.5,
+            },
+        )
+        assert [item["raw_id"] for item in floored.json()["items"]] == ["raw-1"]
+
+        # Lite has no model, so recall answers from the raw layer.
+        monkeypatch.setattr("one_memory.api._recall_model_ready", _model_not_ready)
+        recall = await client.post("/api/v1/recall", json={"query": "喜欢喝绿茶的人"})
+        assert recall.status_code == 200, recall.text
+        assert recall.json()["mode"] == "raw"
+        assert sdk.last("search")["statuses"] == ["active", "shadow"]
+        assert [hit["memory_id"] for hit in recall.json()["hits"]][:1] == ["raw-1"]
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
 @pytest.mark.fresh_sdk
 def test_every_recall_and_ask_option_reaches_the_sdk(tmp_path, monkeypatch) -> None:
     async def scenario(app, sdk, client):
@@ -574,7 +688,12 @@ def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monke
         assert sent["layers"] == ["l1_raw"] and sent["statuses"] == ["active", "shadow"]
         first = next(item for item in body["items"] if item["raw_id"] == "raw-1")
         assert first["status"] == "shadow"
-        assert [message["role"] for message in first["messages"]] == ["user", "assistant"]
+        assert [message["role"] for message in first["messages"]] == [
+            "user",
+            "assistant",
+            "user",
+            "assistant",
+        ]
         assert first["messages"][0]["content"] == "我叫费天翔"
         assert first["role"] is None
         expected = hashlib.sha256(first["content"].encode()).hexdigest()
@@ -584,6 +703,54 @@ def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monke
         second = next(item for item in body["items"] if item["raw_id"] == "raw-2")
         assert second["role"] == "user"
 
+        turns = (await client.get("/api/v2/raw", params={"unit": "turn"})).json()
+        assert turns["unit"] == "turn" and turns["total"] == 3
+        first_turns = [item for item in turns["items"] if item["raw_id"] == "raw-1"]
+        assert [item["turn_index"] for item in first_turns] == [1, 0]
+        assert len({item["turn_id"] for item in first_turns}) == 2
+        assert [[message["content"] for message in item["messages"]] for item in first_turns] == [
+            ["我喜欢绿茶", "也记住了。"],
+            ["我叫费天翔", "好的，记住了。"],
+        ]
+        calls_before_refresh = len(sdk.all("list_memories"))
+        await client.get("/api/v2/raw", params={"unit": "turn"})
+        assert len(sdk.all("list_memories")) == calls_before_refresh + 1
+
+        # A live session puts a new raw on top: one listing call, not a rescan.
+        newest = {
+            **RAW_ROWS["local"][1],
+            "memory_id": "raw-new",
+            "content": "[user]: 刚刚说的",
+            "status": "active",
+        }
+        original = RAW_ROWS["local"]
+        monkeypatch.setitem(RAW_ROWS, "local", [newest, *original])
+        calls_before_append = len(sdk.all("list_memories"))
+        appended = (await client.get("/api/v2/raw", params={"unit": "turn"})).json()
+        assert len(sdk.all("list_memories")) == calls_before_append + 1
+        assert appended["total"] == 4
+        assert [item["raw_id"] for item in appended["items"]][:1] == ["raw-new"]
+        assert [item["turn_id"] for item in appended["items"][1:]] == [
+            item["turn_id"] for item in turns["items"]
+        ]
+        # Anything else (here a raw going away) lists the raw layer again.
+        monkeypatch.setitem(RAW_ROWS, "local", [newest, original[1]])
+        calls_before_delete = len(sdk.all("list_memories"))
+        shrunk = (await client.get("/api/v2/raw", params={"unit": "turn"})).json()
+        assert len(sdk.all("list_memories")) > calls_before_delete + 1
+        assert shrunk["total"] == 2
+        monkeypatch.setitem(RAW_ROWS, "local", original)
+        calls_before_restore = len(sdk.all("list_memories"))
+        restored = (await client.get("/api/v2/raw", params={"unit": "turn"})).json()
+        assert len(sdk.all("list_memories")) > calls_before_restore + 1
+        assert restored["total"] == 3
+
+        ascending = (
+            await client.get("/api/v2/raw", params={"unit": "turn", "order": "asc"})
+        ).json()
+        ascending_first = [item for item in ascending["items"] if item["raw_id"] == "raw-1"]
+        assert [item["turn_index"] for item in ascending_first] == [0, 1]
+
         await client.get(
             "/api/v2/raw",
             params=[("source", "hermes,codex"), ("session_id", "s-1"), ("order", "asc")],

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.13"
+    assert manifest["snapshot_sdk_version"] == "2.0.14"
     paths = {item["path"]: item for item in manifest["files"]}
     assert {
         "management/one-memory.sqlite3",

memhub/apps/core/uv.lock

@@ -991,7 +991,7 @@ wheels = [
 
 [[package]]
 name = "hy-memory"
-version = "2.0.13"
+version = "2.0.14"
 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/88/46/7d5a2d76be135664302d91e15f0d5018cf8e228956e4c80b86583b131cb3/hy_memory-2.0.13.tar.gz", hash = "sha256:f95850d492de74e91891502716da96e8b5297c3fc045dc70ec3474ed240da958", size = 1252802, upload-time = "2026-11-02T09:30:40.331Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/4a/68/932ac0f684a971741643d8dab0a209c6a09d5a132ca607bbb9355e0b8cf2/hy_memory-2.0.14.tar.gz", hash = "sha256:94cf354f51ab8b506c0ac4909a95a6b13fd542f81c894322210beb20996f16fd", size = 1253180, upload-time = "2026-11-04T09:52:17.127Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/99/63/2b4d21d555a5dd50ab838fad944c7fdd6cf90d2dc814e2a8e79eca3d4e03/hy_memory-2.0.13-py3-none-any.whl", hash = "sha256:128cf4729e6ea10b9da3c7fdf21005191294621a65378cd9583827904eea064a", size = 1322154, upload-time = "2026-11-02T09:30:37.746Z" },
+    { url = "https://files.pythonhosted.org/packages/76/7f/9ecbaa2589dae9bd83657adc6388b0c7b0be415c5584fef045db309911eb/hy_memory-2.0.14-py3-none-any.whl", hash = "sha256:6df76d8dc84aaa352149e85055331d22ef904e4acd4540a2328bf058a3f0f2a6", size = 1322364, upload-time = "2026-11-04T09:52:14.096Z" },
 ]
 
 [[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.13" },
+    { name = "hy-memory", specifier = "==2.0.14" },
     { name = "keyring", specifier = ">=25.7.0" },
     { name = "numpy", specifier = "==2.5.2" },
     { name = "onnxruntime", specifier = "==1.23.2" },

memhub/apps/desktop/src/diagnostics-ipc.test.ts

@@ -366,7 +366,7 @@ it("refuses a ui trace from an untrusted window", async () => {
   ).rejects.toThrow(/untrusted_window/);
 });
 
-it("marks a problem and exports a redacted local bundle", async () => {
+it("marks a problem and exports the complete local log bundle", async () => {
   const value = await fixture();
   const mark = await value.handlers.get(
     "one-memory:mark-diagnostic-incident",
@@ -375,6 +375,14 @@ it("marks a problem and exports a redacted local bundle", async () => {
     incident_id: expect.any(String),
     session_id: expect.any(String),
   });
+  await writeFile(
+    path.join(value.dataDir, "logs", "core.log"),
+    "2026-10-14 10:31:00 WARNING database wait\n",
+  );
+  await writeFile(
+    path.join(value.dataDir, "logs", "service-error.log"),
+    "worker failed\n",
+  );
   const exported = await value.handlers.get("one-memory:export-diagnostics")?.(
     {},
   );
@@ -398,6 +406,16 @@ it("marks a problem and exports a redacted local bundle", async () => {
     ),
   );
   expect(summary).toMatchObject({ build: "3800" });
+  const entries = execFileSync(
+    "/usr/bin/unzip",
+    ["-Z1", path.join(value.dataDir, zip)],
+    { encoding: "utf8" },
+  )
+    .trim()
+    .split("\n");
+  expect(entries).toContain("logs/core.log");
+  expect(entries).toContain("logs/service-error.log");
+  expect(entries).toContain("logs/diagnostic-desktop.jsonl");
   value.dispose();
 });
 

memhub/apps/desktop/src/diagnostics-ipc.ts

@@ -196,6 +196,7 @@ export function resolveDiagnosticBuildIdentity(options: {
 
 export async function exportDesktopDiagnostics(
   options: DiagnosticExportOptions,
+  exportOptions: { includeRawLogs?: boolean } = {},
 ) {
   const build = options.buildNumber ?? options.productVersion;
   const target = path.join(
@@ -209,6 +210,7 @@ export async function exportDesktopDiagnostics(
     dataDir: options.dataDir,
     target,
     productVersion: build,
+    includeRawLogs: exportOptions.includeRawLogs === true,
     ...(options.commit === undefined ? {} : { commit: options.commit }),
     ...(options.coreSha256 === undefined
       ? {}
@@ -644,7 +646,7 @@ export function installDiagnosticsIpc(
   options.ipc.handle(EXPORT, async (event, ...args) => {
     if (args.length) throw new Error("diagnostic_arguments_rejected");
     options.requireTrustedWindow(event);
-    return exportDesktopDiagnostics(options);
+    return exportDesktopDiagnostics(options, { includeRawLogs: true });
   });
 
   return () => {

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

@@ -226,6 +226,7 @@ describe("HealthPanel", () => {
     });
     vi.spyOn(window, "prompt").mockReturnValue("蓝莓没出来");
     render(<HealthPanel imports={imports()} />);
+    expect(screen.getByText(/可能包含会话正文和本机路径/)).toBeTruthy();
 
     fireEvent.click(screen.getByRole("button", { name: "记录问题时间点" }));
     expect(await screen.findByText(/问题编号已复制/)).toBeTruthy();

memhub/apps/web/src/HealthPanel.tsx

@@ -298,6 +298,7 @@ export function HealthPanel({ imports }: { imports: ImportState }) {
           <Download size={16} aria-hidden="true" />
           {t("nativeExportLogs")}
         </R6Button>
+        <p>{t("把本机运行日志打包成一个文件，发给开发排查。")}</p>
       </div>
 
       <section className="health-block" aria-label={t("healthLocal")}>

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

@@ -6,7 +6,11 @@ import type {
 } from "@one-memory/contracts";
 import { cleanup, render, screen } from "@testing-library/react";
 import { afterEach, describe, expect, it } from "vitest";
-import { MemoryGraph3D } from "./MemoryGraph3D";
+import {
+  graphNodeTooltip,
+  isGraphNodeSelectable,
+  MemoryGraph3D,
+} from "./MemoryGraph3D";
 
 afterEach(cleanup);
 
@@ -48,6 +52,18 @@ function makeGraph(nodeCount: number): MemoryGraphResponse {
 }
 
 describe("MemoryGraph3D", () => {
+  it("keeps topic annotations hover-only", () => {
+    const topic = {
+      ...makeGraph(1).nodes[0],
+      node_type: "topic" as const,
+      label: "CI",
+    };
+    expect(isGraphNodeSelectable(topic)).toBe(false);
+    expect(graphNodeTooltip(topic, "主题：")).toBe("主题：CI");
+    expect(graphNodeTooltip(topic, "Topic: ")).toBe("Topic: CI");
+    expect(isGraphNodeSelectable({ ...topic, node_type: "memory" })).toBe(true);
+  });
+
   it("没有 WebGL 时给出可访问降级提示", () => {
     render(
       <MemoryGraph3D

memhub/apps/web/src/MemoryGraph3D.tsx

@@ -27,6 +27,20 @@ const SELECT_DURATION_MS = 180;
 const SELECTED_COLOR = new THREE.Color(0x294355);
 const SELECTED_EMISSIVE = new THREE.Color(0x607e97);
 
+/** Topic nodes explain the graph; unlike memories and sources, they do not own a detail view. */
+export function isGraphNodeSelectable(node: MemoryGraphNode): boolean {
+  return node.node_type !== "topic";
+}
+
+export function graphNodeTooltip(
+  node: MemoryGraphNode,
+  topicPrefix: string,
+): string {
+  return node.node_type === "topic"
+    ? `${topicPrefix}${node.label}`
+    : node.label;
+}
+
 export function MemoryGraph3D({
   graph,
   selectedNodeId,
@@ -36,10 +50,20 @@ export function MemoryGraph3D({
   selectedNodeId: string | null;
   onSelectNode(node: MemoryGraphNode): void;
 }) {
-  const { text } = useMemoryLanguage();
+  const { locale, text } = useMemoryLanguage();
   const host = useRef<HTMLDivElement>(null);
   const labelHost = useRef<HTMLDivElement>(null);
   const selectRef = useRef(onSelectNode);
+  const nodesByIdRef = useRef(
+    new Map(graph.nodes.map((node) => [node.node_id, node])),
+  );
+  nodesByIdRef.current = new Map(
+    graph.nodes.map((node) => [node.node_id, node]),
+  );
+  const topicPrefixRef = useRef("");
+  topicPrefixRef.current = `${text("主题")}${
+    locale.startsWith("zh") || locale.startsWith("ja") ? "：" : ": "
+  }`;
   // 选中态通过 ref 传入渲染循环，而不是进 useEffect 依赖。
   // 依赖里带 selectedNodeId 会导致每次点击重建整个场景、重跑布局、重建全部 mesh。
   const selectedRef = useRef(selectedNodeId);
@@ -157,9 +181,7 @@ export function MemoryGraph3D({
       }
     }
     const labels = graph.nodes
-      .filter(
-        (node) => node.node_type === "source" || node.node_type === "topic",
-      )
+      .filter((node) => node.node_type === "source")
       .slice(0, 4)
       .map((node) => {
         const element = document.createElement("span");
@@ -194,7 +216,7 @@ export function MemoryGraph3D({
       }
     };
 
-    const selectable: THREE.Mesh[] = [];
+    const hittable: THREE.Mesh[] = [];
     const materials = new Set<THREE.Material>();
     const geometryByType = {
       source: new THREE.SphereGeometry(0.24, 18, 12),
@@ -237,7 +259,7 @@ export function MemoryGraph3D({
       mesh.scale.setScalar(1);
       mesh.visible = returning || extending;
       scene.add(mesh);
-      selectable.push(mesh);
+      hittable.push(mesh);
       visuals.push({
         mesh,
         material,
@@ -298,7 +320,6 @@ export function MemoryGraph3D({
       });
     }
 
-    const nodesById = new Map(graph.nodes.map((node) => [node.node_id, node]));
     const raycaster = new THREE.Raycaster();
     const pointer = new THREE.Vector2();
     const hitAt = (event: PointerEvent) => {
@@ -307,7 +328,7 @@ export function MemoryGraph3D({
       pointer.x = ((event.clientX - bounds.left) / bounds.width) * 2 - 1;
       pointer.y = -((event.clientY - bounds.top) / bounds.height) * 2 + 1;
       raycaster.setFromCamera(pointer, camera);
-      return raycaster.intersectObjects(selectable, false)[0];
+      return raycaster.intersectObjects(hittable, false)[0];
     };
     let pressed: { x: number; y: number; id: number; moved: boolean } | null =
       null;
@@ -333,8 +354,9 @@ export function MemoryGraph3D({
       )
         return;
       const nodeId = hitAt(event)?.object.userData.nodeId;
-      const node = typeof nodeId === "string" ? nodesById.get(nodeId) : null;
-      if (node) selectRef.current(node);
+      const node =
+        typeof nodeId === "string" ? nodesByIdRef.current.get(nodeId) : null;
+      if (node && isGraphNodeSelectable(node)) selectRef.current(node);
     };
     const hoverNode = (event: PointerEvent) => {
       if (pressed) {
@@ -344,12 +366,18 @@ export function MemoryGraph3D({
         return;
       }
       const nodeId = hitAt(event)?.object.userData.nodeId;
-      const node = typeof nodeId === "string" ? nodesById.get(nodeId) : null;
-      renderer.domElement.style.cursor = node ? "pointer" : "grab";
+      const node =
+        typeof nodeId === "string" ? nodesByIdRef.current.get(nodeId) : null;
+      renderer.domElement.style.cursor =
+        node && isGraphNodeSelectable(node)
+          ? "pointer"
+          : node
+            ? "default"
+            : "grab";
       setHovered(
         node
           ? {
-              label: node.label,
+              label: graphNodeTooltip(node, topicPrefixRef.current),
               x: event.clientX - element.getBoundingClientRect().left,
               y: event.clientY - element.getBoundingClientRect().top,
             }
@@ -629,17 +657,23 @@ export function MemoryGraph3D({
       {!available ? (
         <div className="memory-graph-fallback">
           <p>{text("这台设备打不开 3D 视图，已显示关联列表。")}</p>
-          {graph.nodes.map((node) => (
-            <button
-              type="button"
-              className="secondary"
-              key={node.node_id}
-              aria-pressed={selectedNodeId === node.node_id}
-              onClick={() => onSelectNode(node)}
-            >
-              {node.label}
-            </button>
-          ))}
+          {graph.nodes.map((node) =>
+            isGraphNodeSelectable(node) ? (
+              <button
+                type="button"
+                className="secondary"
+                key={node.node_id}
+                aria-pressed={selectedNodeId === node.node_id}
+                onClick={() => onSelectNode(node)}
+              >
+                {node.label}
+              </button>
+            ) : (
+              <span className="memory-graph-fallback-topic" key={node.node_id}>
+                {graphNodeTooltip(node, topicPrefixRef.current)}
+              </span>
+            ),
+          )}
         </div>
       ) : null}
       {available && (

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

@@ -149,7 +149,7 @@ it("OBS-22: the graph inspector has no editing entry and never says changes cann
   expect(screen.queryByText(/暂不能保存更改/)).toBeNull();
 });
 
-it("replaces a topic detail with an off-page graph memory detail", async () => {
+it("keeps topic annotations out of the inspector and opens detail-bearing graph nodes", async () => {
   const onOpen = vi.fn();
   const topicGraph = {
     total_memories: 2,
@@ -221,13 +221,13 @@ it("replaces a topic detail with an off-page graph memory detail", async () => {
     />,
   );
 
+  await act(async () => {});
+  onOpen.mockClear();
   await userEvent.click(screen.getByRole("button", { name: "图中选择 git" }));
-  expect(screen.getByRole("heading", { name: "git" })).toBeTruthy();
-  expect(screen.getAllByText("第一条").length).toBeGreaterThan(0);
-  expect(screen.queryByText("对应记忆")).toBeNull();
-  expect(screen.queryByText("属于主题")).toBeNull();
-  expect(screen.queryByText("主题")).toBeNull();
-  expect(screen.queryByText("两跳")).toBeNull();
+  expect(onOpen).not.toHaveBeenCalled();
+  expect(screen.queryByRole("heading", { name: "git" })).toBeNull();
+  expect(screen.getByRole("heading", { name: "第一条" })).toBeTruthy();
+  expect(screen.getByTestId("canvas").dataset.selected).toBe("memory:m0");
   await userEvent.click(
     screen.getByRole("button", { name: "图中选择 先验证，再推送" }),
   );

memhub/apps/web/src/MemoryMap.tsx

@@ -684,6 +684,9 @@ export function MemoryMap(props: Props) {
     setGroupKey(key);
   }
   function pickNode(node: MemoryGraphNode) {
+    // Topics are graph annotations, not detail-bearing records. They remain
+    // hoverable in the canvas but must not replace the memory inspector.
+    if (node.node_type === "topic") return;
     if (node.memory_id) {
       pick(node.memory_id);
       return;
@@ -708,11 +711,6 @@ export function MemoryMap(props: Props) {
         if (first) pick(first, false);
       }
     }
-    if (node.node_type === "topic") {
-      if (props.simplified) return;
-      setAxis("topic");
-      setGroupKey(`topic:${node.label}`);
-    }
   }
   const label =
     group?.label ??

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

@@ -1,5 +1,6 @@
 // @vitest-environment jsdom
 
+import { legacyMemoryModePlan } from "@one-memory/operations";
 import {
   cleanup,
   render,
@@ -64,31 +65,9 @@ function prepare(mode: "lite" | "pro" | "ultra" = "pro") {
     running: [],
   });
   vi.mocked(getMemoryProcessing).mockResolvedValue({} as never);
-  vi.mocked(planMemoryMode).mockImplementation(async (target) => ({
-    current: mode,
-    target,
-    effective_target: target,
-    capabilities: {
-      content_plane: target === "lite" ? "raw" : "memory",
-      generation_required: target !== "lite",
-      graph_enabled: target === "ultra",
-      evolve_available: target === "ultra",
-    },
-    effects: [
-      ...(target === "lite" ? (["higher_mode_memories_hidden"] as const) : []),
-      ...(mode === "lite" && target !== "lite"
-        ? (["earlier_raw_not_extracted", "lite_search_stops"] as const)
-        : []),
-      ...(target === "ultra" && mode !== "ultra"
-        ? (["existing_memories_added_to_graph"] as const)
-        : []),
-      ...(mode === "ultra" && target !== "ultra"
-        ? (["evolve_stopped"] as const)
-        : []),
-    ],
-    warnings: [],
-    blockers: [],
-  }));
+  vi.mocked(planMemoryMode).mockImplementation(async (target) =>
+    legacyMemoryModePlan(mode, target, { hasActiveModel: true }),
+  );
 }
 
 async function chooseMode(
@@ -227,6 +206,20 @@ it("uses the isolated legacy plan only for a route-level old-Core 404", async ()
   expect(await screen.findByText(/已有记忆继续保留/)).toBeTruthy();
 });
 
+it("keeps downgrade effects when an old Core falls back from ultra to pro", async () => {
+  prepare("ultra");
+  vi.mocked(planMemoryMode).mockRejectedValue(new Error("404"));
+  vi.mocked(memoryModePlanUnsupported).mockReturnValue(true);
+  const user = userEvent.setup();
+  render(<MemoryModeSettings hasActiveModel onConfigureModel={vi.fn()} />);
+
+  await chooseMode(user, /智能记忆（pro）/);
+
+  const dialog = screen.getByRole("dialog");
+  expect(within(dialog).getAllByText("已有数据")).toHaveLength(1);
+  expect(within(dialog).getByText("每天自动整合")).toBeTruthy();
+});
+
 it("explains model cost, search scope and the next step before switching to ultra", async () => {
   prepare("lite");
   const user = userEvent.setup();
@@ -238,6 +231,8 @@ it("explains model cost, search scope and the next step before switching to ultr
   expect(screen.getByText("搜索范围")).toBeTruthy();
   expect(screen.getByText("切换后")).toBeTruthy();
   expect(screen.getByText(/已经处理完的旧对话不会补做/)).toBeTruthy();
+  expect(screen.getByText("知识图谱")).toBeTruthy();
+  expect(screen.getAllByText("已有数据")).toHaveLength(1);
   expect(screen.getByText(/切换本身不会立即整合/)).toBeTruthy();
 });
 
@@ -248,9 +243,27 @@ it("distinguishes regular memories from higher-level Ultra data when switching t
 
   await chooseMode(user, /智能记忆（pro）/);
 
-  expect(screen.getByText(/普通沉淀记忆仍可使用/)).toBeTruthy();
-  expect(screen.getByText(/更高层记忆和图谱数据继续保留/)).toBeTruthy();
-  expect(screen.getByText(/每天自动整合会停止/)).toBeTruthy();
+  const dialog = screen.getByRole("dialog");
+  expect(within(dialog).getAllByText("已有数据")).toHaveLength(1);
+  expect(within(dialog).getByText(/普通沉淀记忆仍可使用/)).toBeTruthy();
+  expect(within(dialog).getByText(/更高层记忆和图谱数据继续保留/)).toBeTruthy();
+  expect(within(dialog).getByText("每天自动整合")).toBeTruthy();
+  expect(
+    within(dialog).getByText("每天自动整合会停止；已生成的整合结果继续保留。"),
+  ).toBeTruthy();
+});
+
+it("does not show automatic consolidation when switching from lite to pro", async () => {
+  prepare("lite");
+  const user = userEvent.setup();
+  render(<MemoryModeSettings hasActiveModel onConfigureModel={vi.fn()} />);
+
+  await chooseMode(user, /智能记忆（pro）/);
+
+  const dialog = screen.getByRole("dialog");
+  expect(within(dialog).getAllByText("已有数据")).toHaveLength(1);
+  expect(within(dialog).getByText(/已经处理完的旧对话不会补做/)).toBeTruthy();
+  expect(within(dialog).queryByText("每天自动整合")).toBeNull();
 });
 
 it("focuses the Evolve settings after switching to ultra", async () => {

memhub/apps/web/src/MemoryModeSettings.tsx

@@ -85,18 +85,20 @@ function planRows(plan: MemoryModePlan) {
   if (plan.effects.includes("higher_mode_memories_hidden"))
     rows.push({
       label: "已有数据",
-      content: "已有记忆继续保留，但高级记忆在当前模式下不显示、不参与召回。",
+      content:
+        plan.effective_target === "pro"
+          ? "普通沉淀记忆仍可使用；深度记忆新增的更高层记忆和图谱数据继续保留，但在智能记忆下不显示、不参与召回。原始对话不会删除。"
+          : "已有记忆继续保留，但在基础记忆下不显示、不参与召回；原始对话不会删除。",
     });
   if (plan.effects.includes("existing_memories_added_to_graph"))
     rows.push({
-      label: "已有数据",
+      label: "知识图谱",
       content: "已有记忆会在后台加入知识图谱，期间仍可使用。",
     });
   if (plan.effects.includes("evolve_stopped"))
     rows.push({
-      label: "已有数据",
-      content:
-        "普通沉淀记忆仍可使用；更高层记忆和图谱数据继续保留，但不参与当前召回；每天自动整合会停止。",
+      label: "每天自动整合",
+      content: "每天自动整合会停止，已经生成的数据不会删除。",
     });
   if (plan.capabilities.evolve_available)
     rows.push({

memhub/apps/web/src/MemorySpace.r6.test.tsx

@@ -387,6 +387,67 @@ it("pages and filters the raw conversation data plane in lite", async () => {
   );
 });
 
+it("selects and renders one turn when two turns share a raw storage batch", async () => {
+  setup();
+  reads.processing.mockResolvedValue(modeProcessing("lite"));
+  reads.raw.mockResolvedValue({
+    items: [
+      {
+        raw_id: "raw-batch",
+        turn_id: "raw-turn:batch:0",
+        turn_index: 0,
+        turn_count: 2,
+        content: "[user]: 第一问\n[assistant]: 第一答",
+        user_id: "local",
+        agent_id: "all-apps",
+        source: "cursor",
+        session_id: "session-batch",
+        memory_at: null,
+        gmt_created: 1_761_000_000,
+        custom: {},
+        messages: [
+          { role: "user", content: "第一问" },
+          { role: "assistant", content: "第一答" },
+        ],
+      },
+      {
+        raw_id: "raw-batch",
+        turn_id: "raw-turn:batch:1",
+        turn_index: 1,
+        turn_count: 2,
+        content: "[user]: 第二问\n[assistant]: 第二答",
+        user_id: "local",
+        agent_id: "all-apps",
+        source: "cursor",
+        session_id: "session-batch",
+        memory_at: null,
+        gmt_created: 1_761_000_000,
+        custom: {},
+        messages: [
+          { role: "user", content: "第二问" },
+          { role: "assistant", content: "第二答" },
+        ],
+      },
+    ],
+    total: 2,
+    limit: 20,
+    offset: 0,
+    unit: "turn",
+  });
+  const user = userEvent.setup();
+  render(<MemorySpace simplified />);
+
+  await user.click(await screen.findByRole("button", { name: /第二问/ }));
+
+  await waitFor(() => {
+    const detail = document.querySelector(".map-dialogue-reading");
+    expect(detail?.textContent).toContain("第二问");
+    expect(detail?.textContent).toContain("第二答");
+    expect(detail?.textContent).not.toContain("第一问");
+    expect(detail?.textContent).not.toContain("第一答");
+  });
+});
+
 it("switches data planes without keeping rows or details from the prior mode", async () => {
   setup();
   vi.useFakeTimers();
@@ -482,6 +543,27 @@ it("opens a search hit with the complete role transcript", async () => {
           items: [
             {
               raw_id: "raw-search",
+              turn_id: "raw-turn:search:0",
+              turn_index: 0,
+              turn_count: 2,
+              content: "[user]: 第一轮问题\n[assistant]: 第一轮回答",
+              user_id: "local",
+              agent_id: "all-apps",
+              source: "cursor",
+              session_id: "session-search",
+              memory_at: null,
+              gmt_created: 1_761_000_000,
+              custom: {},
+              messages: [
+                { role: "user", content: "第一轮问题" },
+                { role: "assistant", content: "第一轮回答" },
+              ],
+            },
+            {
+              raw_id: "raw-search",
+              turn_id: "raw-turn:search:1",
+              turn_index: 1,
+              turn_count: 2,
               content: "[user]: 完整问题\n[assistant]: 完整回答",
               user_id: "local",
               agent_id: "all-apps",
@@ -496,7 +578,7 @@ it("opens a search hit with the complete role transcript", async () => {
               ],
             },
           ],
-          total: 1,
+          total: 2,
           limit: 500,
           offset: 0,
         }
@@ -506,6 +588,9 @@ it("opens a search hit with the complete role transcript", async () => {
     {
       kind: "raw",
       raw_id: "raw-search",
+      turn_id: "raw-turn:search:1",
+      turn_index: 1,
+      turn_count: 2,
       content: "被截断的搜索摘要",
       source: "cursor",
       session_id: "session-search",
@@ -525,6 +610,9 @@ it("opens a search hit with the complete role transcript", async () => {
 
   expect(await screen.findByText("完整问题")).toBeTruthy();
   expect(screen.getByText("完整回答")).toBeTruthy();
+  expect(
+    document.querySelector(".map-dialogue-reading")?.textContent,
+  ).not.toContain("第一轮问题");
   expect(reads.raw).toHaveBeenCalledWith(
     expect.objectContaining({
       source: "cursor",

memhub/apps/web/src/MemorySpace.tsx

@@ -61,6 +61,7 @@ import {
   type RawMapRecord,
   rawDetail,
   rawLibraryPage,
+  rawRecordId,
   rawSearchPage,
 } from "./sdk-raw-map";
 import { sourceNameOr, UNATTRIBUTED_SOURCE_ID } from "./source-label";
@@ -277,8 +278,11 @@ export function MemorySpace({
         const rememberRaw = (items: RawMapRecord[]) => {
           const next = new Map<string, RawMapRecord>();
           if (selectedRawRef.current)
-            next.set(selectedRawRef.current.raw_id, selectedRawRef.current);
-          for (const item of items) next.set(item.raw_id, item);
+            next.set(
+              rawRecordId(selectedRawRef.current),
+              selectedRawRef.current,
+            );
+          for (const item of items) next.set(rawRecordId(item), item);
           rawRecords.current = next;
         };
         const [page, graph] = lite
@@ -717,7 +721,9 @@ export function MemorySpace({
               offset,
               limit: 500,
             });
-            const full = page.items.find((item) => item.raw_id === memoryId);
+            const full = page.items.find(
+              (item) => rawRecordId(item) === memoryId,
+            );
             if (full) {
               record = full;
               rawRecords.current.set(memoryId, full);

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

@@ -72,6 +72,7 @@ const probeProfile = vi.fn();
 const activate = vi.fn();
 const remove = vi.fn();
 const update = vi.fn();
+const refreshProfiles = vi.fn();
 const planActivation = vi.fn();
 const startPlannedSwitch = vi.fn();
 let switchPlan: EmbeddingSwitchPlan | null = null;
@@ -224,6 +225,7 @@ vi.mock("./use-model-profiles", () => ({
     activate,
     remove,
     update,
+    refreshProfiles,
   }),
 }));
 vi.mock("./use-embedding-switch", () => ({
@@ -279,6 +281,7 @@ describe("ModelSettings", () => {
     switchCurrent = null;
     create.mockResolvedValue(profile);
     update.mockResolvedValue({ ...preset, has_api_key: true, revision: 2 });
+    refreshProfiles.mockResolvedValue(undefined);
     planActivation.mockResolvedValue({
       needed: false,
       reason: "same_embedding",
@@ -576,7 +579,8 @@ describe("ModelSettings", () => {
       name: "高级：自行配置向量模型",
     });
     await user.click(disclosure);
-    await user.selectOptions(screen.getByLabelText("向量配置模式"), "custom");
+    await user.click(screen.getByLabelText("向量配置模式"));
+    await user.click(screen.getByRole("option", { name: "自定义搜索模型" }));
     await user.type(screen.getByLabelText("向量模型名"), "custom-vector");
     await user.type(
       screen.getByLabelText("向量模型 Base URL"),
@@ -666,9 +670,9 @@ describe("ModelSettings", () => {
     ];
     render(<ModelSettings />);
     await userEvent.click(screen.getByRole("button", { name: "编辑配置" }));
-    await userEvent.selectOptions(
-      screen.getByLabelText("向量配置模式"),
-      "local",
+    await userEvent.click(screen.getByLabelText("向量配置模式"));
+    await userEvent.click(
+      screen.getByRole("option", { name: "本地内置（推荐）" }),
     );
     await userEvent.click(screen.getByRole("button", { name: "保存修改" }));
     expect(update).toHaveBeenCalledWith(profile.profile_id, {
@@ -687,9 +691,9 @@ describe("ModelSettings", () => {
       name: "高级：自行配置向量模型",
     });
     await userEvent.click(disclosure);
-    await userEvent.selectOptions(
-      screen.getByLabelText("向量配置模式"),
-      "custom",
+    await userEvent.click(screen.getByLabelText("向量配置模式"));
+    await userEvent.click(
+      screen.getByRole("option", { name: "自定义搜索模型" }),
     );
     await userEvent.click(disclosure);
     await userEvent.click(screen.getByRole("button", { name: "保存配置" }));
@@ -850,6 +854,190 @@ describe("ModelSettings", () => {
     expect(activate).not.toHaveBeenCalled();
   });
 
+  it("completes an embedding change only after the switch commit activates the target", async () => {
+    const onSetupComplete = vi.fn();
+    profiles = [
+      {
+        ...profile,
+        capabilities: {
+          ...profile.capabilities,
+          endpoint_reachable: true,
+          model_available: true,
+          embedding_available: true,
+        },
+      },
+    ];
+    planActivation.mockImplementationOnce(async () => {
+      switchPlan = migrationPlan;
+      return migrationPlan;
+    });
+    const runningJob: EmbeddingSwitchJob = {
+      ...oldFinishedJob,
+      job_id: "switch-job",
+      state: "reembedding",
+      progress: { ...oldFinishedJob.progress, phase: "vectors", done: 0 },
+    };
+    startPlannedSwitch.mockResolvedValue(runningJob);
+    switchCurrent = {
+      schema_version: 1,
+      phase: "idle",
+      writes_paused: false,
+      scan_state: "done",
+      scan_error: null,
+      job: null,
+      last_finished: oldFinishedJob,
+    };
+    const view = render(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    await userEvent.click(screen.getByRole("button", { name: "设为当前模型" }));
+    await userEvent.click(
+      await screen.findByRole("button", { name: "开始更新索引" }),
+    );
+
+    await waitFor(() =>
+      expect(startPlannedSwitch).toHaveBeenCalledExactlyOnceWith(),
+    );
+    expect(onSetupComplete).not.toHaveBeenCalled();
+
+    refreshProfiles.mockImplementationOnce(async () => {
+      profiles = [{ ...profiles[0], is_active: true }];
+    });
+    switchCurrent = {
+      ...switchCurrent,
+      phase: "idle",
+      job: null,
+      last_finished: { ...runningJob, state: "done" },
+    };
+    view.rerender(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    await waitFor(() => expect(onSetupComplete).toHaveBeenCalledOnce());
+  });
+
+  it("does not complete a committed switch until the target revision is active", async () => {
+    const onSetupComplete = vi.fn();
+    profiles = [
+      {
+        ...profile,
+        capabilities: {
+          ...profile.capabilities,
+          endpoint_reachable: true,
+          model_available: true,
+          embedding_available: true,
+        },
+      },
+    ];
+    planActivation.mockImplementationOnce(async () => {
+      switchPlan = migrationPlan;
+      return migrationPlan;
+    });
+    const runningJob: EmbeddingSwitchJob = {
+      ...oldFinishedJob,
+      job_id: "switch-job",
+      state: "reembedding",
+      progress: { ...oldFinishedJob.progress, phase: "vectors", done: 0 },
+    };
+    startPlannedSwitch.mockResolvedValue(runningJob);
+    switchCurrent = {
+      schema_version: 1,
+      phase: "idle",
+      writes_paused: false,
+      scan_state: "done",
+      scan_error: null,
+      job: null,
+      last_finished: oldFinishedJob,
+    };
+    const view = render(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    await userEvent.click(screen.getByRole("button", { name: "设为当前模型" }));
+    await userEvent.click(
+      await screen.findByRole("button", { name: "开始更新索引" }),
+    );
+    await waitFor(() => expect(startPlannedSwitch).toHaveBeenCalledOnce());
+
+    switchCurrent = {
+      ...switchCurrent,
+      phase: "idle",
+      job: null,
+      last_finished: { ...runningJob, state: "done" },
+    };
+    view.rerender(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    expect(
+      await screen.findByText(
+        "配置已保存，尚未确认启用成功。请核对配置后重试。",
+      ),
+    ).toBeTruthy();
+    expect(onSetupComplete).not.toHaveBeenCalled();
+    expect(
+      (
+        screen.getByRole("button", {
+          name: "保存并测试启用",
+        }) as HTMLButtonElement
+      ).disabled,
+    ).toBe(false);
+  });
+
+  it("keeps embedding configuration open when the switch rolls back", async () => {
+    const onSetupComplete = vi.fn();
+    profiles = [
+      {
+        ...profile,
+        capabilities: {
+          ...profile.capabilities,
+          endpoint_reachable: true,
+          model_available: true,
+          embedding_available: true,
+        },
+      },
+    ];
+    planActivation.mockImplementationOnce(async () => {
+      switchPlan = migrationPlan;
+      return migrationPlan;
+    });
+    const runningJob: EmbeddingSwitchJob = {
+      ...oldFinishedJob,
+      job_id: "switch-job",
+      state: "reembedding",
+      progress: { ...oldFinishedJob.progress, phase: "vectors", done: 0 },
+    };
+    startPlannedSwitch.mockResolvedValue(runningJob);
+    switchCurrent = {
+      schema_version: 1,
+      phase: "idle",
+      writes_paused: false,
+      scan_state: "done",
+      scan_error: null,
+      job: null,
+      last_finished: oldFinishedJob,
+    };
+    const view = render(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    await userEvent.click(screen.getByRole("button", { name: "设为当前模型" }));
+    await userEvent.click(
+      await screen.findByRole("button", { name: "开始更新索引" }),
+    );
+    await waitFor(() => expect(startPlannedSwitch).toHaveBeenCalledOnce());
+
+    switchCurrent = {
+      ...switchCurrent,
+      phase: "idle",
+      job: null,
+      last_finished: {
+        ...runningJob,
+        state: "rolled_back",
+        error_code: "embedding_switch_failed",
+      },
+    };
+    view.rerender(<ModelSettings onSetupComplete={onSetupComplete} />);
+
+    expect(
+      await screen.findByText(
+        "索引更新没有完成，仍在使用原来的搜索模型，记忆没有丢失。",
+      ),
+    ).toBeTruthy();
+    expect(onSetupComplete).not.toHaveBeenCalled();
+  });
+
   it("hands a successful save to its card, clears only the key, and never tests or activates", async () => {
     profiles = [];
     const user = userEvent.setup();
@@ -1044,14 +1232,16 @@ describe("ModelSettings", () => {
     await user.click(
       screen.getByRole("button", { name: "高级：自行配置向量模型" }),
     );
-    await user.selectOptions(screen.getByLabelText("向量配置模式"), "custom");
+    await user.click(screen.getByLabelText("向量配置模式"));
+    await user.click(screen.getByRole("option", { name: "自定义搜索模型" }));
     await user.type(
       screen.getByLabelText("配置名称"),
       "Remote with local embedding",
     );
     await user.type(screen.getByLabelText("生成模型名"), "remote-model");
     await user.type(screen.getByLabelText("向量模型名"), "nomic-embed-text");
-    await user.selectOptions(screen.getByLabelText("向量模型来源"), "ollama");
+    await user.click(screen.getByLabelText("向量模型来源"));
+    await user.click(screen.getByRole("option", { name: "本机 Ollama" }));
     await user.type(
       screen.getByLabelText("向量模型 Base URL"),
       "http://127.0.0.1:11434",
@@ -1278,6 +1468,87 @@ describe("ModelSettings", () => {
       expect(nameField().disabled).toBe(false);
     });
   });
+
+  it("only completes compact setup after the backend confirms the profile is active", async () => {
+    const user = userEvent.setup();
+    const onSetupComplete = vi.fn();
+    const capabilities = {
+      endpoint_reachable: true,
+      model_available: true,
+      structured_output: false,
+      embedding_available: true,
+    };
+    const created = {
+      ...profile,
+      profile_id: "created",
+      revision: 1,
+      capabilities,
+    };
+    profiles = [];
+    create.mockImplementation(async () => {
+      profiles = [created];
+      return created;
+    });
+    probeProfile.mockResolvedValue({
+      profile_id: created.profile_id,
+      profile_revision: created.revision,
+      stages: [],
+      capabilities,
+    });
+    activate.mockResolvedValue({ ...created, is_active: true });
+    render(<ModelSettings compactSingle onSetupComplete={onSetupComplete} />);
+
+    await user.type(screen.getByLabelText("配置名称"), "Current model");
+    await user.type(screen.getByLabelText("生成模型名"), "model-v1");
+    await user.type(screen.getByLabelText(/^API Key/), "secret-value");
+    await user.click(screen.getByRole("button", { name: "测试并完成配置" }));
+
+    await waitFor(() => expect(onSetupComplete).toHaveBeenCalledOnce());
+    expect(planActivation).toHaveBeenCalledWith(
+      created.profile_id,
+      created.revision,
+    );
+    expect(activate).toHaveBeenCalledWith(created.profile_id);
+  });
+
+  it("keeps compact setup open when activation is not confirmed", async () => {
+    const user = userEvent.setup();
+    const onSetupComplete = vi.fn();
+    const capabilities = {
+      endpoint_reachable: true,
+      model_available: true,
+      structured_output: false,
+      embedding_available: true,
+    };
+    const created = {
+      ...profile,
+      profile_id: "created",
+      revision: 1,
+      capabilities,
+    };
+    profiles = [];
+    create.mockImplementation(async () => {
+      profiles = [created];
+      return created;
+    });
+    probeProfile.mockResolvedValue({
+      profile_id: created.profile_id,
+      profile_revision: created.revision,
+      stages: [],
+      capabilities,
+    });
+    activate.mockResolvedValue({ ...created, is_active: false });
+    render(<ModelSettings compactSingle onSetupComplete={onSetupComplete} />);
+
+    await user.type(screen.getByLabelText("配置名称"), "Inactive model");
+    await user.type(screen.getByLabelText("生成模型名"), "model-v1");
+    await user.type(screen.getByLabelText(/^API Key/), "secret-value");
+    await user.click(screen.getByRole("button", { name: "测试并完成配置" }));
+
+    await screen.findByText("配置已保存，尚未确认启用成功。请核对配置后重试。");
+    expect(onSetupComplete).not.toHaveBeenCalled();
+    expect(screen.getByRole("button", { name: "测试并完成配置" })).toBeTruthy();
+  });
 });
 
 describe("把一张预置卡设为当前模型", () => {

memhub/apps/web/src/ModelSettings.tsx

@@ -11,6 +11,7 @@ import {
   type FormEvent,
   type Ref,
   useEffect,
+  useId,
   useImperativeHandle,
   useLayoutEffect,
   useRef,
@@ -26,6 +27,7 @@ import {
   presetForGeneration,
   validateGenerationModel,
 } from "./GenerationModelPrimitives";
+import { OptionMenu } from "./OptionMenu";
 import { useOnboardingTranslation } from "./onboarding-language";
 import { PROVIDER_PRESETS } from "./provider-presets";
 import { R6StateRecovery } from "./R6StateRecovery";
@@ -238,6 +240,7 @@ export function ModelSettings({
   onSetupComplete?(): void;
 } = {}) {
   const { text, t } = useOnboardingTranslation();
+  const fieldId = useId();
   const models = useModelProfiles();
   const embeddingSwitch = useEmbeddingSwitch();
   const [provider, setProvider] = useState<ProviderKind>("openai_compatible");
@@ -275,6 +278,9 @@ export function ModelSettings({
   const [setupError, setSetupError] = useState<string | null>(null);
   const [switchTarget, setSwitchTarget] = useState<ModelProfile | null>(null);
   const [switchJobId, setSwitchJobId] = useState<string | null>(null);
+  const [switchRefreshCompleted, setSwitchRefreshCompleted] = useState<
+    string | null
+  >(null);
   const settledSwitchJob = useRef<string | null>(null);
   const flight = useRef(false);
   const alive = useRef(true);
@@ -308,10 +314,11 @@ export function ModelSettings({
       return;
     settledSwitchJob.current = finished.job_id;
     if (finished.state === "done") {
-      void models.refreshProfiles();
+      setSetupPhase("activating");
       setSetupError(null);
-      setSwitchTarget(null);
-      setSwitchJobId(null);
+      void models.refreshProfiles().then(() => {
+        if (alive.current) setSwitchRefreshCompleted(finished.job_id);
+      });
       onProfilesChanged?.();
       return;
     }
@@ -323,6 +330,7 @@ export function ModelSettings({
       );
       setSwitchTarget(null);
       setSwitchJobId(null);
+      setSwitchRefreshCompleted(null);
     }
   }, [
     embeddingSwitch.current?.last_finished,
@@ -331,6 +339,45 @@ export function ModelSettings({
     switchJobId,
     switchTarget,
   ]);
+  useEffect(() => {
+    const finished = embeddingSwitch.current?.last_finished;
+    if (
+      !switchTarget ||
+      !switchJobId ||
+      finished?.job_id !== switchJobId ||
+      finished.state !== "done" ||
+      switchRefreshCompleted !== switchJobId
+    )
+      return;
+    const active = models.profiles.find(
+      (profile) =>
+        profile.profile_id === switchTarget.profile_id &&
+        profile.revision >= switchTarget.revision &&
+        profile.is_active,
+    );
+    if (!active) {
+      setSetupPhase(null);
+      setSetupError("配置已保存，尚未确认启用成功。请核对配置后重试。");
+      setSwitchTarget(null);
+      setSwitchJobId(null);
+      setSwitchRefreshCompleted(null);
+      return;
+    }
+    if (active.name !== PRESET_PROFILE_NAME) setSaved(active);
+    setSetupPhase(null);
+    setSetupError(null);
+    setSwitchTarget(null);
+    setSwitchJobId(null);
+    setSwitchRefreshCompleted(null);
+    onSetupComplete?.();
+  }, [
+    embeddingSwitch.current?.last_finished,
+    models.profiles,
+    onSetupComplete,
+    switchJobId,
+    switchRefreshCompleted,
+    switchTarget,
+  ]);
   useLayoutEffect(() => {
     if (saved) savedCard.current?.focus({ preventScroll: true });
   }, [saved]);
@@ -381,6 +428,7 @@ export function ModelSettings({
       saved);
   const editorLocked =
     saving ||
+    switchJobId !== null ||
     models.busy !== null ||
     !!models.pendingCreation ||
     !!models.pendingProbe ||
@@ -864,6 +912,9 @@ export function ModelSettings({
       if (!alive.current) return;
       if (created) {
         if (thenSetup) {
+          // Retrying after probe, activation, or switch confirmation failed
+          // must update this saved revision instead of creating a duplicate.
+          setEditing(created);
           // The probe decides whether this form closes. A rejected key or a
           // failed endpoint leaves the filled-in form on screen with the
           // reason; only a passing test returns to the saved card.
@@ -1339,20 +1390,37 @@ export function ModelSettings({
               )}
               {focusSection === "embedding" || advancedEmbedding ? (
                 <>
-                  <label>
-                    {text(
-                      focusSection === "embedding"
-                        ? "搜索模型"
-                        : "向量配置模式",
-                    )}
-                    <select
-                      name="embedding-mode"
+                  <div className="model-form-field">
+                    <label
+                      className="model-option-label"
+                      id={`${fieldId}-embedding-mode-label`}
+                      htmlFor={`${fieldId}-embedding-mode`}
+                    >
+                      {text(
+                        focusSection === "embedding"
+                          ? "搜索模型"
+                          : "向量配置模式",
+                      )}
+                    </label>
+                    <OptionMenu
+                      id={`${fieldId}-embedding-mode`}
+                      labelledBy={`${fieldId}-embedding-mode-label`}
                       value={embeddingMode}
                       disabled={formLocked}
-                      onChange={(event) => {
-                        setEmbeddingMode(
-                          event.target.value as "local" | "custom",
-                        );
+                      className="model-option-menu"
+                      panelClassName="model-option-menu-panel"
+                      options={[
+                        {
+                          value: "local",
+                          label: text("本地内置（推荐）"),
+                        },
+                        {
+                          value: "custom",
+                          label: text("自定义搜索模型"),
+                        },
+                      ]}
+                      onChange={(value) => {
+                        setEmbeddingMode(value as "local" | "custom");
                         setEmbeddingModeChanged(true);
                         setFields((previous) => ({
                           ...previous,
@@ -1360,11 +1428,8 @@ export function ModelSettings({
                           "embedding-base-url": undefined,
                         }));
                       }}
-                    >
-                      <option value="local">{text("本地内置（推荐）")}</option>
-                      <option value="custom">{text("自定义搜索模型")}</option>
-                    </select>
-                  </label>
+                    />
+                  </div>
                   <small>{text("技术类型：Embedding")}</small>
                   {embeddingMode === "custom" ? (
                     <>
@@ -1397,25 +1462,38 @@ export function ModelSettings({
                           </small>
                         )}
                       </label>
-                      <label>
-                        {text("向量模型来源")}
-                        <select
-                          name="embedding-provider"
+                      <div className="model-form-field">
+                        <label
+                          className="model-option-label"
+                          id={`${fieldId}-embedding-provider-label`}
+                          htmlFor={`${fieldId}-embedding-provider`}
+                        >
+                          {text("向量模型来源")}
+                        </label>
+                        <OptionMenu
+                          id={`${fieldId}-embedding-provider`}
+                          labelledBy={`${fieldId}-embedding-provider-label`}
                           value={embeddingProvider}
                           disabled={formLocked}
-                          onChange={(event) => {
-                            const next = event.target.value as
+                          className="model-option-menu"
+                          panelClassName="model-option-menu-panel"
+                          options={[
+                            { value: "ollama", label: text("本机 Ollama") },
+                            {
+                              value: "openai_compatible",
+                              label: text(
+                                "同源 OpenAI-compatible（共用 API Key）",
+                              ),
+                            },
+                          ]}
+                          onChange={(value) => {
+                            const next = value as
                               | "openai_compatible"
                               | "ollama";
                             setEmbeddingProvider(next);
                           }}
-                        >
-                          <option value="ollama">{text("本机 Ollama")}</option>
-                          <option value="openai_compatible">
-                            {text("同源 OpenAI-compatible（共用 API Key）")}
-                          </option>
-                        </select>
-                      </label>
+                        />
+                      </div>
                       <label>
                         {text("向量模型 Base URL")}
                         <input
@@ -1669,13 +1747,14 @@ export function ModelSettings({
             embeddingSwitch.clearError();
             setSwitchTarget(null);
             setSwitchJobId(null);
+            setSwitchRefreshCompleted(null);
           }}
           onStart={() => {
             void embeddingSwitch.startPlannedSwitch().then((job) => {
               if (!job || !alive.current) return;
               setSwitchJobId(job.job_id);
+              setSwitchRefreshCompleted(null);
               setSetupError(null);
-              onSetupComplete?.();
             });
           }}
         />

memhub/apps/web/src/OptionMenu.tsx

@@ -111,7 +111,7 @@ export function OptionMenu(props: {
   function choose(index: number) {
     const item = props.options[index];
     if (!item || item.disabled) return;
-    props.onChange(item.value);
+    if (item.value !== props.value) props.onChange(item.value);
     setOpen(false);
     button.current?.focus();
   }

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

@@ -21,6 +21,9 @@ import { LANGUAGE_STORAGE_KEY, LanguageProvider } from "./gui-language";
 import { queryClient, resetClient } from "./query-client";
 import { R6Settings } from "./R6Settings";
 
+const editProfileAction = vi.fn(() => true);
+const addProfileAction = vi.fn(() => true);
+
 it("changes settings copy live without rereading profiles or changing permissions", async () => {
   const user = userEvent.setup();
   localStorage.setItem(LANGUAGE_STORAGE_KEY, "es");
@@ -112,8 +115,8 @@ vi.mock("./ModelSettings", async () => {
     }) => {
       useImperativeHandle(controlsRef, () => ({
         edit: () => false,
-        editProfile: () => false,
-        add: () => false,
+        editProfile: editProfileAction,
+        add: addProfileAction,
         test: () => Promise.resolve(undefined),
       }));
       return createElement(
@@ -166,6 +169,7 @@ it("shows the help and diagnostics group by default and hides it when opted out"
   await user.click(screen.getByRole("button", { name: "帮助与诊断" }));
   expect(screen.getByRole("heading", { name: "帮助与诊断" })).toBeTruthy();
   expect(screen.getByRole("heading", { name: "导出日志" })).toBeTruthy();
+  expect(screen.getByText(/可能包含会话正文和本机路径/)).toBeTruthy();
   expect(screen.getByRole("heading", { name: "关于" })).toBeTruthy();
   expect(screen.getByText("1.0.4")).toBeTruthy();
   expect(screen.getByText("8")).toBeTruthy();
@@ -304,6 +308,7 @@ it("reports a stop that did not confirm instead of showing it stopped", async ()
 });
 
 it("opens the model area even when a pending probe fences the editor, so recovery stays reachable", async () => {
+  editProfileAction.mockReturnValueOnce(false);
   const user = userEvent.setup();
   render(<R6Settings onDiagnostics={vi.fn()} />);
   await user.click(screen.getByRole("button", { name: "模型配置" }));
@@ -321,10 +326,13 @@ it("opens the model area even when a pending probe fences the editor, so recover
   expect(alert.textContent).toContain("待核对");
   // What changed is that the recovery entry no longer hides with the form it
   // is supposed to unlock.
-  expect(screen.getByRole("button", { name: "核对测试结果" })).toBeTruthy();
+  expect(
+    within(generationLane).getByRole("button", { name: "核对测试结果" }),
+  ).toBeTruthy();
 });
 
 it("reaches the same recovery entry when no model has been configured yet", async () => {
+  addProfileAction.mockReturnValueOnce(false);
   vi.mocked(getModelProfiles).mockResolvedValueOnce([]);
   const user = userEvent.setup();
   render(<R6Settings onDiagnostics={vi.fn()} />);
@@ -368,11 +376,46 @@ it("shows separate compact lanes and opens the requested presentation", async ()
   expect(embeddingLane.textContent).toContain("Embedding");
 
   await user.click(within(embeddingLane).getByRole("button", { name: "配置" }));
-  expect(screen.getByText("embedding")).toBeTruthy();
+  expect(await within(embeddingLane).findByText("embedding")).toBeTruthy();
+  expect(editProfileAction).toHaveBeenCalledTimes(1);
+  expect(editProfileAction).toHaveBeenLastCalledWith(
+    expect.objectContaining({ profile_id: "profile-1" }),
+  );
+  expect(embeddingLane.dataset.expanded).toBe("true");
+  expect(
+    within(embeddingLane)
+      .getByRole("button", { name: "收起" })
+      .getAttribute("aria-expanded"),
+  ).toBe("true");
   await user.click(
     within(generationLane).getByRole("button", { name: "配置" }),
   );
-  expect(screen.getByText("generation")).toBeTruthy();
+  expect(await within(generationLane).findByText("generation")).toBeTruthy();
+  expect(editProfileAction).toHaveBeenCalledTimes(2);
+  expect(generationLane.dataset.expanded).toBe("true");
+  expect(within(embeddingLane).queryByText("embedding")).toBeNull();
+
+  await user.click(
+    within(generationLane).getByRole("button", { name: "收起" }),
+  );
+  expect(
+    within(generationLane)
+      .getByText("generation")
+      .closest(".r6-model-owner")
+      ?.hasAttribute("hidden"),
+  ).toBe(true);
+
+  await user.click(
+    within(generationLane).getByRole("button", { name: "配置" }),
+  );
+  expect(editProfileAction).toHaveBeenCalledTimes(2);
+  expect(
+    within(generationLane)
+      .getByText("generation")
+      .closest(".r6-model-owner")
+      ?.hasAttribute("hidden"),
+  ).toBe(false);
+  expect(screen.queryByRole("alert")).toBeNull();
 });
 
 it("leaves the switch unusable when the state cannot be read", async () => {

memhub/apps/web/src/R6Settings.tsx

@@ -78,6 +78,7 @@ export function R6Settings({
   const [configurationOpen, setConfigurationOpen] = useState(false);
   const [configurationFocus, setConfigurationFocus] =
     useState<ModelSettingsFocusSection>("generation");
+  const [configurationRequest, setConfigurationRequest] = useState(0);
   const [memoryMode, setMemoryModeState] = useState<MemoryPipelineMode | null>(
     null,
   );
@@ -101,6 +102,10 @@ export function R6Settings({
   const [sedimentationUnavailable, setSedimentationUnavailable] =
     useState(false);
   const modelActions = useRef<ModelSettingsActions>(null);
+  const configurationStarted = useRef(false);
+  const configurationStartedFocus = useRef<ModelSettingsFocusSection | null>(
+    null,
+  );
   const quitPanel = useRef<HTMLElement>(null);
   const quitOpener = useRef<HTMLElement | null>(null);
   const currentModel = profiles?.find((profile) => profile.is_active) ?? null;
@@ -224,18 +229,49 @@ export function R6Settings({
   function openConfiguration(
     focusSection: ModelSettingsFocusSection = "generation",
   ) {
+    if (
+      configurationStarted.current &&
+      configurationStartedFocus.current === focusSection
+    ) {
+      setConfigurationFocus(focusSection);
+      setConfigurationOpen(true);
+      return;
+    }
     setConfigurationRefusal(false);
     setConfigurationFocus(focusSection);
+    configurationStarted.current = true;
+    configurationStartedFocus.current = focusSection;
+    setConfigurationOpen(true);
+    setConfigurationRequest((value) => value + 1);
+  }
+
+  // The focused editor can move between the two lanes. Fill it only after
+  // React has mounted it in the requested lane, otherwise an edit sent to the
+  // old lane would be lost during the move.
+  // biome-ignore lint/correctness/useExhaustiveDependencies: configurationRequest is the explicit intent trigger.
+  useEffect(() => {
+    if (!configurationOpen || configurationRequest === 0) return;
     const opened = currentModel
       ? modelActions.current?.editProfile(currentModel)
       : modelActions.current?.add();
-    // The area opens even when the editor refuses: the only recovery entry
-    // lives inside it, and hiding both is what locked people out of their own
-    // settings. They needed the form open to reach “核对测试结果”, and that
-    // button to clear the pending record that kept the form shut. The fence
-    // itself is untouched (see ModelSettings): refusing still returns false.
-    setConfigurationOpen(true);
+    // The area stays open even when the editor refuses: its recovery entry is
+    // the only way to reconcile the pending operation.
     if (!opened) setConfigurationRefusal(true);
+  }, [configurationRequest]);
+
+  function toggleConfiguration(focusSection: ModelSettingsFocusSection) {
+    if (configurationOpen && configurationFocus === focusSection) {
+      setConfigurationOpen(false);
+      return;
+    }
+    openConfiguration(focusSection);
+  }
+
+  function completeConfiguration() {
+    configurationStarted.current = false;
+    configurationStartedFocus.current = null;
+    setConfigurationRefusal(false);
+    setConfigurationOpen(false);
   }
 
   async function quitApplication() {
@@ -384,67 +420,156 @@ export function R6Settings({
               </div>
             )}
             <div className="r6-model-overview">
-              <div className="r6-model-lane">
-                <div className="r6-model-lane-copy">
-                  <h3>{text("记忆整理")}</h3>
-                  <p>{text("把会话理解并整理成可长期使用的记忆。")}</p>
-                  <strong>
-                    {currentModel?.name ??
-                      text(
-                        profiles === null
-                          ? "正在读取模型…"
-                          : modelError
-                            ? "模型状态暂不可读"
-                            : "尚未配置模型",
+              <section
+                className="r6-model-lane"
+                data-expanded={
+                  configurationOpen && configurationFocus === "generation"
+                    ? "true"
+                    : undefined
+                }
+              >
+                <div className="r6-model-lane-header">
+                  <div className="r6-model-lane-copy">
+                    <h3>{text("记忆整理")}</h3>
+                    <p>{text("把会话理解并整理成可长期使用的记忆。")}</p>
+                    <strong>
+                      {currentModel?.name ??
+                        text(
+                          profiles === null
+                            ? "正在读取模型…"
+                            : modelError
+                              ? "模型状态暂不可读"
+                              : "尚未配置模型",
+                        )}
+                    </strong>
+                    {currentModel && (
+                      <small>
+                        LLM · {currentModel.model} ·{" "}
+                        {text(safeEndpoint(currentModel.base_url))}
+                      </small>
+                    )}
+                  </div>
+                  <R6Button
+                    variant="quiet"
+                    disabled={profiles === null}
+                    aria-expanded={
+                      configurationOpen && configurationFocus === "generation"
+                    }
+                    aria-controls={
+                      configurationFocus === "generation"
+                        ? `${id}-generation-configuration`
+                        : undefined
+                    }
+                    onClick={() => toggleConfiguration("generation")}
+                  >
+                    {configurationOpen && configurationFocus === "generation"
+                      ? t("wb.collapse")
+                      : text(currentModel ? "配置" : "配置模型")}
+                  </R6Button>
+                </div>
+                {configurationFocus === "generation" && (
+                  <section
+                    id={`${id}-generation-configuration`}
+                    className="r6-model-owner"
+                    hidden={!configurationOpen}
+                    aria-label={text("模型配置表单")}
+                  >
+                    {configurationRefusal && (
+                      <p role="alert" className="r6-settings-error">
+                        {text(
+                          "上次模型测试或创建还待核对，配置暂时不能改动。请先在下方模型配置区核对结果； 现有配置和原文不会被改动。",
+                        )}
+                      </p>
+                    )}
+                    <ModelSettings
+                      compactSingle
+                      focusSection={configurationFocus}
+                      controlsRef={modelActions}
+                      onProfilesChanged={() =>
+                        setModelVersion((value) => value + 1)
+                      }
+                      onSetupComplete={completeConfiguration}
+                    />
+                  </section>
+                )}
+              </section>
+              <section
+                className="r6-model-lane"
+                data-expanded={
+                  configurationOpen && configurationFocus === "embedding"
+                    ? "true"
+                    : undefined
+                }
+              >
+                <div className="r6-model-lane-header">
+                  <div className="r6-model-lane-copy">
+                    <h3>{text("记忆搜索")}</h3>
+                    <p>{text("在需要时找到与你的问题最相关的记忆。")}</p>
+                    <strong>
+                      {text(
+                        currentModel?.embedding_provider_kind &&
+                          currentModel.embedding_provider_kind !==
+                            "local_onnx" &&
+                          currentModel.embedding_model
+                          ? currentModel.embedding_model
+                          : "本地内置（推荐）",
                       )}
-                  </strong>
-                  {currentModel && (
+                    </strong>
                     <small>
-                      LLM · {currentModel.model} ·{" "}
-                      {text(safeEndpoint(currentModel.base_url))}
+                      {text(
+                        currentModel?.embedding_provider_kind &&
+                          currentModel.embedding_provider_kind !== "local_onnx"
+                          ? `Embedding · ${currentModel.embedding_provider_kind}`
+                          : profiles !== null && !currentModel
+                            ? "Embedding · 配置记忆整理后可自定义"
+                            : "Embedding · 本机处理，无需额外配置",
+                      )}
                     </small>
-                  )}
+                  </div>
+                  <R6Button
+                    variant="quiet"
+                    disabled={profiles === null || !currentModel}
+                    aria-expanded={
+                      configurationOpen && configurationFocus === "embedding"
+                    }
+                    aria-controls={
+                      configurationFocus === "embedding"
+                        ? `${id}-embedding-configuration`
+                        : undefined
+                    }
+                    onClick={() => toggleConfiguration("embedding")}
+                  >
+                    {configurationOpen && configurationFocus === "embedding"
+                      ? t("wb.collapse")
+                      : text("配置")}
+                  </R6Button>
                 </div>
-                <R6Button
-                  variant="quiet"
-                  disabled={profiles === null}
-                  onClick={() => openConfiguration("generation")}
-                >
-                  {text(currentModel ? "配置" : "配置模型")}
-                </R6Button>
-              </div>
-              <div className="r6-model-lane">
-                <div className="r6-model-lane-copy">
-                  <h3>{text("记忆搜索")}</h3>
-                  <p>{text("在需要时找到与你的问题最相关的记忆。")}</p>
-                  <strong>
-                    {text(
-                      currentModel?.embedding_provider_kind &&
-                        currentModel.embedding_provider_kind !== "local_onnx" &&
-                        currentModel.embedding_model
-                        ? currentModel.embedding_model
-                        : "本地内置（推荐）",
+                {configurationFocus === "embedding" && (
+                  <section
+                    id={`${id}-embedding-configuration`}
+                    className="r6-model-owner"
+                    hidden={!configurationOpen}
+                    aria-label={text("模型配置表单")}
+                  >
+                    {configurationRefusal && (
+                      <p role="alert" className="r6-settings-error">
+                        {text(
+                          "上次模型测试或创建还待核对，配置暂时不能改动。请先在下方模型配置区核对结果； 现有配置和原文不会被改动。",
+                        )}
+                      </p>
                     )}
-                  </strong>
-                  <small>
-                    {text(
-                      currentModel?.embedding_provider_kind &&
-                        currentModel.embedding_provider_kind !== "local_onnx"
-                        ? `Embedding · ${currentModel.embedding_provider_kind}`
-                        : profiles !== null && !currentModel
-                          ? "Embedding · 配置记忆整理后可自定义"
-                          : "Embedding · 本机处理，无需额外配置",
-                    )}
-                  </small>
-                </div>
-                <R6Button
-                  variant="quiet"
-                  disabled={profiles === null || !currentModel}
-                  onClick={() => openConfiguration("embedding")}
-                >
-                  {text("配置")}
-                </R6Button>
-              </div>
+                    <ModelSettings
+                      compactSingle
+                      focusSection={configurationFocus}
+                      controlsRef={modelActions}
+                      onProfilesChanged={() =>
+                        setModelVersion((value) => value + 1)
+                      }
+                      onSetupComplete={completeConfiguration}
+                    />
+                  </section>
+                )}
+              </section>
             </div>
             {modelError && (
               <R6StateRecovery
@@ -456,13 +581,6 @@ export function R6Settings({
                 onRetry={() => setModelVersion((value) => value + 1)}
               />
             )}
-            {configurationRefusal && (
-              <p role="alert" className="r6-settings-error">
-                {text(
-                  "上次模型测试或创建还待核对，配置暂时不能改动。请先在下方模型配置区核对结果； 现有配置和原文不会被改动。",
-                )}
-              </p>
-            )}
             {currentModel && memoryMode !== "lite" && (
               <div className="r6-setting-row">
                 <div>
@@ -499,19 +617,6 @@ export function R6Settings({
                 )}
               </p>
             )}
-            <section
-              className="r6-model-owner"
-              hidden={!configurationOpen}
-              aria-label={text("模型配置表单")}
-            >
-              <ModelSettings
-                compactSingle
-                focusSection={configurationFocus}
-                controlsRef={modelActions}
-                onProfilesChanged={() => setModelVersion((value) => value + 1)}
-                onSetupComplete={() => setConfigurationOpen(false)}
-              />
-            </section>
           </section>
           <section
             id={`${id}-panel-connections`}

memhub/apps/web/src/R6SettingsLayout.test.ts

@@ -0,0 +1,22 @@
+import { readFileSync } from "node:fs";
+import { expect, it } from "vitest";
+
+function declaration(file: string, selector: string): string {
+  const css = readFileSync(new URL(`./${file}`, import.meta.url), "utf8");
+  const escaped = selector.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+  return css.match(new RegExp(`${escaped}\\s*\\{([^}]*)\\}`))?.[1] ?? "";
+}
+
+it("fills settings controls without widening shared fallback controls", () => {
+  for (const selector of [
+    ".r6-memory-mode-settings .memory-mode-choice-list",
+    ".r6-settings .connection-fallback .custom-harness-name",
+  ]) {
+    const rule = declaration("r6-settings.css", selector);
+    expect(rule, selector).toMatch(/(?:^|[\s;])width:\s*100%\s*;/);
+    expect(rule, selector).toMatch(/(?:^|[\s;])max-width:\s*none\s*;/);
+  }
+  expect(declaration("core-visual.css", ".custom-harness-name")).toMatch(
+    /(?:^|[\s;])max-width:\s*420px\s*;/,
+  );
+});

memhub/apps/web/src/SdkRawHistory.tsx

@@ -3,6 +3,7 @@ import { type FormEvent, useEffect, useRef, useState } from "react";
 import { getSdkRaw, searchSdkRaw } from "./api";
 import { useGuiLanguage } from "./gui-language";
 import { subscribePoll } from "./poll-scheduler";
+import { rawRecordId } from "./sdk-raw-map";
 
 export function SdkRawHistory({ active = true }: { active?: boolean }) {
   const { locale, t, tn } = useGuiLanguage();
@@ -121,7 +122,7 @@ export function SdkRawHistory({ active = true }: { active?: boolean }) {
       ) : null}
       <ol>
         {items.map((item) => (
-          <li key={item.raw_id}>
+          <li key={rawRecordId(item)}>
             <p>{item.content}</p>
             <small>
               {item.source} · {item.session_id}
@@ -140,6 +141,9 @@ function asContextItem(item: SdkRawRecord): SdkRawContextItem {
   return {
     kind: "raw",
     raw_id: item.raw_id,
+    turn_id: item.turn_id,
+    turn_index: item.turn_index,
+    turn_count: item.turn_count,
     content: item.content,
     source: item.source,
     session_id: item.session_id,

memhub/apps/web/src/SelectedRawDetail.tsx

@@ -5,7 +5,7 @@ import type {
 } from "@one-memory/contracts";
 import { clockStamp } from "./clock-stamp";
 import { useGuiLanguage } from "./gui-language";
-import { rawRecordTime } from "./sdk-raw-map";
+import { rawRecordId, rawRecordTime } from "./sdk-raw-map";
 import { sourceNameOr, UNATTRIBUTED_SOURCE_ID } from "./source-label";
 
 type RawRecord = SdkRawRecord | SdkRawContextItem;
@@ -50,7 +50,7 @@ export function SelectedRawDetail({
       <p className="selected-memory-status">{t("memory:原始对话")}</p>
       <ul className="map-transcript">
         {messages.map((message) => (
-          <li key={`${record.raw_id}:${message.role}:${message.content}`}>
+          <li key={`${rawRecordId(record)}:${message.role}:${message.content}`}>
             <small>{authorLabel(message.role, t)}</small>
             <p>{message.content}</p>
           </li>

memhub/apps/web/src/api.raw.test.ts

@@ -32,6 +32,7 @@ it("lists raw conversations for the selected application", async () => {
       session_id: undefined,
       offset: 20,
       limit: 20,
+      unit: "turn",
     },
     signal,
   });
@@ -49,6 +50,7 @@ it("keeps the application filter when searching raw conversations", async () =>
       query: "release plan",
       kinds: ["raw"],
       include_sources: ["claude-code"],
+      raw_unit: "turn",
       limit: 20,
       character_budget: 12_000,
     },

memhub/apps/web/src/api.ts

@@ -986,16 +986,18 @@ export async function getSdkRaw({
   sessionId,
   offset = 0,
   limit = 100,
+  unit = "turn",
   signal,
 }: {
   source?: string;
   sessionId?: string;
   offset?: number;
   limit?: number;
+  unit?: "raw" | "turn";
   signal?: AbortSignal;
 } = {}): Promise<SdkRawPage> {
   const response = await client.get<SdkRawPage>("/api/v2/raw", {
-    params: { source, session_id: sessionId, offset, limit },
+    params: { source, session_id: sessionId, offset, limit, unit },
     signal,
   });
   return response.data;
@@ -1011,6 +1013,7 @@ export async function searchSdkRaw(
       query,
       kinds: ["raw"],
       ...(options.source ? { include_sources: [options.source] } : {}),
+      raw_unit: "turn",
       limit: 20,
       character_budget: 12_000,
     },

memhub/apps/web/src/core-visual.css

@@ -132,12 +132,17 @@ button svg {
   content: "";
   position: absolute;
   right: 18px;
-  top: 22px;
+  top: 50%;
   width: 8px;
   height: 8px;
   border-right: 1.5px solid #4e6478;
   border-bottom: 1.5px solid #4e6478;
-  transform: rotate(45deg);
+  transform: translateY(-65%) rotate(45deg);
+  transform-origin: 50% 50%;
+  transition: transform 150ms ease-out;
+}
+.om-option-menu-button[aria-expanded="true"]::after {
+  transform: translateY(-35%) rotate(225deg);
 }
 .om-option-menu-button:disabled {
   color: #66727b;
@@ -981,6 +986,14 @@ button.map-tool {
   background: #fff;
   box-shadow: var(--shadow-pop);
 }
+.memory-graph-fallback-topic {
+  display: block;
+  width: 100%;
+  margin-block: 6px;
+  padding: 8px 14px;
+  color: var(--tx2);
+  overflow-wrap: anywhere;
+}
 .graph-labels {
   position: absolute;
   inset: 0;

memhub/apps/web/src/onboarding-r6.css

@@ -76,7 +76,6 @@
   text-overflow: ellipsis;
 }
 .onboarding-r6-titlebar .r6-language-option-menu .om-option-menu-button::after {
-  top: 10px;
   right: 11px;
 }
 .onboarding-r6-titlebar

memhub/apps/web/src/r6-settings.css

@@ -66,7 +66,6 @@
   text-overflow: ellipsis;
 }
 .r6-settings .r6-language-option-menu .om-option-menu-button::after {
-  top: 16px;
   right: 16px;
 }
 .r6-settings
@@ -229,7 +228,8 @@
   line-height: 22px;
 }
 .r6-memory-mode-settings .memory-mode-choice-list {
-  max-width: 760px;
+  width: 100%;
+  max-width: none;
 }
 .r6-settings-note {
   margin: 14px 0 0;
@@ -482,7 +482,8 @@
   text-wrap: pretty;
 }
 .r6-settings .connection-fallback .custom-harness-name {
-  max-width: 560px;
+  width: 100%;
+  max-width: none;
   gap: 8px;
   margin: 0 0 14px;
   color: #4f6373;
@@ -596,6 +597,12 @@
   background: #fff;
 }
 .r6-settings .r6-model-lane {
+  min-height: 108px;
+}
+.r6-settings .r6-model-lane[data-expanded="true"] {
+  background: #f8fafb;
+}
+.r6-settings .r6-model-lane-header {
   display: grid;
   grid-template-columns: minmax(0, 1fr) auto;
   align-items: center;
@@ -639,7 +646,7 @@
   font-size: 12px;
   line-height: 20px;
 }
-.r6-settings .r6-model-lane > .r6-button {
+.r6-settings .r6-model-lane-header > .r6-button {
   min-width: 72px;
 }
 .r6-settings .r6-model-recipient {
@@ -690,8 +697,9 @@
   line-height: 28px;
 }
 .r6-settings .r6-model-owner {
-  padding-top: 24px;
+  padding: 24px 20px;
   border-top: 1px solid #d8e2ea;
+  background: #f8fafb;
 }
 .r6-settings .r6-model-owner[hidden] {
   display: none;
@@ -847,7 +855,6 @@
   font-size: 14px;
 }
 .r6-settings .model-form .om-option-menu-button::after {
-  top: 16px;
   right: 16px;
 }
 .r6-settings .model-form .om-option-menu-button[aria-expanded="true"] {
@@ -866,6 +873,39 @@
   outline: 2px solid var(--r6-focus);
   outline-offset: 2px;
 }
+.r6-settings .model-option-label {
+  display: block;
+  margin: 0;
+}
+.model-option-menu-panel {
+  padding: 6px;
+  border-color: var(--r6-control-edge, #c0ccd5);
+  box-shadow: 0 12px 32px rgb(48 71 87 / 13%);
+}
+.model-option-menu-panel [role="option"] {
+  min-height: 40px;
+  padding: 8px 10px;
+  color: #304757;
+  font-size: 14px;
+  line-height: 22px;
+}
+.model-option-menu-panel [role="option"][aria-selected="true"] {
+  background: #dfe9ef;
+  font-weight: 500;
+}
+.om-option-menu-panel.model-option-menu-panel
+  [role="option"][data-active="true"] {
+  color: #304757;
+  background: #e9f0f4;
+}
+.om-option-menu-panel.model-option-menu-panel
+  [role="option"]:hover:not(:disabled) {
+  color: #304757;
+  background: #e9f0f4;
+}
+.model-option-menu-panel [role="option"]:active:not(:disabled) {
+  background: #d4e2ea;
+}
 .r6-settings .model-advanced {
   grid-column: 1 / -1;
 }
@@ -884,7 +924,8 @@
   font-size: 13px;
   line-height: 21px;
 }
-.r6-settings .model-advanced label {
+.r6-settings .model-advanced label,
+.r6-settings .model-advanced .model-form-field {
   display: flex;
   flex-direction: column;
   gap: 8px;
@@ -988,14 +1029,17 @@
     padding: 24px 20px;
     border-radius: 0 0 11px 11px;
   }
-  .r6-settings .r6-model-lane {
+  .r6-settings .r6-model-lane-header {
     grid-template-columns: minmax(0, 1fr);
     gap: 12px;
     padding: 16px;
   }
-  .r6-settings .r6-model-lane > .r6-button {
+  .r6-settings .r6-model-lane-header > .r6-button {
     justify-self: start;
   }
+  .r6-settings .r6-model-owner {
+    padding: 20px 16px;
+  }
   .r6-settings .r6-language-setting-row {
     grid-template-columns: minmax(0, 1fr);
     gap: 8px;

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

@@ -76,4 +76,28 @@ describe("raw map projection", () => {
       history: [],
     });
   });
+
+  it("uses a turn identity while retaining the parent raw id", () => {
+    const turn = {
+      ...record,
+      raw_id: "raw-batch",
+      turn_id: "raw-turn:batch:1",
+      turn_index: 1,
+      turn_count: 2,
+      content: "[user]: second question\n[assistant]: second answer",
+      messages: [
+        { role: "user", content: "second question" },
+        { role: "assistant", content: "second answer" },
+      ],
+    };
+
+    expect(
+      rawLibraryPage({ items: [turn], total: 2, offset: 0, limit: 20 }),
+    ).toMatchObject({
+      items: [{ memory_id: "raw-turn:batch:1", content: turn.content }],
+      source_ids_by_memory: { "raw-turn:batch:1": ["cursor"] },
+    });
+    expect(rawDetail(turn).memory.memory_id).toBe("raw-turn:batch:1");
+    expect(turn.raw_id).toBe("raw-batch");
+  });
 });

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

@@ -9,6 +9,10 @@ import type {
 
 export type RawMapRecord = SdkRawRecord | SdkRawContextItem;
 
+export function rawRecordId(record: RawMapRecord): string {
+  return record.turn_id || record.raw_id;
+}
+
 const EARLIEST_PLAUSIBLE = Date.UTC(2000, 0, 1);
 const LATEST_PLAUSIBLE = () => Date.now() + 86_400_000;
 
@@ -43,7 +47,7 @@ function sourceOf(record: RawMapRecord): string {
 export function rawMapMemory(record: RawMapRecord): GovernedMemory {
   const created = rawRecordTime(record) ?? "";
   return {
-    memory_id: record.raw_id,
+    memory_id: rawRecordId(record),
     user_id: "user_id" in record ? record.user_id : "local",
     kind: "chat",
     status: "confirmed",
@@ -75,17 +79,18 @@ function pageFor(
     { at: string; basis: "conversation" | "import" }
   > = {};
   for (const record of records) {
-    sourceIds[record.raw_id] = [sourceOf(record)];
+    const recordId = rawRecordId(record);
+    sourceIds[recordId] = [sourceOf(record)];
     const conversationAt = isoTime(record.memory_at);
     const importedAt =
       "gmt_created" in record ? isoTime(record.gmt_created) : "";
     if (conversationAt)
-      sourceTime[record.raw_id] = {
+      sourceTime[recordId] = {
         at: conversationAt,
         basis: "conversation",
       };
     else if (importedAt)
-      sourceTime[record.raw_id] = { at: importedAt, basis: "import" };
+      sourceTime[recordId] = { at: importedAt, basis: "import" };
   }
   return {
     items: records.map(rawMapMemory),

memhub/apps/web/src/selected-memory.css

@@ -97,7 +97,6 @@
   border-color: transparent;
 }
 .selected-source-filter .om-option-menu-button::after {
-  top: 18px;
   right: 4px;
 }
 .selected-source-filter details {

memhub/docs/ARCHITECTURE.md

@@ -283,7 +283,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.13` is the current stable SDK dependency; the release
+- `hy-memory==2.0.14` 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/INSTALLATION.md

@@ -1,6 +1,6 @@
 # MemHub 安装与发行
 
-MemHub `v1.0.6`（build 40）的发行目标如下：
+MemHub `v1.0.6`（build 41）的发行目标如下：
 
 | 平台 | 发行物 | 构建与验收 |
 | --- | --- | --- |

memhub/docs/PRODUCT_LOG_TROUBLESHOOTING.md

@@ -10,9 +10,10 @@
 新增受限标量标签，仍应按私人数据处理，不能直接对外发送。`debug` 会让 HY 的
 `hy_memory.log` 记录 DEBUG 行，其中可能包含完整 prompt/response 和获授权的对话正文。
 
-排查时只通过 App 的“导出日志”提供文件：导出过程会按安全白名单重新生成诊断事件，不会打包
-原始日志。不要直接发送 `diagnostic-*.jsonl`、整个 `~/.one-memory` 数据目录或 debug 级别的
-`hy_memory.log`；后者如确需分享，必须由开发人员另行确认并脱敏。
+App 的“导出日志”会同时打包安全投影和 `~/.one-memory/logs/` 下的完整运行日志，供可信开发人员
+还原本机故障。原始日志可能包含会话正文和本机路径，发送前必须确认接收方；不要公开分享，也不要
+发送整个 `~/.one-memory` 数据目录。SDK 的 debug 级 `hy_memory.log` 不在该 ZIP 中；如确需分享，
+必须由开发人员另行确认并脱敏。只需要脱敏事件时使用 `memhub logs export`。
 
 ## 1. Log level 设什么
 
@@ -41,8 +42,8 @@ debug 可能把完整模型 prompt/response 写入本机调试日志，不得长
 
 ## 2. 日志在哪里
 
-App 内推荐直接使用“运行诊断 → 导出日志”。开发人员远程指导时，优先使用
-`memhub logs`，它会再次过滤不安全字段。原始结构化日志只用于本机现场查看，不得作为附件发送。
+App 内推荐直接使用“运行诊断 → 导出日志”，它会生成包含完整运行日志的 ZIP。开发人员远程指导
+且不需要原始日志时，优先使用 `memhub logs`，它会过滤不安全字段。不要手工打包整个数据目录。
 Core 日志默认位于：
 
 ```text

memhub/docs/contracts/diagnostic-logging.md

@@ -45,13 +45,18 @@ host-bridge、desktop 和 cli。单文件 4 MiB，保留 3 份轮转；额外清
 读取器只枚举上述固定文件名的 regular file，不跟随符号链接，不接受任意路径。日常本地查询
 遇到损坏行时跳过并计数，不让一行损坏遮蔽其余证据。
 
-导出是独立的安全边界：它重新读取并校验全部候选事件，只把明确列入导出白名单的字段投影为
+CLI 导出是独立的安全边界：它重新读取并校验全部候选事件，只把明确列入导出白名单的字段投影为
 新的 `diagnostics/events.jsonl`；details 不只检查 key，还按字段约束 value 类型，并拒绝自由文本、
-路径和凭证形态的字符串。绝不复制原始 `diagnostic-*.jsonl`、普通运行日志或崩溃报告。
-导出包还可以包含固定 schema 的系统摘要和问题时间点，但不接受调用方传入的任意对象展开。
+路径和凭证形态的字符串。CLI 不复制原始日志或崩溃报告。
+
+Desktop 设置页的“导出日志”用于开发人员排查本机运行故障。除上述安全投影和状态摘要外，它还会
+把 One Memory Home 的 `logs/` 目录中全部普通文件及轮转文件放入 ZIP 的 `logs/` 目录。它不跟随
+符号链接、不递归读取其他目录，也不包含数据库、SDK 数据目录或系统崩溃报告。原始日志可能包含
+会话正文、本机路径或其他私人信息，ZIP 的 README 必须明确提示只发送给可信的开发人员。
+
 任何事件、marker、安全投影或日志目录读取失败时，导出必须 fail-closed：删除临时文件、不发布
-诊断包，更不能回退为打包原始日志。发布使用已设好权限的临时文件原子创建最终目标，已存在的
-目标不得覆盖，发布后也不存在再 chmod 失败留下伪成功文件的窗口。
+诊断包。发布使用已设好权限的临时文件原子创建最终目标，已存在的目标不得覆盖，发布后也不存在
+再 chmod 失败留下伪成功文件的窗口。
 
 ## 覆盖与节流
 

memhub/docs/contracts/memory-mode-evolve-v1.md

@@ -12,7 +12,7 @@ memory route. `user_id` is optional (default `local`) and must match
 
 | Mode | SDK client | What happens to a conversation | What search reads |
 |---|---|---|---|
-| `lite` | `mode="lite"` | Stored as one raw turn with one embedding. No model call; no generation model needs to be configured. | Raw turns only (the SDK forces `L1_RAW`). |
+| `lite` | `mode="lite"` | Stored in embedded L1 raw batches. No model call; no generation model needs to be configured. The product page projects each batch back into individual conversation turns. | Plain vector RAG over raw (the SDK forces `L1_RAW`), active and shadow, so history pro already processed stays findable. Every hit above the floor is returned by score; the turn view shows the turns containing the query, or else the hit's closest turn. |
 | `pro` (default) | `mode="pro"` | Stored as raw, then a model extracts memories (System 1). The raw becomes SHADOW once extraction succeeds. | Extracted memories (facts, profile, intentions). |
 | `ultra` | `mode="ultra"` | Same as pro, and new facts are also projected into the knowledge graph. Every night, evolve runs System 2 for every user. | Same as pro, plus the graph-backed lanes below. |
 

memhub/package.json

@@ -2,7 +2,7 @@
   "name": "one-memory",
   "version": "1.0.6",
   "release": {
-    "build_number": 40,
+    "build_number": 41,
     "macos_min_version": "13.0",
     "app_update": "manual",
     "runtime_manifest_schema": 1,

memhub/packages/contracts/src/context.ts

@@ -87,7 +87,12 @@ export interface ContextSearchResponse {
 
 export interface SdkRawContextItem {
   kind: "raw";
+  /** Parent SDK L1 raw storage batch. */
   raw_id: string;
+  /** Stable product identity of one turn inside `raw_id`, when requested. */
+  turn_id?: string | null;
+  turn_index?: number | null;
+  turn_count?: number | null;
   content: string;
   source: string;
   session_id: string;

memhub/packages/contracts/src/index.ts

@@ -675,7 +675,12 @@ export interface SdkUserProfileResponse {
 }
 
 export interface SdkRawRecord {
+  /** Parent SDK L1 raw storage batch. */
   raw_id: string;
+  /** Stable product identity of one turn inside `raw_id`, when unit=turn. */
+  turn_id?: string | null;
+  turn_index?: number | null;
+  turn_count?: number | null;
   content: string;
   user_id: string;
   agent_id: string;
@@ -703,6 +708,7 @@ export interface SdkRawPage {
   total: number;
   limit: number;
   offset: number;
+  unit?: "raw" | "turn";
   /** A time range was applied to this page in Core (the SDK list has none). */
   time_filtered?: boolean;
 }
@@ -1112,6 +1118,11 @@ export interface MemoryProcessingSummary {
   llm_backoff_until?: string | null;
   /** The provider failure behind the pause, as a stable code. */
   llm_backoff_error?: string | null;
+  /** The durable write drain: `restarting` after an unexpected exit, until it
+   * runs again. Null before startup recovery started it. */
+  write_drain?: "running" | "restarting" | "stopped" | null;
+  /** Why the drain last exited unexpectedly, as a stable code. */
+  write_drain_error?: string | null;
   /** The active memory pipeline and its graph/evolve status. */
   pipeline?: MemoryPipelineStatus | null;
   /** A running embedding switch or startup merge pauses mode mutations. */

memhub/packages/i18n/src/gui-i18n-catalogs.test.ts

@@ -275,9 +275,10 @@ describe("Mode catalog", () => {
       "搜索沉淀记忆，并用知识图谱补充关联。",
       "搜索沉淀后的记忆。",
       "已经处理完的旧对话不会补做；原始对话全部保留。",
-      "已有记忆继续保留，但高级记忆在当前模式下不显示、不参与召回。",
+      "知识图谱",
+      "普通沉淀记忆仍可使用；深度记忆新增的更高层记忆和图谱数据继续保留，但在智能记忆下不显示、不参与召回。原始对话不会删除。",
       "已有记忆会在后台加入知识图谱，期间仍可使用。",
-      "普通沉淀记忆仍可使用；更高层记忆和图谱数据继续保留，但不参与当前召回；每天自动整合会停止。",
+      "每天自动整合会停止，已经生成的数据不会删除。",
       "记忆维护正在进行，完成后才能切换模式。",
       "模式切换影响暂时无法读取，请重试。",
     ] as const;

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

@@ -877,7 +877,7 @@ const baseEnglishMessages = {
     "Keep syncing, processing and providing memories without opening the app.",
   导出日志: "Export logs",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Package local diagnostic logs into a file to share with support.",
+    "Package all local logs. They may contain conversation text or local paths; share only with trusted developers.",
   "确定退出 One Memory 界面吗？": "Quit the MemHub window?",
   "后台服务会继续同步、整理和提供记忆。":
     "The background service will continue syncing, processing and providing memories.",

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

@@ -47,7 +47,7 @@ const modeChineseMessages = {
   "以前生成的高级记忆会继续保留，但在 lite 下不会显示或参与记忆召回。":
     "高级记忆会保留，但在 lite 下不显示、不参与召回。",
   "每天自动整合会停止，已经生成的数据不会删除。":
-    "自动整合会停止，已有数据保留。",
+    "每天自动整合会停止；已生成的整合结果继续保留。",
   "已有高级记忆会在后台加入知识图谱，期间可以继续使用 MemHub。":
     "已有高级记忆将在后台加入知识图谱。",
   "知识图谱仍在补建，不会重复启动。": "知识图谱仍在补建，不会重复启动。",
@@ -97,6 +97,7 @@ const modeChineseMessages = {
   模型调用与费用: "模型调用与费用",
   搜索范围: "搜索范围",
   已有数据: "已有数据",
+  知识图谱: "知识图谱",
   切换后: "切换后",
   "之后只保存原始对话，不再生成新记忆。":
     "之后只保存原始对话，不再生成新记忆。",
@@ -295,6 +296,7 @@ export const modeEnglishMessages: ModeMessages = {
   模型调用与费用: "Model calls and cost",
   搜索范围: "Search scope",
   已有数据: "Existing data",
+  知识图谱: "Knowledge graph",
   切换后: "After switching",
   "之后只保存原始对话，不再生成新记忆。":
     "New conversations are saved as original text only; no new memories are created.",
@@ -498,6 +500,7 @@ export const modeSpanishMessages: ModeMessages = {
   模型调用与费用: "Llamadas al modelo y coste",
   搜索范围: "Ámbito de búsqueda",
   已有数据: "Datos existentes",
+  知识图谱: "Grafo de conocimiento",
   切换后: "Después del cambio",
   "之后只保存原始对话，不再生成新记忆。":
     "Las conversaciones nuevas se guardan solo como texto original; no se crean memorias nuevas.",
@@ -693,6 +696,7 @@ export const modeTraditionalChineseMessages: ModeMessages = {
   模型调用与费用: "模型呼叫與費用",
   搜索范围: "搜尋範圍",
   已有数据: "現有資料",
+  知识图谱: "知識圖譜",
   切换后: "切換後",
   "之后只保存原始对话，不再生成新记忆。":
     "之後只儲存原始對話，不再產生新記憶。",
@@ -843,6 +847,7 @@ export const modeJapaneseMessages: ModeMessages = {
   模型调用与费用: "モデル呼び出しと料金",
   搜索范围: "検索範囲",
   已有数据: "既存データ",
+  知识图谱: "ナレッジグラフ",
   切换后: "切り替え後",
   "之后只保存原始对话，不再生成新记忆。":
     "以降は元の会話のみを保存し、新しい記憶は生成しません。",
@@ -1043,6 +1048,7 @@ export const modeGermanMessages: ModeMessages = {
   模型调用与费用: "Modellaufrufe und Kosten",
   搜索范围: "Suchbereich",
   已有数据: "Vorhandene Daten",
+  知识图谱: "Wissensgraph",
   切换后: "Nach dem Wechsel",
   "之后只保存原始对话，不再生成新记忆。":
     "Neue Unterhaltungen werden nur noch im Original gespeichert; es werden keine neuen Erinnerungen erstellt.",
@@ -1246,6 +1252,7 @@ export const modePortugueseMessages: ModeMessages = {
   模型调用与费用: "Chamadas ao modelo e custos",
   搜索范围: "Âmbito da pesquisa",
   已有数据: "Dados existentes",
+  知识图谱: "Grafo de conhecimento",
   切换后: "Após a mudança",
   "之后只保存原始对话，不再生成新记忆。":
     "As novas conversas passam a ser guardadas apenas como texto original; não são criadas novas memórias.",
@@ -1451,6 +1458,7 @@ export const modeFrenchMessages: ModeMessages = {
   模型调用与费用: "Appels au modèle et coût",
   搜索范围: "Périmètre de recherche",
   已有数据: "Données existantes",
+  知识图谱: "Graphe de connaissances",
   切换后: "Après le changement",
   "之后只保存原始对话，不再生成新记忆。":
     "Les nouvelles conversations sont uniquement enregistrées dans leur forme originale ; aucune nouvelle mémoire n’est créée.",
@@ -1651,6 +1659,7 @@ export const modeKoreanMessages: ModeMessages = {
   模型调用与费用: "모델 호출 및 비용",
   搜索范围: "검색 범위",
   已有数据: "기존 데이터",
+  知识图谱: "지식 그래프",
   切换后: "전환 후",
   "之后只保存原始对话，不再生成新记忆。":
     "이후 새 대화는 원본만 저장되며 새 메모리는 생성되지 않습니다.",
@@ -1850,6 +1859,7 @@ export const modeRussianMessages: ModeMessages = {
   模型调用与费用: "Вызовы модели и расходы",
   搜索范围: "Область поиска",
   已有数据: "Существующие данные",
+  知识图谱: "Граф знаний",
   切换后: "После переключения",
   "之后只保存原始对话，不再生成新记忆。":
     "Новые диалоги будут сохраняться только в исходном виде; новые воспоминания создаваться не будут.",
@@ -2052,6 +2062,7 @@ export const modeIndonesianMessages: ModeMessages = {
   模型调用与费用: "Panggilan model dan biaya",
   搜索范围: "Cakupan pencarian",
   已有数据: "Data yang ada",
+  知识图谱: "Graf pengetahuan",
   切换后: "Setelah beralih",
   "之后只保存原始对话，不再生成新记忆。":
     "Percakapan baru hanya disimpan sebagai teks asli; tidak ada memori baru yang dibuat.",

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

@@ -251,7 +251,7 @@ export const settingsSpanishMessages = {
     "Sincroniza, organiza y ofrece memorias sin abrir la interfaz.",
   导出日志: "Exportar registros",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Empaqueta los registros locales en un archivo para su diagnóstico.",
+    "Empaqueta todos los registros locales. Pueden contener conversaciones o rutas privadas; compártelos solo con personal de desarrollo de confianza.",
   "确定退出 One Memory 界面吗？": "¿Cerrar la interfaz de MemHub?",
   "后台服务会继续同步、整理和提供记忆。":
     "El servicio en segundo plano seguirá sincronizando, organizando y ofreciendo memorias.",
@@ -362,6 +362,8 @@ export const settingsSpanishMessages = {
 const chineseOverrides: Partial<
   Record<keyof typeof settingsSpanishMessages, string>
 > = {
+  "把本机运行日志打包成一个文件，发给开发排查。":
+    "打包完整本机日志，可能包含会话正文和本机路径；只发送给可信的开发人员。",
   "确定退出 One Memory 界面吗？": "确定退出 MemHub 界面吗？",
   appSettingsTitle: "{name} 应用设置",
   appMemoryScope: "让 {name} 查询已授权范围内的记忆。",

memhub/packages/i18n/src/locales/gui-i18n-de.ts

@@ -1065,7 +1065,7 @@ export const germanMessages: Record<MessageKey, string> = {
     "Erinnerungen weiter synchronisieren, verarbeiten und bereitstellen, ohne die App zu öffnen.",
   导出日志: "Protokolle exportieren",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Packen Sie lokale Diagnoseprotokolle in eine Datei, um sie an den Support zu senden.",
+    "Packt alle lokalen Protokolle. Sie können Gesprächsinhalte oder lokale Pfade enthalten; nur mit vertrauenswürdigen Entwicklern teilen.",
   "确定退出 One Memory 界面吗？": "MemHub-Fenster beenden?",
   "后台服务会继续同步、整理和提供记忆。":
     "Der Hintergrunddienst synchronisiert, verarbeitet und liefert Erinnerungen weiter.",

memhub/packages/i18n/src/locales/gui-i18n-fr.ts

@@ -1078,7 +1078,7 @@ export const frenchMessages: Record<MessageKey, string> = {
     "Continuez à synchroniser, traiter et fournir les mémoires sans ouvrir l’application.",
   导出日志: "Exporter les journaux",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Regroupez les journaux de diagnostic locaux dans un fichier à transmettre à l’assistance.",
+    "Regroupe tous les journaux locaux. Ils peuvent contenir des conversations ou des chemins locaux ; partagez-les uniquement avec des développeurs de confiance.",
   "确定退出 One Memory 界面吗？": "Quitter la fenêtre MemHub ?",
   "后台服务会继续同步、整理和提供记忆。":
     "Le service en arrière-plan continuera de synchroniser, traiter et fournir les mémoires.",

memhub/packages/i18n/src/locales/gui-i18n-id.ts

@@ -1048,7 +1048,7 @@ export const indonesianMessages: Record<MessageKey, string> = {
     "Terus sinkronkan, proses, dan sediakan memori tanpa membuka aplikasi.",
   导出日志: "Ekspor log",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Kemas log diagnostik lokal menjadi satu file untuk dibagikan ke tim dukungan.",
+    "Kemas semua log lokal. Log dapat memuat isi percakapan atau jalur lokal; bagikan hanya kepada pengembang tepercaya.",
   "确定退出 One Memory 界面吗？": "Tutup jendela MemHub?",
   "后台服务会继续同步、整理和提供记忆。":
     "Layanan latar belakang akan terus menyinkronkan, memproses, dan menyediakan memori.",

memhub/packages/i18n/src/locales/gui-i18n-ja.ts

@@ -1045,7 +1045,7 @@ export const japaneseMessages: Record<MessageKey, string> = {
     "アプリを開かなくても、記憶の同期、処理、提供を続けます。",
   导出日志: "ログを書き出す",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "ローカルの診断ログを 1 つのファイルにまとめ、サポートに共有できます。",
+    "ローカルログをすべてまとめます。会話本文やローカルパスが含まれる場合があるため、信頼できる開発者にのみ共有してください。",
   "确定退出 One Memory 界面吗？": "MemHub のウインドウを終了しますか？",
   "后台服务会继续同步、整理和提供记忆。":
     "バックグラウンドサービスは記憶の同期、処理、提供を続けます。",

memhub/packages/i18n/src/locales/gui-i18n-ko.ts

@@ -1021,7 +1021,7 @@ export const koreanMessages: Record<MessageKey, string> = {
     "앱을 열지 않아도 기억을 계속 동기화, 정리하고 제공합니다.",
   导出日志: "로그 내보내기",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "로컬 진단 로그를 파일 하나로 묶어 지원팀에 보냅니다.",
+    "모든 로컬 로그를 묶습니다. 대화 내용이나 로컬 경로가 포함될 수 있으므로 신뢰할 수 있는 개발자에게만 공유하세요.",
   "确定退出 One Memory 界面吗？": "MemHub 창을 종료하시겠습니까?",
   "后台服务会继续同步、整理和提供记忆。":
     "백그라운드 서비스는 계속 기억을 동기화, 정리하고 제공합니다.",

memhub/packages/i18n/src/locales/gui-i18n-pt.ts

@@ -1069,7 +1069,7 @@ export const portugueseMessages: Record<MessageKey, string> = {
     "Continue sincronizando, processando e fornecendo memórias sem abrir o app.",
   导出日志: "Exportar logs",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Empacote os logs de diagnóstico locais em um arquivo para compartilhar com o suporte.",
+    "Empacota todos os logs locais. Eles podem conter conversas ou caminhos locais; compartilhe apenas com desenvolvedores de confiança.",
   "确定退出 One Memory 界面吗？": "Fechar a janela do MemHub?",
   "后台服务会继续同步、整理和提供记忆。":
     "O serviço em segundo plano continuará sincronizando, processando e fornecendo memórias.",

memhub/packages/i18n/src/locales/gui-i18n-ru.ts

@@ -1053,7 +1053,7 @@ export const russianMessages: Record<MessageKey, string> = {
     "Синхронизация, обработка и предоставление памяти без открытия приложения.",
   导出日志: "Экспорт журналов",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "Упакуйте локальные журналы диагностики в файл, чтобы отправить в поддержку.",
+    "Упаковывает все локальные журналы. Они могут содержать текст диалогов или локальные пути; отправляйте их только доверенным разработчикам.",
   "确定退出 One Memory 界面吗？": "Закрыть окно MemHub?",
   "后台服务会继续同步、整理和提供记忆。":
     "Фоновая служба продолжит синхронизацию, обработку и предоставление памяти.",

memhub/packages/i18n/src/locales/gui-i18n-zh-hant.ts

@@ -951,7 +951,7 @@ export const traditionalChineseMessages: Record<MessageKey, string> = {
     "不需開啟視窗，也能持續同步、整理和提供記憶。",
   导出日志: "匯出日誌",
   "把本机运行日志打包成一个文件，发给开发排查。":
-    "把本機執行日誌打包成一個檔案，提供給開發人員排查。",
+    "打包完整本機日誌，可能包含對話正文和本機路徑；只傳送給可信的開發人員。",
   "确定退出 One Memory 界面吗？": "要關閉 MemHub 視窗嗎？",
   "后台服务会继续同步、整理和提供记忆。":
     "背景服務會繼續同步、整理和提供記憶。",

memhub/packages/operations/src/memory-mode-operation.test.ts

@@ -32,7 +32,31 @@ describe("memory mode plan", () => {
     expect(parseMemoryModePlan({ ...valid, blockers: ["unknown"] })).toBeNull();
   });
 
-  it("isolates the old-Core matrix in one helper", () => {
+  it.each([
+    ["lite", "pro", ["earlier_raw_not_extracted", "lite_search_stops"]],
+    [
+      "lite",
+      "ultra",
+      [
+        "earlier_raw_not_extracted",
+        "lite_search_stops",
+        "existing_memories_added_to_graph",
+      ],
+    ],
+    ["pro", "lite", ["higher_mode_memories_hidden"]],
+    ["pro", "ultra", ["existing_memories_added_to_graph"]],
+    ["ultra", "lite", ["higher_mode_memories_hidden", "evolve_stopped"]],
+    ["ultra", "pro", ["higher_mode_memories_hidden", "evolve_stopped"]],
+  ] as const)(
+    "keeps the old-Core %s to %s effects aligned with the plan route",
+    (current, target, effects) => {
+      expect(
+        legacyMemoryModePlan(current, target, { hasActiveModel: true }).effects,
+      ).toEqual(effects);
+    },
+  );
+
+  it("keeps the old-Core model blocker and target capabilities", () => {
     expect(
       legacyMemoryModePlan("lite", "ultra", { hasActiveModel: false }),
     ).toMatchObject({
@@ -42,10 +66,6 @@ describe("memory mode plan", () => {
         graph_enabled: true,
       },
       blockers: ["active_model_required"],
-      effects: expect.arrayContaining([
-        "earlier_raw_not_extracted",
-        "existing_memories_added_to_graph",
-      ]),
     });
   });
 });

memhub/packages/operations/src/memory-mode-operation.ts

@@ -106,12 +106,15 @@ export function legacyMemoryModePlan(
   options: { hasActiveModel: boolean },
 ): MemoryModePlan {
   const effects: MemoryModeChange["effects"] = [];
-  if (target === "lite") {
-    effects.push("higher_mode_memories_hidden");
-    if (current === "ultra") effects.push("evolve_stopped");
-  } else if (current === "lite") {
+  const rank: Record<MemoryPipelineMode, number> = {
+    lite: 0,
+    pro: 1,
+    ultra: 2,
+  };
+  if (current !== target && current === "lite")
     effects.push("earlier_raw_not_extracted", "lite_search_stops");
-  }
+  if (rank[target] < rank[current]) effects.push("higher_mode_memories_hidden");
+  if (current !== target && current === "ultra") effects.push("evolve_stopped");
   if (target === "ultra" && current !== "ultra")
     effects.push("existing_memories_added_to_graph");
   return {

memhub/packages/runtime-control/src/diagnostic-log.test.ts

@@ -735,6 +735,76 @@ describe("diagnostic log", () => {
     expect(all).not.toContain("/Users/");
   });
 
+  it("optionally includes every regular file from the local logs directory", async () => {
+    const dataDir = await root();
+    const outputDir = await root();
+    const logs = path.join(dataDir, "logs");
+    await mkdir(logs, { recursive: true });
+    await writeFile(
+      path.join(logs, "diagnostic-core.jsonl"),
+      `${JSON.stringify({
+        schema_version: 1,
+        timestamp: "2026-10-05T01:00:00.000Z",
+        level: "error",
+        service: "core",
+        component: "core.database",
+        event: "database.error",
+        outcome: "failed",
+        session_id: "11111111-1111-4111-8111-111111111111",
+        error_code: "database_locked",
+      })}\n`,
+    );
+    await writeFile(
+      path.join(logs, "core.log"),
+      "raw private conversation and full local path\n",
+    );
+    await writeFile(path.join(logs, "service-error.log.1"), "rotated error\n");
+    await writeFile(path.join(logs, ".hidden log"), "hidden diagnostic\n");
+    await writeFile(path.join(logs, "服务 日志.log"), "unicode diagnostic\n");
+    await writeFile(path.join(logs, "empty.log"), "");
+    await writeFile(
+      path.join(logs, "large.log"),
+      Buffer.alloc(256 * 1024, 120),
+    );
+    const outside = path.join(await root(), "outside.log");
+    await writeFile(outside, "must not be followed\n");
+    await symlink(outside, path.join(logs, "linked.log"));
+    const target = path.join(outputDir, "MemHub.zip");
+
+    const result = await exportDiagnosticZip({
+      dataDir,
+      target,
+      productVersion: "3800",
+      includeRawLogs: true,
+    });
+
+    expect(result).toEqual({ filename: "MemHub.zip", file_count: 10 });
+    const files = unzipSync(await readFile(target));
+    expect(Object.keys(files).sort()).toEqual([
+      "README.txt",
+      "diagnostics/events.jsonl",
+      "logs/.hidden log",
+      "logs/core.log",
+      "logs/diagnostic-core.jsonl",
+      "logs/empty.log",
+      "logs/large.log",
+      "logs/service-error.log.1",
+      "logs/服务 日志.log",
+      "summary.json",
+    ]);
+    expect(strFromU8(files["logs/core.log"] ?? new Uint8Array())).toContain(
+      "raw private conversation",
+    );
+    expect(strFromU8(files["README.txt"] ?? new Uint8Array())).toContain(
+      "原始日志可能包含会话正文",
+    );
+    expect(files["logs/linked.log"]).toBeUndefined();
+    expect(files["logs/empty.log"]).toHaveLength(0);
+    expect(files["logs/large.log"]).toHaveLength(256 * 1024);
+    if (process.platform !== "win32")
+      expect((await lstat(target)).mode & 0o777).toBe(0o600);
+  });
+
   it("fails closed without publishing a ZIP when a diagnostic line is invalid", async () => {
     const dataDir = await root();
     await mkdir(path.join(dataDir, "logs"), { recursive: true });

memhub/packages/runtime-control/src/diagnostic-log.ts

@@ -1,8 +1,10 @@
 import { randomUUID } from "node:crypto";
+import { once } from "node:events";
 import {
   appendFileSync,
   chmodSync,
   constants,
+  createWriteStream,
   lstatSync,
   mkdirSync,
   renameSync,
@@ -12,6 +14,7 @@ import {
 import {
   appendFile,
   chmod,
+  type FileHandle,
   link,
   lstat,
   mkdir,
@@ -25,7 +28,7 @@ import {
 } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { strToU8, zipSync } from "fflate";
+import { strToU8, Zip, ZipPassThrough, zipSync } from "fflate";
 
 export const DIAGNOSTIC_SCHEMA_VERSION = 1 as const;
 export const DIAGNOSTIC_MAX_FILE_BYTES = 4 * 1024 * 1024;
@@ -627,6 +630,7 @@ export async function exportDiagnosticZip(options: {
   now?: Date;
   markers?: DiagnosticMarker[];
   summary?: Record<string, unknown>;
+  includeRawLogs?: boolean;
 }): Promise<{ filename: string; file_count: number }> {
   const now = options.now ?? new Date();
   const home = options.home ?? os.homedir();
@@ -672,7 +676,14 @@ export async function exportDiagnosticZip(options: {
         "MemHub diagnostics",
         "",
         "包含：最近 7 天通过安全白名单重新投影的诊断事件，以及不含正文的状态摘要。",
-        "不包含：原始日志、崩溃报告、对话原文、记忆内容、API Key、token 或宿主凭证。",
+        ...(options.includeRawLogs
+          ? [
+              "同时包含：One Memory Home 的 logs/ 目录中全部普通日志文件及轮转文件。",
+              "注意：原始日志可能包含会话正文、本机路径或其他私人信息，只应发送给可信的开发人员。",
+            ]
+          : [
+              "不包含：原始日志、崩溃报告、对话原文、记忆内容、API Key、token 或宿主凭证。",
+            ]),
         ...(markers.length
           ? [
               "",
@@ -689,20 +700,30 @@ export async function exportDiagnosticZip(options: {
   const directory = path.dirname(options.target);
   await mkdir(directory, { recursive: true });
   const temporary = path.join(directory, `.memhub-logs-${randomUUID()}.tmp`);
+  const rawLogs = options.includeRawLogs
+    ? await openRawLogSnapshots(options.dataDir)
+    : [];
   try {
-    await writeFile(temporary, zipSync(archive, { level: 6 }), {
-      mode: 0o600,
-      flag: "wx",
-    });
+    if (options.includeRawLogs)
+      await writeStreamingDiagnosticZip(temporary, archive, rawLogs);
+    else
+      await writeFile(temporary, zipSync(archive, { level: 6 }), {
+        mode: 0o600,
+        flag: "wx",
+      });
     await chmod(temporary, 0o600);
     await publishTemporaryFile(temporary, options.target);
   } catch (error) {
     await rm(temporary, { force: true }).catch(() => undefined);
     throw error;
+  } finally {
+    await Promise.all(
+      rawLogs.map((file) => file.handle.close().catch(() => undefined)),
+    );
   }
   return {
     filename: path.basename(options.target),
-    file_count: Object.keys(archive).length,
+    file_count: Object.keys(archive).length + rawLogs.length,
   };
 }
 
@@ -902,6 +923,151 @@ async function readRegularFileNoFollow(
   }
 }
 
+interface RawLogSnapshot {
+  name: string;
+  handle: FileHandle;
+  size: number;
+}
+
+async function openRawLogSnapshots(dataDir: string): Promise<RawLogSnapshot[]> {
+  const directory = diagnosticDirectory(dataDir);
+  let directoryInfo: Awaited<ReturnType<typeof lstat>>;
+  try {
+    directoryInfo = await lstat(directory);
+  } catch (error) {
+    if ((error as NodeJS.ErrnoException).code === "ENOENT") return [];
+    throw error;
+  }
+  if (!directoryInfo.isDirectory() || directoryInfo.isSymbolicLink())
+    fail("diagnostic_directory_unsafe");
+  const entries = await readdir(directory, { withFileTypes: true });
+  const snapshots: RawLogSnapshot[] = [];
+  try {
+    for (const entry of entries) {
+      if (entry.isSymbolicLink()) continue;
+      if (
+        entry.name === "." ||
+        entry.name === ".." ||
+        [...entry.name].some((character) => {
+          const code = character.charCodeAt(0);
+          return (
+            character === "/" || character === "\\" || code < 32 || code === 127
+          );
+        })
+      )
+        fail("diagnostic_log_name_unsafe");
+      const target = path.join(directory, entry.name);
+      const before = await lstat(target).catch(() => null);
+      if (!before?.isFile() || before.isSymbolicLink()) continue;
+      const handle = await open(
+        target,
+        constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0),
+      );
+      let retained = false;
+      try {
+        const opened = await handle.stat();
+        if (
+          !opened.isFile() ||
+          opened.dev !== before.dev ||
+          opened.ino !== before.ino
+        )
+          fail("diagnostic_file_unsafe");
+        snapshots.push({ name: entry.name, handle, size: opened.size });
+        retained = true;
+      } finally {
+        if (!retained) await handle.close().catch(() => undefined);
+      }
+    }
+    const after = await lstat(directory);
+    if (
+      !after.isDirectory() ||
+      after.isSymbolicLink() ||
+      after.dev !== directoryInfo.dev ||
+      after.ino !== directoryInfo.ino
+    )
+      fail("diagnostic_directory_unsafe");
+    return snapshots.sort((left, right) => left.name.localeCompare(right.name));
+  } catch (error) {
+    await Promise.all(
+      snapshots.map((file) => file.handle.close().catch(() => undefined)),
+    );
+    throw error;
+  }
+}
+
+async function writeStreamingDiagnosticZip(
+  target: string,
+  archive: Record<string, Uint8Array>,
+  rawLogs: RawLogSnapshot[],
+): Promise<void> {
+  const output = createWriteStream(target, { flags: "wx", mode: 0o600 });
+  let outputError: Error | null = null;
+  let backpressure = Promise.resolve();
+  const finished = new Promise<void>((resolve, reject) => {
+    output.once("finish", resolve);
+    output.once("error", (error) => {
+      outputError = error;
+      reject(error);
+    });
+  });
+  // The producer may still be reading a source file when the output stream
+  // fails. Attach a rejection handler immediately; the awaited promise below
+  // still carries the same error into the export's fail-closed cleanup.
+  void finished.catch(() => undefined);
+  const zip = new Zip((error, chunk, final) => {
+    if (error) {
+      outputError = error;
+      output.destroy(error);
+      return;
+    }
+    if (chunk.length > 0 && !output.write(chunk))
+      backpressure = once(output, "drain").then(() => undefined);
+    if (final) output.end();
+  });
+  const ensureWritable = async () => {
+    await backpressure;
+    if (outputError) throw outputError;
+  };
+  try {
+    for (const [name, content] of Object.entries(archive)) {
+      await ensureWritable();
+      const input = new ZipPassThrough(name);
+      zip.add(input);
+      input.push(content, true);
+    }
+    for (const file of rawLogs) {
+      await ensureWritable();
+      const input = new ZipPassThrough(`logs/${file.name}`);
+      zip.add(input);
+      let position = 0;
+      while (position < file.size) {
+        await ensureWritable();
+        const buffer = Buffer.allocUnsafe(
+          Math.min(64 * 1024, file.size - position),
+        );
+        const { bytesRead } = await file.handle.read(
+          buffer,
+          0,
+          buffer.length,
+          position,
+        );
+        if (bytesRead < 1) fail("diagnostic_file_changed");
+        position += bytesRead;
+        input.push(buffer.subarray(0, bytesRead), position === file.size);
+      }
+      if (file.size === 0) input.push(new Uint8Array(), true);
+    }
+    await ensureWritable();
+    zip.end();
+    await finished;
+  } catch (error) {
+    zip.terminate();
+    output.destroy(error instanceof Error ? error : undefined);
+    await finished.catch(() => undefined);
+    throw error;
+  }
+}
+
 export async function diagnosticLogFiles(dataDir: string): Promise<string[]> {
   const directory = diagnosticDirectory(dataDir);
   let directoryInfo: Awaited<ReturnType<typeof lstat>>;

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.13"
+HY_MEMORY_VERSION = "2.0.14"
 
 
 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\.13`/);
+  assert.match(readme, /`hy-memory==2\.0\.14`/);
   assert.match(readme, /`v1\.0\.6` 的发行目标是 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.13.dist-info"
+    dist_info = tmp_path / "hy_memory-2.0.14.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.13",
+                "Version: 2.0.14",
                 "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.13"
+    assert notice["version"] == "2.0.14"
     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.13"
+version = "2.0.14"
 description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
 readme = "README.md"
 requires-python = ">=3.9"

tests/test_raw_public_api.py

@@ -27,6 +27,7 @@ class RawStore:
         exclude_sources,
         layers,
         limit,
+        status_filter,
         **_kwargs,
     ):
         rows = []
@@ -41,11 +42,18 @@ class RawStore:
                 continue
             if node.layer not in layers:
                 continue
+            if node.status not in status_filter:
+                continue
             rows.append({"node": node, "score": 0.9})
         return rows[:limit]
 
 
-def raw_node(raw_id: str, session_id: str, source: str) -> MemoryNode:
+def raw_node(
+    raw_id: str,
+    session_id: str,
+    source: str,
+    status: MemoryStatus = MemoryStatus.ACTIVE,
+) -> MemoryNode:
     return MemoryNode(
         node_id=raw_id,
         user_id="u1",
@@ -55,7 +63,7 @@ def raw_node(raw_id: str, session_id: str, source: str) -> MemoryNode:
         layer=MemoryLayer.L1_RAW,
         content=f"content {raw_id}",
         source_type=SourceType.EXPLICIT,
-        status=MemoryStatus.ACTIVE,
+        status=status,
         is_latest=True,
         embedding=[0.1, 0.2],
     )
@@ -102,6 +110,7 @@ async def test_existing_search_contract_can_explicitly_search_raw():
         request_id="search-1",
         t0=0.0,
         query_embedding=[0.1, 0.2],
+        statuses=None,
     )
 
     assert [item["memory_id"] for item in result["memories"]["normal"]] == [
@@ -109,3 +118,33 @@ async def test_existing_search_contract_can_explicitly_search_raw():
     ]
     assert result["memories"]["normal"][0]["layer"] == "l1_raw"
     assert result["memories"]["normal"][0]["source"] == "claude"
+
+
+@pytest.mark.asyncio
+async def test_explicit_raw_search_can_include_shadow_rows():
+    client = HyMemoryClient.__new__(HyMemoryClient)
+    client._vector_store = RawStore(
+        [raw_node("raw-shadow", "s1", "cursor", MemoryStatus.SHADOW)]
+    )
+    client._embed_service = AsyncMock()
+    client._embed_service.embed.return_value = [0.1, 0.2]
+
+    result = await HyMemoryClient._search_explicit_raw_layer.__wrapped__(
+        client,
+        "content",
+        user_ids=["u1"],
+        agent_ids=["a1"],
+        session_ids=[],
+        include_sources=[],
+        exclude_sources=[],
+        limit=5,
+        min_score=0.0,
+        created_after=None,
+        request_id="search-shadow",
+        t0=0.0,
+        query_embedding=[0.1, 0.2],
+        statuses=["active", "shadow"],
+    )
+
+    assert result["memories"]["normal"][0]["memory_id"] == "raw-shadow"
+    assert result["memories"]["normal"][0]["status"] == "shadow"