refactor(doctor): consolidate legacy source ownership (#117796)

2026-08-13 17:07:40 +00:00 · 2026-08-01 21:26:51 -07:00
parent 0745a7e012
commit ea01fffc0c
10 changed files with 237 additions and 422 deletions
--- a/src/infra/push-apns-store.ts
+++ b/src/infra/push-apns-store.ts
@@ -358,7 +358,7 @@ export function apnsRegistrationFromRow(row: ApnsRegistrationRow): ApnsRegistrat
    normalizePersistedRelayOrigin,
  );
  if (!normalized) {
-    throw new Error(`invalid APNs registration row for node ${row.node_id}`);
+    throw new Error("invalid APNs registration row");
  }
  const canonical = apnsRegistrationToRow(normalized);
  if (
@@ -375,7 +375,7 @@ export function apnsRegistrationFromRow(row: ApnsRegistrationRow): ApnsRegistrat
    canonical.token_debug_suffix !== row.token_debug_suffix ||
    canonical.updated_at_ms !== row.updated_at_ms
  ) {
-    throw new Error(`non-canonical APNs registration row for node ${row.node_id}`);
+    throw new Error("non-canonical APNs registration row");
  }
  return normalized;
 }
--- a/src/infra/state-migrations.apns.test.ts
+++ b/src/infra/state-migrations.apns.test.ts
@@ -387,9 +387,10 @@ describe("legacy APNs Doctor migration", () => {

  it("rolls back inserted rows when a later canonical row is invalid", async () => {
    const stateDir = useStateDir();
+    const privateMarker = "must-not-appear-in-doctor-output";
    const sourcePath = await writeLegacyState(stateDir, {
      "legacy-only": directRegistration({ nodeId: "legacy-only" }),
-      "corrupt-node": directRegistration({ nodeId: "corrupt-node" }),
+      [privateMarker]: directRegistration({ nodeId: privateMarker }),
    });
    openOpenClawStateDatabase()
      .db.prepare(
@@ -397,11 +398,12 @@ describe("legacy APNs Doctor migration", () => {
           node_id, transport, topic, environment, updated_at_ms
         ) VALUES (?, ?, ?, ?, ?)`,
      )
-      .run("corrupt-node", "unknown", "ai.openclaw.ios", "sandbox", 1);
+      .run(privateMarker, "unknown", "ai.openclaw.ios", "sandbox", 1);

    const result = await migrate(stateDir);

    expect(result.warnings[0]).toContain("invalid APNs registration row");
+    expect(result.warnings.join("\n")).not.toContain(privateMarker);
    await expect(loadApnsRegistration("legacy-only", stateDir)).resolves.toBeNull();
    expect(fs.existsSync(sourcePath)).toBe(true);
    expect(fs.existsSync(`${sourcePath}.doctor-importing`)).toBe(false);
--- a/src/infra/state-migrations.apns.ts
+++ b/src/infra/state-migrations.apns.ts
@@ -29,8 +29,8 @@ import {
  type LegacyMigrationReceipt,
 } from "./state-migrations.receipts.js";
 import {
+  LegacyMigrationSourceClaim,
  legacyMigrationSourceOrClaimMayExist,
-  legacyMigrationSourceSnapshotsMatch as snapshotsMatch,
  resolveLegacyMigrationRelativePath,
  type LegacyMigrationSourceSnapshot,
 } from "./state-migrations.source-snapshot.js";
@@ -284,19 +284,6 @@ function importAndRecordReceipt(params: {
  );
 }

-async function removePath(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-  removeSource?: (sourcePath: string) => Promise<void> | void;
-}): Promise<void> {
-  if (params.removeSource) {
-    await params.removeSource(params.sourcePath);
-    return;
-  }
-  await params.stateRoot.remove(relativeLegacyPath(params.stateDir, params.sourcePath));
-}
-
 async function cleanupReceiptAuthoritativeSources(params: {
  stateRoot: Root;
  stateDir: string;
@@ -312,7 +299,11 @@ async function cleanupReceiptAuthoritativeSources(params: {
    }
    // Validate ownership and drain the pinned inode before deleting receipt-retired bytes.
    await readLegacySourceSnapshot(params.stateRoot, params.stateDir, candidate);
-    await removePath({ ...params, sourcePath: candidate });
+    if (params.removeSource) {
+      await params.removeSource(candidate);
+    } else {
+      await params.stateRoot.remove(relativeLegacyPath(params.stateDir, candidate));
+    }
    removed += 1;
  }
  if (!params.receipt.removedSource || removed > 0) {
@@ -321,29 +312,6 @@ async function cleanupReceiptAuthoritativeSources(params: {
  return removed;
 }

-async function restoreClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<string | null> {
-  const claimPath = `${params.sourcePath}${APNS_DOCTOR_CLAIM_SUFFIX}`;
-  try {
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))) {
-      return null;
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
-      return `source path already exists: ${params.sourcePath}`;
-    }
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, claimPath),
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
-    return null;
-  } catch (error) {
-    return String(error);
-  }
-}
-
 async function migrateWithExclusiveStateOwnership(params: {
  stateRoot: Root;
  detected: LegacyStateDetection["apns"];
@@ -380,16 +348,25 @@ async function migrateWithExclusiveStateOwnership(params: {
  }

  const sourcePath = params.detected.sourcePath;
-  const claimPath = `${sourcePath}${APNS_DOCTOR_CLAIM_SUFFIX}`;
-  const hasSource = await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath));
-  const hasClaim = await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath));
+  const source = new LegacyMigrationSourceClaim<LegacySourceSnapshot>({
+    stateRoot: params.stateRoot,
+    stateDir: params.stateDir,
+    sourcePath,
+    label: "APNs",
+    includeFilePath: false,
+    claimSuffix: APNS_DOCTOR_CLAIM_SUFFIX,
+    readSnapshot: (snapshotPath) =>
+      readLegacySourceSnapshot(params.stateRoot, params.stateDir, snapshotPath),
+  });
+  const hasSource = await source.exists();
+  const hasClaim = await source.exists(true);
  if (hasSource && hasClaim) {
    return {
      changes,
      warnings: ["Failed migrating legacy APNs state: source and interrupted claim both exist."],
    };
  }
-  const activePath = hasSource ? sourcePath : hasClaim ? claimPath : null;
+  const activePath = hasSource ? sourcePath : hasClaim ? source.claimPath : null;
  if (!activePath) {
    return { changes, warnings };
  }
@@ -420,22 +397,13 @@ async function migrateWithExclusiveStateOwnership(params: {

  if (activePath === sourcePath) {
    try {
-      params.beforeClaim?.();
-      await params.stateRoot.move(
-        relativeLegacyPath(params.stateDir, sourcePath),
-        relativeLegacyPath(params.stateDir, claimPath),
-      );
-      const claimed = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-      if (!snapshotsMatch(snapshot, claimed)) {
-        throw new Error("legacy APNs source changed before Doctor could claim it");
-      }
-      snapshot = claimed;
-    } catch (error) {
-      const restoreError = await restoreClaim({
-        stateRoot: params.stateRoot,
-        stateDir: params.stateDir,
-        sourcePath,
+      snapshot = await source.claim({
+        snapshot,
+        mismatchMessage: "legacy APNs source changed before Doctor could claim it",
+        beforeClaim: params.beforeClaim,
      });
+    } catch (error) {
+      const restoreError = await source.restore();
      warnings.push(
        `Failed migrating legacy APNs state: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
      );
@@ -452,11 +420,7 @@ async function migrateWithExclusiveStateOwnership(params: {
      registrations,
    });
  } catch (error) {
-    const restoreError = await restoreClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
-    });
+    const restoreError = await source.restore();
    warnings.push(
      `Failed migrating legacy APNs state: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
    );
@@ -464,10 +428,10 @@ async function migrateWithExclusiveStateOwnership(params: {
  }

  try {
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath))) {
-      throw new Error("legacy APNs source reappeared during import");
-    }
-    await removePath({ ...params, sourcePath: claimPath });
+    await source.remove({
+      removeSource: params.removeSource,
+      sourceReappearedMessage: "legacy APNs source reappeared during import",
+    });
    markLegacyMigrationSourceRemoved(result.sourceKey, params.env);
  } catch (error) {
    warnings.push(`APNs state is in SQLite, but legacy cleanup failed: ${String(error)}`);
--- a/src/infra/state-migrations.exec-approvals.ts
+++ b/src/infra/state-migrations.exec-approvals.ts
@@ -21,10 +21,10 @@ import {
  resolveLegacyMigrationSourceKey,
 } from "./state-migrations.receipts.js";
 import {
+  LegacyMigrationSourceClaim,
  legacyMigrationSourceOrClaimMayExist,
  legacyMigrationSourceSnapshotsMatch as snapshotsMatch,
  readLegacyMigrationSourceSnapshot,
-  resolveLegacyMigrationRelativePath,
  type LegacyMigrationSourceSnapshot,
 } from "./state-migrations.source-snapshot.js";
 import type { MigrationMessages } from "./state-migrations.types.js";
@@ -58,10 +58,6 @@ export function detectLegacyExecApprovals(params: {
  };
 }

-function relativeLegacyPath(stateDir: string, filePath: string): string {
-  return resolveLegacyMigrationRelativePath(stateDir, filePath, "exec approvals", false);
-}
-
 async function readLegacySourceSnapshot(
  stateRoot: Root,
  stateDir: string,
@@ -194,54 +190,6 @@ function decideAndRecordMigration(params: {
  );
 }

-async function restoreClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<string | null> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  try {
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))) {
-      return null;
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
-      return `source path already exists: ${params.sourcePath}`;
-    }
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, claimPath),
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
-    return null;
-  } catch (error) {
-    return String(error);
-  }
-}
-
-async function recoverInterruptedClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<void> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  const claimRelative = relativeLegacyPath(params.stateDir, claimPath);
-  if (!(await params.stateRoot.exists(claimRelative))) {
-    return;
-  }
-  const sourceRelative = relativeLegacyPath(params.stateDir, params.sourcePath);
-  if (!(await params.stateRoot.exists(sourceRelative))) {
-    await params.stateRoot.move(claimRelative, sourceRelative);
-    return;
-  }
-  const [source, claim] = await Promise.all([
-    readLegacySourceSnapshot(params.stateRoot, params.stateDir, params.sourcePath),
-    readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath),
-  ]);
-  if (source.sha256 !== claim.sha256 || source.size !== claim.size) {
-    throw new Error("legacy exec approvals source and interrupted claim both exist");
-  }
-  await params.stateRoot.remove(claimRelative);
-}
-
 function decisionMessage(decision: MigrationDecision, removeSource: boolean): string {
  switch (decision) {
    case "legacy-imported":
@@ -271,42 +219,48 @@ async function migrateWithExclusiveStateOwnership(params: {
  removeSource?: (sourcePath: string) => Promise<void> | void;
 }): Promise<MigrationMessages> {
  const sourcePath = params.detected.sourcePath;
+  const source = new LegacyMigrationSourceClaim<LegacySourceSnapshot>({
+    stateRoot: params.stateRoot,
+    stateDir: params.stateDir,
+    sourcePath,
+    label: "exec approvals",
+    includeFilePath: false,
+    claimSuffix: DOCTOR_CLAIM_SUFFIX,
+    readSnapshot: (snapshotPath) =>
+      readLegacySourceSnapshot(params.stateRoot, params.stateDir, snapshotPath),
+  });
  try {
-    await recoverInterruptedClaim({ ...params, sourcePath });
+    await source.recover("legacy exec approvals source and interrupted claim both exist");
  } catch (error) {
    return {
      changes: [],
      warnings: [`Failed recovering a legacy exec approvals Doctor claim: ${String(error)}`],
    };
  }
-  const sourceRelative = relativeLegacyPath(params.stateDir, sourcePath);
-  if (!(await params.stateRoot.exists(sourceRelative))) {
+  if (!(await source.exists())) {
    return { changes: [], warnings: [] };
  }

  let snapshot: LegacySourceSnapshot;
  try {
-    snapshot = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    snapshot = await source.read();
  } catch (error) {
    return { changes: [], warnings: [`Failed reading legacy exec approvals: ${String(error)}`] };
  }

-  const claimPath = `${sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  const claimRelative = relativeLegacyPath(params.stateDir, claimPath);
  try {
    params.beforeVerify?.();
-    const current = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    const current = await source.read();
    if (!snapshotsMatch(current, snapshot)) {
      throw new Error("legacy exec approvals changed after migration loaded them");
    }
-    params.beforeClaim?.();
-    await params.stateRoot.move(sourceRelative, claimRelative);
-    const claimed = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-    if (!snapshotsMatch(claimed, snapshot)) {
-      throw new Error("legacy exec approvals changed before migration could claim them");
-    }
+    await source.claim({
+      snapshot,
+      mismatchMessage: "legacy exec approvals changed before migration could claim them",
+      beforeClaim: params.beforeClaim,
+    });
  } catch (error) {
-    const restoreError = await restoreClaim({ ...params, sourcePath });
+    const restoreError = await source.restore();
    return {
      changes: [],
      warnings: [
@@ -323,7 +277,7 @@ async function migrateWithExclusiveStateOwnership(params: {
      snapshot,
    });
  } catch (error) {
-    const restoreError = await restoreClaim({ ...params, sourcePath });
+    const restoreError = await source.restore();
    return {
      changes: [],
      warnings: [
@@ -332,9 +286,8 @@ async function migrateWithExclusiveStateOwnership(params: {
    };
  }

-  const preserveSource = !result.removeSource;
-  if (preserveSource) {
-    const restoreError = await restoreClaim({ ...params, sourcePath });
+  if (!result.removeSource) {
+    const restoreError = await source.restore();
    return {
      changes: [],
      warnings: [
@@ -344,20 +297,11 @@ async function migrateWithExclusiveStateOwnership(params: {
  }

  try {
-    if (await params.stateRoot.exists(sourceRelative)) {
-      throw new Error("legacy exec approvals reappeared during migration cleanup");
-    }
-    if (params.removeSource) {
-      await params.removeSource(claimPath);
-    } else {
-      await params.stateRoot.remove(claimRelative);
-    }
-    if (
-      (await params.stateRoot.exists(sourceRelative)) ||
-      (await params.stateRoot.exists(claimRelative))
-    ) {
-      throw new Error("legacy exec approvals remain after migration cleanup");
-    }
+    await source.remove({
+      removeSource: params.removeSource,
+      sourceReappearedMessage: "legacy exec approvals reappeared during migration cleanup",
+      remainingMessage: "legacy exec approvals remain after migration cleanup",
+    });
  } catch (error) {
    return {
      changes: [],
--- a/src/infra/state-migrations.mcp-oauth.test.ts
+++ b/src/infra/state-migrations.mcp-oauth.test.ts
@@ -373,6 +373,7 @@ describe("legacy MCP OAuth Doctor migration", () => {

  it("rejects malformed and unexpected store shapes without mutation", async () => {
    const { env, stateDir } = useStateDir();
+    const privateMarker = "must-not-appear-in-doctor-output";
    const malformedPath = await writeLegacy({
      stateDir,
      value: validStore({ tokens: { access_token: 42, token_type: "Bearer" } }),
@@ -382,14 +383,21 @@ describe("legacy MCP OAuth Doctor migration", () => {
      fileName: "other-1234567890abcdef.json",
      value: { ...validStore(), unexpected: true },
    });
+    const invalidJsonPath = await writeLegacy({
+      stateDir,
+      fileName: "syntax-1234567890abcdef.json",
+      bytes: Buffer.from(`{"tokens":{"access_token":${privateMarker}}}`, "utf8"),
+    });

    const result = await migrate(stateDir, env);

-    expect(result.warnings).toHaveLength(2);
+    expect(result.warnings).toHaveLength(3);
    expect(result.warnings.join("\n")).toContain("tokens are invalid");
    expect(result.warnings.join("\n")).toContain("unexpected field");
+    expect(result.warnings.join("\n")).not.toContain(privateMarker);
    expect(fs.existsSync(malformedPath)).toBe(true);
    expect(fs.existsSync(unexpectedPath)).toBe(true);
+    expect(fs.existsSync(invalidJsonPath)).toBe(true);
    expect(storeRow(env)).toBeUndefined();
    expect(receipt(env)).toBeUndefined();
  });
--- a/src/infra/state-migrations.mcp-oauth.ts
+++ b/src/infra/state-migrations.mcp-oauth.ts
@@ -23,6 +23,7 @@ import {
  type LegacyMigrationReceipt,
 } from "./state-migrations.receipts.js";
 import {
+  LegacyMigrationSourceClaim,
  legacyMigrationPathMayExist,
  legacyMigrationSourceSnapshotsMatch as snapshotsMatch,
  readLegacyMigrationSourceSnapshot,
@@ -42,6 +43,14 @@ type McpOAuthMigrationDatabase = Pick<OpenClawStateKyselyDatabase, "mcp_oauth_st

 type LegacySourceSnapshot = LegacyMigrationSourceSnapshot & { store: Record<string, unknown> };

+function parseLegacyMcpOAuthJson(buffer: Buffer): unknown {
+  try {
+    return JSON.parse(utf8Decoder.decode(buffer));
+  } catch {
+    throw new Error("legacy MCP OAuth store contains invalid JSON");
+  }
+}
+
 function exactLegacyBaseName(name: string): string | null {
  const baseName = name.endsWith(DOCTOR_CLAIM_SUFFIX)
    ? name.slice(0, -DOCTOR_CLAIM_SUFFIX.length)
@@ -116,7 +125,7 @@ async function readLegacySourceSnapshot(
  const parsed =
    options.parseStore === false
      ? {}
-      : parseLegacyMcpOAuthStore(JSON.parse(utf8Decoder.decode(snapshot.buffer)));
+      : parseLegacyMcpOAuthStore(parseLegacyMcpOAuthJson(snapshot.buffer));
  return { ...snapshot, store: parsed };
 }

@@ -225,19 +234,6 @@ function importAndRecordReceipt(params: {
  );
 }

-async function removePath(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-  removeSource?: (sourcePath: string) => Promise<void> | void;
-}): Promise<void> {
-  if (params.removeSource) {
-    await params.removeSource(params.sourcePath);
-    return;
-  }
-  await params.stateRoot.remove(relativeLegacyPath(params.stateDir, params.sourcePath));
-}
-
 async function cleanupReceiptAuthoritativeSources(params: {
  stateRoot: Root;
  stateDir: string;
@@ -254,7 +250,11 @@ async function cleanupReceiptAuthoritativeSources(params: {
    await readLegacySourceSnapshot(params.stateRoot, params.stateDir, candidate, {
      parseStore: false,
    });
-    await removePath({ ...params, sourcePath: candidate });
+    if (params.removeSource) {
+      await params.removeSource(candidate);
+    } else {
+      await params.stateRoot.remove(relativeLegacyPath(params.stateDir, candidate));
+    }
    removed += 1;
  }
  if (!params.receipt.removedSource || removed > 0) {
@@ -263,29 +263,6 @@ async function cleanupReceiptAuthoritativeSources(params: {
  return removed;
 }

-async function restoreClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<string | null> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  try {
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))) {
-      return null;
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
-      return `source path already exists: ${params.sourcePath}`;
-    }
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, claimPath),
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
-    return null;
-  } catch (error) {
-    return String(error);
-  }
-}
-
 async function migrateOneStore(params: {
  stateRoot: Root;
  stateDir: string;
@@ -313,11 +290,18 @@ async function migrateOneStore(params: {
    return notices.length > 0 ? { changes, warnings, notices } : { changes, warnings };
  }

-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  const hasSource = await params.stateRoot.exists(
-    relativeLegacyPath(params.stateDir, params.sourcePath),
-  );
-  const hasClaim = await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath));
+  const source = new LegacyMigrationSourceClaim<LegacySourceSnapshot>({
+    stateRoot: params.stateRoot,
+    stateDir: params.stateDir,
+    sourcePath: params.sourcePath,
+    label: "MCP OAuth",
+    includeFilePath: false,
+    claimSuffix: DOCTOR_CLAIM_SUFFIX,
+    readSnapshot: (snapshotPath) =>
+      readLegacySourceSnapshot(params.stateRoot, params.stateDir, snapshotPath),
+  });
+  const hasSource = await source.exists();
+  const hasClaim = await source.exists(true);
  if (hasSource && hasClaim) {
    return {
      changes,
@@ -326,7 +310,7 @@ async function migrateOneStore(params: {
      ],
    };
  }
-  const activePath = hasSource ? params.sourcePath : hasClaim ? claimPath : null;
+  const activePath = hasSource ? params.sourcePath : hasClaim ? source.claimPath : null;
  if (!activePath) {
    return { changes, warnings };
  }
@@ -343,18 +327,13 @@ async function migrateOneStore(params: {

  if (activePath === params.sourcePath) {
    try {
-      params.beforeClaim?.(params.sourcePath);
-      await params.stateRoot.move(
-        relativeLegacyPath(params.stateDir, params.sourcePath),
-        relativeLegacyPath(params.stateDir, claimPath),
-      );
-      const claimed = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-      if (!snapshotsMatch(snapshot, claimed)) {
-        throw new Error("legacy MCP OAuth source changed before Doctor could claim it");
-      }
-      snapshot = claimed;
+      snapshot = await source.claim({
+        snapshot,
+        mismatchMessage: "legacy MCP OAuth source changed before Doctor could claim it",
+        beforeClaim: () => params.beforeClaim?.(params.sourcePath),
+      });
    } catch (error) {
-      const restoreError = await restoreClaim(params);
+      const restoreError = await source.restore();
      warnings.push(
        `Failed migrating legacy MCP OAuth store ${path.basename(params.sourcePath)}: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
      );
@@ -370,7 +349,7 @@ async function migrateOneStore(params: {
      snapshot,
    });
  } catch (error) {
-    const restoreError = await restoreClaim(params);
+    const restoreError = await source.restore();
    warnings.push(
      `Failed migrating legacy MCP OAuth store ${path.basename(params.sourcePath)}: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
    );
@@ -378,21 +357,18 @@ async function migrateOneStore(params: {
  }

  try {
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
+    if (await source.exists()) {
      throw new Error("legacy MCP OAuth source reappeared during import");
    }
-    const finalSnapshot = await readLegacySourceSnapshot(
-      params.stateRoot,
-      params.stateDir,
-      claimPath,
-    );
+    const finalSnapshot = await source.read(true);
    if (!snapshotsMatch(snapshot, finalSnapshot)) {
      throw new Error("legacy MCP OAuth claim changed after SQLite import");
    }
-    await removePath({ ...params, sourcePath: claimPath });
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath))) {
-      throw new Error("legacy MCP OAuth Doctor claim remains after cleanup");
-    }
+    await source.remove({
+      removeSource: params.removeSource,
+      claimRemainingMessage: "legacy MCP OAuth Doctor claim remains after cleanup",
+      skipSourceCheck: true,
+    });
    markLegacyMigrationSourceRemoved(result.sourceKey, params.env);
  } catch (error) {
    warnings.push(`MCP OAuth state is in SQLite, but legacy cleanup failed: ${String(error)}`);
--- a/src/infra/state-migrations.restart-sentinel.ts
+++ b/src/infra/state-migrations.restart-sentinel.ts
@@ -19,10 +19,10 @@ import {
 } from "./state-migrations.receipts.js";
 import type { LegacyRestartSentinelDetection } from "./state-migrations.restart-sentinel.types.js";
 import {
+  LegacyMigrationSourceClaim,
  legacyMigrationSourceOrClaimMayExist,
  legacyMigrationSourceSnapshotsMatch as snapshotsMatch,
  readLegacyMigrationSourceSnapshot,
-  resolveLegacyMigrationRelativePath,
  type LegacyMigrationSourceSnapshot as LegacySourceSnapshot,
 } from "./state-migrations.source-snapshot.js";
 import type { MigrationMessages } from "./state-migrations.types.js";
@@ -51,24 +51,6 @@ export function detectLegacyRestartSentinel(params: {
  };
 }

-function relativeLegacyPath(stateDir: string, filePath: string): string {
-  return resolveLegacyMigrationRelativePath(stateDir, filePath, "restart sentinel", false);
-}
-
-async function readLegacySourceSnapshot(
-  stateRoot: Root,
-  stateDir: string,
-  sourcePath: string,
-): Promise<LegacySourceSnapshot> {
-  return readLegacyMigrationSourceSnapshot({
-    stateRoot,
-    stateDir,
-    sourcePath,
-    maxBytes: MAX_LEGACY_RESTART_SENTINEL_BYTES,
-    label: "restart sentinel",
-  });
-}
-
 function parseLegacyEnvelope(snapshot: LegacySourceSnapshot): RestartSentinelEnvelope | null {
  try {
    return parseRestartSentinelEnvelope(JSON.parse(utf8Decoder.decode(snapshot.buffer)));
@@ -139,59 +121,32 @@ function decideAndRecordMigration(params: {
  );
 }

-async function restoreClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<string | null> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  try {
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))) {
-      return null;
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
-      return `source path already exists: ${params.sourcePath}`;
-    }
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, claimPath),
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
-    return null;
-  } catch (error) {
-    return String(error);
-  }
-}
-
 async function recoverInterruptedClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
+  source: LegacyMigrationSourceClaim;
  env: NodeJS.ProcessEnv;
 }): Promise<void> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  const claimRelativePath = relativeLegacyPath(params.stateDir, claimPath);
-  if (!(await params.stateRoot.exists(claimRelativePath))) {
+  if (!(await params.source.exists(true))) {
    return;
  }
-  if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath)))) {
-    await params.stateRoot.move(
-      claimRelativePath,
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
+  if (!(await params.source.exists())) {
+    const restoreError = await params.source.restore();
+    if (restoreError) {
+      throw new Error(restoreError);
+    }
    return;
  }
  // Both paths can only be retired safely when the claimed bytes already have
  // an authoritative decision; otherwise preserve both for operator recovery.
  if (
    !readLegacyMigrationReceipt(
-      resolveLegacyMigrationSourceKey("restart-sentinel-json", params.sourcePath),
+      resolveLegacyMigrationSourceKey("restart-sentinel-json", params.source.sourcePath),
      params.env,
    )
  ) {
    throw new Error("legacy restart sentinel source and interrupted claim both exist");
  }
-  await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-  await params.stateRoot.remove(claimRelativePath);
+  await params.source.read(true);
+  await params.source.remove({ skipSourceCheck: true });
 }

 function decisionChange(decision: MigrationDecision): string {
@@ -224,11 +179,25 @@ async function migrateWithExclusiveStateOwnership(params: {
  const warnings: string[] = [];
  const notices: string[] = [];
  const sourcePath = params.detected.sourcePath;
+  const source = new LegacyMigrationSourceClaim<LegacySourceSnapshot>({
+    stateRoot: params.stateRoot,
+    stateDir: params.stateDir,
+    sourcePath,
+    label: "restart sentinel",
+    includeFilePath: false,
+    claimSuffix: DOCTOR_CLAIM_SUFFIX,
+    readSnapshot: (snapshotPath) =>
+      readLegacyMigrationSourceSnapshot({
+        stateRoot: params.stateRoot,
+        stateDir: params.stateDir,
+        sourcePath: snapshotPath,
+        maxBytes: MAX_LEGACY_RESTART_SENTINEL_BYTES,
+        label: "restart sentinel",
+      }),
+  });
  try {
    await recoverInterruptedClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
+      source,
      env: params.env,
    });
  } catch (error) {
@@ -237,13 +206,13 @@ async function migrateWithExclusiveStateOwnership(params: {
      warnings: [`Failed recovering a legacy restart sentinel Doctor claim: ${String(error)}`],
    };
  }
-  if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath)))) {
+  if (!(await source.exists())) {
    return { changes, warnings };
  }

  let snapshot: LegacySourceSnapshot;
  try {
-    snapshot = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    snapshot = await source.read();
  } catch (error) {
    return {
      changes,
@@ -251,28 +220,19 @@ async function migrateWithExclusiveStateOwnership(params: {
    };
  }
  const envelope = parseLegacyEnvelope(snapshot);
-  const claimPath = `${sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
  try {
    params.beforeVerify?.();
-    const current = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    const current = await source.read();
    if (!snapshotsMatch(current, snapshot)) {
      throw new Error("legacy restart sentinel changed after migration loaded it");
    }
-    params.beforeClaim?.();
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, sourcePath),
-      relativeLegacyPath(params.stateDir, claimPath),
-    );
-    const claimed = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-    if (!snapshotsMatch(claimed, snapshot)) {
-      throw new Error("legacy restart sentinel changed before migration could claim it");
-    }
-  } catch (error) {
-    const restoreError = await restoreClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
+    await source.claim({
+      snapshot,
+      mismatchMessage: "legacy restart sentinel changed before migration could claim it",
+      beforeClaim: params.beforeClaim,
    });
+  } catch (error) {
+    const restoreError = await source.restore();
    return {
      changes,
      warnings: [
@@ -290,11 +250,7 @@ async function migrateWithExclusiveStateOwnership(params: {
      envelope,
    });
  } catch (error) {
-    const restoreError = await restoreClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
-    });
+    const restoreError = await source.restore();
    return {
      changes,
      warnings: [
@@ -304,20 +260,11 @@ async function migrateWithExclusiveStateOwnership(params: {
  }

  try {
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath))) {
-      throw new Error("legacy restart sentinel reappeared during migration cleanup");
-    }
-    if (params.removeSource) {
-      await params.removeSource(claimPath);
-    } else {
-      await params.stateRoot.remove(relativeLegacyPath(params.stateDir, claimPath));
-    }
-    if (
-      (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath))) ||
-      (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))
-    ) {
-      throw new Error("legacy restart sentinel remains after migration cleanup");
-    }
+    await source.remove({
+      removeSource: params.removeSource,
+      sourceReappearedMessage: "legacy restart sentinel reappeared during migration cleanup",
+      remainingMessage: "legacy restart sentinel remains after migration cleanup",
+    });
  } catch (error) {
    warnings.push(`Legacy restart sentinel cleanup failed: ${String(error)}`);
    return { changes, warnings };
--- a/src/infra/state-migrations.source-snapshot.test.ts
+++ b/src/infra/state-migrations.source-snapshot.test.ts
@@ -52,11 +52,24 @@ describe("doctor legacy migration source contract", () => {
    expect(legacyMigrationSourceOrClaimMayExist(sourcePath)).toBe(false);
  });

-  it("rejects source paths outside the trusted migration root", () => {
+  it("rejects source paths outside the trusted migration root without exposing redacted paths", async () => {
    const { stateDir } = createSource();
-    expect(() =>
-      resolveLegacyMigrationRelativePath(stateDir, path.join(stateDir, "..", "escape"), "test"),
-    ).toThrow("outside the state directory");
+    const outsidePath = path.join(stateDir, "..", "private-marker");
+    expect(() => resolveLegacyMigrationRelativePath(stateDir, outsidePath, "test")).toThrow(
+      "outside the state directory",
+    );
+    const stateRoot = await root(stateDir, { hardlinks: "reject", symlinks: "reject" });
+    expect(
+      () =>
+        new LegacyMigrationSourceClaim({
+          stateRoot,
+          stateDir,
+          sourcePath: outsidePath,
+          label: "test",
+          includeFilePath: false,
+          readSnapshot: () => Promise.reject(new Error("unreachable")),
+        }),
+    ).toThrowError(/^legacy test path is outside the state directory$/u);
  });

  it("shares the same pinned source identity across root-bound and sync readers", async () => {
--- a/src/infra/state-migrations.source-snapshot.ts
+++ b/src/infra/state-migrations.source-snapshot.ts
@@ -38,6 +38,7 @@ export class LegacyMigrationSourceClaim<
      claimSuffix?: string;
      readSnapshot: (sourcePath: string) => Promise<TSnapshot>;
      formatError?: (error: unknown) => string;
+      includeFilePath?: boolean;
    },
  ) {
    this.sourcePath = params.sourcePath;
@@ -46,11 +47,13 @@ export class LegacyMigrationSourceClaim<
      params.stateDir,
      this.sourcePath,
      params.label,
+      params.includeFilePath,
    );
    this.claimRelativePath = resolveLegacyMigrationRelativePath(
      params.stateDir,
      this.claimPath,
      params.label,
+      params.includeFilePath,
    );
  }

@@ -113,6 +116,8 @@ export class LegacyMigrationSourceClaim<
      removeSource?: (sourcePath: string) => Promise<void> | void;
      sourceReappearedMessage?: string;
      remainingMessage?: string;
+      sourceRemainingMessage?: string;
+      claimRemainingMessage?: string;
      skipSourceCheck?: boolean;
    } = {},
  ): Promise<void> {
@@ -127,8 +132,13 @@ export class LegacyMigrationSourceClaim<
    } else {
      await this.params.stateRoot.remove(this.claimRelativePath);
    }
-    if (params.remainingMessage && ((await this.exists()) || (await this.exists(true)))) {
-      throw new Error(params.remainingMessage);
+    const sourceRemainingMessage = params.sourceRemainingMessage ?? params.remainingMessage;
+    if (sourceRemainingMessage && (await this.exists())) {
+      throw new Error(sourceRemainingMessage);
+    }
+    const claimRemainingMessage = params.claimRemainingMessage ?? params.remainingMessage;
+    if (claimRemainingMessage && (await this.exists(true))) {
+      throw new Error(claimRemainingMessage);
    }
  }
 }
--- a/src/infra/state-migrations.subagent-registry.ts
+++ b/src/infra/state-migrations.subagent-registry.ts
@@ -3,10 +3,10 @@ import path from "node:path";
 import { root, type Root } from "@openclaw/fs-safe";
 import { withLegacyMigrationStateLock } from "./state-migrations.lock.js";
 import {
+  LegacyMigrationSourceClaim,
  legacyMigrationSourceOrClaimMayExist as sourceOrClaimMayExist,
  legacyMigrationSourceSnapshotsMatch as sourceSnapshotsMatch,
  readLegacyMigrationSourceSnapshot,
-  resolveLegacyMigrationRelativePath,
  type LegacyMigrationSourceSnapshot as LegacySourceSnapshot,
 } from "./state-migrations.source-snapshot.js";
 import {
@@ -34,75 +34,32 @@ export function detectLegacySubagentRegistry(params: {
  };
 }

-function relativeLegacyPath(stateDir: string, filePath: string): string {
-  return resolveLegacyMigrationRelativePath(stateDir, filePath, "subagent registry");
-}
-
-async function readLegacySourceSnapshot(
-  stateRoot: Root,
-  stateDir: string,
-  sourcePath: string,
-): Promise<LegacySourceSnapshot> {
-  return readLegacyMigrationSourceSnapshot({
-    stateRoot,
-    stateDir,
-    sourcePath,
-    maxBytes: LEGACY_SUBAGENT_REGISTRY_MAX_BYTES,
-    label: "subagent registry",
-  });
-}
-
-async function recoverInterruptedClaim(
-  stateRoot: Root,
-  stateDir: string,
-  sourcePath: string,
-  env: NodeJS.ProcessEnv,
-): Promise<void> {
-  const claimPath = `${sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  const claimRelativePath = relativeLegacyPath(stateDir, claimPath);
-  const sourceRelativePath = relativeLegacyPath(stateDir, sourcePath);
-  if (!(await stateRoot.exists(claimRelativePath))) {
+async function recoverInterruptedClaim(params: {
+  source: LegacyMigrationSourceClaim;
+  env: NodeJS.ProcessEnv;
+}): Promise<void> {
+  if (!(await params.source.exists(true))) {
    return;
  }
-  const claimed = await readLegacySourceSnapshot(stateRoot, stateDir, claimPath);
-  if (!(await stateRoot.exists(sourceRelativePath))) {
-    await stateRoot.move(claimRelativePath, sourceRelativePath);
+  const claimed = await params.source.read(true);
+  if (!(await params.source.exists())) {
+    const restoreError = await params.source.restore();
+    if (restoreError) {
+      throw new Error(restoreError);
+    }
    return;
  }
-  await readLegacySourceSnapshot(stateRoot, stateDir, sourcePath);
+  await params.source.read();
  // The interrupted claim and recreated source are two separate retirements.
  // Record the older bytes before deletion; the recreated source is processed next.
  const result = recordLegacySubagentRegistryDiscard({
-    env,
-    sourcePath,
+    env: params.env,
+    sourcePath: params.source.sourcePath,
    sourceSha256: claimed.sha256,
    sourceSize: claimed.size,
  });
-  await stateRoot.remove(claimRelativePath);
-  markLegacySubagentRegistrySourceRemoved(result.sourceKey, env);
-}
-
-async function restoreClaim(params: {
-  stateRoot: Root;
-  stateDir: string;
-  sourcePath: string;
-}): Promise<string | null> {
-  const claimPath = `${params.sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
-  try {
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath)))) {
-      return null;
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, params.sourcePath))) {
-      return `source path already exists: ${params.sourcePath}`;
-    }
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, claimPath),
-      relativeLegacyPath(params.stateDir, params.sourcePath),
-    );
-    return null;
-  } catch (error) {
-    return String(error);
-  }
+  await params.source.remove({ skipSourceCheck: true });
+  markLegacySubagentRegistrySourceRemoved(result.sourceKey, params.env);
 }

 async function migrateWithExclusiveStateOwnership(params: {
@@ -118,44 +75,50 @@ async function migrateWithExclusiveStateOwnership(params: {
  const warnings: string[] = [];
  const notices: string[] = [];
  const sourcePath = params.detected.sourcePath;
+  const source = new LegacyMigrationSourceClaim<LegacySourceSnapshot>({
+    stateRoot: params.stateRoot,
+    stateDir: params.stateDir,
+    sourcePath,
+    label: "subagent registry",
+    claimSuffix: DOCTOR_CLAIM_SUFFIX,
+    readSnapshot: (snapshotPath) =>
+      readLegacyMigrationSourceSnapshot({
+        stateRoot: params.stateRoot,
+        stateDir: params.stateDir,
+        sourcePath: snapshotPath,
+        maxBytes: LEGACY_SUBAGENT_REGISTRY_MAX_BYTES,
+        label: "subagent registry",
+      }),
+  });
  if (!params.detected.hasLegacy) {
    return { changes, warnings };
  }

  let snapshot: LegacySourceSnapshot;
  try {
-    await recoverInterruptedClaim(params.stateRoot, params.stateDir, sourcePath, params.env);
-    if (!(await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath)))) {
+    await recoverInterruptedClaim({ source, env: params.env });
+    if (!(await source.exists())) {
      return { changes, warnings };
    }
-    snapshot = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    snapshot = await source.read();
  } catch (error) {
    warnings.push(`Failed reading legacy subagent registry: ${String(error)}`);
    return { changes, warnings };
  }

-  const claimPath = `${sourcePath}${DOCTOR_CLAIM_SUFFIX}`;
  try {
    params.beforeVerify?.();
-    const current = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, sourcePath);
+    const current = await source.read();
    if (!sourceSnapshotsMatch(current, snapshot)) {
      throw new Error("legacy subagent registry changed after Doctor loaded it");
    }
-    params.beforeClaim?.();
-    await params.stateRoot.move(
-      relativeLegacyPath(params.stateDir, sourcePath),
-      relativeLegacyPath(params.stateDir, claimPath),
-    );
-    const claimed = await readLegacySourceSnapshot(params.stateRoot, params.stateDir, claimPath);
-    if (!sourceSnapshotsMatch(claimed, snapshot)) {
-      throw new Error("legacy subagent registry changed before Doctor could claim it");
-    }
-  } catch (error) {
-    const restoreError = await restoreClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
+    await source.claim({
+      snapshot,
+      mismatchMessage: "legacy subagent registry changed before Doctor could claim it",
+      beforeClaim: params.beforeClaim,
    });
+  } catch (error) {
+    const restoreError = await source.restore();
    warnings.push(
      `Failed migrating legacy subagent registry: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
    );
@@ -171,11 +134,7 @@ async function migrateWithExclusiveStateOwnership(params: {
      sourceSize: snapshot.size,
    });
  } catch (error) {
-    const restoreError = await restoreClaim({
-      stateRoot: params.stateRoot,
-      stateDir: params.stateDir,
-      sourcePath,
-    });
+    const restoreError = await source.restore();
    warnings.push(
      `Failed migrating legacy subagent registry: ${String(error)}${restoreError ? `; restore failure: ${restoreError}` : ""}`,
    );
@@ -183,20 +142,12 @@ async function migrateWithExclusiveStateOwnership(params: {
  }

  try {
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath))) {
-      throw new Error(`legacy subagent registry reappeared during retirement: ${sourcePath}`);
-    }
-    if (params.removeSource) {
-      await params.removeSource(claimPath);
-    } else {
-      await params.stateRoot.remove(relativeLegacyPath(params.stateDir, claimPath));
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, sourcePath))) {
-      throw new Error(`legacy subagent registry reappeared during cleanup: ${sourcePath}`);
-    }
-    if (await params.stateRoot.exists(relativeLegacyPath(params.stateDir, claimPath))) {
-      throw new Error(`legacy subagent registry Doctor claim remains after cleanup: ${claimPath}`);
-    }
+    await source.remove({
+      removeSource: params.removeSource,
+      sourceReappearedMessage: `legacy subagent registry reappeared during retirement: ${sourcePath}`,
+      sourceRemainingMessage: `legacy subagent registry reappeared during cleanup: ${sourcePath}`,
+      claimRemainingMessage: `legacy subagent registry Doctor claim remains after cleanup: ${source.claimPath}`,
+    });
  } catch (error) {
    warnings.push(`Legacy subagent registry retirement cleanup failed: ${String(error)}`);
    return { changes, warnings };