From d89afb38873e66d14f379b212bca6153404acac7 Mon Sep 17 00:00:00 2001 From: alvinfei Date: Mon, 26 Oct 2026 23:59:38 +0800 Subject: [PATCH 1/2] fix(memhub): make Core ready before it resumes unfinished work A home with pending imports and write jobs kept /health closed while the lifespan resumed them, so the Runtime hit its 60s readiness budget, killed a healthy Core with core_start_timeout, and the next Core began the same recovery again. The write drain also started before readiness and held the single writer the rest of startup needed. Readiness is now only "the stores are open and Core answers". After core.ready a recovery task runs, step by step: the post-cutover metadata repair, the durable write drain, raw retention (or the legacy pending-dispatch replay), interrupted imports, live sync, source reconcile and the legacy catch-up. A failed step is logged and named in /health without stopping readiness or the steps after it. /health reports recovery as pending, running, done or failed. Shutdown cancels recovery first and still releases the content-backfill gate. Co-Authored-By: Claude Opus 5.5 (1M context) --- memhub/apps/core/src/one_memory/api.py | 304 ++++++++++++------ memhub/apps/core/tests/TEST_INDEX.md | 3 +- .../core/tests/core_test_classification.py | 5 +- .../tests/test_core_ready_before_recovery.py | 144 +++++++++ memhub/apps/core/tests/test_health.py | 7 + .../tests/test_source_permission_races.py | 2 + .../core/tests/test_source_status_latency.py | 1 + .../core/tests/test_visibility_background.py | 2 + memhub/packages/contracts/src/index.ts | 7 + 9 files changed, 371 insertions(+), 104 deletions(-) create mode 100644 memhub/apps/core/tests/test_core_ready_before_recovery.py diff --git a/memhub/apps/core/src/one_memory/api.py b/memhub/apps/core/src/one_memory/api.py index d18226b..f1303f4 100644 --- a/memhub/apps/core/src/one_memory/api.py +++ b/memhub/apps/core/src/one_memory/api.py @@ -253,6 +253,11 @@ class HealthResponse(BaseModel): # is answering. Reads fall back to exact LIKE while this is not ready. content_candidate_backfill: Literal["idle", "running", "ready", "failed"] = "idle" content_candidate_backfill_failure: str | None = None + # Picking up work an earlier run left unfinished starts after readiness. + # "running" is not degraded: Core is ready and answers throughout; the + # failed steps are named so a stuck recovery is visible without a log. + recovery: Literal["pending", "running", "done", "failed"] = "done" + recovery_failed_steps: tuple[str, ...] = () class BackupRequest(BaseModel): @@ -685,28 +690,8 @@ def create_app( # Only the maintenance Core (restore) may open such a home. raise SdkMigrationRequired() if app.state.memory_authority is MemoryAuthority.SDK: - from one_memory.hy_sdk_migration import ( - repair_sdk_metadata_from_bundle, - ) - - try: - repair = await asyncio.to_thread( - repair_sdk_metadata_from_bundle, - data_dir=runtime_settings.data_dir, - database_path=database_path, - ) - if repair.vector_rows or repair.coding_rows: - logger.info( - "sdk_metadata_repair_completed mapped=%d vector=%d coding=%d", - repair.mapped, - repair.vector_rows, - repair.coding_rows, - ) - except Exception as error: - logger.warning( - "sdk_metadata_repair_failed error_type=%s", - type(error).__name__, - ) + # The metadata repair after a cutover reads every bundle: it is + # recovery work, so it runs after readiness (`recover_after_ready`). app.state.memory_governance_store = RetiredGovernanceStore(database_path) app.state.evidence_store = RetiredEvidenceStore(database_path) else: @@ -921,8 +906,6 @@ def create_app( extraction_settings=app.state.extraction_settings_store, data_dir=runtime_settings.data_dir, ) - if app.state.memory_authority is MemoryAuthority.SDK: - app.state.sdk_memory_runtime.start_raw_retention() # C2 owns the single production budget (handle 5s + reserve 250ms, # total 5.25s, lease 10s, renew 1s, claim 250ms). The composition # root only consumes it: no local formula, and the 250ms claim @@ -1012,11 +995,9 @@ def create_app( source_claim_enabled=lambda source_id: _sedimentation_claim_enabled(app, source_id), ) app.state.memory_write_drain = memory_write_drain - app.state.memory_write_drain_task = watched_task( - asyncio.get_running_loop(), - lambda: memory_write_drain.run_forever(), - name="one-memory-durable-write-drain", - ) + # 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 # 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 @@ -1089,60 +1070,6 @@ def create_app( sdk_authority=lambda: app.state.memory_authority is MemoryAuthority.SDK, ) app.state.import_job_store.replace_never_published_snapshot_failures() - # A file's Evidence and its pending-dispatch row commit atomically, - # but the callback that turns them into a task runs after the - # transaction. A crash in that window leaves durable Evidence with - # no in-memory timer left to notice it; drain those rows once on - # startup through the same permission/debounce path as live work. - if ( - runtime_settings.automatic_memory_processing - and app.state.memory_authority is MemoryAuthority.LEGACY - ): - # Two-phase, like the live runner: enumerate the file - # identities, read each without consuming its pending marker, - # await the durable submit, and only then consume. A crash or - # a failed/cancelled submit leaves the marker for the next - # startup instead of dropping the committed Evidence. The - # legacy `take_all_stable_records()` consumes while reading and - # is deliberately no longer the startup path. - for user_id, source_id, file_key in await asyncio.to_thread( - app.state.evidence_store.list_pending_dispatch_files - ): - try: - for dispatch in await asyncio.to_thread( - app.state.evidence_store.take_stable_file_dispatches, - user_id=user_id, - source_id=source_id, - file_key=file_key, - ): - # Same intake as the live callback, so the crash - # window between Evidence commit and durable submit - # is closed by the same idempotent path rather than - # a second one. - token = live_sync_arrival.set(dispatch.priority == "live") - try: - await enqueue_stable_evidence(dispatch.record) - finally: - live_sync_arrival.reset(token) - except Exception: - # The submit did not succeed, so the pending marker is - # deliberately left in place: the next startup will - # replay this file. Startup continues for the other - # files instead of failing the whole composition root. - logger.warning( - "stable_evidence_startup_replay_deferred source=%s file=%s", - source_id, - file_key, - exc_info=True, - ) - continue - await asyncio.to_thread( - app.state.evidence_store.consume_stable_file_records, - user_id=user_id, - source_id=source_id, - file_key=file_key, - ) - app.state.import_supervisor.resume_interrupted() app.state.source_connection_store = SourceConnectionStore( app.state.source_grant_store, app.state.import_job_store, @@ -1185,11 +1112,6 @@ def create_app( app.state.live_sync_supervisor.set_watched_sources( watch_roots if watching and runtime_settings.live_sync_enabled else () ) - if runtime_settings.live_sync_enabled: - app.state.live_sync_supervisor.start() - if app.state.live_sync_watch is not None: - app.state.live_sync_watch.start(prime=True) - await _reconcile_source_sync_from_config() # Taking up the memory work again — the tasks left unfinished, then # the conversations with no memory yet — reads every conversation # ever imported: 13.5 seconds on this machine's real database, and @@ -1203,9 +1125,6 @@ def create_app( warm_bulk_pool() app.state.memory_backfill = None app.state.memory_visibility_task = None - if app.state.memory_authority is MemoryAuthority.LEGACY: - _start_backfill(app) - _start_visibility_recompute(app) # Both times the loop was blocked here it was found from outside, # by hand, days later. This says so at the time, and says what was # holding it. @@ -1224,15 +1143,182 @@ def create_app( app.state.resource_diagnostics = watched_task( asyncio.get_running_loop(), _emit_resource_snapshots ) + # Readiness is "Core answers": the stores above are open and + # /health, recall and the library work. Everything below picks up + # work an earlier run left unfinished -- the post-cutover metadata + # repair, the durable write queue, interrupted imports, live sync, + # source reconcile. On a library with pending work that took longer + # than the Runtime's readiness budget, so the Runtime killed a + # healthy Core and the next one began the same recovery again. It + # now runs only after readiness, one step at a time, and a failed + # step is reported without taking readiness or the other steps + # with it. + app.state.recovery = "pending" + app.state.recovery_failed_steps = () + + async def recover_after_ready() -> None: + app.state.recovery = "running" + recovery_started = time.monotonic() + failed: list[str] = [] + + async def step(name: str, action: Callable[[], Awaitable[None]]) -> None: + step_started = time.monotonic() + try: + await action() + except asyncio.CancelledError: + raise + except Exception as error: + failed.append(name) + app.state.recovery_failed_steps = tuple(failed) + logger.warning( + "core_recovery_step_failed step=%s error_type=%s", + name, + type(error).__name__, + ) + diagnostic_event( + level="warn", + component="core.recovery", + event="core.recovery_step_failed", + outcome="failed", + error_type=type(error).__name__, + duration_ms=elapsed_ms(step_started), + details={"phase": name}, + ) + + async def repair_sdk_metadata() -> None: + from one_memory.hy_sdk_migration import ( + repair_sdk_metadata_from_bundle, + ) + + repair = await asyncio.to_thread( + repair_sdk_metadata_from_bundle, + data_dir=runtime_settings.data_dir, + database_path=database_path, + ) + if repair.vector_rows or repair.coding_rows: + logger.info( + "sdk_metadata_repair_completed mapped=%d vector=%d coding=%d", + repair.mapped, + repair.vector_rows, + repair.coding_rows, + ) + + async def start_write_drain() -> None: + app.state.memory_write_drain_task = watched_task( + asyncio.get_running_loop(), + lambda: memory_write_drain.run_forever(), + name="one-memory-durable-write-drain", + ) + + async def start_raw_retention() -> None: + app.state.sdk_memory_runtime.start_raw_retention() + + async def replay_pending_dispatch() -> None: + # A file's Evidence and its pending-dispatch row commit atomically, + # but the callback that turns them into a task runs after the + # transaction. A crash in that window leaves durable Evidence with + # no in-memory timer left to notice it; drain those rows once on + # startup through the same permission/debounce path as live work. + if ( + runtime_settings.automatic_memory_processing + and app.state.memory_authority is MemoryAuthority.LEGACY + ): + # Two-phase, like the live runner: enumerate the file + # identities, read each without consuming its pending marker, + # await the durable submit, and only then consume. A crash or + # a failed/cancelled submit leaves the marker for the next + # startup instead of dropping the committed Evidence. The + # legacy `take_all_stable_records()` consumes while reading and + # is deliberately no longer the startup path. + for user_id, source_id, file_key in await asyncio.to_thread( + app.state.evidence_store.list_pending_dispatch_files + ): + try: + for dispatch in await asyncio.to_thread( + app.state.evidence_store.take_stable_file_dispatches, + user_id=user_id, + source_id=source_id, + file_key=file_key, + ): + # Same intake as the live callback, so the crash + # window between Evidence commit and durable submit + # is closed by the same idempotent path rather than + # a second one. + token = live_sync_arrival.set(dispatch.priority == "live") + try: + await enqueue_stable_evidence(dispatch.record) + finally: + live_sync_arrival.reset(token) + except Exception: + # The submit did not succeed, so the pending marker is + # deliberately left in place: the next startup will + # replay this file. Startup continues for the other + # files instead of failing the whole composition root. + logger.warning( + "stable_evidence_startup_replay_deferred source=%s file=%s", + source_id, + file_key, + exc_info=True, + ) + continue + await asyncio.to_thread( + app.state.evidence_store.consume_stable_file_records, + user_id=user_id, + source_id=source_id, + file_key=file_key, + ) + + async def resume_imports() -> None: + app.state.import_supervisor.resume_interrupted() + + async def start_live_sync() -> None: + if runtime_settings.live_sync_enabled: + app.state.live_sync_supervisor.start() + if app.state.live_sync_watch is not None: + app.state.live_sync_watch.start(prime=True) + + async def start_legacy_catch_up() -> None: + _start_backfill(app) + _start_visibility_recompute(app) + + # Yield once first so the HTTP server finishes binding before + # any of this asks the loop for time. + await asyncio.sleep(0) + try: + sdk = app.state.memory_authority is MemoryAuthority.SDK + if sdk: + # Before the drain: the repair restores metadata the + # queued writes are read back against. + await step("sdk_metadata_repair", repair_sdk_metadata) + await step("write_drain", start_write_drain) + if sdk: + await step("raw_retention", start_raw_retention) + else: + await step("pending_dispatch_replay", replay_pending_dispatch) + await step("import_resume", resume_imports) + await step("live_sync", start_live_sync) + await step("source_reconcile", _reconcile_source_sync_from_config) + if not sdk: + await step("legacy_catch_up", start_legacy_catch_up) + finally: + # Startup writes are done. Only now let the derived-content + # rebuild take the single write checkout: it is a long + # background job, and on a large library it would otherwise + # hold the writer while the recovery above still needs it + # (ISSUE-CONTENT-BACKFILL-BLOCKS-STARTUP). + app.state.memory_governance_store.release_content_backfill_startup_gate() + app.state.recovery = "failed" if failed else "done" + diagnostic_event( + level="warn" if failed else "info", + component="core.recovery", + event="core.recovery_finished", + outcome="failed" if failed else "succeeded", + duration_ms=elapsed_ms(recovery_started), + details={"count": len(failed)}, + ) + + app.state.recovery_task = None try: - # Startup writes are done. Only now let the derived-content - # rebuild take the single write checkout: it is a long - # background job, and on a large library it would otherwise - # hold the writer while the store initializations above still - # need it, so startup would wait out the write budget and fail - # with `database is locked` - # (ISSUE-CONTENT-BACKFILL-BLOCKS-STARTUP). - app.state.memory_governance_store.release_content_backfill_startup_gate() diagnostic_event( level="info", component="core.lifecycle", @@ -1241,6 +1327,11 @@ def create_app( instance_id=runtime_settings.instance_id, duration_ms=elapsed_ms(lifecycle_started), ) + app.state.recovery_task = watched_task( + asyncio.get_running_loop(), + recover_after_ready, + name="one-memory-post-ready-recovery", + ) yield finally: shutdown_started = time.monotonic() @@ -1251,6 +1342,14 @@ def create_app( outcome="started", instance_id=runtime_settings.instance_id, ) + # First: recovery may be about to start what is stopped below. + if app.state.recovery_task is not None: + app.state.recovery_task.cancel() + with suppress(asyncio.CancelledError): + await app.state.recovery_task + # Cancelled before it got that far, recovery never let the + # rebuild go; a held rebuild would hold shutdown below too. + app.state.memory_governance_store.release_content_backfill_startup_gate() app.state.main_database_report.cancel() app.state.loop_heartbeat.cancel() with suppress(asyncio.CancelledError): @@ -1278,8 +1377,9 @@ def create_app( await app.state.live_sync_supervisor.shutdown() await app.state.import_supervisor.shutdown() app.state.memory_write_drain.stop() - with suppress(asyncio.CancelledError): - await app.state.memory_write_drain_task + if app.state.memory_write_drain_task is not None: + with suppress(asyncio.CancelledError): + await app.state.memory_write_drain_task await app.state.sdk_memory_runtime.shutdown() await app.state.sdk_write_coordinator.close() if app.state.sdk_client_handle is not None: @@ -1497,6 +1597,8 @@ def create_app( content_candidate_backfill_failure=( governance.content_candidate_backfill_failure() if governance is not None else None ), + recovery=getattr(app.state, "recovery", "pending"), + recovery_failed_steps=getattr(app.state, "recovery_failed_steps", ()), ) @app.get("/health", response_model=HealthResponse, include_in_schema=False) diff --git a/memhub/apps/core/tests/TEST_INDEX.md b/memhub/apps/core/tests/TEST_INDEX.md index f7d1b57..8b61c97 100644 --- a/memhub/apps/core/tests/TEST_INDEX.md +++ b/memhub/apps/core/tests/TEST_INDEX.md @@ -68,6 +68,7 @@ | `test_conversation_search.py` | `integration` | Searching imported conversations before any generation model is configured. | | `test_conversation_seed_ids.py` | `contract` | 覆盖 seed ids read metadata only without hydrating、seed eligibility keeps person and memory content only、latest activity includes nonseed turns and import time fallback 等 13 个场景。 | | `test_conversation_turn_grouping.py` | `contract` | A proposal and the turn that adopts it must reach HY as one turn. | +| `test_core_ready_before_recovery.py` | `integration` | Readiness never waits for picking up work an earlier run left unfinished. | | `test_core_test_reporting.py` | `contract` | 覆盖 report writes sanitized machine readable views、file and marker aggregates are stable、classification suggestions never guess unit contract or integration 等 6 个场景。 | | `test_credential_responsiveness.py` | `integration` | 覆盖 credentials do not block the event loop、disabled automatic processing does not digest onboarding、answering reads credentials off loop 等 5 个场景。 | | `test_cursor_default_root_windows.py` | `contract` | R-29 剩余：Cursor 默认根路径不能写死成 macOS。 | @@ -302,4 +303,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. | -共 297 个测试文件。 +共 298 个测试文件。 diff --git a/memhub/apps/core/tests/core_test_classification.py b/memhub/apps/core/tests/core_test_classification.py index f4da74c..b53476e 100644 --- a/memhub/apps/core/tests/core_test_classification.py +++ b/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 = 297 -EXPECTED_TEST_FILES_SHA256 = "50fc934916818220efdd6259dafb8831c2e568a19fc63e2ea6a9b865d1a75c90" +EXPECTED_TEST_FILE_COUNT = 298 +EXPECTED_TEST_FILES_SHA256 = "3b2bece1e20fc1db0e8a82db65bb2cd189317ca34922c97ae8849d38283761e8" # Pure transformations, validation models, parity checks and state machines. # This list is intentionally small: uncertain files default to contract. @@ -51,6 +51,7 @@ INTEGRATION_FILES = frozenset( { "test_auth.py", "test_backup_endpoint.py", + "test_core_ready_before_recovery.py", "test_backup_restore_api.py", "test_c1_import_via_api_path.py", "test_capture_deadline.py", diff --git a/memhub/apps/core/tests/test_core_ready_before_recovery.py b/memhub/apps/core/tests/test_core_ready_before_recovery.py new file mode 100644 index 0000000..e1ea8b9 --- /dev/null +++ b/memhub/apps/core/tests/test_core_ready_before_recovery.py @@ -0,0 +1,144 @@ +"""Readiness never waits for picking up work an earlier run left unfinished. + +A 1.0.4 home with pending imports and write jobs used to resume all of it +before the lifespan yielded, so /health did not answer within the Runtime's +60s budget and the Runtime killed a healthy Core, which then began the same +recovery again. Recovery now runs after readiness. +""" + +import asyncio +import threading +import time + +import pytest +from httpx import ASGITransport, AsyncClient +from pydantic import SecretStr + +from one_memory import hy_sdk_migration +from one_memory.api import create_app +from one_memory.config import Settings +from one_memory.hy_sdk_migration import MetadataRepairResult +from one_memory.secrets import InMemorySecretStore + +pytestmark = pytest.mark.fresh_sdk + + +def application(tmp_path): + return create_app( + Settings( + data_dir=tmp_path / "data", + source_home=tmp_path, + api_token=SecretStr("ready-test"), + imports_enabled=False, + automatic_memory_processing=False, + live_sync_enabled=False, + ), + secret_store=InMemorySecretStore(), + ) + + +def client_for(app): + return AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + headers={"Authorization": "Bearer ready-test"}, + ) + + +async def health(client): + response = await client.get("/api/v1/health") + assert response.status_code == 200 + return response.json() + + +async def wait_for_recovery(client, *, seconds=10.0): + deadline = time.monotonic() + seconds + while time.monotonic() < deadline: + body = await health(client) + if body["recovery"] in {"done", "failed"}: + return body + await asyncio.sleep(0.01) + raise AssertionError("recovery did not finish") + + +def test_core_is_ready_while_recovery_is_still_running(tmp_path, monkeypatch): + release = threading.Event() + entered = threading.Event() + + def slow_repair(**_): + entered.set() + release.wait(10) + return MetadataRepairResult(None, 0, 0, 0) + + monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", slow_repair) + + async def scenario(): + app = application(tmp_path) + try: + started = time.monotonic() + async with app.router.lifespan_context(app), client_for(app) as client: + # Startup finished although the first recovery step cannot. + assert time.monotonic() - started < 5 + assert await asyncio.to_thread(entered.wait, 5) + body = await health(client) + assert body["status"] == "ok" + assert body["recovery"] == "running" + # Nothing after the blocked step has been started yet: the + # write drain in particular would hold the single writer. + assert app.state.memory_write_drain_task is None + # Reads answer during recovery. + assert (await client.get("/api/v1/sources")).status_code == 200 + release.set() + body = await wait_for_recovery(client) + assert body["recovery"] == "done" + assert body["recovery_failed_steps"] == [] + assert app.state.memory_write_drain_task is not None + finally: + release.set() + + asyncio.run(scenario()) + + +def test_a_failed_recovery_step_does_not_stop_readiness_or_later_steps(tmp_path, monkeypatch): + def broken_repair(**_): + raise OSError("bundle unreadable") + + monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", broken_repair) + + async def scenario(): + app = application(tmp_path) + async with app.router.lifespan_context(app), client_for(app) as client: + body = await wait_for_recovery(client) + assert body["status"] == "ok" + assert body["recovery"] == "failed" + assert body["recovery_failed_steps"] == ["sdk_metadata_repair"] + # The steps after the failed one still ran. + assert app.state.memory_write_drain_task is not None + + asyncio.run(scenario()) + + +def test_shutdown_during_recovery_stops_it_without_starting_more(tmp_path, monkeypatch): + release = threading.Event() + entered = threading.Event() + + def slow_repair(**_): + entered.set() + release.wait(10) + return MetadataRepairResult(None, 0, 0, 0) + + monkeypatch.setattr(hy_sdk_migration, "repair_sdk_metadata_from_bundle", slow_repair) + + async def scenario(): + app = application(tmp_path) + try: + async with app.router.lifespan_context(app): + assert await asyncio.to_thread(entered.wait, 5) + recovery = app.state.recovery_task + # Shutdown cancelled recovery before it could start the drain. + assert recovery.done() + assert app.state.memory_write_drain_task is None + finally: + release.set() + + asyncio.run(scenario()) diff --git a/memhub/apps/core/tests/test_health.py b/memhub/apps/core/tests/test_health.py index b59bc1d..dd17295 100644 --- a/memhub/apps/core/tests/test_health.py +++ b/memhub/apps/core/tests/test_health.py @@ -45,6 +45,9 @@ def test_health_reports_the_single_core() -> None: "memory_backfill_error": None, "content_candidate_backfill": "idle", "content_candidate_backfill_failure": None, + # No lifespan ran, so recovery has not begun. + "recovery": "pending", + "recovery_failed_steps": [], } @@ -210,6 +213,8 @@ def test_lifespan_drains_committed_pending_dispatch_through_enqueue(tmp_path, mo async def scenario() -> None: async with app.router.lifespan_context(app): + # Replayed by the recovery that runs after readiness. + await app.state.recovery_task assert record.evidence_id in seen assert evidence.take_all_stable_records() == () @@ -434,6 +439,7 @@ def test_a_backfill_that_fails_says_so_rather_than_disappearing(tmp_path, monkey app.router.lifespan_context(app), AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client, ): + await app.state.recovery_task await app.state.memory_backfill response = await client.get("/api/v1/health") @@ -497,6 +503,7 @@ def test_a_failed_catch_up_leaves_a_traceback_behind(tmp_path, monkeypatch, capl app.router.lifespan_context(app), AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client, ): + await app.state.recovery_task with suppress(BaseException): await app.state.memory_backfill return (await client.get("/api/v1/health")).json() diff --git a/memhub/apps/core/tests/test_source_permission_races.py b/memhub/apps/core/tests/test_source_permission_races.py index 0e08db6..993c21b 100644 --- a/memhub/apps/core/tests/test_source_permission_races.py +++ b/memhub/apps/core/tests/test_source_permission_races.py @@ -295,6 +295,8 @@ def test_startup_reconciles_enabled_source_from_config(tmp_path, monkeypatch): secret_store=InMemorySecretStore(), ) async with restored.router.lifespan_context(restored): + # The reconcile runs after readiness, in recovery. + await restored.state.recovery_task history = [ job for job in restored.state.import_job_store.list_jobs() diff --git a/memhub/apps/core/tests/test_source_status_latency.py b/memhub/apps/core/tests/test_source_status_latency.py index 56fea1f..283b12c 100644 --- a/memhub/apps/core/tests/test_source_status_latency.py +++ b/memhub/apps/core/tests/test_source_status_latency.py @@ -111,6 +111,7 @@ def test_large_import_status_does_not_scan_source_trees(tmp_path, monkeypatch): app, catalog = build_app(tmp_path, monkeypatch) seed_layouts(catalog) async with app.router.lifespan_context(app), client_for(app) as client: + await app.state.recovery_task await app.state.memory_backfill await asyncio.to_thread(seed_evidence, app) probes = [] diff --git a/memhub/apps/core/tests/test_visibility_background.py b/memhub/apps/core/tests/test_visibility_background.py index e1be6df..f35a06b 100644 --- a/memhub/apps/core/tests/test_visibility_background.py +++ b/memhub/apps/core/tests/test_visibility_background.py @@ -62,6 +62,8 @@ def test_the_switch_on_starts_it(library, monkeypatch): async def scenario() -> None: app = _app(library) async with app.router.lifespan_context(app): + # Started by the recovery that runs after readiness. + await app.state.recovery_task assert app.state.memory_visibility_task is not None asyncio.run(scenario()) diff --git a/memhub/packages/contracts/src/index.ts b/memhub/packages/contracts/src/index.ts index db9d689..69b3650 100644 --- a/memhub/packages/contracts/src/index.ts +++ b/memhub/packages/contracts/src/index.ts @@ -65,6 +65,13 @@ export interface HealthResponse { */ memory_backfill?: "running" | "done" | "failed"; memory_backfill_error?: string | null; + /** + * Work an earlier run left unfinished (write queue, interrupted imports, + * live sync) is picked up after Core is ready. "running" is not degraded: + * readiness never waits for it. Absent from cores older than this field. + */ + recovery?: "pending" | "running" | "done" | "failed"; + recovery_failed_steps?: string[]; } export type ImportStatus = From 567a8b349da3c1512407e48f9f6fc6a863cf6789 Mon Sep 17 00:00:00 2001 From: alvinfei Date: Tue, 27 Oct 2026 09:00:50 +0800 Subject: [PATCH 2/2] test(desktop): let the attach-race winner exist before the loop gives up The test wrote the winning Core's receipt and token from inside the spawn, without awaiting it, while the readiness loop ran with zero delay. On a slow runner the loop finished first, the re-read found a receipt without its token, and start() reported a false conflict. The probe now waits for the winner's files, which is the order the test is about. Co-Authored-By: Claude Opus 5.5 (1M context) --- memhub/apps/desktop/src/core-supervisor.test.ts | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/memhub/apps/desktop/src/core-supervisor.test.ts b/memhub/apps/desktop/src/core-supervisor.test.ts index a90a89a..f646c70 100644 --- a/memhub/apps/desktop/src/core-supervisor.test.ts +++ b/memhub/apps/desktop/src/core-supervisor.test.ts @@ -234,9 +234,14 @@ describe("Core ownership", () => { it("attaches when a standalone Core wins the race with the first spawn", async () => { const dataDir = await mkdtemp(path.join(tmpdir(), "core-attach-race-")); let probes = 0; + // The winner has finished publishing its receipt and token by the time + // our child loses the lock. Writing them from inside the spawn raced the + // readiness loop: on a slow runner the token was not there yet, the + // re-read rejected the receipt, and the test saw a false conflict. + let winner: Promise = Promise.resolve(); const spawnProcess = vi.fn(() => { const child = fakeChild(); - void writeRuntime(dataDir, { + winner = writeRuntime(dataDir, { schema_version: 1, service: "one-memory-core", instance_id: "cli-race-winner", @@ -245,7 +250,9 @@ describe("Core ownership", () => { host: "127.0.0.1", port: 9236, started_at: "2026-09-19T01:00:00Z", - }).then(() => child.emit("exit", 1, null)); + }).then(() => { + child.emit("exit", 1, null); + }); return child; }); const supervisor = new CoreSupervisor({ @@ -256,6 +263,10 @@ describe("Core ownership", () => { spawnProcess, probeHealth: async (port, token) => { probes += 1; + // Our own child never answers: it lost the race. Waiting here for the + // winner's files keeps the readiness loop from concluding before the + // winner exists, which is the order this test is about. + await winner; if (port !== 9236 || token !== "attached-secret") throw new Error("not attached yet"); return {