fix(memhub): align recent activity to card bottom

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

CHANGELOG.md

@@ -4,14 +4,6 @@ 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
@@ -90,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)**！

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,7 +272,7 @@ from one_memory.onboarding import (
     OnboardingState,
     OnboardingStore,
 )
-from one_memory.raw_turns import RawDisplayTurn, raw_display_turns
+from one_memory.raw_turns import RawDisplayTurn, raw_display_turns, raw_prefix_added
 from one_memory.recall import (
     GovernedRecallTarget,
     RecallReceipt,
@@ -610,17 +612,10 @@ class SdkRawItem(HySdkRawRecord):
     turn_count: int | None = Field(default=None, ge=1)
 
     @classmethod
-    def of(
-        cls, record: HySdkRawRecord, turn: RawDisplayTurn | None = None
-    ) -> "SdkRawItem":
+    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
-        )
+        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(exclude={"content"}),
@@ -750,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):
@@ -1509,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,
                 )
@@ -1638,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
@@ -1880,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":
@@ -1956,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:
@@ -1996,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)
@@ -2085,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:
@@ -2823,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(
@@ -3012,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,
@@ -3036,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:
@@ -3074,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.
@@ -3101,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)
@@ -3119,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(
@@ -3193,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(
@@ -3488,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",
@@ -3586,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(
@@ -3595,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
 
@@ -3609,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(
@@ -4532,10 +4548,7 @@ def create_app(
                 )
                 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()
-                        ):
+                        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:
@@ -4546,9 +4559,7 @@ def create_app(
                             offset=0,
                             limit=500,
                         )
-                        first_page = HySdkRawPage.from_memory_list(
-                            value, limit=500, offset=0
-                        )
+                        first_page = HySdkRawPage.from_memory_list(value, limit=500, offset=0)
                         signature = tuple(
                             (record.raw_id, record.status) for record in first_page.raw
                         )
@@ -4560,6 +4571,29 @@ def create_app(
                         ):
                             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)
@@ -4585,16 +4619,11 @@ def create_app(
                             offset=0,
                             limit=500,
                         )
-                        verify = HySdkRawPage.from_memory_list(
-                            verify_value, limit=500, offset=0
-                        )
+                        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
-                        ):
+                        if verify.total != first_page.total or verify_signature != signature:
                             continue
                         projected_turns = tuple(
                             (record, turn)
@@ -4855,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)
@@ -4902,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(
@@ -5699,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)
@@ -5973,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:
@@ -6372,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:
@@ -6399,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 {}),
         },
     )
@@ -7013,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_routes.py

@@ -30,7 +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
+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
@@ -71,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:
@@ -121,9 +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 ()
-            candidate_limit = (
-                min(value.limit * 10, 200) if value.raw_unit == "turn" else value.limit
-            )
+            # 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,
@@ -131,7 +143,7 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
                     "user_ids": [value.user_id],
                     "agent_ids": [],
                     "include_sources": sources,
-                    "limit": candidate_limit,
+                    "limit": value.limit,
                     "min_score": 0.0,
                     "schema_min_score": 0.0,
                     "schema_limit": 0,
@@ -139,59 +151,60 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
                     "scene": "chat",
                     "reader": HY_MEMORY_READER,
                     "layers": ["l1_raw"],
-                    "statuses": ["active", "shadow"],
+                    **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)
             )
-            items: list[SdkRawContextItem] = []
+            # 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
+                )
                 turns = raw_display_turns(hit) if value.raw_unit == "turn" else ()
-                if turns:
-                    query = value.query.casefold()
-                    lexical = tuple(turn for turn in turns if query in turn.content.casefold())
-                    # A batch-level embedding cannot identify which contained
-                    # turn matched semantically. Return only turns grounded by
-                    # the query text instead of assigning every sibling the
-                    # batch score and showing unrelated conversations.
-                    for turn in lexical:
-                        items.append(
-                            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=(
-                                    datetime.fromtimestamp(hit.memory_at, tz=UTC)
-                                    if hit.memory_at is not None
-                                    else None
-                                ),
-                                score=hit.score,
-                            )
-                        )
-                else:
-                    items.append(
+                if not turns:
+                    primary.append(
                         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
-                            ),
+                            memory_at=memory_at,
                             score=hit.score,
                         )
                     )
-                if len(items) >= value.limit:
-                    break
-            return tuple(items[: value.limit])
+                    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))
@@ -3271,6 +3290,11 @@ class DurableWriteDrain:
                 job_id,
                 type(error).__name__,
             )
+            report_exception(
+                "memory.write_queue",
+                "memory.write_queue.executor_failed",
+                error,
+            )
             self._record_database_error(error)
             await self._settle_worker_error(
                 job_id,
@@ -3297,7 +3321,7 @@ class DurableWriteDrain:
         other programming error still escapes to the outer per-job fence.
         """
         try:
-            await in_the_background(
+            await in_the_write_queue(
                 self.queue.settle_worker_error,
                 job_id,
                 error_code=error_code,
@@ -3315,7 +3339,7 @@ class DurableWriteDrain:
 
     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
         )
@@ -3370,6 +3394,13 @@ class DurableWriteDrain:
                             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
@@ -3385,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,
@@ -3497,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,
@@ -3515,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
@@ -3543,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

@@ -9,6 +9,7 @@ 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
@@ -57,6 +58,66 @@ def raw_display_turns(record: HySdkRawRecord) -> tuple[RawDisplayTurn, ...]:
 
 
 def _turn_content(messages: list[tuple[str, str]]) -> str:
-    return "\n".join(
-        f"[{role}]: {content}" if role else content for role, content in messages
-    )
+    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

@@ -5528,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

@@ -91,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."""
 
@@ -109,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"]:
@@ -274,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)
@@ -365,14 +407,56 @@ def test_raw_context_search_can_return_turns_instead_of_storage_batches(
             },
         )
         assert response.status_code == 200, response.text
-        raw_one = [
-            item for item in response.json()["items"] if item["raw_id"] == "raw-1"
-        ]
+        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)
 
@@ -632,12 +716,39 @@ def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monke
         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"
-        ]
+        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(

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/web/src/applications-home.css

@@ -42,6 +42,8 @@
   margin-bottom: 24px;
 }
 .r6-home-overview {
+  display: flex;
+  flex-direction: column;
   min-width: 0;
   min-height: 346px;
   padding: 26px;
@@ -150,6 +152,7 @@
   padding: 8px 0;
 }
 .r6-home-activity {
+  margin-top: auto;
   border-top: 1px solid #d5dfe8;
   padding-top: 12px;
 }

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/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 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. | 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/index.ts

@@ -1118,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/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"]