fix(core): keep initial sync behind bounded history

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

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

@@ -282,11 +282,24 @@ class PollingLiveSyncSupervisor:
                 and job.status
                 in {ImportStatus.PENDING, ImportStatus.RUNNING, ImportStatus.INTERRUPTED}
             ]
+            bounded_history_incomplete = any(
+                job.source_id == source_id
+                and job.operation == "history"
+                and job.history_after is not None
+                and job.status is not ImportStatus.COMPLETED
+                for job in source_jobs
+            )
             # Establish the realtime EOF independently even while history owns
             # this source. Otherwise the P1 job starts from the P2 checkpoint
             # (or from zero) and becomes a second full-history scan.
             adapter = self.import_supervisor.adapters[source_id]
-            await self._live_cursors(source_id, adapter)
+            live_cursors = await self._live_cursors(source_id, adapter)
+            # A bounded onboarding history is the authority for everything up
+            # to its EOF. An adapter without a native live baseline has no safe
+            # realtime cursor until that history finishes; starting a sync from
+            # zero here would bypass `history_after` and import the old corpus.
+            if bounded_history_incomplete and not live_cursors:
+                return None
             # Only one scan of a source may be in flight: they share the
             # cursor, and two readers means every file read twice. With a
             # catch-up still holding the source there is no free moment to ask

memhub/apps/core/tests/test_live_sync.py

@@ -94,6 +94,67 @@ class SettledFixtureAdapter(IncrementalFixtureAdapter):
         }
 
 
+class BoundedOnboardingAdapter:
+    """No native baseline: bounded history must establish the live cursor."""
+
+    descriptor = IncrementalFixtureAdapter.descriptor
+
+    def __init__(self) -> None:
+        self.include_new = False
+
+    async def detect(self) -> bool:
+        return True
+
+    async def scan(self, options: ScanOptions):
+        current = options.cursors.get("e" * 64)
+        checkpoint = current.checkpoint if current is not None else None
+        if checkpoint not in {"byte:10", "byte:20"}:
+            yield ScannedEvidence(
+                evidence=EvidenceInput(
+                    source_id="fixture",
+                    conversation_id="old-conversation",
+                    message_id="old-message",
+                    role="user",
+                    written_by=WrittenBy.PERSON,
+                    created_at=datetime(2020, 1, 1, tzinfo=UTC),
+                    verbatim="old history",
+                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+                ),
+                cursor=SourceCursor(
+                    file_key="e" * 64,
+                    file_identity="bounded",
+                    checkpoint="byte:10",
+                    checkpoint_anchor="f" * 64,
+                    observed_size=20,
+                    observed_modified_ns=1,
+                ),
+            )
+            checkpoint = "byte:10"
+        if self.include_new and checkpoint != "byte:20":
+            yield ScannedEvidence(
+                evidence=EvidenceInput(
+                    source_id="fixture",
+                    conversation_id="new-conversation",
+                    message_id="new-message",
+                    role="user",
+                    written_by=WrittenBy.PERSON,
+                    created_at=datetime(2026, 10, 18, tzinfo=UTC),
+                    verbatim="new append",
+                    origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+                ),
+                cursor=SourceCursor(
+                    file_key="e" * 64,
+                    file_identity="bounded",
+                    checkpoint="byte:20",
+                    checkpoint_anchor="1" * 64,
+                    observed_size=20,
+                    observed_modified_ns=2,
+                    read_to_end=True,
+                ),
+            )
+        yield ScanProgressEvent(phase=ScanPhase.DONE)
+
+
 class HoldingHistoryAdapter:
     """First scan blocks until cancelled; later scans yield one new record."""
 
@@ -328,6 +389,56 @@ def test_live_sync_does_not_resume_a_history_import_the_user_paused(tmp_path) ->
     asyncio.run(scenario())
 
 
