Merge pull request #168 from Tencent-Hunyuan/feat/single-sdk-migrate-command

feat(memhub): migrate every install with one sdk-migrate command

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({

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<boolean>;
+  /** 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<ReturnType<typeof claimCliServiceOwnership>>
       | 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<ReturnType<typeof claimCliServiceOwnership>>
       | undefined;
+    let migrated = false;
     const rollbackUpgrade = async (
       dataDir: string | undefined,
     ): Promise<void> => {
@@ -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<string, string> = {
+  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<SdkMigrationResult> {
+  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,

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 <managed-backup> --yes [--json]"
-            "|migration-bundle --bundle-id <id> --runtime-build <build> --yes [--json]"
-            "|sdk-migrate --migration-id <id> --bundle-id <id> --yes [--json]"
-            "|sdk-cutover --migration-id <id> --bundle-id <id> --yes [--json]"
+            "|sdk-migrate --runtime-build <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 <id> --runtime-build <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 <id> --bundle-id <id>\n")
+    if not runtime_build:
+        stream.write("one-memory: sdk-migrate 需要 --runtime-build <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 <id> --bundle-id <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}"

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,
+    )

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"
-    )

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"}

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:

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 <T>(
-  _dataDir: string,
-  action: () => Promise<T>,
-): Promise<T> => 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 <T>(
-    dataDir: string,
-    action: () => Promise<T>,
-  ): Promise<T> => {
-    fenceCalls.push(dataDir);
-    ordering.push("runtime-fence:start");
-    const result = await action();
-    ordering.push("runtime-fence:end");
-    return result;
-  };
-  const orderedCaptureFence = async <T>(
-    _dataDir: string,
-    action: () => Promise<T>,
-  ): Promise<T> => {
-    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 <T>(
-      _dataDir: string,
-      action: () => Promise<T>,
-    ): Promise<T> => {
-      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();
 });

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<string, unknown>;
-}
-
-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<CommandResult>;
-  fence?: <T>(dataDir: string, action: () => Promise<T>) => Promise<T>;
-  captureFence?: <T>(dataDir: string, action: () => Promise<T>) => Promise<T>;
-}
-
-export async function runStartupSdkMigration(
-  options: StartupMigrationOptions,
-): Promise<StartupMigrationResult> {
-  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<T>(
-  dataDir: string,
-  action: () => Promise<T>,
-): Promise<T> {
-  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<CommandResult> {
-  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<string, unknown>,
-        });
-      } catch {
-        reject(new Error("sdk_migration_response_invalid"));
-      }
-    });
-  });
-}

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。

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` 与

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 <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;

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,

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 <T>(
+  _dataDir: string,
+  action: () => Promise<T>,
+): Promise<T> => action();
+
+function statusThen(
+  migrate: { exitCode: number; document: Record<string, unknown> },
+  status: Record<string, unknown> = { 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 <T>(
+    dataDir: string,
+    action: () => Promise<T>,
+  ): Promise<T> => {
+    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"]);
+});

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<string, unknown>;
+}
+
+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<SdkMigrationCommandResult>;
+  fence?: <T>(dataDir: string, action: () => Promise<T>) => Promise<T>;
+  /** 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<SdkMigrationResult> {
+  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<SdkMigrationCommandResult> {
+  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<string, unknown>,
+        });
+      } 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;
+  }
+}