Skip to content

Commit 8b86645

Browse files
vincentkocopenclaw-clownfish[bot]
vincentkoc
and
openclaw-clownfish[bot]
authored
fix(macos): keep node invokes responsive during system.run (#100842)
* fix(macos): keep app node connected after system.run invoke * fix(macos): document node invoke receive isolation --------- Co-authored-by: openclaw-clownfish[bot] <280122609+openclaw-clownfish[bot]@users.noreply.github.com>
1 parent aabf686 commit 8b86645

2 files changed

Lines changed: 139 additions & 17 deletions

File tree

‎apps/shared/OpenClawKit/Sources/OpenClawKit/GatewayNodeSession.swift‎

Lines changed: 34 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -541,24 +541,45 @@ public actor GatewayNodeSession {
541541
command: request.command,
542542
paramsJSON: request.paramsJSON,
543543
nodeId: request.nodeId)
544-
self.logger.info("node invoke executing id=\(request.id, privacy: .public)")
545-
let response = await Self.invokeWithTimeout(
546-
request: req,
547-
timeoutMs: request.timeoutMs,
548-
onInvoke: onInvoke)
549-
// Invoke output belongs to the requesting channel. A target switch while the device
550-
// command is running must discard it instead of disclosing it to the replacement.
551-
guard self.channelGeneration == channelGeneration,
552-
self.channel === channel
553-
else { return }
554-
self.logger.info(
555-
"node invoke completed id=\(request.id, privacy: .public) ok=\(response.ok, privacy: .public)")
556-
await self.sendInvokeResult(request: request, response: response, channel: channel)
544+
// GatewayChannel waits for push handling before it rearms receive. Run device work
545+
// separately so a long invoke cannot starve heartbeats or later node requests.
546+
Task { [weak self] in
547+
await self?.handleInvokeRequest(
548+
request: request,
549+
bridgeRequest: req,
550+
timeoutMs: request.timeoutMs,
551+
onInvoke: onInvoke,
552+
channel: channel,
553+
channelGeneration: channelGeneration)
554+
}
557555
} catch {
558556
self.logger.error("node invoke decode failed: \(error.localizedDescription, privacy: .public)")
559557
}
560558
}
561559