+def test_bounded_onboarding_history_establishes_cursor_before_first_sync(tmp_path) -> None:
+    async def scenario() -> None:
+        database = tmp_path / "one-memory.sqlite3"
+        evidence = EvidenceStore(database)
+        jobs = ImportJobStore(database)
+        adapter = BoundedOnboardingAdapter()
+        imports = ImportSupervisor(
+            adapters={"fixture": adapter},
+            evidence_store=evidence,
+            job_store=jobs,
+        )
+        live_sync = PollingLiveSyncSupervisor(
+            source_ids=("fixture",),
+            import_supervisor=imports,
+            job_store=jobs,
+        )
+        history = imports.create_job("fixture", operation="history")
+        with sqlite3.connect(database) as connection, connection:
+            connection.execute(
+                "UPDATE import_jobs SET history_after=? WHERE job_id=?",
+                (datetime(2026, 10, 15, tzinfo=UTC).isoformat(), history.job_id),
+            )
+
+        # With no native baseline, sync must wait instead of scanning from zero
+        # and bypassing the onboarding history window.
+        assert await live_sync.trigger("fixture") is None
+        assert not [job for job in jobs.list_jobs() if job.operation == "sync"]
+
+        imports.enqueue(history.job_id)
+        completed = await imports.wait(history.job_id)
+        assert completed.status is ImportStatus.COMPLETED
+        assert completed.stored == 0
+        assert completed.skipped == 1
+        assert evidence.count() == 0
+
+        # Once bounded history has advanced the cursor to EOF, normal realtime
+        # appends are admitted without retaining the three-day cutoff forever.
+        adapter.include_new = True
+        synced = await live_sync.trigger("fixture")
+        assert synced is not None
+        assert synced.status is ImportStatus.COMPLETED
+        assert synced.stored == 1
+        records = evidence.list_extractable()
+        assert [record.verbatim for record in records] == ["new append"]
+        await live_sync.shutdown()
+        await imports.shutdown()
+
+    asyncio.run(scenario())
+
+
 def test_a_sync_left_running_after_its_worker_died_is_started_again(tmp_path) -> None:
     """A running row with no live worker used to be returned as 'already syncing'.
 

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

@@ -174,20 +174,20 @@ it("continues to step3 while sample GET is pending without calling completion AP
   expect(screen.queryByText(/^已生成.*记忆/)).toBeNull();
 });
 it.each([
-  ["pending", "已扫描 100 条，正在筛选"],
-  ["paused", "已扫描 100 条，尚未保存"],
-  ["interrupted", "已扫描 100 条，尚未保存"],
-  ["failed", "已扫描 100 条，尚未保存"],
-  ["cancelled", "已扫描 100 条，尚未保存"],
-  ["completed", "已扫描 100 条，本次没有可保存的对话"],
+  "pending",
+  "paused",
+  "interrupted",
+  "failed",
+  "cancelled",
+  "completed",
 ] as const)(
   "shows %s without mutating jobs or fetching missing content",
-  (status, expected) => {
+  (status) => {
     setup({
       jobs: [{ ...job, status, stored: 0 }],
       sources: [{ ...source, imported_messages: 0 }],
     });
-    expect(screen.getByText(expected)).toBeTruthy();
+    expect(screen.getByText("已扫描 100 条，正在保存")).toBeTruthy();
     expect(api.receipt).not.toHaveBeenCalled();
   },
 );

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

@@ -38,71 +38,6 @@ it("uses the selected job's message count, never the source total or fake percen
   expect(container.textContent).not.toContain("%");
   expect(container.textContent).not.toContain("持续导入中");
 });
-
-it("does not describe scanned-but-skipped history as saved", () => {
-  const { container } = render(
-    <R6ImportProgress
-      sourceIds={["cursor"]}
-      sources={[source]}
-      jobs={[
-        {
-          ...job,
-          status: "completed",
-          discovered: 12_375,
-          stored: 0,
-          skipped: 12_375,
-          history_after: "2026-10-15T00:00:00Z",
-        },
-      ]}
-    />,
-  );
-  expect(
-    screen.getByText("已扫描 12,375 条，本次没有符合时间范围的对话"),
-  ).toBeTruthy();
-  expect(screen.getByText("导入完成")).toBeTruthy();
-  expect(container.textContent).not.toContain("正在保存");
-  expect(container.textContent).not.toContain("已保存 12,375");
-});
-
-it("calls an active bounded scan filtering rather than saving", () => {
-  render(
-    <R6ImportProgress
-      sourceIds={["cursor"]}
-      sources={[source]}
-      jobs={[
-        {
-          ...job,
-          discovered: 3_522,
-          stored: 0,
-          history_after: "2026-10-15T00:00:00Z",
-        },
-      ]}
-    />,
-  );
-  expect(screen.getByText("已扫描 3,522 条，正在筛选最近 3 天")).toBeTruthy();
-});
-
-it("reports only stored rows as saved after a completed partial import", () => {
-  const { container } = render(
-    <R6ImportProgress
-      sourceIds={["cursor"]}
-      sources={[source]}
-      jobs={[
-        {
-          ...job,
-          status: "completed",
-          discovered: 25_177,
-          stored: 20_826,
-          skipped: 4_351,
-        },
-      ]}
-    />,
-  );
-  expect(screen.getByText("已保存 20,826 条消息")).toBeTruthy();
-  expect(screen.getByText("导入完成")).toBeTruthy();
-  expect(container.textContent).not.toContain("25,177");
-});
-
 it.each(["paused", "interrupted", "failed", "completed", "cancelled"] as const)(
   "does not animate %s as active work",
   (status) => {

memhub/apps/web/src/R6ImportProgress.tsx

@@ -3,21 +3,6 @@ import { useEffect, useLayoutEffect, useRef } from "react";
 import { AppMark } from "./app-mark";
 import "./r6-import-progress.css";
 
-function progressText(job: ImportJob): string | null {
-  if (job.stored > 0) return `已保存 ${job.stored.toLocaleString()} 条消息`;
-  if (job.discovered <= 0) return null;
-  const scanned = `已扫描 ${job.discovered.toLocaleString()} 条`;
-  if (job.status === "completed")
-    return job.history_after
-      ? `${scanned}，本次没有符合时间范围的对话`
-      : `${scanned}，本次没有可保存的对话`;
-  if (job.status === "pending" || job.status === "running")
-    return job.history_after
-      ? `${scanned}，正在筛选最近 3 天`
-      : `${scanned}，正在筛选`;
-  return `${scanned}，尚未保存`;
-}
-
 export function R6ImportProgress({
   sourceIds,
   sources,
@@ -101,9 +86,8 @@ export function R6ImportProgress({
           interrupted: "已中断",
           failed: "导入遇到问题",
           cancelled: "已取消",
-          completed: "导入完成",
+          completed: "已保存",
         };
-        const progress = job ? progressText(job) : null;
         return (
           <article
             className="r6-import-progress-card"
@@ -118,8 +102,10 @@ export function R6ImportProgress({
               </div>
             </div>
             <h2>{source?.display_name ?? id}</h2>
-            {progress ? (
-              <p>{progress}</p>
+            {job && job.stored > 0 ? (
+              <p>{`已保存 ${job.stored.toLocaleString()} 条消息`}</p>
+            ) : job && job.discovered > 0 ? (
+              <p>{`已扫描 ${job.discovered.toLocaleString()} 条，正在保存`}</p>
             ) : noImportedHistory ? (
               <p>没有可保存的历史对话</p>
             ) : null}