diff --git a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewayProtocol.kt b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewayProtocol.kt index 71ab52275533..2b1f6b151f2a 100644 --- a/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewayProtocol.kt +++ b/apps/android/app/src/main/java/ai/openclaw/app/gateway/GatewayProtocol.kt @@ -430,6 +430,7 @@ enum class GatewayMethod( NodePluginSurfaceRefresh("node.pluginSurface.refresh"), NodePluginToolsUpdate("node.pluginTools.update"), NodeSkillsUpdate("node.skills.update"), + NodeProtocolFeaturesUpdate("node.protocolFeatures.update"), NodePendingDrain("node.pending.drain"), NodePendingEnqueue("node.pending.enqueue"), NodeInvoke("node.invoke"), diff --git a/src/gateway/node-registry-private.ts b/src/gateway/node-registry-private.ts index 93a1bd2b19cd..b4681a3017f2 100644 --- a/src/gateway/node-registry-private.ts +++ b/src/gateway/node-registry-private.ts @@ -13,10 +13,27 @@ import type { PendingInvoke, PendingSystemRunEvent, } from "./node-registry.invoke-stream.js"; -import type { NodeInvokeResult, NodeRegistry, NodeSession } from "./node-registry.js"; import { normalizeSystemRunTimeoutMs } from "./node-registry.system-run.js"; -type PairingBoundNodeSession = NodeSession & { pairingIdentity: string }; +type NodeRegistryPrivateSession = { + nodeId: string; + connId: string; + pairingIdentity?: string; + pairingGeneration?: string; + client: { invalidated?: boolean }; + clientId?: string; + clientMode?: string; + commands: string[]; +}; + +type NodeInvokeResult = { + ok: boolean; + payload?: unknown; + payloadJSON?: string | null; + error?: { code?: string; message?: string } | null; +}; + +type PairingBoundNodeSession = NodeRegistryPrivateSession & { pairingIdentity: string }; type PairingLeaseResolution = | { status: "current"; session: PairingBoundNodeSession } | { status: "stale"; presenceInvalidated: boolean } @@ -71,12 +88,16 @@ type NodeProtocolFeatureProof = Omit type NodeRegistryPrivateContext = { getNode: (nodeId: string) => PairingBoundNodeSession | undefined; - listCurrentConnected: () => Promise; + listCurrentConnected: () => Promise; hasCurrentPairingStateResolver: boolean; resolvePairingLease: (node: PairingBoundNodeSession) => Promise; pendingInvokes: Map; invokeStreams: NodeInvokeStreamController; - sendEventToSession: (node: NodeSession, event: string, payload: unknown) => boolean; + sendEventToSession: ( + node: NodeRegistryPrivateSession, + event: string, + payload: unknown, + ) => boolean; rememberAuthorizedSystemRunEvent: (event: { nodeId: string; connId: string; @@ -105,7 +126,7 @@ type NodeRegistryPrivateState = { workerSupervisorTransport: NodeWorkerSupervisorTransport; }; -const NODE_REGISTRY_PRIVATE_STATES = new WeakMap(); +const NODE_REGISTRY_PRIVATE_STATES = new WeakMap(); function resolvePendingSystemRunEvent(params: { command: string; @@ -152,7 +173,7 @@ function normalizeSystemRunInvokeParams(params: { command: string; params?: unkn } function resolveWorkerSupervisorProof( - node: NodeSession, + node: NodeRegistryPrivateSession, protocolFeaturesByConn: ReadonlyMap, ): NodeWorkerSupervisorNodeProof | undefined { const proof = protocolFeaturesByConn.get(node.connId); @@ -405,7 +426,7 @@ async function invokeNodeRegistryCore( } export function registerNodeRegistryPrivateRuntime( - nodeRegistry: NodeRegistry, + nodeRegistry: object, context: NodeRegistryPrivateContext, ): void { const state = {} as NodeRegistryPrivateState; @@ -461,8 +482,10 @@ export function registerNodeRegistryPrivateRuntime( NODE_REGISTRY_PRIVATE_STATES.set(nodeRegistry, state); } -export function createNodeRegistryRuntime(create: () => NodeRegistry): { - nodeRegistry: NodeRegistry; +export function createNodeRegistryRuntime( + create: () => TRegistry, +): { + nodeRegistry: TRegistry; nodeWorkerSupervisorTransport: NodeWorkerSupervisorTransport; } { const nodeRegistry = create(); @@ -477,7 +500,7 @@ export function createNodeRegistryRuntime(create: () => NodeRegistry): { } export function invokePublicNodeRegistry( - nodeRegistry: NodeRegistry, + nodeRegistry: object, params: NodeInvokeParams, ): Promise { const state = NODE_REGISTRY_PRIVATE_STATES.get(nodeRegistry); @@ -488,7 +511,7 @@ export function invokePublicNodeRegistry( } export function updateNodeWorkerSupervisorProtocolFeatures(params: { - registry: NodeRegistry; + registry: object; nodeId: string; connId: string | undefined; protocolFeatures: readonly string[]; @@ -503,16 +526,16 @@ export function updateNodeWorkerSupervisorProtocolFeatures(params: { } export function forgetNodeWorkerSupervisorProtocolFeatures( - nodeRegistry: NodeRegistry, + nodeRegistry: object, connId: string, ): void { NODE_REGISTRY_PRIVATE_STATES.get(nodeRegistry)?.protocolFeaturesByConn.delete(connId); } export function isNodeRegistryPendingInvokeConnectionActive(params: { - registry: NodeRegistry; + registry: object; pending: PendingInvoke; - currentNode: NodeSession | undefined; + currentNode: NodeRegistryPrivateSession | undefined; }): boolean { const state = NODE_REGISTRY_PRIVATE_STATES.get(params.registry); const binding = state?.generationBoundInvokes.get(params.pending); @@ -523,7 +546,7 @@ export function isNodeRegistryPendingInvokeConnectionActive(params: { } export function settleNodeRegistryPairingGenerationChange(params: { - registry: NodeRegistry; + registry: object; nodeId: string; connId: string; nextPairingGeneration: string; diff --git a/src/gateway/node-registry.ts b/src/gateway/node-registry.ts index 79e7768c6d21..ee5420ffe083 100644 --- a/src/gateway/node-registry.ts +++ b/src/gateway/node-registry.ts @@ -286,13 +286,28 @@ export class NodeRegistry { getNode: (nodeId) => this.nodesById.get(nodeId), listCurrentConnected: () => this.listCurrentConnected(), hasCurrentPairingStateResolver: Boolean(this.options.resolveCurrentPairingState), - resolvePairingLease: async (node) => - await this.resolvePairingLease(this.capturePairingLease(node), { + resolvePairingLease: async (node) => { + const current = this.nodesById.get(node.nodeId); + if ( + !current || + current.connId !== node.connId || + current.pairingIdentity !== node.pairingIdentity || + current.pairingGeneration !== node.pairingGeneration + ) { + return { status: "stale", presenceInvalidated: false }; + } + return await this.resolvePairingLease(this.capturePairingLease(current), { invalidateStale: false, - }), + }); + }, pendingInvokes: this.pendingInvokes, invokeStreams: this.invokeStreams, - sendEventToSession: (node, event, payload) => this.sendEventToSession(node, event, payload), + sendEventToSession: (node, event, payload) => { + const current = this.nodesById.get(node.nodeId); + return current?.connId === node.connId + ? this.sendEventToSession(current, event, payload) + : false; + }, rememberAuthorizedSystemRunEvent: (event) => this.rememberAuthorizedSystemRunEvent(event), publishActiveNodeContext: () => this.publishActiveNodeContext(), }); diff --git a/src/node-host/node-worker-launch-store.ts b/src/node-host/node-worker-launch-store.ts index 5d2a3b7b9786..877a6c125867 100644 --- a/src/node-host/node-worker-launch-store.ts +++ b/src/node-host/node-worker-launch-store.ts @@ -15,7 +15,7 @@ import { inspectNodeWorkerProcessIdentity, type NodeWorkerProcessIdentity, } from "./node-worker-process-identity.js"; -import type { NodeWorkerSupervisorIdentity } from "./node-worker-supervisor-contract.js"; +import type { NodeWorkerSupervisorIdentity } from "./node-worker-supervisor-identity.js"; type NodeWorkerLaunchState = | "pending" diff --git a/src/node-host/node-worker-supervisor-contract.ts b/src/node-host/node-worker-supervisor-contract.ts index daeaddca790a..d23ef6a53746 100644 --- a/src/node-host/node-worker-supervisor-contract.ts +++ b/src/node-host/node-worker-supervisor-contract.ts @@ -6,6 +6,9 @@ import { type WorkerLaunchDescriptor, } from "../worker/launch-descriptor.js"; import type { NodeWorkerLaunchReceipt } from "./node-worker-launch-store.js"; +import type { NodeWorkerSupervisorIdentity } from "./node-worker-supervisor-identity.js"; + +export type { NodeWorkerSupervisorIdentity } from "./node-worker-supervisor-identity.js"; const IDENTIFIER_MAX_CHARS = 256; const GATEWAY_NAMESPACE_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/u; @@ -19,16 +22,6 @@ export type NodeWorkerLaunchInput = { descriptor: WorkerLaunchDescriptor; }; -export type NodeWorkerSupervisorIdentity = { - launchId: string; - planHash: string; - environmentId: string; - sessionId: string; - ownerEpoch: number; - placementGeneration: number; - runId: string; -}; - export type NodeWorkerSupervisorReceipt = NodeWorkerSupervisorIdentity & { state: "pending" | "running" | "completed" | "failed" | "interrupted" | "cancelled"; }; diff --git a/src/node-host/node-worker-supervisor-identity.ts b/src/node-host/node-worker-supervisor-identity.ts new file mode 100644 index 000000000000..9176ee6e948e --- /dev/null +++ b/src/node-host/node-worker-supervisor-identity.ts @@ -0,0 +1,9 @@ +export type NodeWorkerSupervisorIdentity = { + launchId: string; + planHash: string; + environmentId: string; + sessionId: string; + ownerEpoch: number; + placementGeneration: number; + runId: string; +};