openclaw/apps/macos/Tests/OpenClawIPCTests/MacNodeHostWorkerTests.swift

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
        }
    }
}