mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-18 08:31:49 -06:00
fix(runners): preserve device sessions through lifecycle faults (#124744)
* test(runners): cover paired-node lifecycle wire * fix(runners): preserve retryable device lifecycle * test(runners): keep wire fixture internals local * fix(runners): stop fallback on device capacity
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
5d68a8e1db
commit
91e537da15
@@ -0,0 +1,32 @@
|
||||
title: Paired-device worker lifecycle wire
|
||||
|
||||
scenario:
|
||||
id: paired-node-worker-lifecycle-wire
|
||||
surface: gateway
|
||||
category: gateway.nodes-and-remote-capabilities
|
||||
coverage:
|
||||
secondary:
|
||||
- gateway.node-inventory
|
||||
- gateway.session-apis-session-status
|
||||
objective: Prove paired-device worker faults stay retryable and isolated from Gateway-local work across the real signed node and worker-child wire.
|
||||
successCriteria:
|
||||
- A signed paired node handles private worker invocations through the real bundle installer, supervisor, workspace runtime, and worker child while a Gateway-local session remains usable.
|
||||
- Bundle retention and status capability v1 expose only installed version or missing status through node.list and environments.list, then reinstall an exactly deleted proof bundle after same-identity reconnect.
|
||||
- A pre-turn node disconnect produces a visible runner-offline recovery message, preserves the active placement for same-identity retry, and does not terminalize local work.
|
||||
- Two held real worker children exhaust supervisor capacity 2; a third turn records a bounded capacity failure across RPC, history, and placement, then succeeds after one slot releases.
|
||||
- Node-role removal reconciles the owned environment and placement before returning, fences the old node connection, and leaves Gateway-local execution available.
|
||||
docsRefs:
|
||||
- docs/gateway/protocol.md
|
||||
- docs/gateway/cloud-workers.md
|
||||
- docs/concepts/qa-e2e-automation.md
|
||||
codeRefs:
|
||||
- src/gateway/worker-environments/node-workspace-retain-coordinator.ts
|
||||
- src/gateway/worker-environments/node-launch-adapter.ts
|
||||
- src/gateway/device-worker-revocation.ts
|
||||
- src/node-host/node-worker-bundle-installer.ts
|
||||
- src/node-host/node-worker-supervisor.ts
|
||||
- test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts
|
||||
execution:
|
||||
kind: vitest
|
||||
path: test/e2e/qa-lab/runtime/paired-node-worker-lifecycle-wire.e2e.test.ts
|
||||
summary: Start a real isolated Gateway, pair a signed node host, then prove bundle repair, offline retry, physical capacity recovery, role-removal fencing, and local-session isolation over the real worker wire.
|
||||
@@ -829,10 +829,16 @@ describe("failover-error", () => {
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("returns true for direct and nested runner availability failures", () => {
|
||||
const unavailable = new Error("The device runner is offline");
|
||||
unavailable.name = "WorkerRunnerUnavailableError";
|
||||
for (const error of [unavailable, new Error("worker turn failed", { cause: unavailable })]) {
|
||||
it.each([
|
||||
["availability", "WorkerRunnerUnavailableError", "The device runner is offline"],
|
||||
["capacity", "WorkerRunnerCapacityError", "device worker capacity remained full"],
|
||||
])("returns true for direct and nested runner %s failures", (_label, name, message) => {
|
||||
const coordination = new Error(message);
|
||||
coordination.name = name;
|
||||
for (const error of [
|
||||
coordination,
|
||||
new Error("worker turn failed", { cause: coordination }),
|
||||
]) {
|
||||
expect(isNonProviderRuntimeCoordinationError(error)).toBe(true);
|
||||
expect(resolveModelFallbackError(error)).toEqual({ kind: "coordination", error });
|
||||
}
|
||||
|
||||
@@ -492,8 +492,10 @@ function hasGatewayDrainingFailure(err: unknown): boolean {
|
||||
return errorGraphHasName(err, "GatewayDrainingError");
|
||||
}
|
||||
|
||||
function hasWorkerRunnerUnavailableFailure(err: unknown): boolean {
|
||||
return errorGraphHasName(err, "WorkerRunnerUnavailableError");
|
||||
function hasWorkerRunnerCoordinationFailure(err: unknown): boolean {
|
||||
return ["WorkerRunnerUnavailableError", "WorkerRunnerCapacityError"].some((name) =>
|
||||
errorGraphHasName(err, name),
|
||||
);
|
||||
}
|
||||
|
||||
function hasDirectProviderFailureIdentity(err: unknown): boolean {
|
||||
@@ -891,7 +893,7 @@ export function resolveModelFallbackError(
|
||||
}
|
||||
// Gateway admission can fail before any provider turn starts. Preserve that
|
||||
// identity through wrappers and aggregates so fallback cannot blame a model.
|
||||
if (hasGatewayDrainingFailure(err) || hasWorkerRunnerUnavailableFailure(err)) {
|
||||
if (hasGatewayDrainingFailure(err) || hasWorkerRunnerCoordinationFailure(err)) {
|
||||
return { kind: "coordination", error: err };
|
||||
}
|
||||
const staleLifecycleFailure = hasStaleAgentRunLifecycleFailure(err);
|
||||
|
||||
@@ -2248,6 +2248,46 @@ describe("runWithModelFallback", () => {
|
||||
expect(onFallbackStep).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([
|
||||
[
|
||||
"direct",
|
||||
() =>
|
||||
Object.assign(new Error("device worker capacity remained full"), {
|
||||
name: "WorkerRunnerCapacityError",
|
||||
}),
|
||||
],
|
||||
[
|
||||
"wrapped",
|
||||
() =>
|
||||
new Error("worker turn failed", {
|
||||
cause: Object.assign(new Error("device worker capacity remained full"), {
|
||||
name: "WorkerRunnerCapacityError",
|
||||
}),
|
||||
}),
|
||||
],
|
||||
])("aborts fallback on %s device capacity failures", async (_label, makeError) => {
|
||||
const error = makeError();
|
||||
const run = vi.fn().mockRejectedValueOnce(error).mockResolvedValueOnce("too late");
|
||||
const onError = vi.fn();
|
||||
const onFallbackStep = vi.fn();
|
||||
|
||||
await expect(
|
||||
runWithModelFallback({
|
||||
cfg: undefined,
|
||||
provider: "openai",
|
||||
model: "gpt-5.6-sol",
|
||||
fallbacksOverride: ["openai/gpt-5.4-mini"],
|
||||
skipAuthProfileRuntime: true,
|
||||
run,
|
||||
onError,
|
||||
onFallbackStep,
|
||||
}),
|
||||
).rejects.toBe(error);
|
||||
expect(run).toHaveBeenCalledTimes(1);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
expect(onFallbackStep).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("still advances after a genuine provider rate limit", async () => {
|
||||
const rateLimit = Object.assign(new Error("rate limit exceeded"), { status: 429 });
|
||||
const run = vi.fn().mockRejectedValueOnce(rateLimit).mockResolvedValueOnce("fallback ok");
|
||||
|
||||
@@ -133,6 +133,12 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
|
||||
const workspaceConflictHandlers = createWorkerWorkspaceConflictTranscriptHandlers(
|
||||
loadWorkerPlacementSessionRuntimeModule,
|
||||
);
|
||||
const nodeWorkspaceRetention = createNodeWorkspaceRetainCoordinator({
|
||||
gatewayNamespace: params.gatewayNamespace,
|
||||
placements: params.placements,
|
||||
environments: params.environments,
|
||||
warn: params.warn,
|
||||
});
|
||||
const resolveWorkspacePath = async ({
|
||||
sessionId,
|
||||
sessionKey,
|
||||
@@ -333,6 +339,11 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
|
||||
}
|
||||
return activePlacement;
|
||||
},
|
||||
onActivated: (request) => {
|
||||
if (request.deviceId) {
|
||||
void nodeWorkspaceRetention.schedule(request.deviceId);
|
||||
}
|
||||
},
|
||||
runReclaimBarrier: async ({ sessionId, sessionKey, agentId, reclaim }) => {
|
||||
const sessionRuntime = await loadWorkerPlacementSessionRuntimeModule();
|
||||
const { resolveGatewaySessionStoreTargetWithStore } = sessionRuntime;
|
||||
@@ -409,12 +420,6 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme
|
||||
createSessionEvidenceResolver: createWorkerPlacementSessionEvidenceResolver,
|
||||
warn: params.warn,
|
||||
});
|
||||
const nodeWorkspaceRetention = createNodeWorkspaceRetainCoordinator({
|
||||
gatewayNamespace: params.gatewayNamespace,
|
||||
placements: params.placements,
|
||||
environments: params.environments,
|
||||
warn: params.warn,
|
||||
});
|
||||
const admissionProvider = createWorkerSessionTurnPlacementProvider({
|
||||
environments: params.environments,
|
||||
placements: params.placements,
|
||||
|
||||
@@ -17,7 +17,7 @@ import type {
|
||||
NodeWorkerSupervisorNodeProof,
|
||||
NodeWorkerSupervisorTransport,
|
||||
} from "../node-registry-private.js";
|
||||
import { WorkerRunnerUnavailableError } from "./tunnel-contract.js";
|
||||
import { WorkerRunnerCapacityError, WorkerRunnerUnavailableError } from "./tunnel-contract.js";
|
||||
|
||||
const DEFAULT_RPC_TIMEOUT_MS = 30_000;
|
||||
const DEFAULT_POLL_INTERVAL_MS = 250;
|
||||
@@ -307,11 +307,12 @@ export function createNodeWorkerLaunchAdapter(options: NodeWorkerLaunchAdapterOp
|
||||
const result = await raceWithSignal(operation, signal);
|
||||
if (!result.ok) {
|
||||
const code = result.error?.code ?? "UNAVAILABLE";
|
||||
if (code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE) {
|
||||
throw new WorkerRunnerCapacityError();
|
||||
}
|
||||
throw new NodeWorkerLaunchTransportError(
|
||||
code,
|
||||
code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE
|
||||
? "device worker capacity remained full"
|
||||
: `node worker supervisor invocation failed (${code})`,
|
||||
`node worker supervisor invocation failed (${code})`,
|
||||
);
|
||||
}
|
||||
return parseInvokeReceipt(result.payloadJSON);
|
||||
@@ -496,10 +497,7 @@ export function createNodeWorkerLaunchAdapter(options: NodeWorkerLaunchAdapterOp
|
||||
}
|
||||
// The node authors this result only after its durable claim stayed absent.
|
||||
// Transport dispatch is therefore not launch ambiguity and needs no cancel.
|
||||
if (
|
||||
error instanceof NodeWorkerLaunchTransportError &&
|
||||
error.code === NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE
|
||||
) {
|
||||
if (error instanceof WorkerRunnerCapacityError) {
|
||||
throw error;
|
||||
}
|
||||
if (!mayHaveLaunched) {
|
||||
|
||||
@@ -60,6 +60,7 @@ type WorkerPlacementDispatchOptions = {
|
||||
runLocalBarrier: WorkerLocalDispatchBarrier;
|
||||
runActivationBarrier: WorkerActivationBarrier;
|
||||
runReclaimBarrier: WorkerPlacementReclaimBarrier;
|
||||
onActivated?: (request: WorkerPlacementDispatchRequest) => void;
|
||||
workspaceOperations: WorkerWorkspaceOperationCoordinator;
|
||||
resolveWorkspacePath: (params: {
|
||||
sessionId: string;
|
||||
@@ -253,6 +254,11 @@ export function createWorkerPlacementDispatchService(options: WorkerPlacementDis
|
||||
return activated;
|
||||
},
|
||||
});
|
||||
try {
|
||||
options.onActivated?.(request);
|
||||
} catch {
|
||||
// Maintenance scheduling cannot overturn a durable placement activation.
|
||||
}
|
||||
return activePlacement;
|
||||
} catch (error) {
|
||||
try {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE } from "../../infra/node-commands.js";
|
||||
import type { SpawnResult } from "../../process/exec.js";
|
||||
import type { WorkerLaunchPlan } from "../../worker/launch-descriptor.js";
|
||||
import type { NodeWorkerWorkspaceTransferInput } from "../../worker/node-workspace-transfer-protocol.js";
|
||||
@@ -26,6 +27,15 @@ export class WorkerRunnerUnavailableError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
export class WorkerRunnerCapacityError extends Error {
|
||||
readonly code = NODE_WORKER_CAPACITY_EXHAUSTED_ERROR_CODE;
|
||||
|
||||
constructor() {
|
||||
super("device worker capacity remained full");
|
||||
this.name = "WorkerRunnerCapacityError";
|
||||
}
|
||||
}
|
||||
|
||||
export type WorkerTunnelRequest = {
|
||||
environmentId: string;
|
||||
ownerEpoch: number;
|
||||
|
||||
@@ -8,7 +8,11 @@ import { makeAgentAssistantMessage } from "../../agents/test-helpers/agent-messa
|
||||
import type { SpawnResult } from "../../process/exec.js";
|
||||
import { WORKER_PROVIDER_REPLAY_LOCAL_RETRY_MESSAGE } from "../../worker/transcript-message.js";
|
||||
import type { WorkerSessionPlacementStore } from "./placement-store.js";
|
||||
import { WorkerRunnerUnavailableError, type WorkerTunnelHandle } from "./tunnel-contract.js";
|
||||
import {
|
||||
WorkerRunnerCapacityError,
|
||||
WorkerRunnerUnavailableError,
|
||||
type WorkerTunnelHandle,
|
||||
} from "./tunnel-contract.js";
|
||||
import {
|
||||
ENVIRONMENT_ID,
|
||||
MANIFEST_REF,
|
||||
@@ -237,7 +241,20 @@ describe("worker turn launcher failure recovery", () => {
|
||||
expect(environments.destroy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps the placement active when launch fails before transport dispatch", async () => {
|
||||
it.each([
|
||||
{
|
||||
name: "offline before transport dispatch",
|
||||
error: new WorkerRunnerUnavailableError(),
|
||||
dispatched: false,
|
||||
expectedMessage: "The device runner is offline",
|
||||
},
|
||||
{
|
||||
name: "capacity rejection after transport dispatch",
|
||||
error: new WorkerRunnerCapacityError(),
|
||||
dispatched: true,
|
||||
expectedMessage: "device worker capacity remained full",
|
||||
},
|
||||
])("keeps the placement active after $name", async ({ error, dispatched, expectedMessage }) => {
|
||||
seedActivePlacement();
|
||||
const teardownStates: string[] = [];
|
||||
const observedPlacements: WorkerSessionPlacementStore = {
|
||||
@@ -272,8 +289,11 @@ describe("worker turn launcher failure recovery", () => {
|
||||
resume: vi.fn(async () => {}),
|
||||
})),
|
||||
runWorkspaceCommand: vi.fn(),
|
||||
launchTurn: vi.fn(async () => {
|
||||
throw new WorkerRunnerUnavailableError();
|
||||
launchTurn: vi.fn(async (request) => {
|
||||
if (dispatched) {
|
||||
request.onDispatchReady?.();
|
||||
}
|
||||
throw error;
|
||||
}),
|
||||
syncWorkspace: vi.fn(async () => {
|
||||
throw new Error("unexpected workspace sync");
|
||||
@@ -309,10 +329,10 @@ describe("worker turn launcher failure recovery", () => {
|
||||
turn("run-failed"),
|
||||
runLocal,
|
||||
),
|
||||
).rejects.toThrow("The device runner is offline");
|
||||
).rejects.toThrow(expectedMessage);
|
||||
expect(runLocal).not.toHaveBeenCalled();
|
||||
expect(placements.get(SESSION_ID)).toMatchObject({ state: "active", turnClaim: null });
|
||||
expect(acknowledgeCredentialDelivery).not.toHaveBeenCalled();
|
||||
expect(acknowledgeCredentialDelivery).toHaveBeenCalledTimes(dispatched ? 1 : 0);
|
||||
expect(stopTunnel).not.toHaveBeenCalled();
|
||||
expect(destroy).not.toHaveBeenCalled();
|
||||
expect(teardownStates).toEqual([]);
|
||||
|
||||
@@ -22,7 +22,7 @@ import type {
|
||||
WorkerSessionPlacementStore,
|
||||
WorkerSessionTurnClaim,
|
||||
} from "./placement-store.js";
|
||||
import { WorkerRunnerUnavailableError } from "./tunnel-contract.js";
|
||||
import { WorkerRunnerCapacityError, WorkerRunnerUnavailableError } from "./tunnel-contract.js";
|
||||
import { resolveWorkerBrowserLaunchPlan } from "./worker-browser-launch-plan.js";
|
||||
import {
|
||||
claimWorkerTurn,
|
||||
@@ -527,7 +527,10 @@ export function createWorkerSessionTurnPlacementProvider(options: WorkerTurnLaun
|
||||
await options.recoverPendingWorkspaceResult(placement.environmentId);
|
||||
throw error;
|
||||
}
|
||||
if (error instanceof WorkerRunnerUnavailableError && !handedOff) {
|
||||
if (
|
||||
error instanceof WorkerRunnerCapacityError ||
|
||||
(error instanceof WorkerRunnerUnavailableError && !handedOff)
|
||||
) {
|
||||
await releaseClaimIfOwned(options.placements, turnClaim);
|
||||
throw error;
|
||||
}
|
||||
|
||||
@@ -1,59 +1,40 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import fs from "node:fs/promises";
|
||||
import { createServer, type Server } from "node:http";
|
||||
import path from "node:path";
|
||||
import { performance } from "node:perf_hooks";
|
||||
import { promisify } from "node:util";
|
||||
import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js";
|
||||
import {
|
||||
GATEWAY_CLIENT_MODES,
|
||||
GATEWAY_CLIENT_NAMES,
|
||||
} from "../../../../packages/gateway-protocol/src/client-info.js";
|
||||
import type { DeviceIdentity } from "../../../../src/infra/device-identity.js";
|
||||
import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.js";
|
||||
import { afterEach, describe, expect, it } from "vitest";
|
||||
import {
|
||||
NODE_WORKER_BUNDLE_INSTALL_COMMAND,
|
||||
NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND,
|
||||
NODE_WORKER_SUPERVISOR_STATUS_COMMAND,
|
||||
NODE_WORKER_WORKSPACE_EXEC_COMMAND,
|
||||
} from "../../../../src/infra/node-commands.js";
|
||||
import {
|
||||
NODE_RUNNER_INVENTORY_UPDATE_METHOD,
|
||||
NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE,
|
||||
} from "../../../../src/infra/node-runner-inventory.js";
|
||||
import { handleInvoke, type NodeInvokeRequestPayload } from "../../../../src/node-host/invoke.js";
|
||||
import { NodeWorkerBundleInstaller } from "../../../../src/node-host/node-worker-bundle-installer.js";
|
||||
import { createNodeWorkerSupervisor } from "../../../../src/node-host/node-worker-supervisor.js";
|
||||
import { NodeWorkerWorkspaceRuntime } from "../../../../src/node-host/node-worker-workspace.js";
|
||||
import { VERSION } from "../../../../src/version.js";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js";
|
||||
import {
|
||||
BASELINE_PROMPT,
|
||||
BASELINE_REPLY,
|
||||
MODEL_REF,
|
||||
PROOF_TIMEOUT_MS,
|
||||
startMidturnProvider,
|
||||
} from "./cloud-worker-midturn-loss-fixture.js";
|
||||
import {
|
||||
closeWireServer,
|
||||
connectWireClient,
|
||||
createPairedNodeWorkerHost,
|
||||
createPublishedWireWorkspace,
|
||||
startPairedNodeWorkerGateway,
|
||||
type PairedNodeWorkerHost,
|
||||
type WireGateway,
|
||||
wireMessageText,
|
||||
} from "./paired-node-worker-wire-fixture.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const SESSION_KEY = "agent:qa:node-worker-launch-wire";
|
||||
const NODE_DISPLAY_NAME = "QA Gateway-bundle worker node";
|
||||
const TEST_TIMEOUT_MS = PROOF_TIMEOUT_MS + 60_000;
|
||||
const CONTROL_PROBE_MAX_MS = 4_000;
|
||||
const FINALIZATION_LOAD_CONCURRENCY = 6;
|
||||
|
||||
type Gateway = Awaited<ReturnType<typeof startQaGatewayChild>>;
|
||||
type GatewayEvent = { event: string; payload?: unknown };
|
||||
type NodeRead = {
|
||||
nodeId: string;
|
||||
approvalState?: string;
|
||||
connected?: boolean;
|
||||
paired?: boolean;
|
||||
sessionHost?: boolean;
|
||||
};
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
async function git(cwd: string, ...args: string[]): Promise<string> {
|
||||
@@ -64,261 +45,6 @@ async function git(cwd: string, ...args: string[]): Promise<string> {
|
||||
return stdout.trim();
|
||||
}
|
||||
|
||||
async function closeServer(server: Server): Promise<void> {
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
|
||||
async function createPublishedWorkspace(root: string) {
|
||||
const source = path.join(root, "source");
|
||||
const bare = path.join(root, "repo.git");
|
||||
await fs.mkdir(source, { recursive: true });
|
||||
await execFileAsync("git", ["init", "--bare", bare]);
|
||||
await git(source, "init", "-b", "main");
|
||||
await git(source, "config", "user.name", "OpenClaw QA");
|
||||
await git(source, "config", "user.email", "openclaw-qa@example.invalid");
|
||||
await fs.mkdir(path.join(source, "nested"));
|
||||
await fs.writeFile(path.join(source, "launch-wire.txt"), "local-install launch wire\n");
|
||||
await fs.writeFile(path.join(source, "nested", "tracked.txt"), "nested tracked input\n");
|
||||
await git(source, "add", ".");
|
||||
await git(source, "commit", "-m", "initialize node worker launch wire workspace");
|
||||
await git(source, "remote", "add", "publish", bare);
|
||||
await git(source, "push", "publish", "main");
|
||||
await git(source, "remote", "remove", "publish");
|
||||
await git(bare, "update-server-info");
|
||||
|
||||
const server = createServer((request, response) => {
|
||||
void (async () => {
|
||||
const pathname = decodeURIComponent(new URL(request.url ?? "/", "http://localhost").pathname);
|
||||
if (!pathname.startsWith("/repo.git/")) {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
const candidate = path.resolve(bare, pathname.slice("/repo.git/".length));
|
||||
if (candidate !== bare && !candidate.startsWith(`${bare}${path.sep}`)) {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const contents = await fs.readFile(candidate);
|
||||
response.writeHead(200, {
|
||||
"content-type": pathname.endsWith("/info/refs")
|
||||
? "text/plain; charset=utf-8"
|
||||
: "application/octet-stream",
|
||||
"content-length": String(contents.byteLength),
|
||||
});
|
||||
response.end(request.method === "HEAD" ? undefined : contents);
|
||||
} catch {
|
||||
response.writeHead(404).end();
|
||||
}
|
||||
})();
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
throw new Error("published workspace server did not bind");
|
||||
}
|
||||
const origin = `http://127.0.0.1:${address.port}/repo.git`;
|
||||
await git(source, "remote", "add", "origin", origin);
|
||||
const commit = await git(source, "rev-parse", "HEAD");
|
||||
await git(source, "ls-remote", "--exit-code", origin, "refs/heads/main");
|
||||
return { commit, source: await fs.realpath(source), server };
|
||||
}
|
||||
|
||||
async function connectClient(params: {
|
||||
gateway: Gateway;
|
||||
role: "operator" | "node";
|
||||
identity: DeviceIdentity | null;
|
||||
onEvent?: (event: GatewayEvent) => void;
|
||||
timeoutMs?: number;
|
||||
}): Promise<GatewayClient> {
|
||||
return await new Promise<GatewayClient>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const finish = (error?: Error) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
if (error) {
|
||||
client.stop();
|
||||
reject(error);
|
||||
} else {
|
||||
resolve(client);
|
||||
}
|
||||
};
|
||||
const timeout = setTimeout(
|
||||
() => finish(new Error("Gateway client connection timed out")),
|
||||
params.timeoutMs ?? 30_000,
|
||||
);
|
||||
timeout.unref();
|
||||
const node = params.role === "node";
|
||||
const client = new GatewayClient({
|
||||
url: params.gateway.wsUrl,
|
||||
token: params.gateway.token,
|
||||
env: params.gateway.runtimeEnv,
|
||||
role: params.role,
|
||||
clientName: node ? GATEWAY_CLIENT_NAMES.NODE_HOST : GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT,
|
||||
clientDisplayName: node ? NODE_DISPLAY_NAME : "Node worker launch wire operator",
|
||||
clientVersion: VERSION,
|
||||
platform: node ? "macos" : process.platform,
|
||||
deviceFamily: node ? "Mac" : undefined,
|
||||
mode: node ? GATEWAY_CLIENT_MODES.NODE : GATEWAY_CLIENT_MODES.BACKEND,
|
||||
scopes: node ? [] : ["operator.admin", "operator.pairing", "operator.read", "operator.write"],
|
||||
caps: node ? ["system"] : undefined,
|
||||
commands: node ? [] : undefined,
|
||||
deviceIdentity: params.identity,
|
||||
requestTimeoutMs: PROOF_TIMEOUT_MS,
|
||||
onEvent: params.onEvent,
|
||||
onHelloOk: () => finish(),
|
||||
onConnectError: (error) => finish(error),
|
||||
onClose: (code, reason) => finish(new Error(`Gateway closed (${code}): ${reason}`)),
|
||||
});
|
||||
client.start();
|
||||
});
|
||||
}
|
||||
|
||||
function isPairingRequired(error: unknown): boolean {
|
||||
const details =
|
||||
error && typeof error === "object"
|
||||
? (error as { details?: { code?: unknown } }).details
|
||||
: undefined;
|
||||
return details?.code === "PAIRING_REQUIRED" || String(error).includes("PAIRING_REQUIRED");
|
||||
}
|
||||
|
||||
async function approvePairing(operator: GatewayClient, nodeId: string): Promise<void> {
|
||||
let deviceRequestId: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; deviceId?: string; role?: string }>;
|
||||
}>("device.pair.list", {});
|
||||
deviceRequestId = result.pending?.find(
|
||||
(entry) => entry.deviceId === nodeId || entry.role === "node",
|
||||
)?.requestId;
|
||||
expect(deviceRequestId).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await operator.request("device.pair.approve", { requestId: deviceRequestId });
|
||||
|
||||
await approveNodePairing(operator, nodeId);
|
||||
}
|
||||
|
||||
async function approveNodePairing(operator: GatewayClient, nodeId: string): Promise<void> {
|
||||
let nodeRequestId: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; nodeId?: string }>;
|
||||
}>("node.pair.list", {});
|
||||
nodeRequestId = result.pending?.find((entry) => entry.nodeId === nodeId)?.requestId;
|
||||
expect(nodeRequestId).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await operator.request("node.pair.approve", { requestId: nodeRequestId });
|
||||
}
|
||||
|
||||
async function ensureNodeApproved(operator: GatewayClient, nodeId: string): Promise<boolean> {
|
||||
let approvalState: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{ nodes?: NodeRead[] }>("node.list", {});
|
||||
approvalState = result.nodes?.find((node) => node.nodeId === nodeId)?.approvalState;
|
||||
expect(approvalState).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
if (approvalState !== "approved") {
|
||||
await approveNodePairing(operator, nodeId);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
async function connectPairedNode(params: {
|
||||
gateway: Gateway;
|
||||
operator: GatewayClient;
|
||||
identity: DeviceIdentity;
|
||||
bundlePrewarm?: 1;
|
||||
onEvent: (event: GatewayEvent) => void;
|
||||
}): Promise<GatewayClient> {
|
||||
const connect = () =>
|
||||
connectClient({
|
||||
gateway: params.gateway,
|
||||
role: "node",
|
||||
identity: params.identity,
|
||||
onEvent: params.onEvent,
|
||||
});
|
||||
let client: GatewayClient;
|
||||
try {
|
||||
client = await connect();
|
||||
} catch (error) {
|
||||
if (!isPairingRequired(error)) {
|
||||
throw error;
|
||||
}
|
||||
await approvePairing(params.operator, params.identity.deviceId);
|
||||
client = await connect();
|
||||
}
|
||||
if (await ensureNodeApproved(params.operator, params.identity.deviceId)) {
|
||||
await client.stopAndWait({ timeoutMs: 2_000 });
|
||||
client = await connect();
|
||||
}
|
||||
await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, {
|
||||
protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE],
|
||||
workerHost: {
|
||||
enabled: true,
|
||||
capacity: "available",
|
||||
...(params.bundlePrewarm ? { bundlePrewarm: params.bundlePrewarm } : {}),
|
||||
},
|
||||
});
|
||||
return client;
|
||||
}
|
||||
|
||||
async function waitForApprovedNode(operator: GatewayClient, nodeId: string): Promise<NodeRead> {
|
||||
let approved: NodeRead | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{ nodes?: NodeRead[] }>("node.list", {});
|
||||
approved = result.nodes?.find((node) => node.nodeId === nodeId);
|
||||
expect(approved).toMatchObject({
|
||||
nodeId,
|
||||
approvalState: "approved",
|
||||
connected: true,
|
||||
paired: true,
|
||||
sessionHost: true,
|
||||
});
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
if (!approved) {
|
||||
throw new Error("paired worker node did not become available");
|
||||
}
|
||||
return approved;
|
||||
}
|
||||
|
||||
function messageText(message: unknown): string {
|
||||
const content = (message as { content?: unknown })?.content;
|
||||
if (typeof content === "string") {
|
||||
return content;
|
||||
}
|
||||
return Array.isArray(content)
|
||||
? content
|
||||
.flatMap((part) =>
|
||||
part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string"
|
||||
? [(part as { text: string }).text]
|
||||
: [],
|
||||
)
|
||||
.join("")
|
||||
: "";
|
||||
}
|
||||
|
||||
describe("node worker launch wire", () => {
|
||||
it(
|
||||
"transfers and reconciles a gateway-push workspace through a device runner",
|
||||
@@ -326,48 +52,12 @@ describe("node worker launch wire", () => {
|
||||
async () => {
|
||||
const root = tempDirs.make("openclaw-node-worker-launch-wire-");
|
||||
const provider = await startMidturnProvider();
|
||||
const published = await createPublishedWorkspace(root);
|
||||
const nodeEnv = {
|
||||
...process.env,
|
||||
HOME: path.join(root, "node-home"),
|
||||
NODE_DISABLE_COMPILE_CACHE: undefined,
|
||||
OPENCLAW_STATE_DIR: path.join(root, "node-state"),
|
||||
};
|
||||
await fs.mkdir(nodeEnv.HOME, { recursive: true });
|
||||
const supervisor = createNodeWorkerSupervisor({
|
||||
env: nodeEnv,
|
||||
capacity: FINALIZATION_LOAD_CONCURRENCY,
|
||||
});
|
||||
const bundleInstaller = new NodeWorkerBundleInstaller({ env: nodeEnv });
|
||||
const workspace = new NodeWorkerWorkspaceRuntime({
|
||||
root: path.join(root, "node-workspaces"),
|
||||
env: nodeEnv,
|
||||
});
|
||||
const legacyNodeEnv = {
|
||||
...process.env,
|
||||
HOME: path.join(root, "legacy-node-home"),
|
||||
NODE_DISABLE_COMPILE_CACHE: undefined,
|
||||
OPENCLAW_STATE_DIR: path.join(root, "legacy-node-state"),
|
||||
};
|
||||
await fs.mkdir(legacyNodeEnv.HOME, { recursive: true });
|
||||
const legacySupervisor = createNodeWorkerSupervisor({ env: legacyNodeEnv });
|
||||
const legacyBundleInstaller = new NodeWorkerBundleInstaller({ env: legacyNodeEnv });
|
||||
const legacyWorkspace = new NodeWorkerWorkspaceRuntime({
|
||||
root: path.join(root, "legacy-node-workspaces"),
|
||||
env: legacyNodeEnv,
|
||||
});
|
||||
let gateway: Gateway | undefined;
|
||||
const published = await createPublishedWireWorkspace(root);
|
||||
let gateway: WireGateway | undefined;
|
||||
let operator: GatewayClient | undefined;
|
||||
let node: GatewayClient | undefined;
|
||||
let legacyNode: GatewayClient | undefined;
|
||||
let closing = false;
|
||||
let workerNode: PairedNodeWorkerHost | undefined;
|
||||
let legacyWorkerNode: PairedNodeWorkerHost | undefined;
|
||||
let reconnected = false;
|
||||
const invokeTasks = new Set<Promise<void>>();
|
||||
const invokeErrors: unknown[] = [];
|
||||
const legacyInvokeTasks = new Set<Promise<void>>();
|
||||
const legacyInvokeErrors: unknown[] = [];
|
||||
const commands: string[] = [];
|
||||
const legacyCommands: string[] = [];
|
||||
let bundlePrewarm: unknown;
|
||||
let legacyBundlePrewarm: unknown;
|
||||
let launchId: string | undefined;
|
||||
@@ -379,94 +69,46 @@ describe("node worker launch wire", () => {
|
||||
});
|
||||
|
||||
try {
|
||||
gateway = await startQaGatewayChild({
|
||||
repoRoot: process.cwd(),
|
||||
useRepoCli: true,
|
||||
providerBaseUrl: `${provider.baseUrl}/v1`,
|
||||
providerMode: "mock-openai",
|
||||
primaryModel: MODEL_REF,
|
||||
alternateModel: MODEL_REF,
|
||||
transportBaseUrl: "http://127.0.0.1",
|
||||
controlUiEnabled: false,
|
||||
mutateConfig: (config) => ({
|
||||
...config,
|
||||
nodeHost: {
|
||||
...config.nodeHost,
|
||||
workerRuns: { enabled: true },
|
||||
},
|
||||
}),
|
||||
});
|
||||
operator = await connectClient({ gateway, role: "operator", identity: null });
|
||||
const identity = loadOrCreateDeviceIdentity({
|
||||
path: path.join(root, "node-identity.sqlite"),
|
||||
});
|
||||
const onNodeEvent = (event: GatewayEvent) => {
|
||||
if (event.event !== "node.invoke.request" || !node) {
|
||||
return;
|
||||
}
|
||||
const receiver = node;
|
||||
const frame = event.payload as NodeInvokeRequestPayload;
|
||||
commands.push(frame.command);
|
||||
if (frame.command === NODE_WORKER_WORKSPACE_EXEC_COMMAND && frame.paramsJSON) {
|
||||
const workspaceCommand = JSON.parse(frame.paramsJSON) as {
|
||||
transfer?: { direction?: unknown };
|
||||
};
|
||||
if (
|
||||
observeFinalizationLoad &&
|
||||
workspaceCommand.transfer?.direction === "upload" &&
|
||||
!finalizationStartedAt
|
||||
) {
|
||||
finalizationStartedAt = performance.now();
|
||||
resolveFinalizationStarted();
|
||||
}
|
||||
}
|
||||
if (frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && frame.paramsJSON) {
|
||||
launchId = (JSON.parse(frame.paramsJSON) as { launchId?: string }).launchId;
|
||||
}
|
||||
if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) {
|
||||
bundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown })
|
||||
.bundlePrewarm;
|
||||
}
|
||||
const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, {
|
||||
workerBundleInstaller: bundleInstaller,
|
||||
workerSupervisor: supervisor,
|
||||
workerWorkspace: workspace,
|
||||
gatewayUrl: gateway!.wsUrl,
|
||||
})
|
||||
.then(async () => {
|
||||
if (
|
||||
frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND &&
|
||||
!reconnected &&
|
||||
!closing
|
||||
) {
|
||||
reconnected = true;
|
||||
await receiver.stopAndWait({ timeoutMs: 2_000 });
|
||||
if (!closing) {
|
||||
node = await connectPairedNode({
|
||||
gateway: gateway!,
|
||||
operator: operator!,
|
||||
identity,
|
||||
bundlePrewarm: 1,
|
||||
onEvent: onNodeEvent,
|
||||
});
|
||||
}
|
||||
}
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
invokeErrors.push(error);
|
||||
})
|
||||
.finally(() => invokeTasks.delete(task));
|
||||
invokeTasks.add(task);
|
||||
};
|
||||
node = await connectPairedNode({
|
||||
gateway = await startPairedNodeWorkerGateway({ providerBaseUrl: provider.baseUrl });
|
||||
operator = await connectWireClient({ gateway, role: "operator", identity: null });
|
||||
workerNode = await createPairedNodeWorkerHost({
|
||||
gateway,
|
||||
operator,
|
||||
identity,
|
||||
bundlePrewarm: 1,
|
||||
onEvent: onNodeEvent,
|
||||
root,
|
||||
capacity: FINALIZATION_LOAD_CONCURRENCY,
|
||||
bundlePrewarm: true,
|
||||
onInvoke: (frame) => {
|
||||
if (frame.command === NODE_WORKER_WORKSPACE_EXEC_COMMAND && frame.paramsJSON) {
|
||||
const workspaceCommand = JSON.parse(frame.paramsJSON) as {
|
||||
transfer?: { direction?: unknown };
|
||||
};
|
||||
if (
|
||||
observeFinalizationLoad &&
|
||||
workspaceCommand.transfer?.direction === "upload" &&
|
||||
!finalizationStartedAt
|
||||
) {
|
||||
finalizationStartedAt = performance.now();
|
||||
resolveFinalizationStarted();
|
||||
}
|
||||
}
|
||||
if (frame.command === NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND && frame.paramsJSON) {
|
||||
launchId = (JSON.parse(frame.paramsJSON) as { launchId?: string }).launchId;
|
||||
}
|
||||
if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) {
|
||||
bundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown })
|
||||
.bundlePrewarm;
|
||||
}
|
||||
},
|
||||
afterInvoke: async (frame, host) => {
|
||||
if (frame.command !== NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND || reconnected) {
|
||||
return;
|
||||
}
|
||||
reconnected = true;
|
||||
await host.disconnect();
|
||||
await host.connect();
|
||||
},
|
||||
});
|
||||
const listed = await waitForApprovedNode(operator, identity.deviceId);
|
||||
expect(listed.sessionHost).toBe(true);
|
||||
expect(workerNode.client).toBeTruthy();
|
||||
|
||||
await operator.request("sessions.create", {
|
||||
key: SESSION_KEY,
|
||||
@@ -487,7 +129,7 @@ describe("node worker launch wire", () => {
|
||||
);
|
||||
const dispatched = await gateway.call(
|
||||
"sessions.dispatch",
|
||||
{ key: SESSION_KEY, deviceId: identity.deviceId },
|
||||
{ key: SESSION_KEY, deviceId: workerNode.identity.deviceId },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS },
|
||||
);
|
||||
const placement = (dispatched as { placement?: Record<string, unknown> }).placement;
|
||||
@@ -523,16 +165,18 @@ describe("node worker launch wire", () => {
|
||||
`node worker turn failed: ${JSON.stringify(completed)}\n${gateway.logs().slice(-12_000)}`,
|
||||
);
|
||||
}
|
||||
await Promise.all([...invokeTasks]);
|
||||
expect(invokeErrors).toEqual([]);
|
||||
await workerNode.waitForInvokes();
|
||||
expect(workerNode.invokeErrors).toEqual([]);
|
||||
expect(reconnected).toBe(true);
|
||||
expect(commands).toContain(NODE_WORKER_BUNDLE_INSTALL_COMMAND);
|
||||
expect(commands).toContain(NODE_WORKER_WORKSPACE_EXEC_COMMAND);
|
||||
expect(commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND);
|
||||
expect(commands).toContain(NODE_WORKER_SUPERVISOR_STATUS_COMMAND);
|
||||
expect(workerNode.commands).toContain(NODE_WORKER_BUNDLE_INSTALL_COMMAND);
|
||||
expect(workerNode.commands).toContain(NODE_WORKER_WORKSPACE_EXEC_COMMAND);
|
||||
expect(workerNode.commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND);
|
||||
expect(workerNode.commands).toContain(NODE_WORKER_SUPERVISOR_STATUS_COMMAND);
|
||||
expect(launchId).toBeTruthy();
|
||||
expect(bundlePrewarm).toBe(1);
|
||||
await expect(supervisor.status(launchId!)).resolves.toMatchObject({ state: "completed" });
|
||||
await expect(workerNode.supervisor.status(launchId!)).resolves.toMatchObject({
|
||||
state: "completed",
|
||||
});
|
||||
|
||||
const history = await operator.request<{ messages?: unknown[] }>("chat.history", {
|
||||
sessionKey: SESSION_KEY,
|
||||
@@ -542,7 +186,7 @@ describe("node worker launch wire", () => {
|
||||
history.messages?.filter(
|
||||
(message) =>
|
||||
(message as { role?: unknown }).role === "assistant" &&
|
||||
messageText(message).includes(BASELINE_REPLY),
|
||||
wireMessageText(message).includes(BASELINE_REPLY),
|
||||
),
|
||||
).toHaveLength(1);
|
||||
const described = (await gateway.call("sessions.describe", { key: SESSION_KEY })) as {
|
||||
@@ -563,39 +207,18 @@ describe("node worker launch wire", () => {
|
||||
"device result\n",
|
||||
);
|
||||
|
||||
const legacyIdentity = loadOrCreateDeviceIdentity({
|
||||
path: path.join(root, "legacy-node-identity.sqlite"),
|
||||
});
|
||||
const onLegacyNodeEvent = (event: GatewayEvent) => {
|
||||
if (event.event !== "node.invoke.request" || !legacyNode) {
|
||||
return;
|
||||
}
|
||||
const receiver = legacyNode;
|
||||
const frame = event.payload as NodeInvokeRequestPayload;
|
||||
legacyCommands.push(frame.command);
|
||||
if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) {
|
||||
legacyBundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown })
|
||||
.bundlePrewarm;
|
||||
}
|
||||
const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, {
|
||||
workerBundleInstaller: legacyBundleInstaller,
|
||||
workerSupervisor: legacySupervisor,
|
||||
workerWorkspace: legacyWorkspace,
|
||||
gatewayUrl: gateway!.wsUrl,
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
legacyInvokeErrors.push(error);
|
||||
})
|
||||
.finally(() => legacyInvokeTasks.delete(task));
|
||||
legacyInvokeTasks.add(task);
|
||||
};
|
||||
legacyNode = await connectPairedNode({
|
||||
legacyWorkerNode = await createPairedNodeWorkerHost({
|
||||
gateway,
|
||||
operator,
|
||||
identity: legacyIdentity,
|
||||
onEvent: onLegacyNodeEvent,
|
||||
root,
|
||||
label: "legacy-node",
|
||||
onInvoke: (frame) => {
|
||||
if (frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND && frame.paramsJSON) {
|
||||
legacyBundlePrewarm = (JSON.parse(frame.paramsJSON) as { bundlePrewarm?: unknown })
|
||||
.bundlePrewarm;
|
||||
}
|
||||
},
|
||||
});
|
||||
await waitForApprovedNode(operator, legacyIdentity.deviceId);
|
||||
const legacySessionKey = `${SESSION_KEY}-legacy-node`;
|
||||
await operator.request("sessions.create", {
|
||||
key: legacySessionKey,
|
||||
@@ -607,7 +230,7 @@ describe("node worker launch wire", () => {
|
||||
});
|
||||
await gateway.call(
|
||||
"sessions.dispatch",
|
||||
{ key: legacySessionKey, deviceId: legacyIdentity.deviceId },
|
||||
{ key: legacySessionKey, deviceId: legacyWorkerNode.identity.deviceId },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS },
|
||||
);
|
||||
const legacyRunId = `node-worker-launch-wire-legacy-${Date.now()}`;
|
||||
@@ -624,9 +247,9 @@ describe("node worker launch wire", () => {
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS + 5_000 },
|
||||
),
|
||||
).resolves.toMatchObject({ status: "ok" });
|
||||
await Promise.all([...legacyInvokeTasks]);
|
||||
expect(legacyInvokeErrors).toEqual([]);
|
||||
expect(legacyCommands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND);
|
||||
await legacyWorkerNode.waitForInvokes();
|
||||
expect(legacyWorkerNode.invokeErrors).toEqual([]);
|
||||
expect(legacyWorkerNode.commands).toContain(NODE_WORKER_SUPERVISOR_LAUNCH_COMMAND);
|
||||
expect(legacyBundlePrewarm).toBeUndefined();
|
||||
|
||||
const loadSessions: string[] = [];
|
||||
@@ -642,7 +265,7 @@ describe("node worker launch wire", () => {
|
||||
});
|
||||
await gateway.call(
|
||||
"sessions.dispatch",
|
||||
{ key: sessionKey, deviceId: identity.deviceId },
|
||||
{ key: sessionKey, deviceId: workerNode.identity.deviceId },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS },
|
||||
);
|
||||
loadSessions.push(sessionKey);
|
||||
@@ -675,25 +298,25 @@ describe("node worker launch wire", () => {
|
||||
})();
|
||||
const loadRunIds = await Promise.all(
|
||||
loadSessions.map(async (sessionKey, index) => {
|
||||
const runId = `node-worker-finalization-load-${index}-${Date.now()}`;
|
||||
const started = await operator!.request<{ runId?: string; status?: string }>(
|
||||
const loadRunId = `node-worker-finalization-load-${index}-${Date.now()}`;
|
||||
const loadStarted = await operator!.request<{ runId?: string; status?: string }>(
|
||||
"chat.send",
|
||||
{
|
||||
sessionKey,
|
||||
message: BASELINE_PROMPT,
|
||||
deliver: false,
|
||||
idempotencyKey: runId,
|
||||
idempotencyKey: loadRunId,
|
||||
},
|
||||
);
|
||||
expect(started).toMatchObject({ runId, status: "started" });
|
||||
return runId;
|
||||
expect(loadStarted).toMatchObject({ runId: loadRunId, status: "started" });
|
||||
return loadRunId;
|
||||
}),
|
||||
);
|
||||
const waits = Promise.all(
|
||||
loadRunIds.map(async (runId) => {
|
||||
loadRunIds.map(async (loadRunId) => {
|
||||
const completedLoad = await operator!.request<{ status?: string }>(
|
||||
"agent.wait",
|
||||
{ runId, timeoutMs: PROOF_TIMEOUT_MS },
|
||||
{ runId: loadRunId, timeoutMs: PROOF_TIMEOUT_MS },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS + 5_000 },
|
||||
);
|
||||
expect(completedLoad.status).toBe("ok");
|
||||
@@ -701,7 +324,7 @@ describe("node worker launch wire", () => {
|
||||
);
|
||||
await finalizationStarted;
|
||||
const freshConnectionStartedAt = performance.now();
|
||||
const freshClient = await connectClient({
|
||||
const freshClient = await connectWireClient({
|
||||
gateway,
|
||||
role: "operator",
|
||||
identity: null,
|
||||
@@ -723,18 +346,13 @@ describe("node worker launch wire", () => {
|
||||
);
|
||||
expect(freshConnectionMs).toBeLessThan(CONTROL_PROBE_MAX_MS);
|
||||
} finally {
|
||||
closing = true;
|
||||
const cleanup = await Promise.allSettled([
|
||||
node?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(),
|
||||
legacyNode?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(),
|
||||
workerNode?.stop() ?? Promise.resolve(),
|
||||
legacyWorkerNode?.stop() ?? Promise.resolve(),
|
||||
operator?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(),
|
||||
Promise.allSettled([...invokeTasks]),
|
||||
Promise.allSettled([...legacyInvokeTasks]),
|
||||
supervisor.close(),
|
||||
legacySupervisor.close(),
|
||||
gateway?.stop() ?? Promise.resolve(),
|
||||
provider.stop(),
|
||||
closeServer(published.server),
|
||||
closeWireServer(published.server),
|
||||
]);
|
||||
const failures = cleanup.flatMap((result) =>
|
||||
result.status === "rejected" ? [result.reason] : [],
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createServer, type ServerResponse } from "node:http";
|
||||
|
||||
type Deferred = { promise: Promise<void>; resolve: () => void };
|
||||
|
||||
function createDeferred(): Deferred {
|
||||
let resolve = () => {};
|
||||
const promise = new Promise<void>((settle) => {
|
||||
resolve = settle;
|
||||
});
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
async function readBody(request: AsyncIterable<unknown>): Promise<string> {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of request) {
|
||||
chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk)));
|
||||
}
|
||||
return Buffer.concat(chunks).toString("utf8");
|
||||
}
|
||||
|
||||
function responseText(raw: string): string {
|
||||
const matches = [...raw.matchAll(/Reply exactly:\s*([A-Z0-9_-]+)/gu)];
|
||||
return matches.at(-1)?.[1] ?? "WIRE-OK";
|
||||
}
|
||||
|
||||
function writeEvent(response: ServerResponse, event: unknown): void {
|
||||
response.write(`data: ${JSON.stringify(event)}\n\n`);
|
||||
}
|
||||
|
||||
function writeReply(response: ServerResponse, text: string): void {
|
||||
if (!response.headersSent) {
|
||||
response.writeHead(200, {
|
||||
"content-type": "text/event-stream",
|
||||
"cache-control": "no-store",
|
||||
connection: "keep-alive",
|
||||
});
|
||||
}
|
||||
const id = `msg_${randomUUID()}`;
|
||||
const item = {
|
||||
type: "message",
|
||||
id,
|
||||
role: "assistant",
|
||||
phase: "final_answer",
|
||||
status: "completed",
|
||||
content: [{ type: "output_text", text, annotations: [] }],
|
||||
};
|
||||
writeEvent(response, {
|
||||
type: "response.output_item.added",
|
||||
output_index: 0,
|
||||
item: { ...item, status: "in_progress", content: [] },
|
||||
});
|
||||
writeEvent(response, {
|
||||
type: "response.output_text.delta",
|
||||
item_id: id,
|
||||
output_index: 0,
|
||||
content_index: 0,
|
||||
delta: text,
|
||||
});
|
||||
writeEvent(response, {
|
||||
type: "response.output_text.done",
|
||||
item_id: id,
|
||||
output_index: 0,
|
||||
content_index: 0,
|
||||
text,
|
||||
});
|
||||
writeEvent(response, { type: "response.output_item.done", output_index: 0, item });
|
||||
writeEvent(response, {
|
||||
type: "response.completed",
|
||||
response: {
|
||||
id: `resp_${id}`,
|
||||
status: "completed",
|
||||
output: [item],
|
||||
usage: { input_tokens: 32, output_tokens: 8, total_tokens: 40 },
|
||||
},
|
||||
});
|
||||
response.end("data: [DONE]\n\n");
|
||||
}
|
||||
|
||||
export async function startPairedNodeWorkerLifecycleProvider(holdMarkers: readonly string[]) {
|
||||
const holdSet = new Set(holdMarkers);
|
||||
const releases = new Map(holdMarkers.map((marker) => [marker, createDeferred()] as const));
|
||||
const held = new Set<string>();
|
||||
const pendingResponses = new Set<ServerResponse>();
|
||||
const server = createServer((request, response) => {
|
||||
void (async () => {
|
||||
if (request.method === "GET" && request.url === "/v1/models") {
|
||||
response.writeHead(200, { "content-type": "application/json" });
|
||||
response.end(JSON.stringify({ data: [{ id: "gpt-5.6-luna", object: "model" }] }));
|
||||
return;
|
||||
}
|
||||
if (request.method !== "POST" || request.url !== "/v1/responses") {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
const raw = await readBody(request);
|
||||
const text = responseText(raw);
|
||||
if (holdSet.has(text)) {
|
||||
held.add(text);
|
||||
pendingResponses.add(response);
|
||||
response.once("close", () => pendingResponses.delete(response));
|
||||
response.writeHead(200, {
|
||||
"content-type": "text/event-stream",
|
||||
"cache-control": "no-store",
|
||||
connection: "keep-alive",
|
||||
});
|
||||
response.flushHeaders();
|
||||
await releases.get(text)!.promise;
|
||||
if (!response.destroyed) {
|
||||
writeReply(response, text);
|
||||
}
|
||||
pendingResponses.delete(response);
|
||||
return;
|
||||
}
|
||||
writeReply(response, text);
|
||||
})().catch((error: unknown) => {
|
||||
if (!response.headersSent) {
|
||||
response.writeHead(500);
|
||||
}
|
||||
response.end(error instanceof Error ? error.message : String(error));
|
||||
});
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
throw new Error("paired node lifecycle provider did not bind");
|
||||
}
|
||||
return {
|
||||
baseUrl: `http://127.0.0.1:${address.port}`,
|
||||
hasHeld(marker: string): boolean {
|
||||
return held.has(marker);
|
||||
},
|
||||
release(marker: string): void {
|
||||
releases.get(marker)?.resolve();
|
||||
},
|
||||
releaseAll(): void {
|
||||
for (const release of releases.values()) {
|
||||
release.resolve();
|
||||
}
|
||||
},
|
||||
async stop(): Promise<void> {
|
||||
for (const release of releases.values()) {
|
||||
release.resolve();
|
||||
}
|
||||
for (const response of pendingResponses) {
|
||||
response.destroy();
|
||||
}
|
||||
pendingResponses.clear();
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>((resolve) => server.close(() => resolve()));
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,351 @@
|
||||
import fs from "node:fs/promises";
|
||||
import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { NODE_WORKER_WORKSPACE_RETAIN_COMMAND } from "../../../../src/infra/node-commands.js";
|
||||
import { useAutoCleanupTempDirTracker } from "../../../helpers/temp-dir.js";
|
||||
import { PROOF_TIMEOUT_MS } from "./cloud-worker-midturn-loss-fixture.js";
|
||||
import { startPairedNodeWorkerLifecycleProvider } from "./paired-node-worker-lifecycle-provider.js";
|
||||
import {
|
||||
bundleInstallFrames,
|
||||
closeWireServer,
|
||||
connectWireClient,
|
||||
createPairedNodeWorkerHost,
|
||||
createPublishedWireWorkspace,
|
||||
startPairedNodeWorkerGateway,
|
||||
type PairedNodeWorkerHost,
|
||||
type PublishedWireWorkspace,
|
||||
type WireGateway,
|
||||
type WireNodeRead,
|
||||
wireMessageText,
|
||||
} from "./paired-node-worker-wire-fixture.js";
|
||||
|
||||
const TEST_TIMEOUT_MS = PROOF_TIMEOUT_MS + 180_000;
|
||||
const SESSION_PREFIX = "agent:qa:paired-node-worker-lifecycle";
|
||||
const HOLD_A = "WIRE-HOLD-A";
|
||||
const HOLD_B = "WIRE-HOLD-B";
|
||||
|
||||
type TurnResult = { runId?: string; status?: string; summary?: string };
|
||||
type Placement = { environmentId?: string; state?: string; workerBundleHash?: string };
|
||||
type EnvironmentRead = {
|
||||
id: string;
|
||||
type: string;
|
||||
workerBundle?: WireNodeRead["workerBundle"];
|
||||
worker?: { state?: string };
|
||||
};
|
||||
|
||||
const tempDirs = useAutoCleanupTempDirTracker(afterEach);
|
||||
|
||||
async function createSession(params: {
|
||||
operator: GatewayClient;
|
||||
published: PublishedWireWorkspace;
|
||||
suffix: string;
|
||||
}): Promise<string> {
|
||||
const key = `${SESSION_PREFIX}:${params.suffix}`;
|
||||
await params.operator.request("sessions.create", {
|
||||
key,
|
||||
agentId: "qa",
|
||||
worktree: true,
|
||||
worktreeName: `paired-node-${params.suffix}`,
|
||||
worktreeBaseRef: "main",
|
||||
cwd: params.published.source,
|
||||
});
|
||||
return key;
|
||||
}
|
||||
|
||||
async function dispatchNodeSession(params: {
|
||||
gateway: WireGateway;
|
||||
key: string;
|
||||
nodeId: string;
|
||||
}): Promise<Placement> {
|
||||
const result = (await params.gateway.call(
|
||||
"sessions.dispatch",
|
||||
{ key: params.key, deviceId: params.nodeId },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS },
|
||||
)) as { placement?: Placement };
|
||||
expect(result.placement).toMatchObject({ state: "active" });
|
||||
return result.placement!;
|
||||
}
|
||||
|
||||
async function startTurn(params: {
|
||||
operator: GatewayClient;
|
||||
key: string;
|
||||
marker: string;
|
||||
}): Promise<string> {
|
||||
const runId = `${params.marker.toLowerCase()}-${Date.now()}`;
|
||||
const started = await params.operator.request<TurnResult>("chat.send", {
|
||||
sessionKey: params.key,
|
||||
message: `Reply exactly: ${params.marker}`,
|
||||
deliver: false,
|
||||
idempotencyKey: runId,
|
||||
});
|
||||
expect(started).toMatchObject({ runId, status: "started" });
|
||||
return runId;
|
||||
}
|
||||
|
||||
async function waitForTurn(operator: GatewayClient, runId: string): Promise<TurnResult> {
|
||||
return await operator.request<TurnResult>(
|
||||
"agent.wait",
|
||||
{ runId, timeoutMs: PROOF_TIMEOUT_MS },
|
||||
{ timeoutMs: PROOF_TIMEOUT_MS + 5_000 },
|
||||
);
|
||||
}
|
||||
|
||||
async function expectSuccessfulTurn(params: {
|
||||
operator: GatewayClient;
|
||||
key: string;
|
||||
marker: string;
|
||||
}): Promise<void> {
|
||||
const runId = await startTurn(params);
|
||||
await expect(waitForTurn(params.operator, runId)).resolves.toMatchObject({ status: "ok" });
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const history = await params.operator.request<{ messages?: unknown[] }>("chat.history", {
|
||||
sessionKey: params.key,
|
||||
limit: 100,
|
||||
});
|
||||
expect(
|
||||
history.messages?.some(
|
||||
(message) =>
|
||||
(message as { role?: unknown }).role === "assistant" &&
|
||||
wireMessageText(message).includes(params.marker),
|
||||
),
|
||||
).toBe(true);
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
}
|
||||
|
||||
async function describePlacement(gateway: WireGateway, key: string): Promise<Placement> {
|
||||
const described = (await gateway.call("sessions.describe", { key })) as {
|
||||
session?: { placement?: Placement };
|
||||
};
|
||||
return described.session?.placement ?? {};
|
||||
}
|
||||
|
||||
async function readNode(
|
||||
operator: GatewayClient,
|
||||
nodeId: string,
|
||||
): Promise<WireNodeRead | undefined> {
|
||||
const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {});
|
||||
return result.nodes?.find((node) => node.nodeId === nodeId);
|
||||
}
|
||||
|
||||
async function readEnvironments(operator: GatewayClient): Promise<EnvironmentRead[]> {
|
||||
const result = await operator.request<{ environments?: EnvironmentRead[] }>(
|
||||
"environments.list",
|
||||
{},
|
||||
);
|
||||
return result.environments ?? [];
|
||||
}
|
||||
|
||||
async function expectPublicBundleStatus(params: {
|
||||
operator: GatewayClient;
|
||||
nodeId: string;
|
||||
status: "installed" | "missing";
|
||||
}): Promise<void> {
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const node = await readNode(params.operator, params.nodeId);
|
||||
const environment = (await readEnvironments(params.operator)).find(
|
||||
(entry) => entry.id === `node:${params.nodeId}`,
|
||||
);
|
||||
expect(node?.workerBundle?.status).toBe(params.status);
|
||||
expect(environment?.workerBundle?.status).toBe(params.status);
|
||||
for (const projection of [node?.workerBundle, environment?.workerBundle]) {
|
||||
expect(Object.keys(projection ?? {}).toSorted()).toEqual(
|
||||
params.status === "installed" ? ["status", "version"] : ["status"],
|
||||
);
|
||||
expect(projection).not.toHaveProperty("bundleHash");
|
||||
expect(projection).not.toHaveProperty("path");
|
||||
}
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
}
|
||||
|
||||
describe("paired node worker lifecycle wire", () => {
|
||||
it(
|
||||
"keeps local control usable across bundle loss, disconnect, capacity, and role removal",
|
||||
{ timeout: TEST_TIMEOUT_MS },
|
||||
async () => {
|
||||
const root = tempDirs.make("openclaw-paired-node-worker-lifecycle-");
|
||||
const provider = await startPairedNodeWorkerLifecycleProvider([HOLD_A, HOLD_B]);
|
||||
const published = await createPublishedWireWorkspace(root);
|
||||
let gateway: WireGateway | undefined;
|
||||
let operator: GatewayClient | undefined;
|
||||
let workerNode: PairedNodeWorkerHost | undefined;
|
||||
try {
|
||||
gateway = await startPairedNodeWorkerGateway({ providerBaseUrl: provider.baseUrl });
|
||||
operator = await connectWireClient({ gateway, role: "operator", identity: null });
|
||||
workerNode = await createPairedNodeWorkerHost({
|
||||
gateway,
|
||||
operator,
|
||||
root,
|
||||
capacity: 2,
|
||||
capacityWaitMs: 750,
|
||||
bundlePrewarm: true,
|
||||
bundleRetention: true,
|
||||
bundleStatus: true,
|
||||
});
|
||||
const nodeId = workerNode.identity.deviceId;
|
||||
const localKey = await createSession({ operator, published, suffix: "local-control" });
|
||||
const retainedKey = await createSession({ operator, published, suffix: "retention" });
|
||||
const retainedPlacement = await dispatchNodeSession({
|
||||
gateway,
|
||||
key: retainedKey,
|
||||
nodeId,
|
||||
});
|
||||
const bundleHash = retainedPlacement.workerBundleHash;
|
||||
expect(bundleHash).toMatch(/^[a-f0-9]{64}$/u);
|
||||
|
||||
// Local and node-placed sessions share the Gateway without sharing failure state.
|
||||
await expectSuccessfulTurn({ operator, key: retainedKey, marker: "WIRE-NODE-BASELINE" });
|
||||
await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-BASELINE" });
|
||||
|
||||
// Bundle maintenance reports only the public status/version projection and repairs loss.
|
||||
await expectPublicBundleStatus({ operator, nodeId, status: "installed" });
|
||||
const installedBundle = await workerNode.installedBundleDirectory(bundleHash!);
|
||||
const installCountBeforeLoss = bundleInstallFrames(workerNode).length;
|
||||
await fs.rm(installedBundle, { recursive: true, force: true });
|
||||
await workerNode.disconnect();
|
||||
await workerNode.connect();
|
||||
await expectPublicBundleStatus({ operator, nodeId, status: "missing" });
|
||||
expect(
|
||||
workerNode.commands.filter((command) => command === NODE_WORKER_WORKSPACE_RETAIN_COMMAND)
|
||||
.length,
|
||||
).toBeGreaterThan(0);
|
||||
await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-LOSS" });
|
||||
|
||||
const repairedKey = await createSession({ operator, published, suffix: "reinstalled" });
|
||||
await dispatchNodeSession({ gateway, key: repairedKey, nodeId });
|
||||
await vi.waitFor(
|
||||
() => expect(bundleInstallFrames(workerNode!).length).toBe(installCountBeforeLoss + 1),
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await expectPublicBundleStatus({ operator, nodeId, status: "installed" });
|
||||
|
||||
// An offline runner fails before handoff, leaves the active placement retryable, and
|
||||
// does not terminalize the independent local session.
|
||||
await workerNode.disconnect();
|
||||
const offlineRunId = await startTurn({
|
||||
operator,
|
||||
key: repairedKey,
|
||||
marker: "WIRE-OFFLINE-ATTEMPT",
|
||||
});
|
||||
const offline = await waitForTurn(operator, offlineRunId);
|
||||
expect(offline.status).not.toBe("ok");
|
||||
expect(`${offline.summary ?? ""} ${JSON.stringify(offline)}`).toMatch(
|
||||
/runner-offline|runner is offline|reconnect/iu,
|
||||
);
|
||||
expect(await describePlacement(gateway, repairedKey)).toMatchObject({ state: "active" });
|
||||
await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-OFFLINE" });
|
||||
await workerNode.connect();
|
||||
await expectSuccessfulTurn({ operator, key: repairedKey, marker: "WIRE-OFFLINE-RETRY" });
|
||||
|
||||
// Two nonterminal real worker children consume both physical slots. The third turn
|
||||
// records a bounded capacity failure at the public RPC/history/placement boundaries.
|
||||
const holdAKey = await createSession({ operator, published, suffix: "capacity-a" });
|
||||
const holdBKey = await createSession({ operator, published, suffix: "capacity-b" });
|
||||
const capacityKey = await createSession({ operator, published, suffix: "capacity-c" });
|
||||
await dispatchNodeSession({ gateway, key: holdAKey, nodeId });
|
||||
await dispatchNodeSession({ gateway, key: holdBKey, nodeId });
|
||||
await dispatchNodeSession({ gateway, key: capacityKey, nodeId });
|
||||
const holdARunId = await startTurn({ operator, key: holdAKey, marker: HOLD_A });
|
||||
const holdBRunId = await startTurn({ operator, key: holdBKey, marker: HOLD_B });
|
||||
await vi.waitFor(
|
||||
() => {
|
||||
expect(provider.hasHeld(HOLD_A)).toBe(true);
|
||||
expect(provider.hasHeld(HOLD_B)).toBe(true);
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
const capacityRunId = await startTurn({
|
||||
operator,
|
||||
key: capacityKey,
|
||||
marker: "WIRE-CAPACITY-ATTEMPT",
|
||||
});
|
||||
const capacity = await waitForTurn(operator, capacityRunId);
|
||||
expect(capacity.status).not.toBe("ok");
|
||||
expect(`${capacity.summary ?? ""} ${JSON.stringify(capacity)}`).toMatch(/capacity/iu);
|
||||
expect(await describePlacement(gateway, capacityKey)).toMatchObject({ state: "active" });
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const history = await operator!.request<{ messages?: unknown[] }>("chat.history", {
|
||||
sessionKey: capacityKey,
|
||||
limit: 100,
|
||||
});
|
||||
const messages = history.messages ?? [];
|
||||
expect(
|
||||
messages.some(
|
||||
(message) =>
|
||||
(message as { role?: unknown }).role === "user" &&
|
||||
wireMessageText(message).includes("WIRE-CAPACITY-ATTEMPT"),
|
||||
),
|
||||
).toBe(true);
|
||||
expect(
|
||||
messages.some(
|
||||
(message) =>
|
||||
(message as { role?: unknown }).role === "assistant" &&
|
||||
wireMessageText(message).includes("WIRE-CAPACITY-ATTEMPT"),
|
||||
),
|
||||
).toBe(false);
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AT-CAPACITY" });
|
||||
provider.release(HOLD_A);
|
||||
await expect(waitForTurn(operator, holdARunId)).resolves.toMatchObject({ status: "ok" });
|
||||
await expectSuccessfulTurn({
|
||||
operator,
|
||||
key: capacityKey,
|
||||
marker: "WIRE-CAPACITY-RETRY",
|
||||
});
|
||||
provider.release(HOLD_B);
|
||||
await expect(waitForTurn(operator, holdBRunId)).resolves.toMatchObject({ status: "ok" });
|
||||
|
||||
// Node-role removal waits for environment and placement cleanup before returning,
|
||||
// invalidates the old connection, and leaves local execution available.
|
||||
const removalKey = await createSession({ operator, published, suffix: "role-removal" });
|
||||
const removalPlacement = await dispatchNodeSession({ gateway, key: removalKey, nodeId });
|
||||
expect(workerNode.client).toBeTruthy();
|
||||
await expect(operator.request("node.pair.remove", { nodeId })).resolves.toMatchObject({
|
||||
nodeId,
|
||||
});
|
||||
const removedEnvironment = (await readEnvironments(operator)).find(
|
||||
(entry) => entry.id === removalPlacement.environmentId,
|
||||
);
|
||||
expect(["destroyed", "failed", "orphaned"]).toContain(removedEnvironment?.worker?.state);
|
||||
expect(await describePlacement(gateway, removalKey)).toMatchObject({ state: "failed" });
|
||||
await expect(workerNode.publishInventory()).rejects.toBeTruthy();
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const node = await readNode(operator!, nodeId);
|
||||
expect(node?.connected).not.toBe(true);
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await expectSuccessfulTurn({ operator, key: localKey, marker: "WIRE-LOCAL-AFTER-REMOVAL" });
|
||||
await workerNode.waitForInvokes();
|
||||
expect(workerNode.invokeErrors).toEqual([]);
|
||||
} finally {
|
||||
provider.releaseAll();
|
||||
const cleanup = await Promise.allSettled([
|
||||
workerNode?.stop() ?? Promise.resolve(),
|
||||
operator?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(),
|
||||
gateway?.stop() ?? Promise.resolve(),
|
||||
provider.stop(),
|
||||
closeWireServer(published.server),
|
||||
]);
|
||||
const failures = cleanup.flatMap((result) =>
|
||||
result.status === "rejected" ? [result.reason] : [],
|
||||
);
|
||||
if (failures.length === 1) {
|
||||
throw failures[0];
|
||||
}
|
||||
if (failures.length > 1) {
|
||||
throw new AggregateError(failures, "paired node worker lifecycle cleanup failed");
|
||||
}
|
||||
}
|
||||
},
|
||||
);
|
||||
});
|
||||
@@ -0,0 +1,509 @@
|
||||
import { execFile } from "node:child_process";
|
||||
import fs from "node:fs/promises";
|
||||
import { createServer, type Server } from "node:http";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { GatewayClient } from "openclaw/plugin-sdk/gateway-runtime";
|
||||
import { expect, vi } from "vitest";
|
||||
import { startQaGatewayChild } from "../../../../extensions/qa-lab/api.js";
|
||||
import {
|
||||
GATEWAY_CLIENT_MODES,
|
||||
GATEWAY_CLIENT_NAMES,
|
||||
} from "../../../../packages/gateway-protocol/src/client-info.js";
|
||||
import { WORKER_BUNDLE_PREWARM_VERSION } from "../../../../packages/gateway-protocol/src/schema/worker-admission.js";
|
||||
import type { DeviceIdentity } from "../../../../src/infra/device-identity.js";
|
||||
import { loadOrCreateDeviceIdentity } from "../../../../src/infra/device-identity.js";
|
||||
import { NODE_WORKER_BUNDLE_INSTALL_COMMAND } from "../../../../src/infra/node-commands.js";
|
||||
import {
|
||||
NODE_RUNNER_INVENTORY_UPDATE_METHOD,
|
||||
NODE_WORKER_BUNDLE_RETENTION_VERSION,
|
||||
NODE_WORKER_BUNDLE_STATUS_VERSION,
|
||||
NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE,
|
||||
} from "../../../../src/infra/node-runner-inventory.js";
|
||||
import { handleInvoke, type NodeInvokeRequestPayload } from "../../../../src/node-host/invoke.js";
|
||||
import { NodeWorkerBundleInstaller } from "../../../../src/node-host/node-worker-bundle-installer.js";
|
||||
import { createNodeWorkerSupervisor } from "../../../../src/node-host/node-worker-supervisor.js";
|
||||
import { NodeWorkerWorkspaceRuntime } from "../../../../src/node-host/node-worker-workspace.js";
|
||||
import { VERSION } from "../../../../src/version.js";
|
||||
import { MODEL_REF, PROOF_TIMEOUT_MS } from "./cloud-worker-midturn-loss-fixture.js";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
const NODE_DISPLAY_NAME = "QA Gateway-bundle worker node";
|
||||
|
||||
export type WireGateway = Awaited<ReturnType<typeof startQaGatewayChild>>;
|
||||
type WireGatewayEvent = { event: string; payload?: unknown };
|
||||
export type WireNodeRead = {
|
||||
nodeId: string;
|
||||
approvalState?: string;
|
||||
connected?: boolean;
|
||||
paired?: boolean;
|
||||
sessionHost?: boolean;
|
||||
workerBundle?: { status: "installed"; version: string } | { status: "missing" };
|
||||
};
|
||||
export type PublishedWireWorkspace = {
|
||||
commit: string;
|
||||
source: string;
|
||||
server: Server;
|
||||
};
|
||||
|
||||
async function git(cwd: string, ...args: string[]): Promise<string> {
|
||||
const { stdout } = await execFileAsync("git", ["-C", cwd, ...args], {
|
||||
encoding: "utf8",
|
||||
timeout: 20_000,
|
||||
});
|
||||
return stdout.trim();
|
||||
}
|
||||
|
||||
export async function closeWireServer(server: Server): Promise<void> {
|
||||
server.closeAllConnections();
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.close((error) => (error ? reject(error) : resolve()));
|
||||
});
|
||||
}
|
||||
|
||||
export async function createPublishedWireWorkspace(root: string): Promise<PublishedWireWorkspace> {
|
||||
const source = path.join(root, "source");
|
||||
const bare = path.join(root, "repo.git");
|
||||
await fs.mkdir(source, { recursive: true });
|
||||
await execFileAsync("git", ["init", "--bare", bare]);
|
||||
await git(source, "init", "-b", "main");
|
||||
await git(source, "config", "user.name", "OpenClaw QA");
|
||||
await git(source, "config", "user.email", "openclaw-qa@example.invalid");
|
||||
await fs.mkdir(path.join(source, "nested"));
|
||||
await fs.writeFile(path.join(source, "launch-wire.txt"), "local-install launch wire\n");
|
||||
await fs.writeFile(path.join(source, "nested", "tracked.txt"), "nested tracked input\n");
|
||||
await git(source, "add", ".");
|
||||
await git(source, "commit", "-m", "initialize node worker launch wire workspace");
|
||||
await git(source, "remote", "add", "publish", bare);
|
||||
await git(source, "push", "publish", "main");
|
||||
await git(source, "remote", "remove", "publish");
|
||||
await git(bare, "update-server-info");
|
||||
|
||||
const server = createServer((request, response) => {
|
||||
void (async () => {
|
||||
const pathname = decodeURIComponent(new URL(request.url ?? "/", "http://localhost").pathname);
|
||||
if (!pathname.startsWith("/repo.git/")) {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
const candidate = path.resolve(bare, pathname.slice("/repo.git/".length));
|
||||
if (candidate !== bare && !candidate.startsWith(`${bare}${path.sep}`)) {
|
||||
response.writeHead(404).end();
|
||||
return;
|
||||
}
|
||||
try {
|
||||
const contents = await fs.readFile(candidate);
|
||||
response.writeHead(200, {
|
||||
"content-type": pathname.endsWith("/info/refs")
|
||||
? "text/plain; charset=utf-8"
|
||||
: "application/octet-stream",
|
||||
"content-length": String(contents.byteLength),
|
||||
});
|
||||
response.end(request.method === "HEAD" ? undefined : contents);
|
||||
} catch {
|
||||
response.writeHead(404).end();
|
||||
}
|
||||
})();
|
||||
});
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(0, "127.0.0.1", resolve);
|
||||
});
|
||||
const address = server.address();
|
||||
if (!address || typeof address === "string") {
|
||||
throw new Error("published workspace server did not bind");
|
||||
}
|
||||
const origin = `http://127.0.0.1:${address.port}/repo.git`;
|
||||
await git(source, "remote", "add", "origin", origin);
|
||||
const commit = await git(source, "rev-parse", "HEAD");
|
||||
await git(source, "ls-remote", "--exit-code", origin, "refs/heads/main");
|
||||
return { commit, source: await fs.realpath(source), server };
|
||||
}
|
||||
|
||||
export async function connectWireClient(params: {
|
||||
gateway: WireGateway;
|
||||
role: "operator" | "node";
|
||||
identity: DeviceIdentity | null;
|
||||
onEvent?: (event: WireGatewayEvent) => void;
|
||||
timeoutMs?: number;
|
||||
}): Promise<GatewayClient> {
|
||||
return await new Promise<GatewayClient>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const finish = (error?: Error) => {
|
||||
if (settled) {
|
||||
return;
|
||||
}
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
if (error) {
|
||||
client.stop();
|
||||
reject(error);
|
||||
} else {
|
||||
resolve(client);
|
||||
}
|
||||
};
|
||||
const timeout = setTimeout(
|
||||
() => finish(new Error("Gateway client connection timed out")),
|
||||
params.timeoutMs ?? 30_000,
|
||||
);
|
||||
timeout.unref();
|
||||
const node = params.role === "node";
|
||||
const client = new GatewayClient({
|
||||
url: params.gateway.wsUrl,
|
||||
token: params.gateway.token,
|
||||
env: params.gateway.runtimeEnv,
|
||||
role: params.role,
|
||||
clientName: node ? GATEWAY_CLIENT_NAMES.NODE_HOST : GATEWAY_CLIENT_NAMES.GATEWAY_CLIENT,
|
||||
clientDisplayName: node ? NODE_DISPLAY_NAME : "Paired node worker wire operator",
|
||||
clientVersion: VERSION,
|
||||
platform: node ? "macos" : process.platform,
|
||||
deviceFamily: node ? "Mac" : undefined,
|
||||
mode: node ? GATEWAY_CLIENT_MODES.NODE : GATEWAY_CLIENT_MODES.BACKEND,
|
||||
scopes: node ? [] : ["operator.admin", "operator.pairing", "operator.read", "operator.write"],
|
||||
caps: node ? ["system"] : undefined,
|
||||
commands: node ? [] : undefined,
|
||||
deviceIdentity: params.identity,
|
||||
requestTimeoutMs: PROOF_TIMEOUT_MS,
|
||||
onEvent: params.onEvent,
|
||||
onHelloOk: () => finish(),
|
||||
onConnectError: (error) => finish(error),
|
||||
onClose: (code, reason) => finish(new Error(`Gateway closed (${code}): ${reason}`)),
|
||||
});
|
||||
client.start();
|
||||
});
|
||||
}
|
||||
|
||||
function isPairingRequired(error: unknown): boolean {
|
||||
const details =
|
||||
error && typeof error === "object"
|
||||
? (error as { details?: { code?: unknown } }).details
|
||||
: undefined;
|
||||
return details?.code === "PAIRING_REQUIRED" || String(error).includes("PAIRING_REQUIRED");
|
||||
}
|
||||
|
||||
async function approveNodePairing(operator: GatewayClient, nodeId: string): Promise<void> {
|
||||
let nodeRequestId: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; nodeId?: string }>;
|
||||
}>("node.pair.list", {});
|
||||
nodeRequestId = result.pending?.find((entry) => entry.nodeId === nodeId)?.requestId;
|
||||
expect(nodeRequestId).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await operator.request("node.pair.approve", { requestId: nodeRequestId });
|
||||
}
|
||||
|
||||
async function approvePairing(operator: GatewayClient, nodeId: string): Promise<void> {
|
||||
let deviceRequestId: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{
|
||||
pending?: Array<{ requestId?: string; deviceId?: string; role?: string }>;
|
||||
}>("device.pair.list", {});
|
||||
deviceRequestId = result.pending?.find(
|
||||
(entry) => entry.deviceId === nodeId || entry.role === "node",
|
||||
)?.requestId;
|
||||
expect(deviceRequestId).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
await operator.request("device.pair.approve", { requestId: deviceRequestId });
|
||||
await approveNodePairing(operator, nodeId);
|
||||
}
|
||||
|
||||
async function ensureNodeApproved(operator: GatewayClient, nodeId: string): Promise<boolean> {
|
||||
let approvalState: string | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {});
|
||||
approvalState = result.nodes?.find((node) => node.nodeId === nodeId)?.approvalState;
|
||||
expect(approvalState).toBeTruthy();
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
if (approvalState !== "approved") {
|
||||
await approveNodePairing(operator, nodeId);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
async function waitForApprovedWireNode(
|
||||
operator: GatewayClient,
|
||||
nodeId: string,
|
||||
): Promise<WireNodeRead> {
|
||||
let approved: WireNodeRead | undefined;
|
||||
await vi.waitFor(
|
||||
async () => {
|
||||
const result = await operator.request<{ nodes?: WireNodeRead[] }>("node.list", {});
|
||||
approved = result.nodes?.find((node) => node.nodeId === nodeId);
|
||||
expect(approved).toMatchObject({
|
||||
nodeId,
|
||||
approvalState: "approved",
|
||||
connected: true,
|
||||
paired: true,
|
||||
sessionHost: true,
|
||||
});
|
||||
},
|
||||
{ timeout: 30_000, interval: 100 },
|
||||
);
|
||||
if (!approved) {
|
||||
throw new Error("paired worker node did not become available");
|
||||
}
|
||||
return approved;
|
||||
}
|
||||
|
||||
type WireWorkerHostOptions = {
|
||||
gateway: WireGateway;
|
||||
operator: GatewayClient;
|
||||
root: string;
|
||||
label?: string;
|
||||
capacity?: number;
|
||||
capacityWaitMs?: number;
|
||||
bundlePrewarm?: boolean;
|
||||
bundleRetention?: boolean;
|
||||
bundleStatus?: boolean;
|
||||
onInvoke?: (frame: NodeInvokeRequestPayload) => void;
|
||||
afterInvoke?: (frame: NodeInvokeRequestPayload, host: PairedNodeWorkerHost) => Promise<void>;
|
||||
};
|
||||
|
||||
export type PairedNodeWorkerHost = {
|
||||
readonly identity: DeviceIdentity;
|
||||
readonly commands: string[];
|
||||
readonly frames: NodeInvokeRequestPayload[];
|
||||
readonly invokeErrors: unknown[];
|
||||
readonly supervisor: ReturnType<typeof createNodeWorkerSupervisor>;
|
||||
readonly bundleInstaller: NodeWorkerBundleInstaller;
|
||||
readonly workspace: NodeWorkerWorkspaceRuntime;
|
||||
readonly client: GatewayClient | undefined;
|
||||
connect(): Promise<void>;
|
||||
disconnect(): Promise<void>;
|
||||
publishInventory(): Promise<void>;
|
||||
waitForInvokes(): Promise<void>;
|
||||
installedBundleDirectory(bundleHash: string): Promise<string>;
|
||||
stop(): Promise<void>;
|
||||
};
|
||||
|
||||
export async function createPairedNodeWorkerHost(
|
||||
options: WireWorkerHostOptions,
|
||||
): Promise<PairedNodeWorkerHost> {
|
||||
const label = options.label ?? "node";
|
||||
const nodeStateDir = path.join(options.root, `${label}-state`);
|
||||
const nodeHostRoot = path.join(nodeStateDir, "node-host");
|
||||
const nodeEnv = {
|
||||
...process.env,
|
||||
HOME: path.join(options.root, `${label}-home`),
|
||||
NODE_DISABLE_COMPILE_CACHE: undefined,
|
||||
OPENCLAW_STATE_DIR: nodeStateDir,
|
||||
};
|
||||
await fs.mkdir(nodeEnv.HOME, { recursive: true });
|
||||
const workspace = new NodeWorkerWorkspaceRuntime({ root: nodeHostRoot, env: nodeEnv });
|
||||
const bundleInstaller = new NodeWorkerBundleInstaller({ root: nodeHostRoot, env: nodeEnv });
|
||||
let capacityAvailable = true;
|
||||
let client: GatewayClient | undefined;
|
||||
let closing = false;
|
||||
const invokeTasks = new Set<Promise<void>>();
|
||||
const invokeErrors: unknown[] = [];
|
||||
const commands: string[] = [];
|
||||
const frames: NodeInvokeRequestPayload[] = [];
|
||||
const identity = loadOrCreateDeviceIdentity({
|
||||
path: path.join(options.root, `${label}-identity.sqlite`),
|
||||
});
|
||||
|
||||
const inventory = () => ({
|
||||
protocolFeatures: [NODE_WORKER_SUPERVISOR_PROTOCOL_FEATURE],
|
||||
workerHost: {
|
||||
enabled: true as const,
|
||||
capacity: capacityAvailable ? ("available" as const) : ("full" as const),
|
||||
...(options.bundlePrewarm ? { bundlePrewarm: WORKER_BUNDLE_PREWARM_VERSION } : {}),
|
||||
...(options.bundleRetention ? { bundleRetention: NODE_WORKER_BUNDLE_RETENTION_VERSION } : {}),
|
||||
...(options.bundleStatus ? { bundleStatus: NODE_WORKER_BUNDLE_STATUS_VERSION } : {}),
|
||||
},
|
||||
});
|
||||
|
||||
const supervisor = createNodeWorkerSupervisor({
|
||||
env: nodeEnv,
|
||||
workspace,
|
||||
capacity: options.capacity,
|
||||
capacityWaitMs: options.capacityWaitMs,
|
||||
onAvailabilityChanged: (available) => {
|
||||
capacityAvailable = available;
|
||||
},
|
||||
});
|
||||
|
||||
let host!: PairedNodeWorkerHost;
|
||||
const onEvent = (event: WireGatewayEvent) => {
|
||||
if (closing || event.event !== "node.invoke.request" || !client) {
|
||||
return;
|
||||
}
|
||||
const receiver = client;
|
||||
const frame = event.payload as NodeInvokeRequestPayload;
|
||||
commands.push(frame.command);
|
||||
frames.push(frame);
|
||||
options.onInvoke?.(frame);
|
||||
const task = handleInvoke(frame, receiver, { current: async () => [] }, undefined, {
|
||||
workerBundleInstaller: bundleInstaller,
|
||||
workerSupervisor: supervisor,
|
||||
workerWorkspace: workspace,
|
||||
gatewayUrl: options.gateway.wsUrl,
|
||||
})
|
||||
.then(async () => await options.afterInvoke?.(frame, host))
|
||||
.catch((error: unknown) => {
|
||||
invokeErrors.push(error);
|
||||
})
|
||||
.finally(() => invokeTasks.delete(task));
|
||||
invokeTasks.add(task);
|
||||
};
|
||||
|
||||
const connect = async () => {
|
||||
if (closing) {
|
||||
throw new Error("paired worker node is closing");
|
||||
}
|
||||
const open = () =>
|
||||
connectWireClient({
|
||||
gateway: options.gateway,
|
||||
role: "node",
|
||||
identity,
|
||||
onEvent,
|
||||
});
|
||||
let next: GatewayClient;
|
||||
try {
|
||||
next = await open();
|
||||
} catch (error) {
|
||||
if (!isPairingRequired(error)) {
|
||||
throw error;
|
||||
}
|
||||
await approvePairing(options.operator, identity.deviceId);
|
||||
next = await open();
|
||||
}
|
||||
client = next;
|
||||
if (await ensureNodeApproved(options.operator, identity.deviceId)) {
|
||||
await client.stopAndWait({ timeoutMs: 2_000 });
|
||||
client = await open();
|
||||
}
|
||||
await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, inventory());
|
||||
};
|
||||
const drainInvokeTasks = async () => {
|
||||
while (invokeTasks.size > 0) {
|
||||
await Promise.allSettled([...invokeTasks]);
|
||||
}
|
||||
};
|
||||
|
||||
host = {
|
||||
identity,
|
||||
commands,
|
||||
frames,
|
||||
invokeErrors,
|
||||
supervisor,
|
||||
bundleInstaller,
|
||||
workspace,
|
||||
get client() {
|
||||
return client;
|
||||
},
|
||||
connect,
|
||||
async disconnect() {
|
||||
const current = client;
|
||||
client = undefined;
|
||||
await current?.stopAndWait({ timeoutMs: 2_000 });
|
||||
},
|
||||
async publishInventory() {
|
||||
if (!client) {
|
||||
throw new Error("paired worker node is disconnected");
|
||||
}
|
||||
await client.request(NODE_RUNNER_INVENTORY_UPDATE_METHOD, inventory());
|
||||
},
|
||||
async waitForInvokes() {
|
||||
await drainInvokeTasks();
|
||||
},
|
||||
async installedBundleDirectory(bundleHash) {
|
||||
const namespaces = await fs.readdir(nodeHostRoot, { withFileTypes: true });
|
||||
const matches: string[] = [];
|
||||
for (const entry of namespaces) {
|
||||
if (!entry.isDirectory()) {
|
||||
continue;
|
||||
}
|
||||
const candidate = path.join(nodeHostRoot, entry.name, "bundles", bundleHash);
|
||||
try {
|
||||
if ((await fs.stat(candidate)).isDirectory()) {
|
||||
matches.push(candidate);
|
||||
}
|
||||
} catch {
|
||||
// This namespace does not own the proof bundle.
|
||||
}
|
||||
}
|
||||
if (matches.length !== 1) {
|
||||
throw new Error(`expected one proof-owned installed bundle, found ${matches.length}`);
|
||||
}
|
||||
return matches[0]!;
|
||||
},
|
||||
async stop() {
|
||||
closing = true;
|
||||
const current = client;
|
||||
client = undefined;
|
||||
const connectionCleanup = await Promise.allSettled([
|
||||
current?.stopAndWait({ timeoutMs: 2_000 }) ?? Promise.resolve(),
|
||||
]);
|
||||
await drainInvokeTasks();
|
||||
const cleanup = await Promise.allSettled([supervisor.close()]);
|
||||
const failures = [...connectionCleanup, ...cleanup].flatMap((result) =>
|
||||
result.status === "rejected" ? [result.reason] : [],
|
||||
);
|
||||
if (failures.length === 1) {
|
||||
throw failures[0];
|
||||
}
|
||||
if (failures.length > 1) {
|
||||
throw new AggregateError(failures, "paired node worker cleanup failed");
|
||||
}
|
||||
},
|
||||
};
|
||||
|
||||
await supervisor.initialize();
|
||||
await connect();
|
||||
await waitForApprovedWireNode(options.operator, identity.deviceId);
|
||||
return host;
|
||||
}
|
||||
|
||||
export async function startPairedNodeWorkerGateway(params: {
|
||||
providerBaseUrl: string;
|
||||
}): Promise<WireGateway> {
|
||||
return await startQaGatewayChild({
|
||||
repoRoot: process.cwd(),
|
||||
useRepoCli: true,
|
||||
providerBaseUrl: `${params.providerBaseUrl}/v1`,
|
||||
providerMode: "mock-openai",
|
||||
primaryModel: MODEL_REF,
|
||||
alternateModel: MODEL_REF,
|
||||
transportBaseUrl: "http://127.0.0.1",
|
||||
controlUiEnabled: false,
|
||||
mutateConfig: (config) => ({
|
||||
...config,
|
||||
nodeHost: {
|
||||
...config.nodeHost,
|
||||
workerRuns: { enabled: true },
|
||||
},
|
||||
}),
|
||||
});
|
||||
}
|
||||
|
||||
export function wireMessageText(message: unknown): string {
|
||||
const content = (message as { content?: unknown })?.content;
|
||||
if (typeof content === "string") {
|
||||
return content;
|
||||
}
|
||||
return Array.isArray(content)
|
||||
? content
|
||||
.flatMap((part) =>
|
||||
part && typeof part === "object" && typeof (part as { text?: unknown }).text === "string"
|
||||
? [(part as { text: string }).text]
|
||||
: [],
|
||||
)
|
||||
.join("")
|
||||
: "";
|
||||
}
|
||||
|
||||
export function bundleInstallFrames(host: PairedNodeWorkerHost): NodeInvokeRequestPayload[] {
|
||||
return host.frames.filter((frame) => frame.command === NODE_WORKER_BUNDLE_INSTALL_COMMAND);
|
||||
}
|
||||
Reference in New Issue
Block a user