test: stabilize reembed merge fixture timestamps

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

CHANGELOG.md

@@ -4,24 +4,14 @@ Hy Memory and MemHub are versioned and released independently.
 
 ## Unreleased
 
-- MemHub App now exposes the complete memory-mode lifecycle. New installs
-  explicitly choose lite or pro during onboarding; pro completes only after a
-  generation model is active, while ultra remains an advanced Settings choice.
-  Settings shows mode effects, graph backfill, the default 03:00 evolve
-  schedule and manual evolve. The memory page shows original conversations in
-  lite and advanced memories in pro/ultra. Existing installs remain pro unless
-  the user changes them, and embedding maintenance blocks mode/evolve changes.
-- 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.
+## Hy Memory 2.0.14 — 2026-11-04
+
+- `search()` / `async_search()` take an optional `statuses` for explicit
+  L1_RAW search (`layers=["l1_raw"]`, or any search in lite mode), e.g.
+  `statuses=["active", "shadow"]`, so raw turns a pro extraction already
+  shadowed can still be searched. Default stays active-only. Raw hits now
+  report their real `status`. Passing `statuses` to a non-raw search or to
+  `scene="all"` raises `ValueError` instead of being ignored.
 
 ## Hy Memory 2.0.13 — 2026-11-02
 
@@ -92,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.14`. 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

memhub/README.md

@@ -19,7 +19,7 @@ If you run into a problem, please open an Issue with the version, platform,repro
 ## 发行与平台合同
 
 源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.14`。MemHub 源码版本为 `1.0.6`
-（build 40），GitHub 仓库为
+（build 41），GitHub 仓库为
 [`Tencent-Hunyuan/Hy-Memory`](https://github.com/Tencent-Hunyuan/Hy-Memory)，npm 包名为
 `memhub-cli`。当前代码架构见
 [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)。

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/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,
@@ -743,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):
@@ -1502,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,
                 )
@@ -1631,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
@@ -1873,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":
@@ -1949,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:
@@ -1989,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)
@@ -2078,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:
@@ -2816,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(
@@ -3005,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,
@@ -3029,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:
@@ -3067,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.
@@ -3094,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)
@@ -3112,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(
@@ -3186,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(
@@ -3481,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",
@@ -3579,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(
@@ -3588,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
 
@@ -3602,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(
@@ -4861,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)
@@ -4908,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(
@@ -5705,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)
@@ -5979,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:
@@ -7021,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/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/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/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 个场景。 |
@@ -322,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. |
 
-共 317 个测试文件。
+共 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 = 317
-EXPECTED_TEST_FILES_SHA256 = "b1370c1ea537ecb3f9837d0a6354e27c8ebe7c588a5b2927044e6ac27532fd26"
+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",

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_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/web/src/model-setup-next.css

@@ -134,6 +134,11 @@
   background: transparent;
   color: #4e6478;
 }
+.model-settings .generation-technical-toggle:hover:not(:disabled) {
+  border-color: transparent;
+  background: #e9f0f4;
+  color: #304757;
+}
 .model-settings .generation-technical-fields {
   display: grid;
   grid-template-columns: repeat(2, minmax(0, 1fr));

memhub/docs/INSTALLATION.md

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

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. */

pyproject.toml

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

tests/test_reembed.py

@@ -9,6 +9,7 @@ import hashlib
 import json
 import math
 import random
+from datetime import datetime
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -461,6 +462,8 @@ async def _two_spaces(path):
     """A live 1024 space and a stray 512 space sharing some ids."""
     from hy_memory.data.vector_store_chroma import ChromaVectorStore
 
+    stable_time = datetime(2025, 1, 1)
+
     async def _store(base, dims, model, contents):
         cfg = MemoryConfig()
         cfg.vector_store.provider = "chroma"
@@ -475,6 +478,7 @@ async def _two_spaces(path):
             node = MemoryNode(
                 node_id=node_id, user_id=user, agent_id="a", session_id="s",
                 layer=MemoryLayer.L2_FACT, content=content, status=status,
+                gmt_created=stable_time, gmt_modified=stable_time, valid_from=stable_time,
             )
             node.embedding = emb.vector(content)
             nodes.append(node)