From e44dde218f1629d87f585bf8bfeabda85e1a21f3 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 14 Aug 2026 10:10:12 -0700 Subject: [PATCH] fix(gateway): revoke device workers when pairing is removed (#123696) * fix(gateway): revoke device workers on unpair * refactor(backup): split sqlite verification * refactor(backup): narrow sqlite verifier exports * test(gateway): track worker startup temp state * refactor(backup): preserve sqlite architecture boundary * test(gateway): follow worker startup logger binding * fix(gateway): revoke workers during pairing pruning --- docs/nodes/index.md | 6 +- docs/plan/runners.md | 7 +- src/commands/backup-verify-manifest.ts | 149 +++++++++++++++++ src/commands/backup-verify.ts | 158 ++---------------- src/gateway/device-pairing-prune.test.ts | 50 +++++- src/gateway/device-pairing-prune.ts | 8 +- src/gateway/device-worker-revocation.ts | 26 +++ src/gateway/server-import-boundary.test.ts | 2 +- src/gateway/server-methods/devices.test.ts | 30 ++++ src/gateway/server-methods/devices.ts | 2 + src/gateway/server-methods/nodes.pairing.ts | 8 +- src/gateway/server-methods/nodes.test.ts | 32 ++++ .../server-worker-environment-startup.test.ts | 89 ++++++++++ .../server-worker-environment-startup.ts | 32 +++- .../device-provider.test.ts | 17 +- .../worker-environments/device-provider.ts | 17 ++ .../provider-reconciliation.test.ts | 35 ++++ src/gateway/worker-environments/service.ts | 40 +++-- src/gateway/worker-environments/store.test.ts | 16 ++ src/gateway/worker-environments/store.ts | 3 + 20 files changed, 552 insertions(+), 175 deletions(-) create mode 100644 src/commands/backup-verify-manifest.ts create mode 100644 src/gateway/device-worker-revocation.ts create mode 100644 src/gateway/server-worker-environment-startup.test.ts diff --git a/docs/nodes/index.md b/docs/nodes/index.md index b06ff62fb226..354f0293469f 100644 --- a/docs/nodes/index.md +++ b/docs/nodes/index.md @@ -454,7 +454,11 @@ disconnect; at that boundary its old worker environment is treated as gone and the session placement reconciles normally. Pairing itself remains, so a later reconnect can provision a fresh environment. Legacy pairings without exact node disconnect history are retained fail-safe rather than expired from unrelated -device activity. +device activity. Removing the device pairing, silently pruning a superseded +pairing, or removing only its node role invalidates clients first, then runs +targeted environment and placement reconciliation; explicit removal waits for +the credential fence before returning success, and the periodic sweep retries +failed provider or placement cleanup. See [Anthropic: Claude sessions across computers](/providers/anthropic#claude-sessions-across-computers) for the Control UI behavior and storage sources. diff --git a/docs/plan/runners.md b/docs/plan/runners.md index e327d026db3c..32ed708b44f0 100644 --- a/docs/plan/runners.md +++ b/docs/plan/runners.md @@ -186,7 +186,12 @@ stated honestly (revision 1 undersold this): (unpaired or ceiling elapsed → normal orphan/reap path). A device-environment reaper keyed on unpair/dormancy — not on provider teardown proof — cleans rows, - credentials, and staged refs. Unreferenced terminal environment rows retain + credentials, and staged refs. Explicit device removal, node-role removal, and + silent superseded-pairing pruning share one client-invalidation, credential, + environment, and placement reconciliation flow. Explicit RPCs wait for the + credential fence before success returns; periodic reconciliation retries + failed provider or placement cleanup. Unreferenced + terminal environment rows retain seven days of operator diagnostics, then prune in bounded post-reconcile batches; any surviving placement keeps its environment provenance. Device-side GC of per-session workspace dirs and superseded bundles is a diff --git a/src/commands/backup-verify-manifest.ts b/src/commands/backup-verify-manifest.ts new file mode 100644 index 000000000000..7286f0ae74d3 --- /dev/null +++ b/src/commands/backup-verify-manifest.ts @@ -0,0 +1,149 @@ +import path from "node:path"; +import { readStringValue } from "@openclaw/normalization-core/string-coerce"; +import { + isArchivePathWithin, + normalizeArchivePath, + normalizeArchiveRoot, +} from "../infra/backup-archive-path-policy.js"; +import { isRecord } from "../utils.js"; + +export type BackupManifest = { + schemaVersion: number; + createdAt: string; + archiveRoot: string; + runtimeVersion: string; + platform: string; + nodeVersion: string; + options?: { + includeWorkspace?: boolean; + }; + paths?: { + stateDir?: string; + configPath?: string; + oauthDir?: string; + workspaceDirs?: string[]; + }; + assets: Array<{ + kind: string; + sourcePath: string; + archivePath: string; + }>; + skipped?: Array<{ + kind?: string; + sourcePath?: string; + reason?: string; + coveredBy?: string; + }>; +}; + +export function parseBackupManifest(raw: string): BackupManifest { + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (err) { + throw new Error("Backup manifest is not valid JSON.", { cause: err }); + } + + if (!isRecord(parsed)) { + throw new Error("Backup manifest must be an object."); + } + if (parsed.schemaVersion !== 1) { + throw new Error(`Unsupported backup manifest schemaVersion: ${String(parsed.schemaVersion)}`); + } + if (typeof parsed.archiveRoot !== "string" || !parsed.archiveRoot.trim()) { + throw new Error("Backup manifest is missing archiveRoot."); + } + if (typeof parsed.createdAt !== "string" || !parsed.createdAt.trim()) { + throw new Error("Backup manifest is missing createdAt."); + } + if (!Array.isArray(parsed.assets)) { + throw new Error("Backup manifest is missing assets."); + } + + const assets: BackupManifest["assets"] = []; + for (const asset of parsed.assets) { + if (!isRecord(asset)) { + throw new Error("Backup manifest contains a non-object asset."); + } + if (typeof asset.kind !== "string" || !asset.kind.trim()) { + throw new Error("Backup manifest asset is missing kind."); + } + if (typeof asset.sourcePath !== "string" || !asset.sourcePath.trim()) { + throw new Error("Backup manifest asset is missing sourcePath."); + } + if (typeof asset.archivePath !== "string" || !asset.archivePath.trim()) { + throw new Error("Backup manifest asset is missing archivePath."); + } + assets.push({ + kind: asset.kind, + sourcePath: asset.sourcePath, + archivePath: asset.archivePath, + }); + } + + return { + schemaVersion: 1, + archiveRoot: parsed.archiveRoot, + createdAt: parsed.createdAt, + runtimeVersion: + typeof parsed.runtimeVersion === "string" && parsed.runtimeVersion.trim() + ? parsed.runtimeVersion + : "unknown", + platform: typeof parsed.platform === "string" ? parsed.platform : "unknown", + nodeVersion: typeof parsed.nodeVersion === "string" ? parsed.nodeVersion : "unknown", + options: isRecord(parsed.options) + ? { includeWorkspace: parsed.options.includeWorkspace as boolean | undefined } + : undefined, + paths: isRecord(parsed.paths) + ? { + stateDir: readStringValue(parsed.paths.stateDir), + configPath: readStringValue(parsed.paths.configPath), + oauthDir: readStringValue(parsed.paths.oauthDir), + workspaceDirs: Array.isArray(parsed.paths.workspaceDirs) + ? parsed.paths.workspaceDirs.filter( + (entry): entry is string => typeof entry === "string", + ) + : undefined, + } + : undefined, + assets, + skipped: Array.isArray(parsed.skipped) ? parsed.skipped : undefined, + }; +} + +export function isRootBackupManifestEntry(entryPath: string): boolean { + const parts = entryPath.split("/"); + return parts.length === 2 && parts[0] !== "" && parts[1] === "manifest.json"; +} + +export function verifyBackupManifestEntries(manifest: BackupManifest, entries: Set): void { + const archiveRoot = normalizeArchiveRoot(manifest.archiveRoot); + const manifestEntryPath = path.posix.join(archiveRoot, "manifest.json"); + const normalizedEntries = [...entries]; + const normalizedEntrySet = new Set(normalizedEntries); + + if (!normalizedEntrySet.has(manifestEntryPath)) { + throw new Error(`Archive is missing manifest entry: ${manifestEntryPath}`); + } + + for (const entry of normalizedEntries) { + if (!isArchivePathWithin(entry, archiveRoot)) { + throw new Error(`Archive entry is outside the declared archive root: ${entry}`); + } + } + + const payloadRoot = path.posix.join(archiveRoot, "payload"); + for (const asset of manifest.assets) { + const assetArchivePath = normalizeArchivePath(asset.archivePath, "Backup manifest asset path"); + if (!isArchivePathWithin(assetArchivePath, payloadRoot)) { + throw new Error(`Manifest asset path is outside payload root: ${asset.archivePath}`); + } + const exact = normalizedEntrySet.has(assetArchivePath); + const nested = normalizedEntries.some( + (entry) => entry !== assetArchivePath && isArchivePathWithin(entry, assetArchivePath), + ); + if (!exact && !nested) { + throw new Error(`Archive is missing payload for manifest asset: ${assetArchivePath}`); + } + } +} diff --git a/src/commands/backup-verify.ts b/src/commands/backup-verify.ts index 8950e58b4ab6..2393c408ff14 100644 --- a/src/commands/backup-verify.ts +++ b/src/commands/backup-verify.ts @@ -4,7 +4,6 @@ import os from "node:os"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import { toStringifiedError } from "@openclaw/normalization-core/error-coercion"; -import { readStringValue } from "@openclaw/normalization-core/string-coerce"; import * as tar from "tar"; import { loadSqliteVecExtension } from "../../packages/memory-host-sdk/src/engine-storage.js"; import { @@ -18,45 +17,20 @@ import { formatDiskSpaceBytes, tryReadDiskSpace } from "../infra/disk-space.js"; import { openNodeSqliteDatabase } from "../infra/node-sqlite.js"; import { assertSqliteIntegrity } from "../infra/sqlite-integrity.js"; import { type RuntimeEnv, writeRuntimeJson } from "../runtime.js"; -import { isRecord, resolveUserPath } from "../utils.js"; +import { resolveUserPath } from "../utils.js"; import { BACKUP_MAX_DECOMPRESSION_RATIO, buildBackupArchivePath } from "./backup-shared.js"; +import { + type BackupManifest, + isRootBackupManifestEntry, + parseBackupManifest, + verifyBackupManifestEntries, +} from "./backup-verify-manifest.js"; const MAX_MANIFEST_BYTES = 1024 * 1024; const MAX_SQLITE_SNAPSHOT_EXTRACT_BYTES = 64 * 1024 * 1024 * 1024; const SQLITE_SNAPSHOT_FREE_SPACE_RESERVE_BYTES = 256 * 1024 * 1024; const SQLITE_SNAPSHOT_SIDECAR_SUFFIXES = ["-wal", "-shm", "-journal"] as const; -type BackupManifestAsset = { - kind: string; - sourcePath: string; - archivePath: string; -}; - -type BackupManifest = { - schemaVersion: number; - createdAt: string; - archiveRoot: string; - runtimeVersion: string; - platform: string; - nodeVersion: string; - options?: { - includeWorkspace?: boolean; - }; - paths?: { - stateDir?: string; - configPath?: string; - oauthDir?: string; - workspaceDirs?: string[]; - }; - assets: BackupManifestAsset[]; - skipped?: Array<{ - kind?: string; - sourcePath?: string; - reason?: string; - coveredBy?: string; - }>; -}; - type BackupVerifyOptions = { archive: string; json?: boolean; @@ -93,81 +67,6 @@ type SqliteSnapshotEntry = NormalizedArchiveEntry & { type ExpectedSqliteRole = "agent" | "global"; -function parseManifest(raw: string): BackupManifest { - let parsed: unknown; - try { - parsed = JSON.parse(raw); - } catch (err) { - throw new Error("Backup manifest is not valid JSON.", { cause: err }); - } - - if (!isRecord(parsed)) { - throw new Error("Backup manifest must be an object."); - } - if (parsed.schemaVersion !== 1) { - throw new Error(`Unsupported backup manifest schemaVersion: ${String(parsed.schemaVersion)}`); - } - if (typeof parsed.archiveRoot !== "string" || !parsed.archiveRoot.trim()) { - throw new Error("Backup manifest is missing archiveRoot."); - } - if (typeof parsed.createdAt !== "string" || !parsed.createdAt.trim()) { - throw new Error("Backup manifest is missing createdAt."); - } - if (!Array.isArray(parsed.assets)) { - throw new Error("Backup manifest is missing assets."); - } - - const assets: BackupManifestAsset[] = []; - for (const asset of parsed.assets) { - if (!isRecord(asset)) { - throw new Error("Backup manifest contains a non-object asset."); - } - if (typeof asset.kind !== "string" || !asset.kind.trim()) { - throw new Error("Backup manifest asset is missing kind."); - } - if (typeof asset.sourcePath !== "string" || !asset.sourcePath.trim()) { - throw new Error("Backup manifest asset is missing sourcePath."); - } - if (typeof asset.archivePath !== "string" || !asset.archivePath.trim()) { - throw new Error("Backup manifest asset is missing archivePath."); - } - assets.push({ - kind: asset.kind, - sourcePath: asset.sourcePath, - archivePath: asset.archivePath, - }); - } - - return { - schemaVersion: 1, - archiveRoot: parsed.archiveRoot, - createdAt: parsed.createdAt, - runtimeVersion: - typeof parsed.runtimeVersion === "string" && parsed.runtimeVersion.trim() - ? parsed.runtimeVersion - : "unknown", - platform: typeof parsed.platform === "string" ? parsed.platform : "unknown", - nodeVersion: typeof parsed.nodeVersion === "string" ? parsed.nodeVersion : "unknown", - options: isRecord(parsed.options) - ? { includeWorkspace: parsed.options.includeWorkspace as boolean | undefined } - : undefined, - paths: isRecord(parsed.paths) - ? { - stateDir: readStringValue(parsed.paths.stateDir), - configPath: readStringValue(parsed.paths.configPath), - oauthDir: readStringValue(parsed.paths.oauthDir), - workspaceDirs: Array.isArray(parsed.paths.workspaceDirs) - ? parsed.paths.workspaceDirs.filter( - (entry): entry is string => typeof entry === "string", - ) - : undefined, - } - : undefined, - assets, - skipped: Array.isArray(parsed.skipped) ? parsed.skipped : undefined, - }; -} - async function listArchiveEntries(archivePath: string): Promise { const entries: ArchiveEntry[] = []; await tar.t({ @@ -215,43 +114,6 @@ async function extractManifest(params: { return content.toString("utf8"); } -function isRootManifestEntry(entryPath: string): boolean { - const parts = entryPath.split("/"); - return parts.length === 2 && parts[0] !== "" && parts[1] === "manifest.json"; -} - -function verifyManifestAgainstEntries(manifest: BackupManifest, entries: Set): void { - const archiveRoot = normalizeArchiveRoot(manifest.archiveRoot); - const manifestEntryPath = path.posix.join(archiveRoot, "manifest.json"); - const normalizedEntries = [...entries]; - const normalizedEntrySet = new Set(normalizedEntries); - - if (!normalizedEntrySet.has(manifestEntryPath)) { - throw new Error(`Archive is missing manifest entry: ${manifestEntryPath}`); - } - - for (const entry of normalizedEntries) { - if (!isArchivePathWithin(entry, archiveRoot)) { - throw new Error(`Archive entry is outside the declared archive root: ${entry}`); - } - } - - const payloadRoot = path.posix.join(archiveRoot, "payload"); - for (const asset of manifest.assets) { - const assetArchivePath = normalizeArchivePath(asset.archivePath, "Backup manifest asset path"); - if (!isArchivePathWithin(assetArchivePath, payloadRoot)) { - throw new Error(`Manifest asset path is outside payload root: ${asset.archivePath}`); - } - const exact = normalizedEntrySet.has(assetArchivePath); - const nested = normalizedEntries.some( - (entry) => entry !== assetArchivePath && isArchivePathWithin(entry, assetArchivePath), - ); - if (!exact && !nested) { - throw new Error(`Archive is missing payload for manifest asset: ${assetArchivePath}`); - } - } -} - function verifyHardlinkTargetsAgainstArchiveRoot( hardlinkTargets: Array<{ entryPath: string; normalized: string }>, archiveRoot: string, @@ -705,7 +567,7 @@ export async function verifyBackupArchive(archive: string): Promise ({ entryPath: entry.path, linkpath: entry.linkpath })); const normalizedEntrySet = new Set(entries.map((entry) => entry.normalized)); - const manifestMatches = entries.filter((entry) => isRootManifestEntry(entry.normalized)); + const manifestMatches = entries.filter((entry) => isRootBackupManifestEntry(entry.normalized)); if (manifestMatches.length !== 1) { throw new Error(`Expected exactly one backup manifest entry, found ${manifestMatches.length}.`); } @@ -725,8 +587,8 @@ export async function verifyBackupArchive(archive: string): Promise }; type PruneContext = Parameters[0]["context"]; -function createPruneContext(params?: { connectedDeviceIds?: string[] }) { +function createPruneContext(params?: { + connectedDeviceIds?: string[]; + workerEnvironmentIds?: Record; +}) { const broadcasts: BroadcastCall[] = []; const invalidated: string[] = []; const disconnected: string[] = []; const logs: string[] = []; const warnings: string[] = []; const clearedSurfaces: string[] = []; + const revokedWorkers: string[] = []; + const reconciledPlacements: string[] = []; + const order: string[] = []; const connected = new Set(params?.connectedDeviceIds ?? []); + const workerEnvironmentService = {} as NonNullable; + bindDeviceWorkerReconciliation(workerEnvironmentService, async (deviceId) => { + revokedWorkers.push(deviceId); + order.push(`worker:${deviceId}`); + return params?.workerEnvironmentIds?.[deviceId] ?? []; + }); + const workerPlacementDispatchService = { + reconcileActive: async (environmentId: string) => { + reconciledPlacements.push(environmentId); + order.push(`placement:${environmentId}`); + }, + } as NonNullable; const context: PruneContext = { broadcast: (event, payload) => { broadcasts.push({ event, payload: payload as Record }); @@ -50,10 +69,14 @@ function createPruneContext(params?: { connectedDeviceIds?: string[] }) { hasConnectedClientsForDevice: (deviceId: string) => connected.has(deviceId), invalidateClientsForDevice: (deviceId: string) => { invalidated.push(deviceId); + order.push(`invalidate:${deviceId}`); }, disconnectClientsForDevice: (deviceId: string) => { disconnected.push(deviceId); + order.push(`disconnect:${deviceId}`); }, + workerEnvironmentService, + workerPlacementDispatchService, nodeRegistry: { updateSurface: (nodeId: string) => { clearedSurfaces.push(nodeId); @@ -61,7 +84,18 @@ function createPruneContext(params?: { connectedDeviceIds?: string[] }) { }, }, }; - return { broadcasts, invalidated, disconnected, logs, warnings, clearedSurfaces, context }; + return { + broadcasts, + invalidated, + disconnected, + logs, + warnings, + clearedSurfaces, + revokedWorkers, + reconciledPlacements, + order, + context, + }; } async function pairSilentDevice(params: { @@ -166,7 +200,9 @@ describe("pruneSupersededSilentPairingsAfterApproval", () => { }); const wakeLifecycle = captureNodeWakeLifecycle("node-stale"); - const harness = createPruneContext(); + const harness = createPruneContext({ + workerEnvironmentIds: { "node-stale": ["environment-node-stale"] }, + }); const pruned = await pruneSupersededSilentPairingsAfterApproval({ deviceId: anchor.deviceId, context: harness.context, @@ -185,7 +221,15 @@ describe("pruneSupersededSilentPairingsAfterApproval", () => { expect(listPendingNodeActions({ nodeId: "node-stale", ttlMs: 60_000 })).toEqual([]); await expect(loadApnsRegistration("node-stale", baseDir)).resolves.toBeNull(); expect(harness.invalidated).toEqual(["node-stale"]); + expect(harness.revokedWorkers).toEqual(["node-stale"]); + expect(harness.reconciledPlacements).toEqual(["environment-node-stale"]); expect(harness.disconnected).toEqual(["node-stale"]); + expect(harness.order).toEqual([ + "invalidate:node-stale", + "worker:node-stale", + "placement:environment-node-stale", + "disconnect:node-stale", + ]); expect(harness.clearedSurfaces).toEqual(["node-stale"]); expect(harness.warnings).toEqual([]); expect(harness.broadcasts).toEqual([ diff --git a/src/gateway/device-pairing-prune.ts b/src/gateway/device-pairing-prune.ts index dddd5b762fd5..3a0fb397e458 100644 --- a/src/gateway/device-pairing-prune.ts +++ b/src/gateway/device-pairing-prune.ts @@ -3,6 +3,7 @@ import { pruneSupersededSilentPairedDevices, type PrunedSupersededPairedDevice, } from "../infra/device-pairing.js"; +import { reconcileRevokedDeviceWorker } from "./device-worker-revocation.js"; import { clearRemovedNodeRuntimeState } from "./node-runtime-state.js"; import type { GatewayRequestContext } from "./server-methods/types.js"; @@ -12,6 +13,8 @@ type PruneContext = Pick< | "hasConnectedClientsForDevice" | "invalidateClientsForDevice" | "disconnectClientsForDevice" + | "workerEnvironmentService" + | "workerPlacementDispatchService" > & { logGateway: Pick; nodeRegistry: Pick; @@ -45,9 +48,10 @@ export async function pruneSupersededSilentPairingsAfterApproval(params: { // queues, wake lifecycles, and runtime metadata before session teardown. clearRemovedNodeRuntimeState({ nodeId: entry.deviceId, context }); } - // Invalidate before disconnect so buffered frames from a racing reconnect - // fail authorization, mirroring device.pair.remove ordering. + // Invalidate before credential and placement teardown so racing reconnects + // fail authorization through the same owner used by explicit removal. context.invalidateClientsForDevice?.(entry.deviceId, { reason: "device-pair-removed" }); + await reconcileRevokedDeviceWorker(context, entry.deviceId); if (entry.roles.includes("node")) { context.broadcast( "node.pair.resolved", diff --git a/src/gateway/device-worker-revocation.ts b/src/gateway/device-worker-revocation.ts new file mode 100644 index 000000000000..e05b2b5260b5 --- /dev/null +++ b/src/gateway/device-worker-revocation.ts @@ -0,0 +1,26 @@ +import type { GatewayRequestContext } from "./server-methods/shared-types.js"; +import { reconcileDeviceWorker } from "./worker-environments/device-provider.js"; + +/** Reconciles worker authority after pairing removal without delaying token invalidation. */ +type DeviceWorkerRevocationContext = Pick< + GatewayRequestContext, + "workerEnvironmentService" | "workerPlacementDispatchService" +> & { + logGateway: Pick; +}; + +export async function reconcileRevokedDeviceWorker( + context: DeviceWorkerRevocationContext, + deviceId: string, +): Promise { + const environmentIds = await reconcileDeviceWorker(context.workerEnvironmentService, deviceId); + for (const environmentId of environmentIds) { + try { + await context.workerPlacementDispatchService?.reconcileActive?.(environmentId); + } catch { + context.logGateway.warn( + `device worker placement reconciliation failed device=${deviceId} environment=${environmentId}`, + ); + } + } +} diff --git a/src/gateway/server-import-boundary.test.ts b/src/gateway/server-import-boundary.test.ts index b45d3b9efde4..9c27d457c940 100644 --- a/src/gateway/server-import-boundary.test.ts +++ b/src/gateway/server-import-boundary.test.ts @@ -220,7 +220,7 @@ describe("gateway startup import boundaries", () => { ); const identityStart = workerStartup.indexOf("resolveSshIdentity: async", serviceStart); const bootstrapStart = workerStartup.indexOf("bootstrapWorker: async", serviceStart); - const loggerStart = workerStartup.indexOf("logger: params.log.child", bootstrapStart); + const loggerStart = workerStartup.indexOf("logger: workerEnvironmentLog", bootstrapStart); expect(prepareStart).toBeGreaterThan(-1); expect(serviceStart).toBeGreaterThan(prepareStart); diff --git a/src/gateway/server-methods/devices.test.ts b/src/gateway/server-methods/devices.test.ts index bc796c938d0a..e0615d5acecf 100644 --- a/src/gateway/server-methods/devices.test.ts +++ b/src/gateway/server-methods/devices.test.ts @@ -19,6 +19,7 @@ import { getNodeWakeStateSnapshot, resetNodeWakeStateForTest, } from "../node-wake-state.test-support.js"; +import { bindDeviceWorkerReconciliation } from "../worker-environments/device-provider.js"; import { deviceHandlers } from "./devices.js"; import type { GatewayRequestHandlerOptions } from "./types.js"; @@ -257,6 +258,35 @@ describe("deviceHandlers", () => { expect(disconnect).toHaveBeenCalledWith("device-1"); }); + it("reconciles device worker authority before reporting pairing removal", async () => { + removePairedDeviceMock.mockResolvedValue({ deviceId: "device-1" }); + const opts = createOptions("device.pair.remove", { deviceId: "device-1" }); + const order: string[] = []; + const workerEnvironmentService = {}; + bindDeviceWorkerReconciliation(workerEnvironmentService, async () => { + order.push("environment"); + return ["environment-1"]; + }); + const reconcileActive = vi.fn(async () => { + order.push("placement"); + }); + Object.assign(opts.context, { + workerEnvironmentService, + workerPlacementDispatchService: { reconcileActive }, + }); + vi.mocked(opts.respond).mockImplementation(() => { + order.push("respond"); + }); + + await expectDefined( + deviceHandlers["device.pair.remove"], + 'deviceHandlers["device.pair.remove"] test invariant', + )(opts); + + expect(reconcileActive).toHaveBeenCalledWith("environment-1"); + expect(order).toEqual(["environment", "placement", "respond"]); + }); + it("does not disconnect clients when device removal fails", async () => { removePairedDeviceMock.mockResolvedValue(null); const opts = createOptions("device.pair.remove", { deviceId: "device-1" }); diff --git a/src/gateway/server-methods/devices.ts b/src/gateway/server-methods/devices.ts index afc5869d8981..3d97d8d7463f 100644 --- a/src/gateway/server-methods/devices.ts +++ b/src/gateway/server-methods/devices.ts @@ -28,6 +28,7 @@ import { updatePairedDeviceMetadata, } from "../../infra/device-pairing.js"; import type { DiagnosticSecurityEventInput } from "../../infra/diagnostic-events.js"; +import { reconcileRevokedDeviceWorker } from "../device-worker-revocation.js"; import { clearRemovedNodeRuntimeState } from "../node-runtime-state.js"; import { invalidateNodeWakeState } from "../node-wake-state.js"; import { @@ -522,6 +523,7 @@ export const deviceHandlers: GatewayRequestHandlers = { context.invalidateClientsForDevice?.(removed.deviceId, { reason: "device-pair-removed", }); + await reconcileRevokedDeviceWorker(context, removed.deviceId); context.logGateway.info(`device pairing removed device=${removed.deviceId}`); emitDevicePairingLifecycleSecurityEvent({ action: "device.pairing.removed", diff --git a/src/gateway/server-methods/nodes.pairing.ts b/src/gateway/server-methods/nodes.pairing.ts index 54af3cebdcee..8bea506c19b1 100644 --- a/src/gateway/server-methods/nodes.pairing.ts +++ b/src/gateway/server-methods/nodes.pairing.ts @@ -21,6 +21,7 @@ import { listApprovedPairedDeviceRoles, removePairedDeviceRole, } from "../../infra/device-pairing.js"; +import { reconcileRevokedDeviceWorker } from "../device-worker-revocation.js"; import { resolveNodePairingCommandAllowlist, normalizeDeclaredNodeCommands, @@ -143,7 +144,11 @@ async function removePairedDeviceBackedNode(params: { client: GatewayClient | null; context: Pick< GatewayRequestContext, - "disconnectClientsForDevice" | "invalidateClientsForDevice" | "logGateway" + | "disconnectClientsForDevice" + | "invalidateClientsForDevice" + | "logGateway" + | "workerEnvironmentService" + | "workerPlacementDispatchService" >; }): Promise< | { @@ -208,6 +213,7 @@ async function removePairedDeviceBackedNode(params: { role: "node", reason: "device-pair-removed", }); + await reconcileRevokedDeviceWorker(params.context, removed.deviceId); return { status: "removed", nodeId: removed.deviceId, diff --git a/src/gateway/server-methods/nodes.test.ts b/src/gateway/server-methods/nodes.test.ts index 03fe434729f2..2ec7e6e1905b 100644 --- a/src/gateway/server-methods/nodes.test.ts +++ b/src/gateway/server-methods/nodes.test.ts @@ -45,6 +45,7 @@ import { getNodeWakeStateSnapshot, resetNodeWakeStateForTest, } from "../node-wake-state.test-support.js"; +import { bindDeviceWorkerReconciliation } from "../worker-environments/device-provider.js"; import { nodeHandlers } from "./nodes.js"; import { createWorkerSupervisorNodeClient } from "./nodes.runner-inventory.test-support.js"; import type { GatewayRequestHandlerOptions } from "./types.js"; @@ -680,6 +681,37 @@ describe("nodeHandlers node.pair.remove", () => { await expect(loadApnsRegistration(nodeId)).resolves.toBeNull(); }); + it("reconciles device worker authority before reporting node-role removal", async () => { + const state = await createState("node-remove-worker-reconcile"); + const nodeId = "worker-node-remove"; + await pairAndroidNodeDevice(state.stateDir, nodeId); + const { opts } = createOptions({ nodeId }); + const order: string[] = []; + const workerEnvironmentService = {}; + bindDeviceWorkerReconciliation(workerEnvironmentService, async () => { + order.push("environment"); + return ["environment-1"]; + }); + const reconcileActive = vi.fn(async () => { + order.push("placement"); + }); + Object.assign(opts.context, { + workerEnvironmentService, + workerPlacementDispatchService: { reconcileActive }, + }); + vi.mocked(opts.respond).mockImplementation(() => { + order.push("respond"); + }); + + await expectDefined( + nodeHandlers["node.pair.remove"], + 'nodeHandlers["node.pair.remove"] test invariant', + )(opts); + + expect(reconcileActive).toHaveBeenCalledWith("environment-1"); + expect(order).toEqual(["environment", "placement", "respond"]); + }); + it("preserves an APNs registration created after node-role removal commits", async () => { const state = await createState("node-remove-apns-registration-race"); const nodeId = "ios-node-registration-race"; diff --git a/src/gateway/server-worker-environment-startup.test.ts b/src/gateway/server-worker-environment-startup.test.ts new file mode 100644 index 000000000000..a83e3b08fb45 --- /dev/null +++ b/src/gateway/server-worker-environment-startup.test.ts @@ -0,0 +1,89 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { withEnvAsync } from "../test-utils/env.js"; +import { createDesktopSessionRegistry } from "./desktop/session-registry.js"; +import { + createGatewayWorkerEnvironmentRuntime, + loadGatewayWorkerEnvironmentStartupState, +} from "./server-worker-environment-startup.js"; +import { hashWorkerCredential } from "./worker-environments/credential.js"; +import { + DEVICE_WORKER_PROVIDER_ID, + reconcileDeviceWorker, +} from "./worker-environments/device-provider.js"; + +const DEVICE_ID = "revoked-device"; +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +afterEach(() => { + closeOpenClawStateDatabaseForTest(); +}); + +describe("gateway worker environment startup", () => { + it("binds device revocation to the persisted profile settings", async () => { + const stateDir = tempDirs.make("openclaw-worker-startup-"); + try { + await withEnvAsync({ OPENCLAW_STATE_DIR: stateDir }, async () => { + const startup = await loadGatewayWorkerEnvironmentStartupState(); + startup.store.createIntent({ + environmentId: "device-environment", + providerId: DEVICE_WORKER_PROVIDER_ID, + profileId: `device:${DEVICE_ID}`, + profileSnapshot: { install: "bundle", settings: { device: DEVICE_ID } }, + provisionOperationId: "provision:device-environment", + }); + startup.store.transition({ + environmentId: "device-environment", + from: "requested", + to: "provisioning", + }); + startup.store.transition({ + environmentId: "device-environment", + from: "provisioning", + to: "ready", + patch: { + leaseId: "device-lease", + sshEndpoint: null, + sharedHost: true, + bootstrapReceipt: { + bundleHash: "a".repeat(64), + openclawVersion: "2026.8.14", + protocolFeatures: ["worker-heartbeat-v1"], + installKind: "local", + }, + credential: { + credentialHash: hashWorkerCredential("device-credential"), + sessionId: null, + rpcSetVersion: 1, + expiresAtMs: Date.now() + 60_000, + }, + }, + }); + + const runtime = await createGatewayWorkerEnvironmentRuntime({ + getPluginRegistry: () => ({ workerProviders: new Map() }), + resolveWorkerGateway: () => undefined, + desktopSessionRegistry: createDesktopSessionRegistry({ lingerMs: 1 }), + startup, + log: { child: () => ({ warn: () => {} }) }, + }); + const service = runtime.workerEnvironmentService; + if (!service) { + throw new Error("worker environment service was not created"); + } + try { + await expect(reconcileDeviceWorker(service, DEVICE_ID)).resolves.toEqual([ + "device-environment", + ]); + expect(startup.store.getCredential("device-environment")).toBeUndefined(); + expect(startup.store.get("device-environment")?.state).toBe("orphaned"); + } finally { + await service.stop(); + } + }); + } finally { + closeOpenClawStateDatabaseForTest(); + } + }); +}); diff --git a/src/gateway/server-worker-environment-startup.ts b/src/gateway/server-worker-environment-startup.ts index c569a25c35f9..d1359df68e67 100644 --- a/src/gateway/server-worker-environment-startup.ts +++ b/src/gateway/server-worker-environment-startup.ts @@ -1,3 +1,4 @@ +import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; import { getRuntimeConfig } from "../config/config.js"; import { loadOrCreateProcessDeviceIdentity } from "../infra/device-identity.js"; @@ -13,6 +14,7 @@ import type { NodeWorkerSupervisorTransport } from "./node-registry-private.js"; import type { WorkerBundleProducer, WorkerNpmArtifact } from "./worker-environments/bundle.js"; import { bindDeviceWorkerAvailability, + bindDeviceWorkerReconciliation, createDeviceWorkerRuntime, DEVICE_WORKER_PROVIDER_ID, } from "./worker-environments/device-provider.js"; @@ -193,6 +195,7 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { let dispatchChild: WorkerPlacementDispatchContract["dispatch"] = async () => { throw new Error("Worker session dispatch is unavailable"); }; + const workerEnvironmentLog = params.log.child("worker-environments"); const workerEnvironmentServiceBase = createWorkerEnvironmentService({ store: params.startup.store, getConfig: getRuntimeConfig, @@ -253,10 +256,37 @@ export async function createGatewayWorkerEnvironmentRuntime(params: { { signal, resolveIdentity }, ); }, - logger: params.log.child("worker-environments"), + logger: workerEnvironmentLog, }); const workerEnvironmentService = workerEnvironmentServiceBase; bindDeviceWorkerAvailability(workerEnvironmentService, deviceRuntime.isAvailable); + bindDeviceWorkerReconciliation(workerEnvironmentService, async (deviceId) => { + const environmentIds = params.startup.store + .listForReconcile() + .filter((record) => { + const settings = record.profileSnapshot.settings; + const profileDeviceId = isRecord(settings) ? settings.device : undefined; + return ( + record.providerId === DEVICE_WORKER_PROVIDER_ID && + typeof profileDeviceId === "string" && + profileDeviceId.trim() === deviceId + ); + }) + .map((record) => record.environmentId); + for (const environmentId of environmentIds) { + params.startup.store.revokeEnvironmentCredential(environmentId); + } + await Promise.all( + environmentIds.map(async (environmentId) => { + await workerEnvironmentService.reconcileEnvironment(environmentId).catch(() => { + workerEnvironmentLog.warn( + `Device worker reconcile failed (${deviceId}, ${environmentId}); periodic cleanup will retry`, + ); + }); + }), + ); + return environmentIds; + }); executeSessionTool = createWorkerSessionToolExecutor({ placements: params.startup.placementStore, environments: workerEnvironmentService, diff --git a/src/gateway/worker-environments/device-provider.test.ts b/src/gateway/worker-environments/device-provider.test.ts index 7c7a752fd6ab..cad80ca5298c 100644 --- a/src/gateway/worker-environments/device-provider.test.ts +++ b/src/gateway/worker-environments/device-provider.test.ts @@ -7,7 +7,11 @@ import type { PairedDevice } from "../../infra/device-pairing.types.js"; import { NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE } from "../../infra/node-runner-inventory.js"; import { WorkerProviderError } from "../../plugins/types.js"; import type { NodeWorkerSupervisorNodeProof } from "../node-registry-private.js"; -import { createDeviceWorkerRuntime } from "./device-provider.js"; +import { + bindDeviceWorkerReconciliation, + createDeviceWorkerRuntime, + reconcileDeviceWorker, +} from "./device-provider.js"; const DEVICE_ID = "device-session-host"; const DAY_MS = 24 * 60 * 60 * 1_000; @@ -76,6 +80,17 @@ function deviceRuntime(params: { } describe("device worker provider", () => { + it("binds targeted device reconciliation to the active worker service", async () => { + const service = {}; + const reconcile = async (deviceId: string) => [`environment:${deviceId}`]; + + await expect(reconcileDeviceWorker(service, DEVICE_ID)).resolves.toEqual([]); + bindDeviceWorkerReconciliation(service, reconcile); + await expect(reconcileDeviceWorker(service, DEVICE_ID)).resolves.toEqual([ + `environment:${DEVICE_ID}`, + ]); + }); + it("provisions deterministic node leases only for connected paired session hosts", async () => { const provider = deviceRuntime({ getPairedDevice: async (deviceId) => pairedDevice(deviceId), diff --git a/src/gateway/worker-environments/device-provider.ts b/src/gateway/worker-environments/device-provider.ts index cd36d50cfdf3..809112c39fb1 100644 --- a/src/gateway/worker-environments/device-provider.ts +++ b/src/gateway/worker-environments/device-provider.ts @@ -21,7 +21,9 @@ type DeviceWorkerRuntimeOptions = { }; type DeviceWorkerAvailability = (deviceId: string) => Promise; +type DeviceWorkerReconciliation = (deviceId: string) => Promise; const DEVICE_WORKER_AVAILABILITY = new WeakMap(); +const DEVICE_WORKER_RECONCILIATION = new WeakMap(); export function bindDeviceWorkerAvailability( service: object, @@ -38,6 +40,21 @@ export async function isDeviceWorkerAvailable( return isAvailable ? await isAvailable(deviceId) : false; } +export function bindDeviceWorkerReconciliation( + service: object, + reconcile: DeviceWorkerReconciliation, +): void { + DEVICE_WORKER_RECONCILIATION.set(service, reconcile); +} + +export async function reconcileDeviceWorker( + service: object | undefined, + deviceId: string, +): Promise { + const reconcile = service ? DEVICE_WORKER_RECONCILIATION.get(service) : undefined; + return reconcile ? await reconcile(deviceId) : []; +} + function requireDeviceId(profile: WorkerProfile): string { const deviceId = profile.device; if (typeof deviceId !== "string" || !deviceId.trim()) { diff --git a/src/gateway/worker-environments/provider-reconciliation.test.ts b/src/gateway/worker-environments/provider-reconciliation.test.ts index e011b25c03bd..7b926d97e1ea 100644 --- a/src/gateway/worker-environments/provider-reconciliation.test.ts +++ b/src/gateway/worker-environments/provider-reconciliation.test.ts @@ -67,6 +67,41 @@ describe("worker environment service", () => { expect(support.testState.bootstrapWorker).toHaveBeenCalledTimes(1); }); + it("reconciles one exact environment without sweeping its siblings", async () => { + support.seedReady("worker-target"); + support.seedReady("worker-sibling"); + const inspected: string[] = []; + const workerService = support.createService( + support.createProvider({ + inspect: async (lease) => { + inspected.push(lease.leaseId); + return { status: "active" }; + }, + }), + ); + + await workerService.reconcileEnvironment("worker-target"); + + expect(inspected).toEqual(["lease:worker-target"]); + }); + + it("targeted reconciliation revokes a disappeared worker credential", async () => { + const environmentId = "worker-revoked"; + support.seedReady(environmentId); + const workerService = support.createService( + support.createProvider({ inspect: async () => ({ status: "unknown" }) }), + ); + const admitted = await workerService.admitWorker(support.admissionFor(environmentId)); + if (!admitted.ok) { + throw new Error("fixture worker admission failed"); + } + + await workerService.reconcileEnvironment(environmentId); + + expect(support.testState.store.get(environmentId)?.state).toBe("orphaned"); + expect(workerService.validateWorkerConnection(admitted.identity)).toBe("credential-replaced"); + }); + it("skips an active lease whose durable receipt matches the lifecycle bundle", async () => { support.seedReady("worker-current"); diff --git a/src/gateway/worker-environments/service.ts b/src/gateway/worker-environments/service.ts index c2a2c4e50da8..a408bc2da8e1 100644 --- a/src/gateway/worker-environments/service.ts +++ b/src/gateway/worker-environments/service.ts @@ -318,23 +318,30 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService withLock, }); + const reconcileEnvironment = async (environmentId: string) => { + if (stopping) { + return; + } + await withLock(environmentId, async () => { + const current = store.get(environmentId); + if (!current || inState(current, "destroyed", "failed", "orphaned")) { + return; + } + await providerLifecycle.reconcileRecord(current); + }); + }; + const reconcilePass = async () => { - const tasks = store.listForReconcile().map( - (candidate) => () => - withLock(candidate.environmentId, async () => { - const current = store.get(candidate.environmentId); - if (!current || inState(current, "destroyed", "failed")) { - return; - } - await providerLifecycle - .reconcileRecord(current) - .catch(() => - warn( - `Worker environment reconcile failed (${current.environmentId}, ${current.providerId})`, - ), - ); - }), - ); + const tasks = store + .listForReconcile() + .map( + (candidate) => () => + reconcileEnvironment(candidate.environmentId).catch(() => + warn( + `Worker environment reconcile failed (${candidate.environmentId}, ${candidate.providerId})`, + ), + ), + ); await runTasksWithConcurrency({ tasks, limit: 8 }); store.pruneTerminalEnvironments(); }; @@ -430,6 +437,7 @@ export function createWorkerEnvironmentService(options: WorkerEnvironmentService acknowledgeCredentialDelivery: credentialBroker.acknowledgeCredentialDelivery, startTunnel: environmentAccess.startTunnel, stopTunnel: environmentAccess.stopTunnel, + reconcileEnvironment, reconcileOnce, start, stop, diff --git a/src/gateway/worker-environments/store.test.ts b/src/gateway/worker-environments/store.test.ts index beded62af86e..ebed8e72f3a9 100644 --- a/src/gateway/worker-environments/store.test.ts +++ b/src/gateway/worker-environments/store.test.ts @@ -644,6 +644,22 @@ describe("worker environment store", () => { ).toThrow("owner epoch changed"); }); + it("revokes one environment credential without changing lifecycle state", () => { + const bootstrapping = seedBootstrapping("worker-revocation", "lease-revocation"); + store.transition({ + environmentId: bootstrapping.environmentId, + from: bootstrapping.state, + to: "ready", + patch: readyPatch(), + }); + expect(store.getCredential(bootstrapping.environmentId)).toBeDefined(); + + store.revokeEnvironmentCredential(bootstrapping.environmentId); + + expect(store.getCredential(bootstrapping.environmentId)).toBeUndefined(); + expect(store.get(bootstrapping.environmentId)?.state).toBe("ready"); + }); + it("allocates globally distinct owner epochs when a session moves environments", () => { const makeReady = (environmentId: string, leaseId: string) => { const bootstrapping = seedBootstrapping(environmentId, leaseId); diff --git a/src/gateway/worker-environments/store.ts b/src/gateway/worker-environments/store.ts index e6e9b8f32800..5881335384e6 100644 --- a/src/gateway/worker-environments/store.ts +++ b/src/gateway/worker-environments/store.ts @@ -900,6 +900,9 @@ export function createWorkerEnvironmentStore( getCredential: (environmentId: string) => findCredential(read(), required(environmentId, "id")), getTransferOwner: (environmentId: string) => findTransferOwner(read(), required(environmentId, "id")), + revokeEnvironmentCredential(environmentId: string): void { + return write((db) => revokeCredential(db, required(environmentId, "id"))); + }, findCredentialByHash: (credentialHash: string) => findCredentialByHash(read(), normalizeCredentialHash(credentialHash)), list: (): WorkerEnvironmentRecord[] => listRows(read(), false),