diff --git a/memhub/apps/core/src/one_memory/imports/live_sync.py b/memhub/apps/core/src/one_memory/imports/live_sync.py index 026d0c1..7ce56f4 100644 --- a/memhub/apps/core/src/one_memory/imports/live_sync.py +++ b/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 diff --git a/memhub/apps/core/tests/test_live_sync.py b/memhub/apps/core/tests/test_live_sync.py index c38dbe4..4a1ed13 100644 --- a/memhub/apps/core/tests/test_live_sync.py +++ b/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'.