import Darwin import Foundation import OpenClawKit 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: original.launch, 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: [ 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, "canvas.plugin.render"]) func `worker cannot bypass the canvas namespace 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 `capability union preserves native order and adds worker commands once`() { #expect(MacNodeModeCoordinator.mergingUnique( ["canvas", "screen", "system"], ["system", "mcp"]) == ["canvas", "screen", "system", "mcp"]) } @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 forwards terminal input and cancellation frames`() async throws { let worker = MacNodeHostWorker(session: GatewayNodeSession()) 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 invoke IFS= read -r input IFS= read -r cancel printf '%s' "$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' "$cancel" | grep -q '"type":"invoke-cancel"' || exit 44 printf '%s' "$cancel" | grep -q '"invokeId":"terminal-1"' || exit 45 printf '%s\\n' '{"type":"invoke-result","result":{"id":"terminal-1","ok":true}}' while IFS= read -r line; do :; done """ _ = try await worker.start(launch: MacNodeHostWorkerLaunch( command: ["/bin/sh", "-c", script])) await worker.handleInput(invokeId: "terminal-1", seq: 7, payloadJSON: #"{"data":"x"}"#) await worker.cancel(invokeId: "terminal-1") let response = await worker.invoke(BridgeInvokeRequest( id: "terminal-1", command: "codex.terminal.resume.v1")) #expect(response.ok) await worker.stop() } @Test func `ready worker exit notifies its route owner`() async throws { try await confirmation("unexpected worker exit") { confirmed in let exitGate = AsyncTestGate() let worker = MacNodeHostWorker(session: GatewayNodeSession()) { 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])) await exitGate.wait() } } @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 `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, await second.value] }) await worker.stop() #expect(responses.allSatisfy { $0.ok }) } catch { await worker.stop() throw error } } }