diff --git a/memhub/apps/cli/src/cli.test.ts b/memhub/apps/cli/src/cli.test.ts index 2cd2358..4ec5fb3 100644 --- a/memhub/apps/cli/src/cli.test.ts +++ b/memhub/apps/cli/src/cli.test.ts @@ -57,6 +57,11 @@ const runtimeMocks = vi.hoisted(() => ({ })), uninstallUserService: vi.fn(async (..._args: unknown[]) => "removed"), restartUserService: vi.fn(async () => undefined), + runSdkMigration: vi.fn( + async (): Promise<{ state: string; code?: string }> => ({ + state: "not_required", + }), + ), rollbackLocalInstall: vi.fn(async () => undefined), planCliLink: vi.fn(async ({ home, installRoot }) => ({ schema_version: 1, @@ -126,6 +131,7 @@ vi.mock("@one-memory/runtime-control", async (importOriginal) => { installUserService: runtimeMocks.installUserService, uninstallUserService: runtimeMocks.uninstallUserService, restartUserService: runtimeMocks.restartUserService, + runSdkMigration: runtimeMocks.runSdkMigration, rollbackLocalInstall: runtimeMocks.rollbackLocalInstall, planCliLink: runtimeMocks.planCliLink, installCliLink: runtimeMocks.installCliLink, @@ -889,6 +895,108 @@ describe("public One Memory CLI", () => { expect(runtimeMocks.restartUserService).toHaveBeenCalledTimes(2); }); + const upgradeArguments = [ + "upgrade", + "--source", + "/tmp/dist", + "--home", + "/tmp/home", + "--install-root", + "/tmp/install", + "--data-dir", + "/tmp/custom data", + "--yes", + ]; + const presentCliLink = { + schema_version: 1 as const, + platform: "linux" as const, + home: "/tmp/home", + install_root: "/tmp/install", + bin_directory: "/tmp/bin", + status: "present" as const, + blocked_reason: null, + link_path: "/tmp/bin/one-memory", + target_path: "/tmp/install/current/bin/one-memory", + }; + + it("migrates the home with the new Core before restarting the Runtime", async () => { + vi.clearAllMocks(); + runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink); + const order: string[] = []; + runtimeMocks.runSdkMigration.mockImplementationOnce(async () => { + order.push("migrate"); + return { state: "cutover" }; + }); + runtimeMocks.restartUserService.mockImplementationOnce(async () => { + order.push("restart"); + }); + const output = captureStdout(); + let code: number; + try { + code = await run( + upgradeArguments, + {}, + { waitForRuntimeReady: async () => true }, + ); + } finally { + output.restore(); + } + expect(code!).toBe(0); + expect(order).toEqual(["migrate", "restart"]); + expect(runtimeMocks.runSdkMigration).toHaveBeenCalledWith( + expect.objectContaining({ dataDir: "/tmp/custom data" }), + ); + }); + + it("rolls back to the previous version when the migration does not finish", async () => { + vi.clearAllMocks(); + runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink); + runtimeMocks.runSdkMigration.mockResolvedValueOnce({ + state: "deferred", + code: "sdk_migration_disk_space_insufficient", + }); + const streams = captureStreams(); + let code: number; + try { + code = await run( + upgradeArguments, + {}, + { waitForRuntimeReady: async () => true }, + ); + } finally { + streams.restore(); + } + expect(code!).toBe(1); + expect(JSON.parse(streams.stderr())).toMatchObject({ + command: "upgrade", + code: "sdk_migration_disk_space_insufficient", + }); + expect(runtimeMocks.rollbackLocalInstall).toHaveBeenCalledOnce(); + }); + + it("keeps the new version when a migrated home is slow to come up", async () => { + vi.clearAllMocks(); + runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink); + runtimeMocks.runSdkMigration.mockResolvedValueOnce({ state: "cutover" }); + const streams = captureStreams(); + let code: number; + try { + code = await run( + upgradeArguments, + {}, + { waitForRuntimeReady: async () => false }, + ); + } finally { + streams.restore(); + } + expect(code!).toBe(1); + expect(JSON.parse(streams.stderr())).toMatchObject({ + command: "upgrade", + code: "runtime_not_ready_after_sdk_migration", + }); + expect(runtimeMocks.rollbackLocalInstall).not.toHaveBeenCalled(); + }); + it("reports success once the upgraded Runtime is healthy", async () => { vi.clearAllMocks(); runtimeMocks.planCliLink.mockResolvedValueOnce({ diff --git a/memhub/apps/cli/src/cli.ts b/memhub/apps/cli/src/cli.ts index 443ff20..b451ad7 100644 --- a/memhub/apps/cli/src/cli.ts +++ b/memhub/apps/cli/src/cli.ts @@ -39,7 +39,9 @@ import { rollbackCliLink, rollbackLocalInstall, runRuntimeCli, + runSdkMigration, runtimeInstallLayout, + type SdkMigrationResult, stopRuntimeProcess, } from "@one-memory/runtime-control"; import axios from "axios"; @@ -354,6 +356,9 @@ export interface RunControls { /** Wait for a restarted Runtime to become healthy after `upgrade`. * Injectable so the upgrade tests do not spend real seconds polling. */ waitForRuntimeReady?: (dataDir: string) => Promise; + /** Run the SDK data migration before install/upgrade starts the new + * Runtime. Injectable so the tests do not spawn a real Core. */ + migrateSdkData?: typeof migrateSdkDataBeforeStart; restartManagedRuntime?: typeof restartManagedService; stopRuntime?: typeof stopRuntimeProcess; ensureRuntime?: typeof ensureRuntimeProcess; @@ -3149,6 +3154,7 @@ async function runCommand( null; let installReceiptWritten = false; let serviceApplied = false; + let migrated = false; let previousOwner: | Awaited> | undefined; @@ -3159,6 +3165,22 @@ async function runCommand( } shellPathReceipt = await applyShellPath(shellPath); previousOwner = await claimCliServiceOwnership(layout.dataDir); + // `install` over an existing home (reinstall, App-to-CLI takeover) must + // migrate it too, before the new Runtime starts serving it. + const migration = await ( + controls.migrateSdkData ?? migrateSdkDataBeforeStart + )({ + coreCommand: layout.coreCommand, + dataDir: layout.dataDir, + runtimeBuild: String(plan.build_number ?? plan.version), + asJson: true, + }); + if (migration.state === "deferred") { + throw new SdkMigrationDeferred( + migration.code ?? "sdk_migration_incomplete", + ); + } + migrated = migration.state === "cutover"; await applyPreparedServiceInstall(preparedService); serviceApplied = true; // The receipt exists so a later command can find the data dir without @@ -3210,6 +3232,13 @@ async function runCommand( ); } }; + if (migrated) { + // An older Runtime refuses a migrated home: keep this install. + process.stderr.write( + `${JSON.stringify(notReadyAfterMigration("install"), null, 2)}\n`, + ); + return EXIT_FAILED; + } if (serviceApplied) { await rollbackPreparedServiceInstall( preparedService, @@ -3219,7 +3248,13 @@ async function runCommand( await rollbackFilesAndOwner(); } process.stderr.write( - `${JSON.stringify(serviceErrorDocument("install", error), null, 2)}\n`, + `${JSON.stringify( + error instanceof SdkMigrationDeferred + ? sdkMigrationFailure("install", error.code) + : serviceErrorDocument("install", error), + null, + 2, + )}\n`, ); return EXIT_FAILED; } @@ -3581,6 +3616,7 @@ async function runCommand( let previousOwner: | Awaited> | undefined; + let migrated = false; const rollbackUpgrade = async ( dataDir: string | undefined, ): Promise => { @@ -3612,6 +3648,24 @@ async function runCommand( memhubReceipts.push(await installManagedCliLink(link)); } previousOwner = await claimCliServiceOwnership(layout.dataDir); + const migration = await ( + controls.migrateSdkData ?? migrateSdkDataBeforeStart + )({ + coreCommand: layout.coreCommand, + dataDir: parsed.dataDir ?? layout.dataDir, + runtimeBuild: String(plan.build_number ?? plan.version), + asJson, + }); + if (migration.state === "deferred") { + await rollbackUpgrade(parsed.dataDir); + process.stderr.write( + `${JSON.stringify(sdkMigrationFailure("upgrade", migration.code ?? "sdk_migration_incomplete"), null, 2)}\n`, + ); + return EXIT_FAILED; + } + // An older Runtime refuses a migrated home, so from here a failure + // keeps the new version instead of rolling back to one that cannot start. + migrated = migration.state === "cutover"; await restartUserService(layout, serviceOptions); // A restart that returns without error only means launchd/systemd // accepted the command. O-22: the first `upgrade` implementation @@ -3623,6 +3677,12 @@ async function runCommand( const ready = await (controls.waitForRuntimeReady ?? waitForRuntimeReady)( dataDir, ); + if (!ready && migrated) { + process.stderr.write( + `${JSON.stringify(notReadyAfterMigration("upgrade"), null, 2)}\n`, + ); + return EXIT_FAILED; + } if (!ready) { await rollbackUpgrade(parsed.dataDir); process.stderr.write( @@ -3650,9 +3710,9 @@ async function runCommand( }); } } catch (error) { - await rollbackUpgrade(parsed.dataDir); + if (!migrated) await rollbackUpgrade(parsed.dataDir); process.stderr.write( - `${JSON.stringify(serviceErrorDocument("upgrade", error), null, 2)}\n`, + `${JSON.stringify(migrated ? notReadyAfterMigration("upgrade") : serviceErrorDocument("upgrade", error), null, 2)}\n`, ); return EXIT_FAILED; } @@ -4020,6 +4080,71 @@ export function runtimeIsReady( ); } +const SDK_MIGRATION_STEPS: Record = { + bundle: "正在备份记忆数据，数据较多时需要几分钟……", + prepare: "正在整理待处理的写入任务……", + cutover: "正在切换记忆数据源……", + cleanup: "正在清理旧数据表……", +}; + +/** + * Migrate a home written by an older release before the new Runtime serves it. + * + * Runs the new Runtime's own Core with the Runtime stopped. Progress goes to + * stderr so stdout stays one result document. + */ +async function migrateSdkDataBeforeStart(options: { + coreCommand: string; + dataDir: string; + runtimeBuild: string; + asJson: boolean; +}): Promise { + return runSdkMigration({ + dataDir: options.dataDir, + core: { command: options.coreCommand }, + runtimeBuild: options.runtimeBuild, + onProgress: (step) => { + process.stderr.write( + options.asJson + ? `${JSON.stringify({ schema_version: 1, product: PRODUCT, event: "sdk_migration.progress", step })}\n` + : `${SDK_MIGRATION_STEPS[step] ?? step}\n`, + ); + }, + }); +} + +class SdkMigrationDeferred extends Error { + constructor(readonly code: string) { + super(code); + } +} + +function notReadyAfterMigration(command: "install" | "upgrade") { + return { + schema_version: 1, + product: PRODUCT, + command, + status: "error", + code: "runtime_not_ready_after_sdk_migration", + message: + "记忆数据已迁移到新版本，但新版本 Runtime 没有在限定时间内就绪。旧版本无法打开已迁移的数据，所以没有回滚。请运行 memhub restart，或 memhub doctor 查看原因。", + }; +} + +function sdkMigrationFailure(command: "install" | "upgrade", code: string) { + return { + schema_version: 1, + product: PRODUCT, + command, + status: "error", + code, + message: + code === "sdk_migration_disk_space_insufficient" + ? "磁盘空间不足，记忆数据迁移没有完成，已回滚到上一版本。请腾出与记忆数据同等大小的空间后重试；原有记忆数据不会丢失。" + : "记忆数据迁移没有完成，已回滚到上一版本。请重试；如果仍然失败，请把这个错误码反馈给我们。原有记忆数据不会丢失。", + }; +} + async function waitForRuntimeReady( dataDir: string, timeoutMs = 45_000, diff --git a/memhub/apps/core/src/one_memory/cli.py b/memhub/apps/core/src/one_memory/cli.py index 91fee4b..c5676de 100644 --- a/memhub/apps/core/src/one_memory/cli.py +++ b/memhub/apps/core/src/one_memory/cli.py @@ -1,5 +1,3 @@ -import asyncio -import hashlib import os import signal import socket @@ -115,16 +113,10 @@ def run(arguments: Sequence[str], *, output: TextIO | None = None) -> int: command, *options = arguments if command == "restore": return restore(options, stream) - if command == "migration-bundle": - return migration_bundle(options, stream) if command == "sdk-migrate": return sdk_migrate(options, stream) - if command == "sdk-cutover": - return sdk_cutover(options, stream) if command == "sdk-migration-status": return sdk_migration_status(options, stream) - if command == "sdk-cleanup": - return sdk_cleanup(options, stream) if command == "boundary" and not options: from one_memory.workbuddy_boundary import write_hook @@ -145,11 +137,8 @@ def run(arguments: Sequence[str], *, output: TextIO | None = None) -> int: stream.write( "Usage: one-memory [serve [--maintenance|--reader]|status [--json]|doctor [--json]" "|restore --source --yes [--json]" - "|migration-bundle --bundle-id --runtime-build --yes [--json]" - "|sdk-migrate --migration-id --bundle-id --yes [--json]" - "|sdk-cutover --migration-id --bundle-id --yes [--json]" + "|sdk-migrate --runtime-build --yes [--json]" "|sdk-migration-status [--json]" - "|sdk-cleanup --yes [--json]" "|boundary [--peek|--ack TOKEN]|capture SOURCE]\n" ) return 2 @@ -291,19 +280,20 @@ def _write_restore( stream.write(f"{text}\n") -def migration_bundle(options: Sequence[str], stream: TextIO) -> int: - """Create a complete management + SDK bundle while Runtime is stopped.""" - bundle_id: str | None = None +def sdk_migrate(options: Sequence[str], stream: TextIO) -> int: + """Bring a stopped home to SDK authority in one resumable command. + + Backs up the home, makes queue work independent of the legacy tables, + commits SDK authority and drops those tables. The SDK already holds every + memory and raw turn, so nothing is copied. With `--json`, each step is one + progress line and the last line is the result. + """ runtime_build: str | None = None confirmed = False as_json = False index = 0 while index < len(options): option = options[index] - if option == "--bundle-id" and index + 1 < len(options): - bundle_id = options[index + 1] - index += 2 - continue if option == "--runtime-build" and index + 1 < len(options): runtime_build = options[index + 1] index += 2 @@ -318,133 +308,8 @@ def migration_bundle(options: Sequence[str], stream: TextIO) -> int: continue stream.write(f"one-memory: 参数无效：{option}\n") return 2 - if not bundle_id or not runtime_build: - stream.write("one-memory: migration-bundle 需要 --bundle-id --runtime-build \n") - return 2 - - settings = load_settings() - status = inspect_runtime(settings.data_dir) - if status.state != "stopped": - document = { - "schema_version": 1, - "state": "blocked", - "code": "sdk_migration_bundle_runtime_running", - "runtime_state": status.state, - } - _write_restore(stream, as_json, document, "Runtime 仍在运行；不能创建迁移备份。") - return 4 - if not confirmed: - document = { - "schema_version": 1, - "state": "preview", - "bundle_id": bundle_id, - "requires_confirmation": True, - } - _write_restore(stream, as_json, document, "预览通过；加 --yes 后创建完整迁移备份。") - return 4 - - from one_memory.sdk_migration_bundle import ( - MigrationBundleRejected, - create_sdk_migration_bundle, - inspect_sdk_migration_bundle, - ) - - try: - result = create_sdk_migration_bundle( - data_dir=settings.data_dir, - bundle_id=bundle_id, - runtime_build=runtime_build, - runtime_stopped=True, - ) - except MigrationBundleRejected as error: - if error.code == "sdk_migration_bundle_exists": - try: - manifest = inspect_sdk_migration_bundle( - data_dir=settings.data_dir, - bundle_id=bundle_id, - ) - except MigrationBundleRejected: - pass - else: - document = { - "schema_version": 1, - "state": "reused", - "bundle_id": bundle_id, - "file_count": manifest.get("file_count"), - "total_bytes": manifest.get("total_bytes"), - } - _write_restore( - stream, - as_json, - document, - f"已验证并复用迁移备份：{bundle_id}", - ) - return 0 - document = { - "schema_version": 1, - "state": "failed", - "code": migration_failure_code(error, fallback=error.code), - } - _write_restore(stream, as_json, document, f"迁移备份失败：{document['code']}") - return 1 - except OSError as error: - # A full disk while copying is the most common first-launch failure and - # the only one the person fixes by deleting something. - code = migration_failure_code(error, fallback="sdk_migration_bundle_copy_failed") - document = {"schema_version": 1, "state": "failed", "code": code} - _write_restore(stream, as_json, document, f"迁移备份失败：{code}") - return 1 - document = { - "schema_version": 1, - "state": "created", - "bundle_id": result.bundle_id, - "bundle_path": str(result.bundle_path), - "file_count": result.file_count, - "total_bytes": result.total_bytes, - } - _write_restore(stream, as_json, document, f"完整迁移备份已创建：{result.bundle_path}") - return 0 - - -def cutover_failure_code(error: BaseException) -> str: - code = getattr(error, "code", "") - if isinstance(code, str) and _SAFE_MIGRATION_CODE.match(code): - return code - return "sdk_migration_cutover_failed" - - -def sdk_migrate(options: Sequence[str], stream: TextIO) -> int: - """Make legacy queue work self-contained while keeping legacy authority. - - The SDK already holds every memory and raw turn, so nothing is copied. - """ - migration_id: str | None = None - bundle_id: str | None = None - confirmed = False - as_json = False - index = 0 - while index < len(options): - option = options[index] - if option == "--migration-id" and index + 1 < len(options): - migration_id = options[index + 1] - index += 2 - continue - if option == "--bundle-id" and index + 1 < len(options): - bundle_id = options[index + 1] - index += 2 - continue - if option == "--yes": - confirmed = True - index += 1 - continue - if option == "--json": - as_json = True - index += 1 - continue - stream.write(f"one-memory: 参数无效：{option}\n") - return 2 - if not migration_id or not bundle_id: - stream.write("one-memory: sdk-migrate 需要 --migration-id --bundle-id \n") + if not runtime_build: + stream.write("one-memory: sdk-migrate 需要 --runtime-build \n") return 2 settings = load_settings() @@ -459,178 +324,53 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int: _write_restore(stream, as_json, document, "Runtime 仍在运行；不能迁移。") return 4 if not confirmed: - document = { - "schema_version": 1, - "state": "preview", - "migration_id": migration_id, - "bundle_id": bundle_id, - "requires_confirmation": True, - "commits_cutover": False, - } - _write_restore( - stream, - as_json, - document, - "预览通过；加 --yes 后准备迁移，但不会切换 authority。", - ) + document = {"schema_version": 1, "state": "preview", "requires_confirmation": True} + _write_restore(stream, as_json, document, "预览通过；加 --yes 后执行迁移。") return 4 - from one_memory.database_access import close_main_databases from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE, HY_MEMORY_VERSION - from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate - from one_memory.memory_server.migrations import MIGRATIONS - from one_memory.sdk_data_plane_migration import SdkDataPlaneMigrationStore - from one_memory.sdk_migration_bundle import ( - MigrationBundleRejected, - inspect_sdk_migration_bundle, - ) - from one_memory.sdk_migration_runner import ( - PREPARED_PHASE, - prepare_legacy_work_for_cutover, - recover_abandoned_staging, - ) + from one_memory.sdk_migration_runner import run_startup_migration - database = settings.data_dir / "one-memory.sqlite3" - try: - manifest = inspect_sdk_migration_bundle( - data_dir=settings.data_dir, - bundle_id=bundle_id, - ) - asyncio.run(migrate(SqliteMigrationStore(database), MIGRATIONS)) - close_main_databases() - recover_abandoned_staging( - data_dir=settings.data_dir, - database_path=database, - namespace=HY_MEMORY_NAMESPACE, - ) - store = SdkDataPlaneMigrationStore(database) - store.start( - migration_id=migration_id, - source_runtime_build=str(manifest.get("runtime_build") or ""), - target_sdk_version=HY_MEMORY_VERSION, - source_schema_version=MIGRATIONS[-1].version, - namespace_fingerprint=hashlib.sha256(HY_MEMORY_NAMESPACE.encode()).hexdigest(), - backend_fingerprint=hashlib.sha256(b"chroma:kuzu:sqlite").hexdigest(), - backup_bundle_id=bundle_id, - ) - preparation = prepare_legacy_work_for_cutover(database) - store.set_phase(migration_id, phase=PREPARED_PHASE, state="ready") - except MigrationBundleRejected as error: - code = error.code - except Exception as error: - code = migration_failure_code(error, fallback="sdk_migration_preflight_failed") - else: - document = { - "schema_version": 1, - "state": "ready", - "phase": PREPARED_PHASE, - "migration_id": migration_id, - **preparation.as_document(), - "commits_cutover": False, - } - _write_restore(stream, as_json, document, "迁移准备已完成；authority 尚未切换。") - return 0 - - document = {"schema_version": 1, "state": "failed", "code": code} - _write_restore(stream, as_json, document, f"迁移准备失败：{code}") - return 1 - - -def sdk_cutover(options: Sequence[str], stream: TextIO) -> int: - """Commit SDK authority once the backup is verified and the queue is self-contained.""" - migration_id: str | None = None - bundle_id: str | None = None - confirmed = False - as_json = False - index = 0 - while index < len(options): - option = options[index] - if option == "--migration-id" and index + 1 < len(options): - migration_id = options[index + 1] - index += 2 - continue - if option == "--bundle-id" and index + 1 < len(options): - bundle_id = options[index + 1] - index += 2 - continue - if option == "--yes": - confirmed = True - index += 1 - continue - if option == "--json": - as_json = True - index += 1 - continue - stream.write(f"one-memory: 参数无效：{option}\n") - return 2 - if not migration_id or not bundle_id: - stream.write("one-memory: sdk-cutover 需要 --migration-id --bundle-id \n") - return 2 - settings = load_settings() - status = inspect_runtime(settings.data_dir) - if status.state != "stopped": - document = { - "schema_version": 1, - "state": "blocked", - "code": "sdk_cutover_runtime_running", - "runtime_state": status.state, - } - _write_restore(stream, as_json, document, "Runtime 仍在运行；不能切换 authority。") - return 4 - if not confirmed: - document = { - "schema_version": 1, - "state": "preview", - "migration_id": migration_id, - "bundle_id": bundle_id, - "requires_confirmation": True, - } - _write_restore(stream, as_json, document, "预览通过；加 --yes 后提交 SDK authority。") - return 4 - - from one_memory.hy_runtime_config import HY_MEMORY_VERSION - from one_memory.sdk_data_plane_migration import DataPlaneMigrationError - from one_memory.sdk_migration_bundle import MigrationBundleRejected - from one_memory.sdk_migration_runner import commit_sdk_cutover, resume_legacy_cleanup - - try: - database_path = settings.data_dir / "one-memory.sqlite3" - marker = commit_sdk_cutover( - data_dir=settings.data_dir, - database_path=database_path, - migration_id=migration_id, - bundle_id=bundle_id, - sdk_version=HY_MEMORY_VERSION, - ) - resume_legacy_cleanup( - data_dir=settings.data_dir, - database_path=database_path, - ) - except (DataPlaneMigrationError, MigrationBundleRejected, OSError) as error: - code = cutover_failure_code(error) - document = { - "schema_version": 1, - "state": "blocked", - "code": code, - "migration_id": migration_id, - "bundle_id": bundle_id, - } - _write_restore(stream, as_json, document, f"SDK authority 未切换：{code}") - return 4 - document = { - "schema_version": 1, - "state": "cutover", - "migration_id": marker.migration_id, - "bundle_id": marker.backup_bundle_id, - "sdk_version": marker.sdk_version, - "completed_at": marker.completed_at, + step_names = { + "bundle": "正在备份记忆数据……", + "prepare": "正在整理待处理的写入任务……", + "cutover": "正在切换数据源……", + "cleanup": "正在删除旧表……", } - _write_restore(stream, as_json, document, "SDK authority 已切换。") + + def progress(step: str) -> None: + if as_json: + stream.write( + json.dumps({"schema_version": 1, "state": "running", "step": step}) + "\n" + ) + else: + stream.write(f"{step_names.get(step, step)}\n") + stream.flush() + + try: + result = run_startup_migration( + data_dir=settings.data_dir, + runtime_build=runtime_build, + sdk_version=HY_MEMORY_VERSION, + namespace=HY_MEMORY_NAMESPACE, + progress=progress, + ) + except Exception as error: + code = migration_failure_code(error) + document = {"schema_version": 1, "state": "failed", "code": code} + _write_restore(stream, as_json, document, f"迁移未完成：{code}") + return 1 + messages = { + "not_required": "不需要迁移。", + "complete": "迁移已完成，旧表已清理。", + "cutover": "迁移完成：记忆数据已以 SDK 为准。", + } + _write_restore(stream, as_json, result.as_document(), messages[result.state]) return 0 def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int: - """Read-only startup probe used before taking the Runtime maintenance fence.""" + """Read-only probe: does this home need `sdk-migrate`?""" if list(options) not in ([], ["--json"]): stream.write("one-memory: sdk-migration-status 只接受 --json\n") return 2 @@ -638,16 +378,12 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int: settings = load_settings() database_path = settings.data_dir / "one-memory.sqlite3" if not database_path.is_file(): - document = { + document: dict[str, object] = { "schema_version": 1, - "authority": "legacy", - "state": "empty", - "migration_id": None, + "authority": "sdk", + "state": "not_required", } else: - import sqlite3 - - from one_memory.database_access import main_database from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE from one_memory.sdk_data_plane_migration import ( MemoryAuthority, @@ -656,70 +392,19 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int: from one_memory.sdk_migration_runner import abandoned_staging_present from one_memory.target_schema import legacy_tables_present - migration_store = SdkDataPlaneMigrationStore(database_path) - marker = migration_store.cutover() - authority = migration_store.authority() - stat = database_path.stat() - generation = hashlib.sha256(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:16] + authority = SdkDataPlaneMigrationStore(database_path).authority() if authority is MemoryAuthority.SDK: - legacy_tables = legacy_tables_present(database_path) - staging_cleanup = abandoned_staging_present( - data_dir=settings.data_dir, - namespace=HY_MEMORY_NAMESPACE, - ) - document = { - "schema_version": 1, - "authority": "sdk", - "state": ("cleanup_required" if legacy_tables or staging_cleanup else "complete"), - "migration_id": marker.migration_id if marker else None, - "generation": generation, - "legacy_tables": list(legacy_tables), - } - else: - with main_database(database_path).read(row_factory=sqlite3.Row) as database: - tables = { - str(row["name"]) - for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'") - } - counts = { - "memory": ( - int( - database.execute("SELECT COUNT(*) FROM governed_memories").fetchone()[0] - ) - if "governed_memories" in tables - else 0 - ), - "raw": ( - int(database.execute("SELECT COUNT(*) FROM evidence").fetchone()[0]) - if "evidence" in tables - else 0 - ), - } - active_migration = ( - database.execute( - "SELECT migration_id,backup_bundle_id " - "FROM sdk_data_plane_migrations " - "WHERE state IN ('running','blocked','ready') " - "ORDER BY created_at DESC LIMIT 1" - ).fetchone() - if "sdk_data_plane_migrations" in tables - else None + leftovers = bool( + legacy_tables_present(database_path) + or abandoned_staging_present( + data_dir=settings.data_dir, + namespace=HY_MEMORY_NAMESPACE, ) - document = { - "schema_version": 1, - "authority": "legacy", - "state": "required" if any(counts.values()) else "empty", - "migration_id": ( - str(active_migration["migration_id"]) if active_migration is not None else None - ), - "bundle_id": ( - str(active_migration["backup_bundle_id"]) - if active_migration is not None - else None - ), - "counts": counts, - "generation": generation, - } + ) + state = "required" if leftovers else "not_required" + else: + state = "required" + document = {"schema_version": 1, "authority": authority.value, "state": state} if as_json: stream.write(f"{json.dumps(document, ensure_ascii=False)}\n") else: @@ -727,65 +412,6 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int: return 0 -def sdk_cleanup(options: Sequence[str], stream: TextIO) -> int: - if set(options) - {"--yes", "--json"} or options.count("--yes") > 1: - stream.write("one-memory: sdk-cleanup 只接受 --yes [--json]\n") - return 2 - confirmed = "--yes" in options - as_json = "--json" in options - settings = load_settings() - status = inspect_runtime(settings.data_dir) - if status.state != "stopped": - document = { - "schema_version": 1, - "state": "blocked", - "code": "sdk_cleanup_runtime_running", - } - _write_restore(stream, as_json, document, "Runtime 仍在运行；不能清理旧表。") - return 4 - if not confirmed: - _write_restore( - stream, - as_json, - { - "schema_version": 1, - "state": "preview", - "requires_confirmation": True, - }, - "加 --yes 后删除已迁移旧表。", - ) - return 4 - from one_memory.sdk_data_plane_migration import DataPlaneMigrationError - from one_memory.sdk_migration_bundle import MigrationBundleRejected - from one_memory.sdk_migration_runner import resume_legacy_cleanup - - try: - dropped = resume_legacy_cleanup( - data_dir=settings.data_dir, - database_path=settings.data_dir / "one-memory.sqlite3", - ) - except (DataPlaneMigrationError, MigrationBundleRejected) as error: - code = getattr(error, "code", str(error)) - _write_restore( - stream, - as_json, - {"schema_version": 1, "state": "blocked", "code": code}, - f"旧表清理未完成：{code}", - ) - return 4 - _write_restore( - stream, - as_json, - { - "schema_version": 1, - "state": "clean", - "dropped_tables": list(dropped), - }, - "旧表已删除。", - ) - return 0 - - def _status_text(command: str, status: RuntimeStatus) -> str: if status.state == "running": location = f"{status.host}:{status.port}" diff --git a/memhub/apps/core/src/one_memory/sdk_migration_runner.py b/memhub/apps/core/src/one_memory/sdk_migration_runner.py index f0ecf9e..670bb72 100644 --- a/memhub/apps/core/src/one_memory/sdk_migration_runner.py +++ b/memhub/apps/core/src/one_memory/sdk_migration_runner.py @@ -8,10 +8,14 @@ commits SDK authority and drops those tables. from __future__ import annotations +import asyncio import hashlib import json +import os import shutil import sqlite3 +from collections.abc import Callable, Iterator +from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path @@ -23,15 +27,22 @@ from one_memory.memory_server.contracts import ( WriteRequest, internal_session_id_for, ) -from one_memory.memory_server.durable_queue import DurableWriteQueue +from one_memory.memory_server.durable_queue import ( + CANCELLED, + COMPLETED, + FAILED, + REAPED_PAYLOAD_MARKER, + DurableWriteQueue, +) from one_memory.memory_server.unified_entry import UnifiedWriteIntake from one_memory.sdk_data_plane_migration import ( CutoverMarker, DataPlaneMigrationError, + MemoryAuthority, SdkDataPlaneMigrationStore, ) from one_memory.sdk_migration_bundle import inspect_sdk_migration_bundle -from one_memory.target_schema import drop_legacy_tables +from one_memory.target_schema import drop_legacy_tables, legacy_tables_present PREPARED_PHASE = "prepared" @@ -101,18 +112,26 @@ def prepare_legacy_write_queue_for_cutover(database_path: Path) -> tuple[int, in if "evidence" not in tables: return (0, 0) jobs = ( - tuple(database.execute("SELECT job_id,payload_json FROM memory_write_jobs")) + tuple( + database.execute("SELECT job_id,state,payload_json FROM memory_write_jobs") + ) if "memory_write_jobs" in tables else () ) evidence = EvidenceStore(database_path) frozen = 0 for row in jobs: - request = WriteRequest.model_validate_json(str(row["payload_json"])) + decoded = json.loads(str(row["payload_json"])) + if isinstance(decoded, dict) and REAPED_PAYLOAD_MARKER in decoded: + # A finished job whose input the queue already released. + continue + request = WriteRequest.model_validate(decoded) if request.input_records: continue records = tuple(evidence.get(identifier) for identifier in request.evidence_ids) if any(record is None for record in records): + if str(row["state"]) in {COMPLETED, FAILED, CANCELLED}: + continue raise DataPlaneMigrationError(f"sdk_data_plane_write_job_input_missing:{row['job_id']}") materialized = tuple(record for record in records if record is not None) updated = request.model_copy(update={"input_records": materialized}) @@ -307,3 +326,167 @@ def resume_legacy_cleanup(*, data_dir: Path, database_path: Path) -> tuple[str, namespace=HY_MEMORY_NAMESPACE, ) return drop_legacy_tables(database_path) + + +MIGRATION_PREFIX = "sdk-authority-v2" +CAPTURE_FENCE = ".sdk-migration-capture-fence.json" + + +@dataclass(frozen=True) +class StartupMigrationResult: + """`not_required`, `cutover` (this run switched authority) or `complete`.""" + + state: str + migration_id: str | None = None + bundle_id: str | None = None + preparation: CutoverPreparation | None = None + dropped_tables: tuple[str, ...] = () + + def as_document(self) -> dict[str, object]: + document: dict[str, object] = {"schema_version": 1, "state": self.state} + if self.migration_id is not None: + document["migration_id"] = self.migration_id + document["bundle_id"] = self.bundle_id + if self.preparation is not None: + document.update(self.preparation.as_document()) + if self.dropped_tables: + document["dropped_tables"] = list(self.dropped_tables) + return document + + +def database_generation(database_path: Path) -> str: + stat = database_path.stat() + return hashlib.sha256(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:16] + + +def _resumable_identity(database_path: Path) -> tuple[str, str]: + """Resume the newest unfinished run, including one build 19 left blocked.""" + with main_database(database_path).read(row_factory=sqlite3.Row) as database: + row = ( + database.execute( + "SELECT migration_id,backup_bundle_id FROM sdk_data_plane_migrations " + "WHERE state IN ('running','blocked','ready') AND migration_id LIKE ? " + "ORDER BY created_at DESC LIMIT 1", + (f"{MIGRATION_PREFIX}-%",), + ).fetchone() + if "sdk_data_plane_migrations" in _tables(database) + else None + ) + if row is not None and row["backup_bundle_id"] == f"{row['migration_id']}-pre-cutover": + return str(row["migration_id"]), str(row["backup_bundle_id"]) + migration_id = f"{MIGRATION_PREFIX}-{database_generation(database_path)}" + return migration_id, f"{migration_id}-pre-cutover" + + +@contextmanager +def capture_fence(data_dir: Path) -> Iterator[None]: + """Host capture hooks return silently while this process is alive.""" + marker = data_dir / CAPTURE_FENCE + temporary = marker.with_name(f"{marker.name}.{os.getpid()}.tmp") + temporary.write_text( + json.dumps({"schema_version": 1, "pid": os.getpid()}) + "\n", + encoding="utf-8", + ) + os.chmod(temporary, 0o600) + temporary.replace(marker) + try: + yield + finally: + marker.unlink(missing_ok=True) + temporary.unlink(missing_ok=True) + + +def _ensure_bundle(*, data_dir: Path, bundle_id: str, runtime_build: str) -> dict[str, object]: + from one_memory.sdk_migration_bundle import ( + MigrationBundleRejected, + create_sdk_migration_bundle, + ) + + try: + create_sdk_migration_bundle( + data_dir=data_dir, + bundle_id=bundle_id, + runtime_build=runtime_build, + runtime_stopped=True, + ) + except MigrationBundleRejected as error: + if error.code != "sdk_migration_bundle_exists": + raise + return inspect_sdk_migration_bundle(data_dir=data_dir, bundle_id=bundle_id) + + +def run_startup_migration( + *, + data_dir: Path, + runtime_build: str, + sdk_version: str, + namespace: str, + progress: Callable[[str], None] = lambda _step: None, +) -> StartupMigrationResult: + """Bring one stopped home to SDK authority. Safe to repeat after any failure. + + The caller must stop Runtime and keep its service from restarting. + """ + database_path = data_dir / "one-memory.sqlite3" + if not database_path.is_file(): + return StartupMigrationResult(state="not_required") + store = SdkDataPlaneMigrationStore(database_path) + if store.authority() is MemoryAuthority.SDK: + if not legacy_tables_present(database_path) and not abandoned_staging_present( + data_dir=data_dir, namespace=namespace + ): + return StartupMigrationResult(state="not_required") + progress("cleanup") + return StartupMigrationResult( + state="complete", + dropped_tables=resume_legacy_cleanup(data_dir=data_dir, database_path=database_path), + ) + + from one_memory.database_access import close_main_databases + from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate + from one_memory.memory_server.migrations import MIGRATIONS + + migration_id, bundle_id = _resumable_identity(database_path) + with capture_fence(data_dir): + progress("bundle") + manifest = _ensure_bundle( + data_dir=data_dir, + bundle_id=bundle_id, + runtime_build=runtime_build, + ) + progress("prepare") + asyncio.run(migrate(SqliteMigrationStore(database_path), MIGRATIONS)) + close_main_databases() + recover_abandoned_staging( + data_dir=data_dir, + database_path=database_path, + namespace=namespace, + ) + store.start( + migration_id=migration_id, + source_runtime_build=str(manifest.get("runtime_build") or ""), + target_sdk_version=sdk_version, + source_schema_version=MIGRATIONS[-1].version, + namespace_fingerprint=hashlib.sha256(namespace.encode()).hexdigest(), + backend_fingerprint=hashlib.sha256(b"chroma:kuzu:sqlite").hexdigest(), + backup_bundle_id=bundle_id, + ) + preparation = prepare_legacy_work_for_cutover(database_path) + store.set_phase(migration_id, phase=PREPARED_PHASE, state="ready") + progress("cutover") + commit_sdk_cutover( + data_dir=data_dir, + database_path=database_path, + migration_id=migration_id, + bundle_id=bundle_id, + sdk_version=sdk_version, + ) + progress("cleanup") + dropped = resume_legacy_cleanup(data_dir=data_dir, database_path=database_path) + return StartupMigrationResult( + state="cutover", + migration_id=migration_id, + bundle_id=bundle_id, + preparation=preparation, + dropped_tables=dropped, + ) diff --git a/memhub/apps/core/tests/test_cli_migration_error_codes.py b/memhub/apps/core/tests/test_cli_migration_error_codes.py index 260a94b..0c609c3 100644 --- a/memhub/apps/core/tests/test_cli_migration_error_codes.py +++ b/memhub/apps/core/tests/test_cli_migration_error_codes.py @@ -11,7 +11,7 @@ from pathlib import Path import pytest -from one_memory.cli import cutover_failure_code, migration_failure_code +from one_memory.cli import migration_failure_code class _Coded(RuntimeError): @@ -99,14 +99,3 @@ def test_the_codes_the_desktop_translates_stay_in_sync(code: str) -> None: desktop = Path(__file__).resolve().parents[2] / "desktop" / "src" / "stable-runtime.ts" assert desktop.is_file(), desktop assert f'"{code}"' in desktop.read_text(encoding="utf-8") - - -def test_cutover_codes_never_carry_free_text() -> None: - assert ( - cutover_failure_code(_Coded("sdk_data_plane_cutover_missing")) - == "sdk_data_plane_cutover_missing" - ) - assert ( - cutover_failure_code(RuntimeError("/Users/someone/private")) - == "sdk_migration_cutover_failed" - ) diff --git a/memhub/apps/core/tests/test_cli_restore.py b/memhub/apps/core/tests/test_cli_restore.py index d3a98e4..87bc116 100644 --- a/memhub/apps/core/tests/test_cli_restore.py +++ b/memhub/apps/core/tests/test_cli_restore.py @@ -2,7 +2,6 @@ import asyncio import io import json import sqlite3 -from types import SimpleNamespace from one_memory import cli from one_memory.config import Settings @@ -115,56 +114,7 @@ def test_restore_sdk_bundle_uses_full_data_plane_restore(tmp_path, monkeypatch) assert '"restored_files":4' in document -def test_migration_bundle_refuses_while_runtime_is_running(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("running", "runtime_verified"), - ) - output = io.StringIO() - - code = cli.run( - [ - "migration-bundle", - "--bundle-id", - "migration-1", - "--runtime-build", - "build-13", - "--yes", - "--json", - ], - output=output, - ) - assert code == 4 - assert "sdk_migration_bundle_runtime_running" in output.getvalue() - - -def test_migration_bundle_previews_before_writing(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), - ) - output = io.StringIO() - - code = cli.run( - [ - "migration-bundle", - "--bundle-id", - "migration-1", - "--runtime-build", - "build-13", - "--json", - ], - output=output, - ) - assert code == 4 - assert '"state": "preview"' in output.getvalue() or '"state":"preview"' in output.getvalue() - - -def test_migration_bundle_success_reports_bundle_identity(tmp_path, monkeypatch) -> None: +def _stopped(monkeypatch, tmp_path) -> None: monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) monkeypatch.setattr( cli, @@ -172,221 +122,8 @@ def test_migration_bundle_success_reports_bundle_identity(tmp_path, monkeypatch) lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), ) - class Result: - bundle_id = "migration-1" - bundle_path = tmp_path / "backups" / "migration-1.sdk-bundle" - file_count = 4 - total_bytes = 123 - monkeypatch.setattr( - "one_memory.sdk_migration_bundle.create_sdk_migration_bundle", - lambda **_kwargs: Result(), - ) - output = io.StringIO() - code = cli.run( - [ - "migration-bundle", - "--bundle-id", - "migration-1", - "--runtime-build", - "build-13", - "--yes", - "--json", - ], - output=output, - ) - assert code == 0 - document = output.getvalue().replace(" ", "") - assert '"state":"created"' in document - assert '"bundle_id":"migration-1"' in document - - -def test_migration_bundle_reuses_an_existing_verified_bundle(tmp_path, monkeypatch) -> None: - from one_memory.sdk_migration_bundle import MigrationBundleRejected - - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), - ) - - def exists(**_kwargs): - raise MigrationBundleRejected("sdk_migration_bundle_exists") - - monkeypatch.setattr( - "one_memory.sdk_migration_bundle.create_sdk_migration_bundle", - exists, - ) - monkeypatch.setattr( - "one_memory.sdk_migration_bundle.inspect_sdk_migration_bundle", - lambda **_kwargs: {"file_count": 4, "total_bytes": 123}, - ) - output = io.StringIO() - - code = cli.run( - [ - "migration-bundle", - "--bundle-id", - "migration-1", - "--runtime-build", - "build-13", - "--yes", - "--json", - ], - output=output, - ) - - assert code == 0 - assert '"state": "reused"' in output.getvalue() - - -def test_sdk_migrate_preview_never_commits_cutover(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), - ) - output = io.StringIO() - code = cli.run( - [ - "sdk-migrate", - "--migration-id", - "migration-1", - "--bundle-id", - "bundle-1", - "--json", - ], - output=output, - ) - assert code == 4 - document = output.getvalue().replace(" ", "") - assert '"state":"preview"' in document - assert '"commits_cutover":false' in document - - -def test_sdk_migrate_prepares_without_a_model_or_cutover(tmp_path, monkeypatch) -> None: - """No model profile, no SDK client: the SDK already holds every memory.""" - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), - ) - monkeypatch.setattr( - "one_memory.sdk_migration_bundle.inspect_sdk_migration_bundle", - lambda **_kwargs: {"runtime_build": "v1.0.4"}, - ) - output = io.StringIO() - code = cli.run( - [ - "sdk-migrate", - "--migration-id", - "migration-1", - "--bundle-id", - "bundle-1", - "--yes", - "--json", - ], - output=output, - ) - assert code == 0, output.getvalue() - document = json.loads(output.getvalue()) - assert document["state"] == "ready" - assert document["commits_cutover"] is False - assert document["abandoned"] == {} - - -def test_sdk_cutover_is_offline_confirmed_and_returns_marker(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), - ) - monkeypatch.setattr( - "one_memory.sdk_migration_runner.commit_sdk_cutover", - lambda **_kwargs: SimpleNamespace( - migration_id="migration-1", - backup_bundle_id="bundle-1", - sdk_version="2.0.5", - completed_at="2026-10-21T00:00:00Z", - ), - ) - monkeypatch.setattr( - "one_memory.sdk_migration_runner.resume_legacy_cleanup", - lambda **_kwargs: (), - ) - output = io.StringIO() - - code = cli.run( - [ - "sdk-cutover", - "--migration-id", - "migration-1", - "--bundle-id", - "bundle-1", - "--yes", - "--json", - ], - output=output, - ) - - assert code == 0 - document = output.getvalue().replace(" ", "") - assert '"state":"cutover"' in document - assert '"sdk_version":"2.0.5"' in document - - -def test_sdk_cutover_never_runs_while_runtime_is_live(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - monkeypatch.setattr( - cli, - "inspect_runtime", - lambda _data_dir: RuntimeStatus("running", "healthy"), - ) - output = io.StringIO() - - code = cli.run( - [ - "sdk-cutover", - "--migration-id", - "migration-1", - "--bundle-id", - "bundle-1", - "--yes", - "--json", - ], - output=output, - ) - - assert code == 4 - assert '"code": "sdk_cutover_runtime_running"' in output.getvalue() - - -def test_sdk_migration_status_distinguishes_empty_and_legacy_data(tmp_path, monkeypatch) -> None: - monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) - empty = io.StringIO() - assert cli.run(["sdk-migration-status", "--json"], output=empty) == 0 - assert '"state": "empty"' in empty.getvalue() - - database = tmp_path / "one-memory.sqlite3" - with sqlite3.connect(database) as connection: - connection.execute("CREATE TABLE evidence (evidence_id TEXT PRIMARY KEY)") - connection.execute("INSERT INTO evidence VALUES ('e1')") - required = io.StringIO() - assert cli.run(["sdk-migration-status", "--json"], output=required) == 0 - assert '"state": "required"' in required.getvalue() - assert '"raw": 1' in required.getvalue() - - -def test_sdk_migration_status_exposes_a_resumable_active_identity(tmp_path, monkeypatch) -> None: - monkeypatch.setattr( - cli, - "load_settings", - lambda: Settings(_env_file=None, data_dir=tmp_path), - ) +def _legacy_home(tmp_path) -> None: database = tmp_path / "one-memory.sqlite3" asyncio.run(migrate(SqliteMigrationStore(database), MIGRATIONS)) with sqlite3.connect(database) as connection: @@ -398,22 +135,82 @@ def test_sdk_migration_status_exposes_a_resumable_active_identity(tmp_path, monk ") VALUES ('e1','k','local','cursor','s','raw_conversation'," "'user','person','hello','hash',1,'now','now')" ) - SdkDataPlaneMigrationStore(database).start( - migration_id="sdk-authority-v2-0123456789abcdef", - source_runtime_build="17", - target_sdk_version="2.0.5", - source_schema_version=MIGRATIONS[-1].version, - namespace_fingerprint="namespace", - backend_fingerprint="backend", - backup_bundle_id=( - "sdk-authority-v2-0123456789abcdef-pre-cutover" - ), + + +def test_sdk_migrate_never_runs_while_runtime_is_live(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) + monkeypatch.setattr( + cli, + "inspect_runtime", + lambda _data_dir: RuntimeStatus("running", "healthy"), + ) + output = io.StringIO() + code = cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output) + assert code == 4 + assert json.loads(output.getvalue())["code"] == "sdk_migration_runtime_running" + + +def test_sdk_migrate_previews_without_confirmation(tmp_path, monkeypatch) -> None: + _stopped(monkeypatch, tmp_path) + _legacy_home(tmp_path) + output = io.StringIO() + assert cli.run(["sdk-migrate", "--runtime-build", "20", "--json"], output=output) == 4 + assert json.loads(output.getvalue())["state"] == "preview" + assert SdkDataPlaneMigrationStore(tmp_path / "one-memory.sqlite3").authority().value == ( + "legacy" ) + +def test_sdk_migrate_requires_the_runtime_build(tmp_path, monkeypatch) -> None: + _stopped(monkeypatch, tmp_path) + assert cli.run(["sdk-migrate", "--yes"], output=io.StringIO()) == 2 + + +def test_sdk_migrate_reports_each_step_then_the_result(tmp_path, monkeypatch) -> None: + _stopped(monkeypatch, tmp_path) + _legacy_home(tmp_path) output = io.StringIO() - assert cli.run(["sdk-migration-status", "--json"], output=output) == 0 - document = json.loads(output.getvalue()) - assert document["migration_id"] == "sdk-authority-v2-0123456789abcdef" - assert document["bundle_id"] == ( - "sdk-authority-v2-0123456789abcdef-pre-cutover" - ) + + code = cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output) + + assert code == 0, output.getvalue() + lines = [json.loads(line) for line in output.getvalue().splitlines()] + assert [line["step"] for line in lines[:-1]] == ["bundle", "prepare", "cutover", "cleanup"] + assert lines[-1]["state"] == "cutover" + assert lines[-1]["bundle_id"] == f"{lines[-1]['migration_id']}-pre-cutover" + assert not (tmp_path / ".sdk-migration-capture-fence.json").exists() + + again = io.StringIO() + assert cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=again) == 0 + assert json.loads(again.getvalue())["state"] == "not_required" + + +def test_sdk_migrate_failure_is_one_named_code(tmp_path, monkeypatch) -> None: + import errno + + _stopped(monkeypatch, tmp_path) + + def full_disk(**_kwargs): + raise OSError(errno.ENOSPC, "No space left on device") + + monkeypatch.setattr("one_memory.sdk_migration_runner.run_startup_migration", full_disk) + output = io.StringIO() + assert cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output) == 1 + assert json.loads(output.getvalue()) == { + "schema_version": 1, + "state": "failed", + "code": "sdk_migration_disk_space_insufficient", + } + + +def test_sdk_migration_status_says_whether_migrate_has_work(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path)) + + def status() -> dict[str, object]: + output = io.StringIO() + assert cli.run(["sdk-migration-status", "--json"], output=output) == 0 + return json.loads(output.getvalue()) + + assert status()["state"] == "not_required" + _legacy_home(tmp_path) + assert status() == {"schema_version": 1, "authority": "legacy", "state": "required"} diff --git a/memhub/apps/core/tests/test_sdk_data_plane_migration.py b/memhub/apps/core/tests/test_sdk_data_plane_migration.py index a864637..31e0196 100644 --- a/memhub/apps/core/tests/test_sdk_data_plane_migration.py +++ b/memhub/apps/core/tests/test_sdk_data_plane_migration.py @@ -33,6 +33,7 @@ from one_memory.sdk_migration_runner import ( prepare_legacy_write_queue_for_cutover, recover_abandoned_staging, resume_legacy_cleanup, + run_startup_migration, ) from one_memory.target_schema import legacy_tables_present @@ -246,6 +247,42 @@ def test_unstarted_legacy_task_moves_to_surviving_write_queue(store): ).fetchone() == ("cancelled", "migrated_to_memory_write_job") +def test_finished_jobs_without_input_do_not_stop_the_migration(store): + """Seen on a real home: completed jobs whose payload the queue released.""" + from one_memory.memory_server.durable_queue import REAPED_PAYLOAD_MARKER + + queue = DurableWriteQueue(store.database_path) + + def job(key: str) -> str: + return queue.submit( + WriteRequest( + internal_session_id="session-1", + source_id="cursor", + batch_identity=BatchIdentity( + internal_session_id="session-1", first_round=1, last_round=1 + ), + evidence_ids=["gone"], + idempotency_key=key, + ) + ).job_id + + reaped, finished = job("reaped"), job("finished") + with main_database(store.database_path).write() as database, database: + database.execute( + "UPDATE memory_write_jobs SET state='completed',payload_json=? WHERE job_id=?", + (json.dumps({REAPED_PAYLOAD_MARKER: "digest"}), reaped), + ) + database.execute( + "UPDATE memory_write_jobs SET state='completed' WHERE job_id=?", (finished,) + ) + + assert prepare_legacy_write_queue_for_cutover(store.database_path) == (0, 0) + + job("pending-without-input") + with pytest.raises(DataPlaneMigrationError, match="write_job_input_missing"): + prepare_legacy_write_queue_for_cutover(store.database_path) + + def test_pending_dispatch_becomes_one_write_job_per_conversation(store): _insert_evidence(store, "a1", conversation="a", seq=1) _insert_evidence(store, "a2", conversation="a", role="assistant", seq=2) @@ -356,6 +393,25 @@ def test_stale_protective_root_after_authority_is_discarded(tmp_path, store): assert not _protective(home).exists() +def test_host_capture_is_fenced_for_the_whole_migration(store): + from one_memory.live_capture import sdk_migration_capture_fenced + + home = store.database_path.parent + fenced: dict[str, bool] = {} + + result = run_startup_migration( + data_dir=home, + runtime_build="20", + sdk_version="2.0.5", + namespace="governed-v2", + progress=lambda step: fenced.__setitem__(step, sdk_migration_capture_fenced(home)), + ) + + assert result.state == "cutover" + assert fenced == {"bundle": True, "prepare": True, "cutover": True, "cleanup": True} + assert not sdk_migration_capture_fenced(home) + + def test_a_home_blocked_by_build_19_finishes_on_the_next_startup(tmp_path, monkeypatch, store): """The quarantined ledger, its staging copy and pending queue work all resolve.""" home = store.database_path.parent @@ -404,21 +460,14 @@ def test_a_home_blocked_by_build_19_finishes_on_the_next_startup(tmp_path, monke "inspect_runtime", lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"), ) - migration = "sdk-authority-v2-73b3561be69371e0" - bundle = f"{migration}-pre-cutover" - - def run(*arguments): - output = io.StringIO() - code = cli.run([*arguments, "--yes", "--json"], output=output) - return code, json.loads(output.getvalue()) - - assert run("migration-bundle", "--bundle-id", bundle, "--runtime-build", "19")[0] == 0 - code, prepared = run("sdk-migrate", "--migration-id", migration, "--bundle-id", bundle) - assert (code, prepared["state"]) == (0, "ready"), prepared - assert prepared["frozen_jobs"] == 1 - code, cutover = run("sdk-cutover", "--migration-id", migration, "--bundle-id", bundle) - assert (code, cutover["state"]) == (0, "cutover"), cutover + output = io.StringIO() + code = cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output) + result = json.loads(output.getvalue().splitlines()[-1]) + assert (code, result["state"]) == (0, "cutover"), output.getvalue() + assert result["migration_id"] == "sdk-authority-v2-73b3561be69371e0" + assert result["bundle_id"] == "sdk-authority-v2-73b3561be69371e0-pre-cutover" + assert result["frozen_jobs"] == 1 assert store.authority() is MemoryAuthority.SDK assert legacy_tables_present(store.database_path) == () with main_database(store.database_path).read() as database: diff --git a/memhub/apps/desktop/src/startup-sdk-migration.test.ts b/memhub/apps/desktop/src/startup-sdk-migration.test.ts index 8219e96..058a885 100644 --- a/memhub/apps/desktop/src/startup-sdk-migration.test.ts +++ b/memhub/apps/desktop/src/startup-sdk-migration.test.ts @@ -1,242 +1,17 @@ -import { access, mkdtemp, readFile } from "node:fs/promises"; -import os from "node:os"; -import path from "node:path"; -import { expect, it, vi } from "vitest"; -import { - requireCompletedStartupSdkMigration, - runStartupSdkMigration, - withSdkMigrationCaptureFence, -} from "./startup-sdk-migration"; +import { expect, it } from "vitest"; +import { requireCompletedStartupSdkMigration } from "./startup-sdk-migration"; -const core = { command: "/core/one-memory", args: [] }; -const captureFence = async ( - _dataDir: string, - action: () => Promise, -): Promise => action(); - -it("fences hook capture only while startup migration is running", async () => { - const dataDir = await mkdtemp( - path.join(os.tmpdir(), "memhub-sdk-migration-"), - ); - const marker = path.join(dataDir, ".sdk-migration-capture-fence.json"); - - await withSdkMigrationCaptureFence(dataDir, async () => { - const document = JSON.parse(await readFile(marker, "utf8")) as { - pid: number; - }; - expect(document.pid).toBe(process.pid); - }); - - await expect(access(marker)).rejects.toThrow(); -}); - -it("runs bundle backfill and cutover inside the startup maintenance fence", async () => { - const calls: string[][] = []; - const ordering: string[] = []; - const execute = vi.fn(async (_core, args: string[]) => { - calls.push(args); - const command = args[0]; - return { - exitCode: 0, - document: - command === "sdk-migration-status" - ? { - authority: "legacy", - state: "required", - generation: "0123456789abcdef", - } - : command === "sdk-migrate" - ? { state: "ready" } - : command === "sdk-cutover" - ? { state: "cutover" } - : { state: "created" }, - }; - }); - const fenceCalls: string[] = []; - const fence = async ( - dataDir: string, - action: () => Promise, - ): Promise => { - fenceCalls.push(dataDir); - ordering.push("runtime-fence:start"); - const result = await action(); - ordering.push("runtime-fence:end"); - return result; - }; - const orderedCaptureFence = async ( - _dataDir: string, - action: () => Promise, - ): Promise => { - ordering.push("capture-fence:start"); - const result = await action(); - ordering.push("capture-fence:end"); - return result; - }; - - const result = await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-13", - execute, - fence, - captureFence: orderedCaptureFence, - }); - - expect(result).toEqual({ state: "cutover" }); - expect(fenceCalls).toEqual(["/data"]); - expect(ordering).toEqual([ - "capture-fence:start", - "runtime-fence:start", - "runtime-fence:end", - "capture-fence:end", - ]); - expect(calls.map((args) => args[0])).toEqual([ - "sdk-migration-status", - "migration-bundle", - "sdk-migrate", - "sdk-cutover", - ]); -}); - -it("does not stop Runtime when authority is already SDK or the home is empty", async () => { - for (const document of [ - { authority: "sdk", state: "complete" }, - { authority: "legacy", state: "empty" }, - ]) { - let fenced = false; - const fence = async ( - _dataDir: string, - action: () => Promise, - ): Promise => { - fenced = true; - return action(); - }; - const result = await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-13", - execute: vi.fn(async () => ({ exitCode: 0, document })), - fence, - captureFence, - }); - expect(result).toEqual({ state: "not_required" }); - expect(fenced).toBe(false); - } -}); - -it("reports an unreadable migration status as deferred", async () => { - const result = await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-13", - execute: vi.fn(async () => ({ exitCode: 1, document: {} })), - fence: async (_dataDir, action) => action(), - captureFence, - }); - - expect(result).toEqual({ - state: "deferred", - code: "sdk_migration_status_unavailable", - }); - expect(() => requireCompletedStartupSdkMigration(result)).toThrow( - "sdk_migration_status_unavailable", - ); -}); - -it("resumes legacy table cleanup under the maintenance fence", async () => { - const calls: string[] = []; - const result = await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-13", - execute: vi.fn(async (_core, args: string[]) => { - calls.push(args[0]); - return args[0] === "sdk-migration-status" - ? { - exitCode: 0, - document: { - authority: "sdk", - state: "cleanup_required", - }, - } - : { exitCode: 0, document: { state: "clean" } }; +it("refuses to open the app on a deferred migration, naming its code", () => { + expect(() => + requireCompletedStartupSdkMigration({ + state: "deferred", + code: "sdk_migration_disk_space_insufficient", }), - fence: async (_dataDir, action) => action(), - captureFence, - }); - - expect(result).toEqual({ state: "cutover" }); - expect(calls).toEqual(["sdk-migration-status", "sdk-cleanup"]); -}); - -it("keeps legacy authority when backfill is blocked", async () => { - const execute = vi.fn(async (_core, args: string[]) => { - const command = args[0]; - return { - exitCode: command === "sdk-migrate" ? 4 : 0, - document: - command === "sdk-migration-status" - ? { - authority: "legacy", - state: "required", - generation: "0123456789abcdef", - } - : command === "sdk-migrate" - ? { state: "blocked", code: "quarantine" } - : { state: "created" }, - }; - }); - - const result = await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-13", - execute, - fence: async (_dataDir, action) => action(), - captureFence, - }); - - expect(result).toEqual({ state: "deferred", code: "quarantine" }); - expect(execute.mock.calls.map((call) => call[1][0])).not.toContain( - "sdk-cutover", - ); -}); - -it("reuses one v2 migration identity after a blocked startup", async () => { - const calls: string[][] = []; - const migrationId = "sdk-authority-v2-0123456789abcdef"; - const bundleId = `${migrationId}-pre-cutover`; - const execute = vi.fn(async (_core, args: string[]) => { - calls.push(args); - return args[0] === "sdk-migration-status" - ? { - exitCode: 0, - document: { - authority: "legacy", - state: "required", - generation: "ffffffffffffffff", - migration_id: migrationId, - bundle_id: bundleId, - }, - } - : args[0] === "sdk-migrate" - ? { exitCode: 4, document: { state: "blocked", code: "quarantine" } } - : { exitCode: 0, document: { state: "reused" } }; - }); - - await runStartupSdkMigration({ - dataDir: "/data", - core, - runtimeBuild: "build-17", - execute, - fence: async (_dataDir, action) => action(), - captureFence, - }); - - expect(calls.find((args) => args[0] === "migration-bundle")).toContain( - bundleId, - ); - expect(calls.find((args) => args[0] === "sdk-migrate")).toContain( - migrationId, - ); + ).toThrow("sdk_migration_disk_space_insufficient"); + expect(() => + requireCompletedStartupSdkMigration({ state: "deferred" }), + ).toThrow("sdk_migration_incomplete"); + expect(() => + requireCompletedStartupSdkMigration({ state: "cutover" }), + ).not.toThrow(); }); diff --git a/memhub/apps/desktop/src/startup-sdk-migration.ts b/memhub/apps/desktop/src/startup-sdk-migration.ts index e49d249..cce658d 100644 --- a/memhub/apps/desktop/src/startup-sdk-migration.ts +++ b/memhub/apps/desktop/src/startup-sdk-migration.ts @@ -1,16 +1,10 @@ -import { spawn } from "node:child_process"; -import { mkdir, rename, rm, writeFile } from "node:fs/promises"; -import path from "node:path"; -import { withRuntimeMaintenanceFence } from "@one-memory/runtime-control"; +import { + runSdkMigration, + type SdkMigrationResult, +} from "@one-memory/runtime-control"; -const MIGRATION_PREFIX = "sdk-authority-v2"; -const MAX_OUTPUT_BYTES = 1024 * 1024; -const CAPTURE_FENCE = ".sdk-migration-capture-fence.json"; - -export interface StartupMigrationResult { - state: "not_required" | "cutover" | "deferred"; - code?: string; -} +export type StartupMigrationResult = SdkMigrationResult; +export const runStartupSdkMigration = runSdkMigration; export function requireCompletedStartupSdkMigration( result: StartupMigrationResult, @@ -19,215 +13,3 @@ export function requireCompletedStartupSdkMigration( throw new Error(result.code ?? "sdk_migration_incomplete"); } } - -interface CommandResult { - exitCode: number; - document: Record; -} - -interface CoreCommand { - command: string; - args?: string[]; - cwd?: string; - env?: NodeJS.ProcessEnv; -} - -interface StartupMigrationOptions { - dataDir: string; - core: CoreCommand; - runtimeBuild: string; - execute?: ( - launch: CoreCommand, - args: string[], - dataDir: string, - ) => Promise; - fence?: (dataDir: string, action: () => Promise) => Promise; - captureFence?: (dataDir: string, action: () => Promise) => Promise; -} - -export async function runStartupSdkMigration( - options: StartupMigrationOptions, -): Promise { - const execute = options.execute ?? executeCoreCommand; - const fence = options.fence ?? withRuntimeMaintenanceFence; - const captureFence = options.captureFence ?? withSdkMigrationCaptureFence; - const status = await execute( - options.core, - ["sdk-migration-status", "--json"], - options.dataDir, - ); - if (status.exitCode !== 0) { - return { state: "deferred", code: "sdk_migration_status_unavailable" }; - } - if ( - status.document.authority === "sdk" && - status.document.state === "cleanup_required" - ) { - return captureFence(options.dataDir, () => - fence(options.dataDir, async () => { - const cleanup = await execute( - options.core, - ["sdk-cleanup", "--yes", "--json"], - options.dataDir, - ); - return cleanup.exitCode === 0 && cleanup.document.state === "clean" - ? { state: "cutover" } - : deferred(cleanup); - }), - ); - } - if ( - status.document.authority === "sdk" || - status.document.state !== "required" - ) { - return { state: "not_required" }; - } - const generation = status.document.generation; - if (typeof generation !== "string" || !/^[a-f0-9]{16}$/.test(generation)) { - return { state: "deferred", code: "sdk_migration_generation_invalid" }; - } - const previousMigrationId = status.document.migration_id; - const previousBundleId = status.document.bundle_id; - const canResume = - typeof previousMigrationId === "string" && - previousMigrationId.startsWith(`${MIGRATION_PREFIX}-`) && - typeof previousBundleId === "string" && - previousBundleId === `${previousMigrationId}-pre-cutover`; - const migrationId = canResume - ? previousMigrationId - : `${MIGRATION_PREFIX}-${generation}`; - const bundleId = canResume ? previousBundleId : `${migrationId}-pre-cutover`; - return captureFence(options.dataDir, () => - fence(options.dataDir, async () => { - const bundle = await execute( - options.core, - [ - "migration-bundle", - "--bundle-id", - bundleId, - "--runtime-build", - options.runtimeBuild, - "--yes", - "--json", - ], - options.dataDir, - ); - if (bundle.exitCode !== 0) return deferred(bundle); - const backfill = await execute( - options.core, - [ - "sdk-migrate", - "--migration-id", - migrationId, - "--bundle-id", - bundleId, - "--yes", - "--json", - ], - options.dataDir, - ); - if (backfill.exitCode !== 0 || backfill.document.state !== "ready") { - return deferred(backfill); - } - const cutover = await execute( - options.core, - [ - "sdk-cutover", - "--migration-id", - migrationId, - "--bundle-id", - bundleId, - "--yes", - "--json", - ], - options.dataDir, - ); - if (cutover.exitCode !== 0 || cutover.document.state !== "cutover") { - return deferred(cutover); - } - return { state: "cutover" }; - }), - ); -} - -export async function withSdkMigrationCaptureFence( - dataDir: string, - action: () => Promise, -): Promise { - await mkdir(dataDir, { recursive: true, mode: 0o700 }); - const marker = path.join(dataDir, CAPTURE_FENCE); - const temporary = `${marker}.${process.pid}.tmp`; - await writeFile( - temporary, - `${JSON.stringify({ schema_version: 1, pid: process.pid })}\n`, - { encoding: "utf8", mode: 0o600 }, - ); - await rename(temporary, marker); - try { - return await action(); - } finally { - await rm(marker, { force: true }); - await rm(temporary, { force: true }); - } -} - -function deferred(result: CommandResult): StartupMigrationResult { - return { - state: "deferred", - code: - typeof result.document.code === "string" - ? result.document.code - : "sdk_migration_incomplete", - }; -} - -async function executeCoreCommand( - launch: CoreCommand, - args: string[], - dataDir: string, -): Promise { - return new Promise((resolve, reject) => { - const child = spawn(launch.command, [...(launch.args ?? []), ...args], { - cwd: launch.cwd, - env: { ...process.env, ...launch.env, ONE_MEMORY_DATA_DIR: dataDir }, - stdio: ["ignore", "pipe", "pipe"], - }); - let stdout = ""; - let stderrBytes = 0; - child.stdout.setEncoding("utf8"); - child.stdout.on("data", (chunk: string) => { - stdout += chunk; - if (Buffer.byteLength(stdout) > MAX_OUTPUT_BYTES) { - child.kill("SIGKILL"); - reject(new Error("sdk_migration_output_too_large")); - } - }); - child.stderr.on("data", (chunk: Buffer) => { - stderrBytes += chunk.length; - if (stderrBytes > MAX_OUTPUT_BYTES) { - child.kill("SIGKILL"); - reject(new Error("sdk_migration_error_output_too_large")); - } - }); - child.once("error", reject); - child.once("close", (code) => { - try { - const line = stdout.trim().split("\n").at(-1) ?? "{}"; - const document = JSON.parse(line) as unknown; - if ( - !document || - typeof document !== "object" || - Array.isArray(document) - ) { - throw new Error("sdk_migration_response_invalid"); - } - resolve({ - exitCode: code ?? 1, - document: document as Record, - }); - } catch { - reject(new Error("sdk_migration_response_invalid")); - } - }); - }); -} diff --git a/memhub/docs/ARCHITECTURE.md b/memhub/docs/ARCHITECTURE.md index 3cc5da8..a8d03e8 100644 --- a/memhub/docs/ARCHITECTURE.md +++ b/memhub/docs/ARCHITECTURE.md @@ -135,7 +135,7 @@ This exception does not extend to VDB, Coding, prompt, reconcile, or lifecycle. - Context wire：cutover marker 提交前只使用 legacy Context v1；提交后 Host Bridge 只使用 SDK-native Context v2。两种 schema 原样校验、原样返回，不把 SDK item 补造成 v1 的 `revision`、`evidence_ids`、`paths` 或 `adoption`，已选择的 v2 路径失败时也不回退 legacy 权威。 - Ask wire：marker 前保留 Governance Ask v1；marker 后 Web / Host Bridge 只调用 SDK Ask v2。v2 的引用只含 SDK 原生 `memory_id/content/layer/status/source/gmt_created`，生成过程中重复 `get` 校验同一 SDK 快照，不补造 revision、Evidence 或 scopes，SDK 失败不回退 v1。 - Matter：未形成已发布产品合同，不迁移占位数据；所有 `matter*` 表在升级清理时物理删除，fresh install 不创建，Runtime 不注册 Matter API。 -- 旧用户迁移：由 App 在可见的启动 loading 阶段统一执行；此时 Core 尚未对外提供产品读写。所有已发布版本都先经 SDK 写入，Governance/Evidence 只是镜像，所以迁移**不复制、不比对任何 memory 或 raw**：只创建并校验完整 bundle、让写入队列脱离旧表、提交 marker、删除旧表。普通用户不需要另跑 CLI；App 不得在 migration `deferred` 状态继续启动 Runtime。 +- 旧用户迁移：由 App 在可见的启动 loading 阶段统一执行；此时 Core 尚未对外提供产品读写。所有已发布版本都先经 SDK 写入，Governance/Evidence 只是镜像，所以迁移**不复制、不比对任何 memory 或 raw**：只创建并校验完整 bundle、让写入队列脱离旧表、提交 marker、删除旧表。由一个 Core 命令 `sdk-migrate` 完成；Desktop 启动与 CLI `memhub install`/`upgrade`（macOS、Linux）在启动新 Runtime 前调用它，普通用户不需要另跑命令。App 不得在 migration `deferred` 状态继续启动 Runtime。 - 迁移期间宿主 capture hook 由 Desktop 活跃 PID marker 静默挡住，不再追加 Evidence；cutover 后先从宿主 History checkpoint 补齐遗漏 turn，再恢复 live sync。 - Matter 若未来重新开发，需以新的 SDK-native 合同重新设计，不继承当前占位 schema。 diff --git a/memhub/docs/INSTALLATION.md b/memhub/docs/INSTALLATION.md index a9e52d3..db54844 100644 --- a/memhub/docs/INSTALLATION.md +++ b/memhub/docs/INSTALLATION.md @@ -100,7 +100,7 @@ Downloader 使用标准 `HTTPS_PROXY`/`https_proxy` 与 `NO_PROXY`/`no_proxy` - `npm uninstall -g memhub-cli`：只删除 npm CLI；不停止或删除受管 Runtime，不删除 Memory。 完整 fake Release 验收会在 Runtime 运行中执行真实 npm uninstall，并再次验证 systemd、`current`、受管 `memhub`、doctor 和 SQLite 均仍可用。 -- `memhub upgrade`：下载并校验新 Runtime，健康检查失败时恢复旧版本。 +- `memhub upgrade`：下载并校验新 Runtime；旧版本写过的数据会先迁移到 SDK 数据面（先做完整备份，终端显示进度），再启动新 Runtime。迁移失败时恢复旧版本；迁移完成后旧版本无法打开数据，健康检查失败时保留新版本并提示 `memhub restart` / `memhub doctor`。 - `memhub uninstall`：停止并移除受管服务、Runtime versions/current 和受管 CLI 链接； 默认保留数据。 - `memhub delete-data`：唯一删除 Memory 的命令；Runtime 必须已停止，并要求 `--yes` 与 diff --git a/memhub/docs/contracts/sdk-data-plane-migration-v1.md b/memhub/docs/contracts/sdk-data-plane-migration-v1.md index 39e3923..445345f 100644 --- a/memhub/docs/contracts/sdk-data-plane-migration-v1.md +++ b/memhub/docs/contracts/sdk-data-plane-migration-v1.md @@ -32,8 +32,9 @@ and no raw turn, and does not compare the mirror with the SDK: ## 2. Authority before and after cutover Before the cutover marker is committed, legacy tables remain the product read -authority and Core may still run on them (CLI-managed Runtime). The App does -not start Runtime while migration is incomplete. +authority; a Runtime that was never upgraded through Desktop or `memhub +install`/`upgrade` still serves them. Desktop does not start Runtime while +migration is incomplete. After the cutover marker is committed: @@ -49,27 +50,44 @@ the authority. A fresh home bootstraps the latest management schema with SDK authority directly and never creates legacy tables. -## 3. Startup steps +## 3. Who runs it, and how -The packaged App runs these Core commands while the startup placeholder is -visible, under the Runtime maintenance fence and the capture fence (host -capture hooks return silently instead of writing Evidence): +One Core command does the whole migration: -1. `migration-bundle`: copy `one-memory.sqlite3`, the SDK data plane and the - non-secret configuration into a versioned bundle, then verify it by an - isolated restore. An existing verified bundle for the same migration is - reused. -2. `sdk-migrate`: apply the management schema migrations, discard SDK staging - left by builds through 19 (see §5), and make legacy queue work - self-contained (see §4). It needs no model profile and no SDK client. The run - is recorded as `phase=prepared, state=ready`. -3. `sdk-cutover`: re-verify the bundle, repeat §4 (a CLI-managed Runtime may - have run since step 2), commit SDK authority, and drop the legacy tables. +```text +one-memory sdk-migration-status --json # read-only: is there work? +one-memory sdk-migrate --runtime-build --yes --json +``` -Each step is safe to repeat. A retry resumes the same `sdk-authority-v2-*` -migration ID and bundle, including a run that build 19 left in -`review_required/blocked`; earlier per-item ledger rows are ignored and dropped -with the legacy tables. +`sdk-migrate` requires a stopped Runtime. It takes the host capture fence +(hooks return silently instead of writing Evidence while its PID is alive), +then: + +1. **bundle**: copies `one-memory.sqlite3`, the SDK data plane and non-secret + configuration into a versioned bundle and verifies it by an isolated + restore; an existing verified bundle of the same run is reused; +2. **prepare**: applies the management schema migrations, discards SDK staging + left by builds through 19 (§5) and makes legacy queue work self-contained + (§4); it needs no model profile and no SDK client; +3. **cutover**: re-verifies the bundle, repeats §4, commits SDK authority; +4. **cleanup**: drops the legacy tables (including the `evidence_search` index). + +With `--json` each step prints `{"state":"running","step":…}` and the last line +is the result: `cutover`, `complete` (only cleanup was left), `not_required`, or +`failed` with a stable `code`. Every step is safe to repeat; a retry resumes the +newest unfinished `sdk-authority-v2-*` run and its bundle, including a run that +build 19 left in `review_required/blocked`. + +Callers only keep the Runtime stopped, and its service from restarting, around +that command (`runSdkMigration` in `packages/runtime-control`): + +- **Desktop** runs it in the startup loading phase before attaching the UI; a + failure keeps the App closed with the code's sentence. +- **CLI `memhub install` / `memhub upgrade`** (macOS and Linux) run it with the + new Runtime's Core after switching `current` and before starting the service. + A failure before cutover rolls back to the previous version. After cutover an + older Runtime refuses the home, so a later readiness failure keeps the new + version and reports `runtime_not_ready_after_sdk_migration`. After cutover, History catch-up scans forward from each source checkpoint and recovers turns skipped while capture was fenced. @@ -127,6 +145,9 @@ The release gate covers: - a legacy home reaching cutover with its SDK stores byte-for-byte unchanged; - a home left blocked by build 19 (quarantined ledger and staging copy) completing on the next startup; +- CLI upgrade migrating before restart, rolling back on a failed migration, + and keeping the new version when a migrated home is slow to come up; +- host capture fenced for the whole command; - pending job freezing, legacy task and pending dispatch conversion; - invoked/unknown legacy tasks and pending deletions not blocking cutover; - interrupted directory swap recovery before and after authority; diff --git a/memhub/packages/runtime-control/src/index.ts b/memhub/packages/runtime-control/src/index.ts index cd1331d..8127a18 100644 --- a/memhub/packages/runtime-control/src/index.ts +++ b/memhub/packages/runtime-control/src/index.ts @@ -174,6 +174,13 @@ export { RuntimeSupervisor, type RuntimeSupervisorOptions, } from "./runtime-supervisor.js"; +export { + type CoreCommand, + runSdkMigration, + type SdkMigrationCommandResult, + type SdkMigrationOptions, + type SdkMigrationResult, +} from "./sdk-migration.js"; export { installUserService, parseDisabledOverride, diff --git a/memhub/packages/runtime-control/src/sdk-migration.test.ts b/memhub/packages/runtime-control/src/sdk-migration.test.ts new file mode 100644 index 0000000..2ff71fe --- /dev/null +++ b/memhub/packages/runtime-control/src/sdk-migration.test.ts @@ -0,0 +1,203 @@ +import { mkdtemp, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { expect, it, vi } from "vitest"; +import { runSdkMigration } from "./sdk-migration.js"; + +const core = { command: "/core/one-memory", args: [] }; +const noFence = async ( + _dataDir: string, + action: () => Promise, +): Promise => action(); + +function statusThen( + migrate: { exitCode: number; document: Record }, + status: Record = { authority: "legacy", state: "required" }, +) { + return vi.fn(async (_core, args: string[]) => + args[0] === "sdk-migration-status" + ? { exitCode: 0, document: status } + : migrate, + ); +} + +it("runs the one migration command inside the runtime maintenance fence", async () => { + const ordering: string[] = []; + const execute = vi.fn(async (_core, args: string[]) => { + ordering.push(args[0] ?? ""); + return args[0] === "sdk-migration-status" + ? { exitCode: 0, document: { authority: "legacy", state: "required" } } + : { exitCode: 0, document: { state: "cutover" } }; + }); + const fence = async ( + dataDir: string, + action: () => Promise, + ): Promise => { + ordering.push(`fence:start:${dataDir}`); + const result = await action(); + ordering.push("fence:end"); + return result; + }; + + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute, + fence, + }); + + expect(result).toEqual({ state: "cutover" }); + expect(ordering).toEqual([ + "sdk-migration-status", + "fence:start:/data", + "sdk-migrate", + "fence:end", + ]); + expect(execute.mock.calls[1]?.[1]).toEqual([ + "sdk-migrate", + "--runtime-build", + "20", + "--yes", + "--json", + ]); +}); + +it("does not stop Runtime when nothing needs migrating", async () => { + let fenced = false; + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: statusThen( + { exitCode: 0, document: {} }, + { authority: "sdk", state: "not_required" }, + ), + fence: async (_dataDir, action) => { + fenced = true; + return action(); + }, + }); + + expect(result).toEqual({ state: "not_required" }); + expect(fenced).toBe(false); +}); + +it("treats a finished cleanup as a completed migration", async () => { + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: statusThen( + { exitCode: 0, document: { state: "complete" } }, + { authority: "sdk", state: "required" }, + ), + fence: noFence, + }); + + expect(result).toEqual({ state: "cutover" }); +}); + +it("reports an unreadable migration status as deferred", async () => { + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: vi.fn(async () => ({ exitCode: 1, document: {} })), + fence: noFence, + }); + + expect(result).toEqual({ + state: "deferred", + code: "sdk_migration_status_unavailable", + }); +}); + +it("passes the migration failure code to the startup dialog", async () => { + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: statusThen({ + exitCode: 1, + document: { + state: "failed", + code: "sdk_migration_disk_space_insufficient", + }, + }), + fence: noFence, + }); + + expect(result).toEqual({ + state: "deferred", + code: "sdk_migration_disk_space_insufficient", + }); +}); + +it("never reports success for a result it does not recognise", async () => { + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: statusThen({ exitCode: 0, document: { state: "running" } }), + fence: noFence, + }); + + expect(result).toEqual({ + state: "deferred", + code: "sdk_migration_incomplete", + }); +}); + +it("names a Core that could not be run instead of throwing", async () => { + const result = await runSdkMigration({ + dataDir: "/data", + core, + runtimeBuild: "20", + execute: vi.fn(async (_core, args: string[]) => { + if (args[0] === "sdk-migration-status") { + return { + exitCode: 0, + document: { authority: "legacy", state: "required" }, + }; + } + throw new Error("sdk_migration_response_invalid"); + }), + fence: noFence, + }); + + expect(result).toEqual({ + state: "deferred", + code: "sdk_migration_response_invalid", + }); +}); + +it("streams Core's progress lines to the caller", async () => { + const directory = await mkdtemp(path.join(os.tmpdir(), "sdk-migrate-")); + const fake = path.join(directory, "core.sh"); + await writeFile( + fake, + [ + "#!/bin/sh", + 'if [ "$1" = "sdk-migration-status" ]; then', + ' echo \'{"authority":"legacy","state":"required"}\'; exit 0', + "fi", + 'echo \'{"state":"running","step":"bundle"}\'', + 'echo \'{"state":"running","step":"cutover"}\'', + 'echo \'{"state":"cutover"}\'', + ].join("\n"), + { mode: 0o755 }, + ); + const steps: string[] = []; + + const result = await runSdkMigration({ + dataDir: directory, + core: { command: fake }, + runtimeBuild: "20", + fence: noFence, + onProgress: (step) => steps.push(step), + }); + + expect(result).toEqual({ state: "cutover" }); + expect(steps).toEqual(["bundle", "cutover"]); +}); diff --git a/memhub/packages/runtime-control/src/sdk-migration.ts b/memhub/packages/runtime-control/src/sdk-migration.ts new file mode 100644 index 0000000..8efc685 --- /dev/null +++ b/memhub/packages/runtime-control/src/sdk-migration.ts @@ -0,0 +1,174 @@ +import { spawn } from "node:child_process"; +import { withRuntimeMaintenanceFence } from "./runtime-maintenance.js"; + +const MAX_OUTPUT_BYTES = 1024 * 1024; + +export interface SdkMigrationResult { + state: "not_required" | "cutover" | "deferred"; + code?: string; +} + +export interface SdkMigrationCommandResult { + exitCode: number; + document: Record; +} + +export interface CoreCommand { + command: string; + args?: string[]; + cwd?: string; + env?: NodeJS.ProcessEnv; +} + +export interface SdkMigrationOptions { + dataDir: string; + core: CoreCommand; + runtimeBuild: string; + execute?: ( + launch: CoreCommand, + args: string[], + dataDir: string, + onProgress?: (step: string) => void, + ) => Promise; + fence?: (dataDir: string, action: () => Promise) => Promise; + /** Called with `bundle`, `prepare`, `cutover`, `cleanup` as Core reaches them. */ + onProgress?: (step: string) => void; +} + +/** + * Bring a home to SDK memory authority before a Runtime serves it. + * + * Core's `sdk-migrate` does the whole migration and resumes after any failure; + * the caller (Desktop startup, CLI install/upgrade) only keeps the Runtime + * stopped, and its service from restarting, while it runs. + */ +export async function runSdkMigration( + options: SdkMigrationOptions, +): Promise { + const execute = options.execute ?? executeCoreCommand; + const fence = options.fence ?? withRuntimeMaintenanceFence; + const status = await execute( + options.core, + ["sdk-migration-status", "--json"], + options.dataDir, + ).catch(() => null); + if (status === null || status.exitCode !== 0) { + return { state: "deferred", code: "sdk_migration_status_unavailable" }; + } + if (status.document.state !== "required") { + return { state: "not_required" }; + } + return fence(options.dataDir, async () => { + let migrated: SdkMigrationCommandResult; + try { + migrated = await execute( + options.core, + [ + "sdk-migrate", + "--runtime-build", + options.runtimeBuild, + "--yes", + "--json", + ], + options.dataDir, + options.onProgress, + ); + } catch (error) { + const code = error instanceof Error ? error.message : ""; + return { + state: "deferred", + code: /^[a-z][a-z0-9_]{0,95}$/.test(code) + ? code + : "sdk_migration_response_invalid", + }; + } + if (migrated.exitCode !== 0) return deferred(migrated); + const state = migrated.document.state; + if (state === "cutover" || state === "complete") + return { state: "cutover" }; + if (state === "not_required") return { state: "not_required" }; + return deferred(migrated); + }); +} + +function deferred(result: SdkMigrationCommandResult): SdkMigrationResult { + return { + state: "deferred", + code: + typeof result.document.code === "string" + ? result.document.code + : "sdk_migration_incomplete", + }; +} + +async function executeCoreCommand( + launch: CoreCommand, + args: string[], + dataDir: string, + onProgress?: (step: string) => void, +): Promise { + return new Promise((resolve, reject) => { + const child = spawn(launch.command, [...(launch.args ?? []), ...args], { + cwd: launch.cwd, + env: { ...process.env, ...launch.env, ONE_MEMORY_DATA_DIR: dataDir }, + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let reported = 0; + let stderrBytes = 0; + child.stdout.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + stdout += chunk; + const complete = stdout.lastIndexOf("\n"); + if (onProgress && complete > reported) { + for (const line of stdout.slice(reported, complete).split("\n")) { + const step = progressStep(line); + if (step !== null) onProgress(step); + } + reported = complete + 1; + } + if (Buffer.byteLength(stdout) > MAX_OUTPUT_BYTES) { + child.kill("SIGKILL"); + reject(new Error("sdk_migration_output_too_large")); + } + }); + child.stderr.on("data", (chunk: Buffer) => { + stderrBytes += chunk.length; + if (stderrBytes > MAX_OUTPUT_BYTES) { + child.kill("SIGKILL"); + reject(new Error("sdk_migration_error_output_too_large")); + } + }); + child.once("error", reject); + child.once("close", (code) => { + try { + const line = stdout.trim().split("\n").at(-1) ?? "{}"; + const document = JSON.parse(line) as unknown; + if ( + !document || + typeof document !== "object" || + Array.isArray(document) + ) { + throw new Error("sdk_migration_response_invalid"); + } + resolve({ + exitCode: code ?? 1, + document: document as Record, + }); + } catch { + reject(new Error("sdk_migration_response_invalid")); + } + }); + }); +} + +function progressStep(line: string): string | null { + try { + const document = JSON.parse(line) as { state?: unknown; step?: unknown }; + return document.state === "running" && typeof document.step === "string" + ? document.step + : null; + } catch { + return null; + } +}