248 lines
11 KiB
TypeScript
248 lines
11 KiB
TypeScript
import { mkdir } from "node:fs/promises";
|
|
import { denFetch } from "@openwork/behaviors";
|
|
import type { DenSession } from "@openwork/behaviors";
|
|
import {
|
|
eventually,
|
|
needs,
|
|
queryDenDatabase,
|
|
test,
|
|
unmetNeeds,
|
|
} from "@openwork/testkit";
|
|
import type { TestNeeds } from "@openwork/testkit";
|
|
import { expect } from "vitest";
|
|
import { createDesktopAutomationRunner } from "../../apps/desktop/electron/automation-runner.mjs";
|
|
import { bootCloudModelInfra } from "../../worlds/cloud-model-infra.ts";
|
|
import { bootRemoteSession } from "../../worlds/remote-session.ts";
|
|
|
|
const WORLD_WORKSPACE = "/tmp/openwork-remote-session-world";
|
|
const REQUEST_TIMEOUT_MS = 30_000;
|
|
const requirements: TestNeeds = {
|
|
optIn: ["OPENWORK_EVAL_E2E_TESTS"],
|
|
placement: "local",
|
|
};
|
|
const missingRequirements = unmetNeeds(requirements, process.env);
|
|
const title = missingRequirements.length > 0
|
|
? `remote desktop runner delivery skipped — needs: ${missingRequirements.join(", ")}`
|
|
: "a capable real desktop runner delivers an MCP command into a native local session";
|
|
|
|
interface McpToolResult {
|
|
isError: boolean;
|
|
payload: Record<string, unknown>;
|
|
text: string;
|
|
}
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function auth(session: DenSession): Record<string, string> {
|
|
return { authorization: `Bearer ${session.token}` };
|
|
}
|
|
|
|
function orgHeaders(session: DenSession, orgId: string): Record<string, string> {
|
|
return { ...auth(session), "x-openwork-org-id": orgId };
|
|
}
|
|
|
|
function parseMcpToolResult(result: unknown): McpToolResult {
|
|
if (!isRecord(result)) throw new Error(`MCP tools/call returned a non-object result: ${JSON.stringify(result)}`);
|
|
const content = Array.isArray(result.content) ? result.content.filter(isRecord) : [];
|
|
const text = content
|
|
.map((entry) => (typeof entry.text === "string" ? entry.text : ""))
|
|
.filter((entry) => entry.length > 0)
|
|
.join("\n");
|
|
const parsed: unknown = JSON.parse(text || "{}");
|
|
return { isError: result.isError === true, payload: isRecord(parsed) ? parsed : {}, text };
|
|
}
|
|
|
|
test.skipIf(missingRequirements.length > 0)(title, { timeout: 15 * 60_000 }, async ({ evidence, place }) => {
|
|
needs(requirements);
|
|
await using stack = new AsyncDisposableStack();
|
|
const world = await bootCloudModelInfra(stack, place, {
|
|
daytonaApiUrl: "http://127.0.0.1:9/daytona-guard",
|
|
});
|
|
const admin = world.admin;
|
|
const orgId = world.org.id;
|
|
const databaseUrl = world.den.database?.url;
|
|
if (!databaseUrl) throw new Error("The remote-session runner world did not expose its database.");
|
|
|
|
await mkdir(WORLD_WORKSPACE, { recursive: true });
|
|
const workerWorldName = `remote-session-runner-worker-${process.pid}`;
|
|
const worker = await bootRemoteSession(stack, {
|
|
name: workerWorldName,
|
|
workspace: WORLD_WORKSPACE,
|
|
replace: true,
|
|
});
|
|
expect(worker.reused).toBe(false);
|
|
const manifest = worker.manifest;
|
|
const localHeaders = {
|
|
Accept: "application/json",
|
|
Authorization: `Bearer ${manifest.token}`,
|
|
};
|
|
await eventually(
|
|
async () => (await fetch(manifest.healthUrl).catch(() => null))?.ok === true,
|
|
{ within: 90_000, intervalMs: 1_000, label: "desktop openwork-server healthy" },
|
|
);
|
|
await eventually(
|
|
async () => (await fetch(`${manifest.openworkUrl}/opencode/config`, {
|
|
headers: localHeaders,
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
}).catch(() => null))?.ok === true,
|
|
{ within: 180_000, intervalMs: 2_000, label: "desktop managed OpenCode engine attached" },
|
|
);
|
|
|
|
const runnerId = `eval-runner-${process.pid}-${Date.now().toString(36)}`;
|
|
const registration = await denFetch(admin, "/v1/automation-runners/token", {
|
|
method: "POST",
|
|
headers: orgHeaders(admin, orgId),
|
|
body: JSON.stringify({
|
|
runnerId,
|
|
protocolVersion: 1,
|
|
supportedExecutionTargets: ["desktop"],
|
|
capabilities: ["remote_session_v1"],
|
|
appVersion: "0.0.0-eval",
|
|
platform: "darwin",
|
|
concurrency: 1,
|
|
}),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const runnerToken = isRecord(registration.body) && typeof registration.body.token === "string"
|
|
? registration.body.token
|
|
: "";
|
|
expect(registration.response.ok, registration.text.slice(0, 500)).toBe(true);
|
|
expect(runnerToken.length).toBeGreaterThan(0);
|
|
const runnerRows = await queryDenDatabase(
|
|
databaseUrl,
|
|
"SELECT capabilities FROM automation_runner WHERE id = ?",
|
|
[runnerId],
|
|
);
|
|
expect(JSON.stringify(runnerRows)).toContain("remote_session_v1");
|
|
|
|
const runnerLogs: string[] = [];
|
|
const runner = createDesktopAutomationRunner({
|
|
getLocalRuntime: async () => ({ baseUrl: manifest.openworkUrl, token: manifest.token }),
|
|
fetchImpl: fetch,
|
|
log: (message: string) => runnerLogs.push(message),
|
|
});
|
|
stack.defer(() => runner.stop());
|
|
runner.configure({ baseUrl: world.den.ref.apiUrl, token: runnerToken, runnerId });
|
|
|
|
const mcpTokenResult = await denFetch(admin, "/v1/mcp/token", {
|
|
method: "POST",
|
|
headers: orgHeaders(admin, orgId),
|
|
body: JSON.stringify({ scopes: ["mcp:read", "mcp:write"] }),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const mcpToken = isRecord(mcpTokenResult.body) && typeof mcpTokenResult.body.token === "string"
|
|
? mcpTokenResult.body.token
|
|
: "";
|
|
expect(mcpTokenResult.response.ok, mcpTokenResult.text.slice(0, 500)).toBe(true);
|
|
expect(mcpToken.length).toBeGreaterThan(0);
|
|
let requestId = 0;
|
|
const callCapability = async (name: "remote-session:create" | "remote-session:read", body: Record<string, unknown>) => {
|
|
const response = await fetch(`${world.den.ref.apiUrl}/mcp/agent`, {
|
|
method: "POST",
|
|
headers: {
|
|
authorization: `Bearer ${mcpToken}`,
|
|
"content-type": "application/json",
|
|
accept: "application/json, text/event-stream",
|
|
},
|
|
body: JSON.stringify({
|
|
jsonrpc: "2.0",
|
|
id: ++requestId,
|
|
method: "tools/call",
|
|
params: { name: "execute_capability", arguments: { name, body } },
|
|
}),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const raw = await response.text();
|
|
if (!response.ok) throw new Error(`MCP tools/call failed: HTTP ${response.status} ${raw.slice(0, 500)}`);
|
|
const dataLine = raw.split("\n").find((line) => line.startsWith("data:"));
|
|
if (!dataLine) throw new Error(`MCP tools/call returned no SSE data frame: ${raw.slice(0, 500)}`);
|
|
const frame: unknown = JSON.parse(dataLine.slice(5));
|
|
if (!isRecord(frame) || frame.error) throw new Error(`MCP tools/call failed: ${raw.slice(0, 500)}`);
|
|
return parseMcpToolResult(frame.result);
|
|
};
|
|
|
|
const sessionTitle = `Desktop delivery ${Date.now().toString(36)}`;
|
|
const queued = await callCapability("remote-session:create", { target: "desktop", title: sessionTitle });
|
|
const commandId = typeof queued.payload.commandId === "string" ? queued.payload.commandId : "";
|
|
expect(queued.isError, queued.text).toBe(false);
|
|
expect(queued.payload.state).toBe("queued");
|
|
expect(commandId).toMatch(/^rsc_/);
|
|
expect(runner.wake()).toEqual({ polled: true });
|
|
evidence.recordAssertionEvidence(
|
|
"Capability-aware desktop presence accepts the command",
|
|
`Runner ${runnerId} persisted remote_session_v1 and remote-session:create queued ${commandId}.`,
|
|
runnerToken.length > 0 && queued.payload.state === "queued" && commandId.startsWith("rsc_"),
|
|
);
|
|
|
|
const delivered = await eventually(
|
|
() => callCapability("remote-session:read", { commandId }),
|
|
{
|
|
within: 120_000,
|
|
intervalMs: 500,
|
|
label: `desktop runner delivers command; logs=${runnerLogs.join(" | ")}`,
|
|
until: (result) => result.isError === false && result.payload.state === "delivered",
|
|
},
|
|
);
|
|
const sessionId = typeof delivered.payload.sessionId === "string" ? delivered.payload.sessionId : "";
|
|
const workspaceId = typeof delivered.payload.workspaceId === "string" ? delivered.payload.workspaceId : "";
|
|
expect(sessionId.length).toBeGreaterThan(0);
|
|
expect(workspaceId.length).toBeGreaterThan(0);
|
|
expect(delivered.payload.state).not.toBe("failed");
|
|
expect(delivered.payload.state).not.toBe("expired");
|
|
const commandRows = await queryDenDatabase(
|
|
databaseUrl,
|
|
"SELECT status, claimed_by_runner_id, session_id, workspace_id FROM remote_session_command WHERE id = ?",
|
|
[commandId],
|
|
);
|
|
expect(commandRows.find(isRecord)).toMatchObject({
|
|
status: "delivered",
|
|
claimed_by_runner_id: runnerId,
|
|
session_id: sessionId,
|
|
workspace_id: workspaceId,
|
|
});
|
|
evidence.recordAssertionEvidence(
|
|
"The actual desktop runner claims and completes the command",
|
|
`The command row names ${runnerId} as claimant and remote-session:read returned delivered with session ${sessionId} in workspace ${workspaceId}.`,
|
|
commandRows.find(isRecord)?.claimed_by_runner_id === runnerId
|
|
&& delivered.payload.state === "delivered" && sessionId.length > 0 && workspaceId.length > 0,
|
|
);
|
|
|
|
const sessionsResponse = await fetch(
|
|
`${manifest.openworkUrl}/workspace/${encodeURIComponent(workspaceId)}/opencode/session`,
|
|
{ headers: localHeaders, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) },
|
|
);
|
|
expect(sessionsResponse.ok).toBe(true);
|
|
const sessionsBody: unknown = await sessionsResponse.json();
|
|
const sessions = Array.isArray(sessionsBody)
|
|
? sessionsBody.filter(isRecord)
|
|
: [];
|
|
const localSession = sessions.find((item) => item.id === sessionId);
|
|
expect(localSession).toMatchObject({ id: sessionId, title: sessionTitle });
|
|
const sessionBase = `${manifest.openworkUrl}/workspace/${encodeURIComponent(workspaceId)}/opencode/session`;
|
|
const encodedSessionId = encodeURIComponent(sessionId);
|
|
const snapshotResponses = await Promise.all([
|
|
fetch(`${sessionBase}/${encodedSessionId}`, { headers: localHeaders, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }),
|
|
fetch(`${sessionBase}/${encodedSessionId}/message`, { headers: localHeaders, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }),
|
|
fetch(`${sessionBase}/${encodedSessionId}/todo`, { headers: localHeaders, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }),
|
|
fetch(`${sessionBase}/status`, { headers: localHeaders, signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS) }),
|
|
]);
|
|
expect(snapshotResponses.every((response) => response.ok)).toBe(true);
|
|
const [snapshotSession, snapshotMessages, snapshotTodos, snapshotStatuses]: unknown[] = await Promise.all(
|
|
snapshotResponses.map((response) => response.json()),
|
|
);
|
|
const status = isRecord(snapshotStatuses) && isRecord(snapshotStatuses[sessionId])
|
|
? snapshotStatuses[sessionId]
|
|
: { type: "idle" };
|
|
const snapshot = { session: snapshotSession, messages: snapshotMessages, todos: snapshotTodos, status };
|
|
expect(status.type).toBe("idle");
|
|
expect(Array.isArray(snapshot.messages) ? snapshot.messages : []).toEqual([]);
|
|
evidence.recordAssertionEvidence(
|
|
"The delivery receipt names a genuine web-visible native session",
|
|
`The real source-first server returned receipt session ${sessionId} with title ${sessionTitle}; its snapshot was idle with no messages, proving omission of prompt/model created but did not start execution.`,
|
|
localSession?.id === sessionId && localSession.title === sessionTitle
|
|
&& status.type === "idle" && Array.isArray(snapshot.messages) && snapshot.messages.length === 0,
|
|
);
|
|
});
|