import Darwin import Foundation import OpenClawKit import OpenClawProtocol import Testing @testable import OpenClaw private struct WorkerBackpressureTimeout: Error {} private actor StubMacNodeHostWorker: MacNodeHostWorking { let manifest: MacNodeHostManifest private var requests: [BridgeInvokeRequest] = [] init(commands: [String] = ["system.run", "mcp.tools.call.v1"]) { self.manifest = MacNodeHostManifest( version: "test", caps: ["system", "mcp"], commands: commands, pathEnv: "/usr/bin:/bin") } func start(launch _: MacNodeHostWorkerLaunch) async throws -> MacNodeHostManifest { self.manifest } func supports(_ command: String) async -> Bool { self.manifest.commands.contains(command) } func invoke(_ request: BridgeInvokeRequest) async -> BridgeInvokeResponse { self.requests.append(request) return BridgeInvokeResponse(id: request.id, ok: true, payloadJSON: #"{"owner":"cli"}"#) } func handleInput(invokeId _: String, seq _: Int, payloadJSON _: String) async {} func cancel(invokeId _: String) async {} func setRoute(_: GatewayNodeSessionRoute?, authorityGeneration _: UInt64) async -> Bool { true } func publishInventory(ifCurrentRoute _: GatewayNodeSessionRoute) async {} func stop() async {} func invokedCommands() -> [String] { self.requests.map(\.command) } } @Suite(.serialized) struct MacNodeHostWorkerTests { @Test func `worker crash retry budget is bounded and exponentially delayed`() throws { let input = MacNodeHostWorkerRetryPolicy.Input( launch: MacNodeHostWorkerLaunch( command: ["/usr/local/bin/openclaw", "node", "worker"], configurationGeneration: 4)) var policy = MacNodeHostWorkerRetryPolicy(maximumRetryCount: 5) try policy.prepareForStart(input) let dispositions = (0..<20).map { _ in policy.recordUnexpectedExit(for: input) } #expect(dispositions.prefix(5) == [ .retry(attempt: 1, delayNanoseconds: 1_000_000_000), .retry(attempt: 2, delayNanoseconds: 2_000_000_000), .retry(attempt: 3, delayNanoseconds: 4_000_000_000), .retry(attempt: 4, delayNanoseconds: 8_000_000_000), .retry(attempt: 5, delayNanoseconds: 10_000_000_000), ]) #expect(dispositions.dropFirst(5).allSatisfy { $0 == .giveUp(unexpectedExitCount: 6) }) #expect(throws: MacNodeHostWorkerRetryPolicy.RetryBudgetExhausted.self) { try policy.prepareForStart(input) } } @Test func `new worker input resets an exhausted crash retry budget`() throws { let original = MacNodeHostWorkerRetryPolicy.Input( launch: MacNodeHostWorkerLaunch( command: ["/usr/local/bin/openclaw", "node", "worker"], configurationGeneration: 4)) let updated = MacNodeHostWorkerRetryPolicy.Input( launch: MacNodeHostWorkerLaunch( command: original.launch.command, configurationGeneration: 5)) var policy = MacNodeHostWorkerRetryPolicy(maximumRetryCount: 1) try policy.prepareForStart(original) #expect(policy.recordUnexpectedExit(for: original) == .retry(attempt: 1, delayNanoseconds: 1_000_000_000)) #expect(policy.recordUnexpectedExit(for: original) == .giveUp(unexpectedExitCount: 2)) try policy.prepareForStart(updated) #expect(policy.recordUnexpectedExit(for: updated) == .retry(attempt: 1, delayNanoseconds: 1_000_000_000)) } @Test func `worker allows a generous cold-start window`() async throws { #expect(MacNodeHostWorker.defaultStartupTimeout == 300) let worker = MacNodeHostWorker(session: GatewayNodeSession(), startupTimeout: 1) let script = """ sleep 0.1 printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"/usr/bin:/bin"}}' while IFS= read -r line; do :; done """ let manifest = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script])) #expect(manifest.version == "test") await worker.stop() } @Test func `worker launches in the selected checkout`() async throws { let checkout = try makeTempDirForTests().resolvingSymlinksInPath() let worker = MacNodeHostWorker(session: GatewayNodeSession(), startupTimeout: 1) let script = """ printf '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"%s"}}\\n' \ "$(/bin/pwd -P)" while IFS= read -r line; do :; done """ let manifest = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script], currentDirectoryURL: checkout)) let expectedCurrentDirectory = try #require(realpath(checkout.path, nil)) defer { free(expectedCurrentDirectory) } #expect(manifest.pathEnv == String(cString: expectedCurrentDirectory)) await worker.stop() } @Test(arguments: [ OpenClawSystemCommand.run.rawValue, "mcp.tools.call.v1", "codex.terminal.resume.v1", ]) func `Mac runtime forwards worker-owned commands to the shared worker`(command: String) async { let worker = StubMacNodeHostWorker(commands: [command]) let runtime = MacNodeRuntime(nodeHostWorker: worker) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "worker-run", command: command, paramsJSON: #"{"command":["/usr/bin/true"]}"#)) #expect(response.ok) #expect(response.payloadJSON == #"{"owner":"cli"}"#) #expect(await worker.invokedCommands() == [command]) } @Test(arguments: [MacNodeScreenCommand.snapshot.rawValue, OpenClawComputerCommand.act.rawValue]) func `selected CUA provider gives the command pair exclusively to the worker`(command: String) async { let worker = StubMacNodeHostWorker(commands: [command]) let runtime = MacNodeRuntime( nodeHostWorker: worker, computerControlEnabled: { true }, computerControlProvider: { .cua }) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "cua-owned", command: command, paramsJSON: "{}")) #expect(response.ok) #expect(response.payloadJSON == #"{"owner":"cli"}"#) #expect(await worker.invokedCommands() == [command]) } @Test func `selected CUA provider never falls back to native snapshot`() async { let worker = StubMacNodeHostWorker(commands: []) let runtime = MacNodeRuntime( nodeHostWorker: worker, computerControlEnabled: { true }, computerControlProvider: { .cua }) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "cua-unavailable", command: MacNodeScreenCommand.snapshot.rawValue)) #expect(!response.ok) #expect(response.error?.message == "UNAVAILABLE: selected CUA provider is not ready") #expect(await worker.invokedCommands().isEmpty) } @Test(arguments: [ MacNodeCodexThreadCatalogContract.listCommand, MacNodeCodexThreadCatalogContract.turnsCommand, MacNodeClaudeSessionCatalogContract.listCommand, MacNodeClaudeSessionCatalogContract.readCommand, ]) func `native session catalogs own commands shared with the worker`(command: String) async { let worker = StubMacNodeHostWorker(commands: [command]) let payload = #"{"owner":"native"}"# let runtime = MacNodeRuntime( nodeHostWorker: worker, codexThreadCatalogEnabled: { true }, codexThreadListRequest: { _ in payload }, codexThreadTurnsRequest: { _ in payload }, claudeSessionCatalogEnabled: { true }, claudeSessionListRequest: { _ in payload }, claudeSessionReadRequest: { _ in payload }) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "native-catalog", command: command, paramsJSON: #"{"limit":1}"#)) #expect(response.ok) #expect(response.payloadJSON == payload) #expect(await worker.invokedCommands().isEmpty) } @Test(arguments: [ ( MacNodeCodexThreadCatalogContract.listCommand, "UNAVAILABLE: Codex session catalog is disabled"), ( MacNodeCodexThreadCatalogContract.turnsCommand, "UNAVAILABLE: Codex session catalog is disabled"), ( MacNodeClaudeSessionCatalogContract.listCommand, "UNAVAILABLE: Claude session catalog is disabled"), ( MacNodeClaudeSessionCatalogContract.readCommand, "UNAVAILABLE: Claude session catalog is disabled"), ( OpenClawComputerCommand.act.rawValue, "COMPUTER_DISABLED: enable Computer Control in Settings"), ]) func `worker cannot bypass native capability consent`(command: String, expectedMessage: String) async { let worker = StubMacNodeHostWorker(commands: [command]) let runtime = MacNodeRuntime( nodeHostWorker: worker, computerControlEnabled: { false }, codexThreadCatalogEnabled: { false }, claudeSessionCatalogEnabled: { false }) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "native-disabled", command: command)) #expect(!response.ok) #expect(response.error?.code == .unavailable) #expect(response.error?.message == expectedMessage) #expect(await worker.invokedCommands().isEmpty) } @Test(arguments: [ OpenClawCanvasCommand.present.rawValue, OpenClawCanvasCommand.hide.rawValue, OpenClawCanvasCommand.navigate.rawValue, ]) func `worker cannot bypass the canvas presenter consent gate`(command: String) async { await TestIsolation.withUserDefaultsValues([canvasEnabledKey: false]) { let worker = StubMacNodeHostWorker(commands: [command]) let runtime = MacNodeRuntime(nodeHostWorker: worker) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "canvas-disabled", command: command)) #expect(!response.ok) #expect(response.error?.code == .unavailable) #expect(response.error?.message == "CANVAS_DISABLED: enable Canvas in Settings") #expect(await worker.invokedCommands().isEmpty) } } @Test func `worker cannot claim commands in the retired canvas namespace`() async { await TestIsolation.withUserDefaultsValues([canvasEnabledKey: true]) { let command = "canvas.plugin.render" let worker = StubMacNodeHostWorker(commands: [command]) let runtime = MacNodeRuntime(nodeHostWorker: worker) let response = await runtime.handleInvoke(BridgeInvokeRequest( id: "canvas-retired", command: command)) #expect(!response.ok) #expect(response.error?.code == .invalidRequest) #expect(response.error?.message == "INVALID_REQUEST: unknown command") #expect(await worker.invokedCommands().isEmpty) } } @Test func `capability union preserves native order and adds worker commands once`() { #expect(MacNodeModeCoordinator.mergingUnique( ["canvas", "screen", "system"], ["system", "mcp"]) == ["canvas", "screen", "system", "mcp"]) } @Test func `provider selection filters command ownership and publishes each provider descriptor`() throws { let descriptor = OpenClawProtocol.AnyCodable([ "contractVersion": OpenClawProtocol.AnyCodable(2), ]) let manifest = MacNodeHostManifest( version: "test", caps: ["screen", "computer"], commands: [MacNodeScreenCommand.snapshot.rawValue, OpenClawComputerCommand.act.rawValue], computerUse: descriptor, pathEnv: "/usr/bin:/bin") let peekaboo = try #require(MacNodeModeCoordinator.workerManifest(manifest, for: .peekaboo)) #expect(!peekaboo.commands.contains(MacNodeScreenCommand.snapshot.rawValue)) #expect(!peekaboo.commands.contains(OpenClawComputerCommand.act.rawValue)) #expect(peekaboo.computerUse == nil) let peekabooDescriptor = try #require(MacNodeModeCoordinator.computerUseDescriptor( provider: .peekaboo, commands: [MacNodeScreenCommand.snapshot.rawValue, OpenClawComputerCommand.act.rawValue], workerManifest: peekaboo)) let peekabooJSON = try JSONEncoder().encode(peekabooDescriptor) let peekabooObject = try #require( JSONSerialization.jsonObject(with: peekabooJSON) as? [String: Any]) #expect(peekabooObject["contractVersion"] as? Int == 2) #expect((peekabooObject["provider"] as? [String: Any])?["id"] as? String == "peekaboo") let actions = try #require(peekabooObject["actions"] as? [String]) #expect(actions.contains("get_window_state")) #expect(actions.contains("invoke_menu")) #expect(!actions.contains("zoom")) #expect(!actions.contains("get_browser_state")) #expect(!actions.contains("start_recording")) let features = try #require(peekabooObject["features"] as? [String: Any]) #expect(features["recording"] as? Bool == false) #expect(features["agentCursor"] as? Bool == false) #expect(features["multiDisplay"] as? Bool == true) let cua = try #require(MacNodeModeCoordinator.workerManifest(manifest, for: .cua)) #expect(cua.commands == manifest.commands) #expect(MacNodeModeCoordinator.computerUseDescriptor( provider: .cua, commands: cua.commands, workerManifest: cua) == descriptor) } @Test func `elevation host never advertises a persisted CUA provider`() throws { let suiteName = "MacNodeElevationHostProviderTests.\(UUID().uuidString)" let defaults = try #require(UserDefaults(suiteName: suiteName)) defer { defaults.removePersistentDomain(forName: suiteName) } defaults.set(true, forKey: computerControlEnabledKey) defaults.set(ComputerControlProvider.cua.rawValue, forKey: computerControlProviderKey) let provider = ComputerControlProvider.current( defaults: defaults, cuaAvailable: true, launchPlan: AppLaunchRuntimePlan(arguments: ["OpenClaw", "--elevation-host"])) #expect(provider == .peekaboo) let cuaDescriptor = OpenClawProtocol.AnyCodable(["provider": "cua"]) let manifest = MacNodeHostManifest( version: "test", caps: ["screen", "computer"], commands: [MacNodeScreenCommand.snapshot.rawValue, OpenClawComputerCommand.act.rawValue], computerUse: cuaDescriptor, pathEnv: "/usr/bin:/bin") let workerManifest = try #require(MacNodeModeCoordinator.workerManifest(manifest, for: provider)) #expect(workerManifest.computerUse == nil) let advertised = try #require(MacNodeModeCoordinator.computerUseDescriptor( provider: provider, commands: manifest.commands, workerManifest: workerManifest)) let data = try JSONEncoder().encode(advertised) let object = try #require(JSONSerialization.jsonObject(with: data) as? [String: Any]) #expect((object["provider"] as? [String: Any])?["id"] as? String == "peekaboo") } @Test func `stale route updates cannot replace newer worker authority`() { #expect(MacNodeHostWorker.routeUpdateIsCurrent(candidateGeneration: 4, currentGeneration: 4)) #expect(MacNodeHostWorker.routeUpdateIsCurrent(candidateGeneration: 5, currentGeneration: 4)) #expect(!MacNodeHostWorker.routeUpdateIsCurrent(candidateGeneration: 3, currentGeneration: 4)) } @Test func `worker forces app exec host without fallback`() async throws { let worker = MacNodeHostWorker(session: GatewayNodeSession()) let script = """ test "$OPENCLAW_NODE_EXEC_HOST" = app || exit 42 test "$OPENCLAW_NODE_EXEC_FALLBACK" = 0 || exit 43 printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":["system"],"commands":["system.run"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' printf '%s\\n' '{"type":"gateway-request","id":"gateway-1","method":"node.invoke.progress","params":{"invokeId":"terminal-1","nodeId":"node-1","seq":0,"chunk":"hello"},"timeoutMs":1000}' IFS= read -r unavailable printf '%s' "$unavailable" | grep -q '"type":"gateway-response"' || exit 44 printf '%s' "$unavailable" | grep -q '"ok":false' || exit 45 while IFS= read -r line; do case "$line" in *'"type":"invoke"'*) printf '%s\\n' '{"type":"invoke-result","result":{"id":"worker-run","ok":true,"payload":{"owner":"cli"}}}' ;; esac done """ let manifest = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script])) #expect(manifest.commands == ["system.run"]) let response = await worker.invoke(BridgeInvokeRequest( id: "worker-run", command: "system.run", paramsJSON: #"{"command":["/usr/bin/true"]}"#)) #expect(response.ok) #expect(response.payload != nil) await worker.stop() } @Test func `worker strips inherited CUA values and receives only the app-provided endpoint`() async throws { let endpoint = CuaDriverWorkerEndpoint( socketPath: "/private/test/cua.sock", binaryPath: "/Applications/OpenClaw.app/Contents/Resources/cua-driver") let endpointValue = try endpoint.environmentValue() let inheritedKeys = [CuaDriverWorkerEnvironment.endpoint] + CuaDriverWorkerEnvironment.inheritedFamilyPrefixes.flatMap { [$0 + "SOCKET_PATH", $0 + "BINARY_PATH"] } let inheritedEnvironment = Dictionary(uniqueKeysWithValues: inheritedKeys.map { ($0, Optional("inherited")) }) let script = """ test "$OPENCLAW_CUA_DRIVER_ENDPOINT" = "$1" || exit 41 test "$(env | grep -Ec '^(OPENCLAW_)?CUA_DRIVER_')" = 1 || exit 42 printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' while IFS= read -r line; do :; done """ try await TestIsolation.withEnvValues(inheritedEnvironment) { let worker = MacNodeHostWorker(session: GatewayNodeSession()) _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script, "worker", endpointValue], environment: [ CuaDriverWorkerEnvironment.endpoint: endpointValue, ])) await worker.stop() } } @Test func `unbound worker strips every inherited CUA endpoint value`() async throws { let inheritedKeys = [CuaDriverWorkerEnvironment.endpoint] + CuaDriverWorkerEnvironment.inheritedFamilyPrefixes.flatMap { [$0 + "SOCKET_PATH", $0 + "BINARY_PATH"] } let inheritedEnvironment = Dictionary(uniqueKeysWithValues: inheritedKeys.map { ($0, Optional("inherited")) }) let script = """ test "$(env | grep -Ec '^(OPENCLAW_)?CUA_DRIVER_')" = 0 || exit 41 printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' while IFS= read -r line; do :; done """ try await TestIsolation.withEnvValues(inheritedEnvironment) { let worker = MacNodeHostWorker(session: GatewayNodeSession()) _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script])) await worker.stop() } } @Test func `worker cancellation settles when the child suppresses its result`() async throws { let worker = MacNodeHostWorker(session: GatewayNodeSession()) let marker = FileManager.default.temporaryDirectory .appendingPathComponent("openclaw-worker-cancel-\(UUID().uuidString)") defer { try? FileManager.default.removeItem(at: marker) } let script = """ printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":["terminal"],"commands":["codex.terminal.resume.v1"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' IFS= read -r buffered_invoke IFS= read -r input IFS= read -r buffered_cancel printf '%s' "$buffered_invoke" | grep -q '"id":"terminal-1"' || exit 40 printf '%s' "$input" | grep -q '"type":"invoke-input"' || exit 41 printf '%s' "$input" | grep -q '"invokeId":"terminal-1"' || exit 42 printf '%s' "$input" | grep -q '"seq":7' || exit 43 printf '%s' "$buffered_cancel" | grep -q '"type":"invoke-cancel"' || exit 44 printf '%s' "$buffered_cancel" | grep -q '"invokeId":"terminal-1"' || exit 45 IFS= read -r active_invoke printf '%s' "$active_invoke" | grep -q '"id":"terminal-2"' || exit 46 printf '%s\\n' "$$" > "$1" IFS= read -r active_cancel printf '%s' "$active_cancel" | grep -q '"type":"invoke-cancel"' || exit 47 printf '%s' "$active_cancel" | grep -q '"invokeId":"terminal-2"' || exit 48 while IFS= read -r line; do :; done """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script, "worker", marker.path])) await worker.handleInput(invokeId: "terminal-1", seq: 7, payloadJSON: #"{"data":"x"}"#) await worker.cancel(invokeId: "terminal-1") do { let buffered = try await AsyncTimeout.withTimeout( seconds: 1, onTimeout: { WorkerBackpressureTimeout() }, operation: { await worker.invoke(BridgeInvokeRequest( id: "terminal-1", command: "codex.terminal.resume.v1")) }) #expect(!buffered.ok) #expect(buffered.error?.message == "UNAVAILABLE: node-host worker invocation cancelled") let invoking = Task { await worker.invoke(BridgeInvokeRequest( id: "terminal-2", command: "codex.terminal.resume.v1")) } _ = try await TestProcessSupport.waitForPID(in: marker) await worker.cancel(invokeId: "terminal-2") let active = try await AsyncTimeout.withTimeout( seconds: 1, onTimeout: { WorkerBackpressureTimeout() }, operation: { await invoking.value }) await worker.stop() #expect(!active.ok) #expect(active.error?.message == "UNAVAILABLE: node-host worker invocation cancelled") } catch { await worker.stop() throw error } } @Test func `ready worker exit notifies its route owner`() async throws { try await confirmation("unexpected worker exit") { confirmed in let exitGate = AsyncTestGate() let expectedGeneration: UInt64 = 42 let worker = MacNodeHostWorker(session: GatewayNodeSession()) { generation in #expect(generation == expectedGeneration) confirmed() exitGate.open() } let script = """ printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":["system"],"commands":["system.run"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' sleep 0.05 exit 7 """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script], configurationGeneration: expectedGeneration)) await exitGate.wait() } } @Test func `worker deinit terminates its owned process`() async throws { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("openclaw-worker-deinit-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) let pidFile = directory.appendingPathComponent("worker.pid") defer { TestProcessSupport.killLeakedProcesses(in: [pidFile]) try? FileManager.default.removeItem(at: directory) } var worker: MacNodeHostWorker? = MacNodeHostWorker(session: GatewayNodeSession()) let script = """ printf '%s\n' "$$" > "$1" printf '%s\n' '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"/bin"}}' while :; do /bin/sleep 1; done """ _ = try await #require(worker).start(launch: MacNodeHostWorkerLaunch(command: [ "/bin/sh", "-c", script, "worker", pidFile.path, ])) let pid = try await TestProcessSupport.waitForPID(in: pidFile) worker = nil #expect(await TestProcessSupport.waitUntilGone(pid)) } @Test func `changed worker command replaces the running process`() async throws { let worker = MacNodeHostWorker(session: GatewayNodeSession()) let firstScript = """ printf '%s\\n' '{"type":"ready","version":"first","manifest":{"caps":["system"],"commands":["system.run"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' while IFS= read -r line; do :; done """ let secondScript = """ printf '%s\\n' '{"type":"ready","version":"second","manifest":{"caps":["system"],"commands":["system.run"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' while IFS= read -r line; do :; done """ let first = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", firstScript])) let second = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", secondScript])) #expect(first.version == "first") #expect(second.version == "second") await worker.stop() } @Test func `stop cancels a changed launch during worker cleanup`() async throws { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("openclaw-worker-restart-stop-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) let cleanupStartedPIDFile = directory.appendingPathComponent("cleanup-started.pid") let replacementPIDFile = directory.appendingPathComponent("replacement.pid") defer { TestProcessSupport.killLeakedProcesses(in: [replacementPIDFile, cleanupStartedPIDFile]) try? FileManager.default.removeItem(at: directory) } let worker = MacNodeHostWorker(session: GatewayNodeSession()) let firstScript = """ trap 'printf "%s\n" "$$" > "$1"; /bin/sleep 0.2; exit 0' TERM printf '%s\n' '{"type":"ready","version":"first","manifest":{"caps":[],"commands":[],"pathEnv":"/bin"}}' while :; do /bin/sleep 1; done """ let replacementScript = """ printf '%s\n' "$$" > "$1" printf '%s\n' '{"type":"ready","version":"replacement","manifest":{"caps":[],"commands":[],"pathEnv":"/bin"}}' while :; do /bin/sleep 1; done """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch(command: [ "/bin/sh", "-c", firstScript, "worker", cleanupStartedPIDFile.path, ])) let replacement = Task { try await worker.start(launch: MacNodeHostWorkerLaunch(command: [ "/bin/sh", "-c", replacementScript, "worker", replacementPIDFile.path, ])) } _ = try await TestProcessSupport.waitForPID(in: cleanupStartedPIDFile) await worker.stop() switch await replacement.result { case .success: Issue.record("changed launch succeeded after stop returned") await worker.stop() case .failure: break } #expect(TestProcessSupport.pollPID(in: replacementPIDFile) == nil) } @Test func `stop waits for the worker leader and reaps its descendants`() async throws { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("openclaw-worker-stop-\(UUID().uuidString)", isDirectory: true) try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) let leaderPIDFile = directory.appendingPathComponent("leader.pid") let descendantPIDFile = directory.appendingPathComponent("descendant.pid") let termCompleteFile = directory.appendingPathComponent("term-complete") defer { TestProcessSupport.killLeakedProcesses(in: [descendantPIDFile, leaderPIDFile]) try? FileManager.default.removeItem(at: directory) } let worker = MacNodeHostWorker(session: GatewayNodeSession()) let script = """ printf '%s\n' "$$" > "$1" /bin/sh -c 'trap "" HUP TERM; printf "%s\\n" "$$" > "$1"; while :; do /bin/sleep 1; done' \ descendant "$2" /dev/null 2>&1 & while [ ! -s "$2" ]; do /bin/sleep 0.01; done trap 'touch "$3"; /bin/sleep 0.2; exit 0' TERM printf '%s\n' '{"type":"ready","version":"test","manifest":{"caps":[],"commands":[],"pathEnv":"/bin"}}' while :; do /bin/sleep 1; done """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch(command: [ "/bin/sh", "-c", script, "worker", leaderPIDFile.path, descendantPIDFile.path, termCompleteFile.path, ])) let leaderPID = try await TestProcessSupport.waitForPID(in: leaderPIDFile) let descendantPID = try await TestProcessSupport.waitForPID(in: descendantPIDFile) await worker.stop() #expect(FileManager.default.fileExists(atPath: termCompleteFile.path)) #expect(TestProcessSupport.processIsGone(leaderPID)) #expect(TestProcessSupport.processIsGone(descendantPID)) } @Test func `worker drains stdout while a large stdin frame is backpressured`() async throws { let worker = MacNodeHostWorker(session: GatewayNodeSession()) let script = """ printf '%s\\n' '{"type":"ready","version":"test","manifest":{"caps":["system"],"commands":["system.run"],"pathEnv":"/usr/bin:/bin"},"inventory":{"skills":null,"pluginTools":[]}}' IFS= read -r first printf '{"type":"invoke-result","result":{"id":"first","ok":true,"payload":{"blob":"' head -c 2097152 /dev/zero | tr '\\000' x printf '"}}}\\n' IFS= read -r second printf '%s\\n' '{"type":"invoke-result","result":{"id":"second","ok":true,"payload":{"done":true}}}' """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script])) let first = Task { await worker.invoke(BridgeInvokeRequest( id: "first", command: "system.run", paramsJSON: #"{"command":["/usr/bin/true"]}"#)) } try await Task.sleep(for: .milliseconds(20)) let largeParams = #"{"blob":""# + String(repeating: "x", count: 2 * 1024 * 1024) + #""}"# let second = Task { await worker.invoke(BridgeInvokeRequest( id: "second", command: "system.run", paramsJSON: largeParams)) } do { let responses = try await AsyncTimeout.withTimeout( seconds: 5, onTimeout: { WorkerBackpressureTimeout() }, operation: { [first, second] in await [first.value, second.value] }) await worker.stop() let allResponsesSucceeded = responses.allSatisfy(\.ok) #expect(allResponsesSucceeded) } catch { await worker.stop() throw error } } }