560+
private func handleInvokeRequest(
561+
request: NodeInvokeRequestPayload,
562+
bridgeRequest: BridgeInvokeRequest,
563+
timeoutMs: Int?,
564+
onInvoke: @escaping @Sendable (BridgeInvokeRequest) async -> BridgeInvokeResponse,
565+
channel: GatewayChannelActor,
566+
channelGeneration: UInt64) async
567+
{
568+
self.logger.info("node invoke executing id=\(request.id, privacy: .public)")
569+
let response = await Self.invokeWithTimeout(
570+
request: bridgeRequest,
571+
timeoutMs: timeoutMs,
572+
onInvoke: onInvoke)
573+
// Invoke output belongs to the requesting channel. A target switch while the device
574+
// command is running must discard it instead of disclosing it to the replacement.
575+
guard self.channelGeneration == channelGeneration,
576+
self.channel === channel
577+
else { return }
578+
self.logger.info(
579+
"node invoke completed id=\(request.id, privacy: .public) ok=\(response.ok, privacy: .public)")
580+
await self.sendInvokeResult(request: request, response: response, channel: channel)
581+
}
582+
562583
private func decodeInvokeRequest(from payload: OpenClawProtocol.AnyCodable) throws -> NodeInvokeRequestPayload {
563584
do {
564585
let data = try self.encoder.encode(payload)

‎apps/shared/OpenClawKit/Tests/OpenClawKitTests/GatewayNodeSessionTests.swift‎

Lines changed: 105 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -94,6 +94,7 @@ private final class FakeGatewayWebSocketTask: WebSocketTasking, @unchecked Senda
9494
private var connectAuth: [String: Any]?
9595
private var connectDevice: [String: Any]?
9696
private var sentRequestMethods: [String] = []
97+
private var sentRequestPayloads: [[String: Any]] = []
9798
private var receivePhase = 0
9899
private var pendingReceiveHandler:
99100
(@Sendable (Result<URLSessionWebSocketTask.Message, Error>) -> Void)?
@@ -142,7 +143,10 @@ private final class FakeGatewayWebSocketTask: WebSocketTasking, @unchecked Senda
142143
obj["type"] as? String == "req",
143144
let method = obj["method"] as? String
144145
{
145-
self.lock.withLock { self.sentRequestMethods.append(method) }
146+
self.lock.withLock {
147+
self.sentRequestMethods.append(method)
148+
self.sentRequestPayloads.append(obj)
149+
}
146150
guard method == "connect", let id = obj["id"] as? String else { return }
147151
let params = obj["params"] as? [String: Any]
148152
let auth = (params?["auth"] as? [String: Any]) ?? [:]
@@ -167,6 +171,16 @@ private final class FakeGatewayWebSocketTask: WebSocketTasking, @unchecked Senda
167171
self.lock.withLock { self.sentRequestMethods.count(where: { $0 == method }) }
168172
}
169173

174+
func sentRequests(method: String) -> [[String: Any]] {
175+
self.lock.withLock {
176+
self.sentRequestPayloads.filter { $0["method"] as? String == method }
177+
}
178+
}
179+
180+
func hasPendingReceiveHandler() -> Bool {
181+
self.lock.withLock { self.pendingReceiveHandler != nil }
182+
}
183+
170184
func sendPing(pongReceiveHandler: @escaping @Sendable (Error?) -> Void) {
171185
pongReceiveHandler(nil)
172186
}
@@ -215,14 +229,21 @@ private final class FakeGatewayWebSocketTask: WebSocketTasking, @unchecked Senda
215229
}
216230

217231
func emitInvokeRequest(id: String, command: String) {
232+
self.emitInvokeRequest(id: id, command: command, paramsJSON: "{}")
233+
}
234+
235+
func emitInvokeRequest(id: String, command: String, paramsJSON: String?) {
218236
let handler = self.lock.withLock { () -> (@Sendable (Result<
219237
URLSessionWebSocketTask.Message,
220238
Error,
221239
>) -> Void)? in
222240
defer { self.pendingReceiveHandler = nil }
223241
return self.pendingReceiveHandler
224242
}
225-
handler?(.success(.data(Self.invokeRequestData(id: id, command: command))))
243+
handler?(.success(.data(Self.invokeRequestData(
244+
id: id,
245+
command: command,
246+
paramsJSON: paramsJSON))))
226247
}
227248

228249
private static func connectChallengeData(nonce: String) -> Data {
@@ -284,15 +305,15 @@ private final class FakeGatewayWebSocketTask: WebSocketTasking, @unchecked Senda
284305
return (try? JSONSerialization.data(withJSONObject: frame)) ?? Data()
285306
}
286307

287-
private static func invokeRequestData(id: String, command: String) -> Data {
308+
private static func invokeRequestData(id: String, command: String, paramsJSON: String?) -> Data {
288309
let frame: [String: Any] = [
289310
"type": "event",
290311
"event": "node.invoke.request",
291312
"payload": [
292313
"id": id,
293314
"nodeId": "test-node",
294315
"command": command,
295-
"paramsJSON": "{}",
316+
"paramsJSON": paramsJSON ?? NSNull(),
296317
],
297318
]
298319
return (try? JSONSerialization.data(withJSONObject: frame)) ?? Data()
@@ -620,6 +641,86 @@ struct GatewayNodeSessionTests {
620641
await gateway.disconnect()
621642
}
622643

644+
@Test
645+
func `node invoke requests keep receiving while system run is blocked`() async throws {
646+
let session = FakeGatewayWebSocketSession()
647+
let gateway = GatewayNodeSession()
648+
let systemRunStarted = AsyncStream<Void>.makeStream()
649+
var startedIterator = systemRunStarted.stream.makeAsyncIterator()
650+
let systemRunRelease = AsyncStream<Void>.makeStream()
651+
let options = GatewayConnectOptions(
652+
role: "node",
653+
scopes: [],
654+
caps: [],
655+
commands: ["system.run"],
656+
permissions: [:],
657+
clientId: "openclaw-macos",
658+
clientMode: "node",
659+
clientDisplayName: "macOS Test",
660+
includeDeviceIdentity: false)
661+
662+
try await gateway.connect(
663+
url: #require(URL(string: "ws://example.invalid")),
664+
token: nil,
665+
bootstrapToken: nil,
666+
password: nil,
667+
connectOptions: options,
668+
sessionBox: WebSocketSessionBox(session: session),
669+
onConnected: {},
670+
onDisconnected: { _ in },
671+
onInvoke: { request in
672+
if request.id == "system-run-blocked" {
673+
systemRunStarted.continuation.yield()
674+
for await _ in systemRunRelease.stream {
675+
return BridgeInvokeResponse(
676+
id: request.id,
677+
ok: false,
678+
error: OpenClawNodeError(
679+
code: .unavailable,
680+
message: "UNSUPPORTED: system.run unavailable"))
681+
}
682+
}
683+
return BridgeInvokeResponse(id: request.id, ok: true, payloadJSON: #"{"ok":true}"#)
684+
})
685+
let task = try #require(session.latestTask())
686+
687+
task.emitInvokeRequest(
688+
id: "system-run-blocked",
689+
command: "system.run",
690+
paramsJSON: #"{"command":["/bin/echo","ok"]}"#)
691+
_ = await startedIterator.next()
692+
try await waitUntil("receive loop rearmed during system.run") {
693+
task.hasPendingReceiveHandler()
694+
}
695+
task.emitInvokeRequest(id: "camera-after-system-run", command: "camera.snap")
696+
697+
try await waitUntil("second invoke result while system.run is blocked") {
698+
task.sentRequestCount(method: "node.invoke.result") == 1
699+
}
700+
let earlyResults = task.sentRequests(method: "node.invoke.result")
701+
#expect(earlyResults.count == 1)
702+
let earlyParams = try #require(earlyResults.first?["params"] as? [String: Any])
703+
#expect(earlyParams["id"] as? String == "camera-after-system-run")
704+
#expect(earlyParams["ok"] as? Bool == true)
705+
706+
systemRunRelease.continuation.yield()
707+
systemRunRelease.continuation.finish()
708+
try await waitUntil("blocked system.run result") {
709+
task.sentRequestCount(method: "node.invoke.result") == 2
710+
}
711+
let finalResults = task.sentRequests(method: "node.invoke.result")
712+
#expect(finalResults.count == 2)
713+
let blockedResult = try #require(finalResults.first {
714+
($0["params"] as? [String: Any])?["id"] as? String == "system-run-blocked"
715+
})
716+
let blockedParams = try #require(blockedResult["params"] as? [String: Any])
717+
#expect(blockedParams["ok"] as? Bool == false)
718+
let error = try #require(blockedParams["error"] as? [String: Any])
719+
#expect(error["code"] as? String == OpenClawNodeErrorCode.unavailable.rawValue)
720+
721+
await gateway.disconnect()
722+
}
723+
623724
@Test(.stateDirectoryIsolated)
624725
func `scanned setup code prefers bootstrap auth over stored device token`() async throws {
625726
let tempDir = FileManager.default.temporaryDirectory

0 commit comments

Comments
 (0)