mtfyun/Hy-Memory
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #181 from Tencent-Hunyuan/fix/core-ready-before-recovery

Browse Source 
fix(memhub): make Core ready before it resumes unfinished work
This commit is contained in:
feitianxiang authored and GitHub committed 2026-10-27 09:13:25 +08:00
2 parents 0abf87d4b6 567a8b349d
commit 0dd2bcd6b1
10 files changed
+384 -106
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

memhub/apps/core/src/one_memory/api.py
+203 -101
View File
@@ -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)
memhub/apps/core/tests/TEST_INDEX.md
+2 -1
View File
@@ -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 个测试文件。
memhub/apps/core/tests/core_test_classification.py
+3 -2
View File
@@ -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",
memhub/apps/core/tests/test_core_ready_before_recovery.py
+144
View File
@@ -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())
memhub/apps/core/tests/test_health.py
+7
View File
@@ -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()
memhub/apps/core/tests/test_source_permission_races.py
+2
View File
@@ -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()
memhub/apps/core/tests/test_source_status_latency.py
+1
View File
@@ -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 = []
memhub/apps/core/tests/test_visibility_background.py
+2
View File
@@ -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())
memhub/apps/desktop/src/core-supervisor.test.ts
+13 -2
View File
@@ -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<void> = 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 {
memhub/packages/contracts/src/index.ts
+7
View File
@@ -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 =