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/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 { 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 =