diff --git a/src/infra/push-apns-store.ts b/src/infra/push-apns-store.ts index b2dfb48568c0..6977b36f87b5 100644 --- 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; } diff --git a/src/infra/state-migrations.apns.test.ts b/src/infra/state-migrations.apns.test.ts index 84feb5cd6776..5e0894372026 100644 --- 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); diff --git a/src/infra/state-migrations.apns.ts b/src/infra/state-migrations.apns.ts index 1acdfc94eda4..e3c1700629a1 100644 --- 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; -}): Promise { - 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 { - 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({ + 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)}`); diff --git a/src/infra/state-migrations.exec-approvals.ts b/src/infra/state-migrations.exec-approvals.ts index 3daaaf9c82d5..78430aec7f68 100644 --- 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 { - 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 { - 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; }): Promise { const sourcePath = params.detected.sourcePath; + const source = new LegacyMigrationSourceClaim({ + 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: [], diff --git a/src/infra/state-migrations.mcp-oauth.test.ts b/src/infra/state-migrations.mcp-oauth.test.ts index 56f1fa5356b0..fb835026556f 100644 --- 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(); }); diff --git a/src/infra/state-migrations.mcp-oauth.ts b/src/infra/state-migrations.mcp-oauth.ts index f523579da1ac..f6328000ed7c 100644 --- 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 }; +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; -}): Promise { - 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 { - 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({ + 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)}`); diff --git a/src/infra/state-migrations.restart-sentinel.ts b/src/infra/state-migrations.restart-sentinel.ts index 4400fedea76c..aa3f896266cd 100644 --- 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 { - 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 { - 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 { - 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({ + 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 }; diff --git a/src/infra/state-migrations.source-snapshot.test.ts b/src/infra/state-migrations.source-snapshot.test.ts index 72356c055864..0756806ff16c 100644 --- 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 () => { diff --git a/src/infra/state-migrations.source-snapshot.ts b/src/infra/state-migrations.source-snapshot.ts index 80314e002b3a..5d84ddb829a5 100644 --- 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; 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; sourceReappearedMessage?: string; remainingMessage?: string; + sourceRemainingMessage?: string; + claimRemainingMessage?: string; skipSourceCheck?: boolean; } = {}, ): Promise { @@ -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); } } } diff --git a/src/infra/state-migrations.subagent-registry.ts b/src/infra/state-migrations.subagent-registry.ts index ef95a8325760..5b1e8fdea462 100644 --- 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 { - 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 { - 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 { + 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 { - 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({ + 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 };