805 lines
34 KiB
TypeScript
805 lines
34 KiB
TypeScript
import { createServer } from "node:http";
|
|
import { mkdtemp, rm } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { expect, onTestFinished } from "vitest";
|
|
import {
|
|
clickButton,
|
|
clickText,
|
|
denFetch,
|
|
evalIn,
|
|
go,
|
|
readAvailableModels,
|
|
visibleText,
|
|
waitForText,
|
|
} from "@openwork/behaviors";
|
|
import type { DenSession } from "@openwork/behaviors";
|
|
import type { Surface } from "@openwork/cdp";
|
|
import {
|
|
app,
|
|
eventually,
|
|
needs,
|
|
server,
|
|
test,
|
|
} from "@openwork/testkit";
|
|
|
|
const PROVIDER_NAME = "Automation Reliability Gateway";
|
|
const PROVIDER_KEY = "automation-reliability-gateway";
|
|
const PROVIDER_ENV = "AUTOMATION_RELIABILITY_API_KEY";
|
|
const MODEL_ID = "automation-reliability-model";
|
|
const MODEL_NAME = "Automation Reliability Model";
|
|
const REPLY = "The synthetic Automation lifecycle completed successfully.";
|
|
const PROVIDER_API_KEY = "sk-automation-reliability-local-only";
|
|
const REQUEST_TIMEOUT_MS = 10_000;
|
|
const RUN_TIMEOUT_MS = 180_000;
|
|
const AUTOMATION_MODEL_ATTENTION_CAPABILITY = "model_attention_v1";
|
|
|
|
function isRecord(value: unknown): value is Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function records(value: unknown): Record<string, unknown>[] {
|
|
return Array.isArray(value) ? value.filter(isRecord) : [];
|
|
}
|
|
|
|
function auth(session: DenSession): Record<string, string> {
|
|
return { authorization: `Bearer ${session.token}` };
|
|
}
|
|
|
|
function startProviderMock(
|
|
completionBodies: unknown[],
|
|
control: { unavailable?: boolean } = {},
|
|
): Promise<string> {
|
|
const mock = createServer((request, response) => {
|
|
const url = request.url ?? "";
|
|
if (request.method === "GET" && url.startsWith("/v1/models")) {
|
|
response.writeHead(200, { "content-type": "application/json" });
|
|
response.end(JSON.stringify({ object: "list", data: [{ id: MODEL_ID, object: "model" }] }));
|
|
return;
|
|
}
|
|
if (request.method === "POST" && (url === "/v1/chat/completions" || url === "/chat/completions")) {
|
|
let rawBody = "";
|
|
request.setEncoding("utf8");
|
|
request.on("data", (chunk: string) => { rawBody += chunk; });
|
|
request.on("end", () => {
|
|
let body: unknown;
|
|
try {
|
|
body = JSON.parse(rawBody);
|
|
} catch {
|
|
response.writeHead(400, { "content-type": "application/json" });
|
|
response.end(JSON.stringify({ error: { message: "invalid JSON request body" } }));
|
|
return;
|
|
}
|
|
completionBodies.push(body);
|
|
if (control.unavailable) {
|
|
response.writeHead(503, { "content-type": "application/json" });
|
|
response.end(JSON.stringify({
|
|
error: { message: "Synthetic provider is temporarily unavailable", type: "provider_unavailable" },
|
|
}));
|
|
return;
|
|
}
|
|
const chunks = [
|
|
{ id: `chatcmpl-automation-${completionBodies.length}`, object: "chat.completion.chunk", choices: [{ index: 0, delta: { role: "assistant" }, finish_reason: null }] },
|
|
{ id: `chatcmpl-automation-${completionBodies.length}`, object: "chat.completion.chunk", choices: [{ index: 0, delta: { content: REPLY }, finish_reason: null }] },
|
|
{ id: `chatcmpl-automation-${completionBodies.length}`, object: "chat.completion.chunk", choices: [{ index: 0, delta: {}, finish_reason: "stop" }] },
|
|
];
|
|
response.writeHead(200, {
|
|
"content-type": "text/event-stream",
|
|
"cache-control": "no-cache",
|
|
connection: "keep-alive",
|
|
});
|
|
for (const chunk of chunks) response.write(`data: ${JSON.stringify(chunk)}\n\n`);
|
|
response.write("data: [DONE]\n\n");
|
|
response.end();
|
|
});
|
|
return;
|
|
}
|
|
response.writeHead(404, { "content-type": "application/json" });
|
|
response.end(JSON.stringify({ error: { message: "not found" } }));
|
|
});
|
|
return new Promise((resolve, reject) => {
|
|
mock.once("error", reject);
|
|
mock.listen(0, "127.0.0.1", () => {
|
|
const address = mock.address();
|
|
if (!address || typeof address === "string") {
|
|
reject(new Error("Automation provider mock did not bind a TCP port."));
|
|
return;
|
|
}
|
|
onTestFinished(async () => {
|
|
await new Promise<void>((closeResolve, closeReject) => {
|
|
mock.close((error) => error ? closeReject(error) : closeResolve());
|
|
mock.closeAllConnections();
|
|
});
|
|
});
|
|
resolve(`http://127.0.0.1:${address.port}/v1`);
|
|
});
|
|
});
|
|
}
|
|
|
|
async function organizationId(session: DenSession): Promise<string> {
|
|
const result = await denFetch(session, "/v1/me/orgs", {
|
|
headers: auth(session),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const organizations = isRecord(result.body) ? records(result.body.orgs) : [];
|
|
const active = organizations.find((organization) => organization.isActive === true) ?? organizations[0];
|
|
const id = active && typeof active.id === "string" ? active.id : "";
|
|
expect(result.response.status, result.text).toBe(200);
|
|
expect(id).not.toBe("");
|
|
return id;
|
|
}
|
|
|
|
async function createProvider(
|
|
admin: DenSession,
|
|
orgId: string,
|
|
baseUrl: string,
|
|
apiKey: string,
|
|
): Promise<string> {
|
|
const result = await denFetch(admin, "/v1/llm-providers", {
|
|
method: "POST",
|
|
headers: { ...auth(admin), "x-openwork-org-id": orgId },
|
|
body: JSON.stringify({
|
|
name: PROVIDER_NAME,
|
|
source: "custom",
|
|
customConfig: {
|
|
id: PROVIDER_KEY,
|
|
name: PROVIDER_NAME,
|
|
npm: "@ai-sdk/openai-compatible",
|
|
env: [PROVIDER_ENV],
|
|
api: baseUrl,
|
|
models: [{ id: MODEL_ID, name: MODEL_NAME }],
|
|
},
|
|
apiKey,
|
|
allMembers: true,
|
|
memberIds: [],
|
|
teamIds: [],
|
|
}),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const provider = isRecord(result.body) && isRecord(result.body.llmProvider)
|
|
? result.body.llmProvider
|
|
: null;
|
|
const id = provider && typeof provider.id === "string" ? provider.id : "";
|
|
expect(result.response.status, result.text).toBe(201);
|
|
expect(id).not.toBe("");
|
|
return id;
|
|
}
|
|
|
|
async function createAutomation(
|
|
admin: DenSession,
|
|
orgId: string,
|
|
input: { name: string; instructions: string; providerId: string },
|
|
): Promise<{ automationId: string; revisionId: string }> {
|
|
const result = await denFetch(admin, "/v1/automations", {
|
|
method: "POST",
|
|
headers: { ...auth(admin), "x-openwork-org-id": orgId },
|
|
body: JSON.stringify({
|
|
name: input.name,
|
|
instructions: input.instructions,
|
|
schedule: { kind: "daily", timezone: "UTC", hour: 23, minute: 59 },
|
|
model: { providerId: input.providerId, modelId: MODEL_ID, variant: null },
|
|
}),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
const automation = isRecord(result.body) && isRecord(result.body.automation)
|
|
? result.body.automation
|
|
: null;
|
|
const revision = isRecord(result.body) && isRecord(result.body.revision)
|
|
? result.body.revision
|
|
: null;
|
|
const automationId = automation && typeof automation.id === "string" ? automation.id : "";
|
|
const revisionId = revision && typeof revision.id === "string" ? revision.id : "";
|
|
expect(result.response.status, result.text).toBe(201);
|
|
expect(automation?.state).toBe("active");
|
|
expect(automationId).not.toBe("");
|
|
expect(revisionId).not.toBe("");
|
|
expect(revision?.schedule).toEqual({ kind: "daily", timezone: "UTC", hour: 23, minute: 59 });
|
|
expect(revision?.model).toEqual({ providerId: input.providerId, modelId: MODEL_ID, variant: null });
|
|
return { automationId, revisionId };
|
|
}
|
|
|
|
async function listRuns(admin: DenSession, automationId: string): Promise<Record<string, unknown>[]> {
|
|
const result = await denFetch(admin, `/v1/automations/${encodeURIComponent(automationId)}/runs`, {
|
|
headers: auth(admin),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
expect(result.response.status, result.text).toBe(200);
|
|
return isRecord(result.body) ? records(result.body.items) : [];
|
|
}
|
|
|
|
async function waitForNewRun(
|
|
admin: DenSession,
|
|
automationId: string,
|
|
before: Set<string>,
|
|
trigger: "manual" | "scheduled",
|
|
): Promise<Record<string, unknown>> {
|
|
const run = await eventually(async () => {
|
|
const runs = await listRuns(admin, automationId);
|
|
return runs.find((run) => run.trigger === trigger
|
|
&& typeof run.id === "string"
|
|
&& !before.has(run.id));
|
|
}, {
|
|
within: RUN_TIMEOUT_MS,
|
|
intervalMs: 500,
|
|
label: `new ${trigger} Automation run`,
|
|
until: (run) => isRecord(run),
|
|
});
|
|
if (!run) throw new Error(`The ${trigger} Automation run disappeared after it was observed.`);
|
|
return run;
|
|
}
|
|
|
|
async function waitForTerminalReceipt(
|
|
admin: DenSession,
|
|
runId: string,
|
|
): Promise<Record<string, unknown>> {
|
|
return eventually(async () => {
|
|
const result = await denFetch(admin, `/v1/automation-runs/${encodeURIComponent(runId)}`, {
|
|
headers: auth(admin),
|
|
signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
|
|
});
|
|
expect(result.response.status, result.text).toBe(200);
|
|
return isRecord(result.body) ? result.body : {};
|
|
}, {
|
|
within: RUN_TIMEOUT_MS,
|
|
intervalMs: 500,
|
|
label: `terminal receipt for ${runId}`,
|
|
until: (receipt) => {
|
|
const run = isRecord(receipt.run) ? receipt.run : null;
|
|
return run !== null && ["succeeded", "failed", "cancelled", "skipped"].includes(String(run.status));
|
|
},
|
|
});
|
|
}
|
|
|
|
async function assertSucceededReceipt(
|
|
receipt: Record<string, unknown>,
|
|
expected: { automationId: string; revisionId?: string },
|
|
): Promise<{ runId: string; sessionId: string }> {
|
|
const run = isRecord(receipt.run) ? receipt.run : {};
|
|
const thread = isRecord(run.executionThread) ? run.executionThread : {};
|
|
const events = records(receipt.events);
|
|
const runId = typeof run.id === "string" ? run.id : "";
|
|
const sessionId = typeof thread.nativeThreadId === "string" ? thread.nativeThreadId : "";
|
|
expect(run.status).toBe("succeeded");
|
|
expect(run.automationId).toBe(expected.automationId);
|
|
if (expected.revisionId) expect(run.revisionId).toBe(expected.revisionId);
|
|
expect(run.error).toBeNull();
|
|
expect(run.resultSummary).toContain(REPLY);
|
|
expect(thread).toMatchObject({
|
|
threadKind: "automation",
|
|
executionLocation: "desktop",
|
|
automationId: expected.automationId,
|
|
automationRunId: runId,
|
|
engineKind: "openwork-desktop-runner-v1",
|
|
});
|
|
expect(sessionId).not.toBe("");
|
|
expect(typeof thread.workspaceId).toBe("string");
|
|
expect(events.map((event) => event.type)).toEqual(["user", "assistant", "usage", "terminal"]);
|
|
expect(events.map((event) => event.sequence)).toEqual([1, 2, 3, 4]);
|
|
return { runId, sessionId };
|
|
}
|
|
|
|
async function openAutomation(surface: Surface, automationId: string, name: string): Promise<void> {
|
|
for (let attempt = 0; attempt < 3; attempt += 1) {
|
|
await go(surface, "/automations");
|
|
try {
|
|
await eventually(() => evalIn(surface, "window.location.hash"), {
|
|
within: 30_000,
|
|
intervalMs: 250,
|
|
label: "Automations list route",
|
|
until: (hash) => typeof hash === "string" && /^#\/automations(?:\?|$)/.test(hash),
|
|
});
|
|
await clickText(surface, name, {
|
|
selector: `button[data-automation-id="${automationId}"]`,
|
|
timeoutMs: 30_000,
|
|
});
|
|
await waitForText(surface, "Run now", { timeoutMs: 30_000 });
|
|
return;
|
|
} catch (error) {
|
|
if (attempt === 2) throw error;
|
|
}
|
|
}
|
|
}
|
|
|
|
async function waitForSyntheticModel(surface: Surface): Promise<void> {
|
|
const availableModels = await eventually(() => readAvailableModels(surface), {
|
|
within: 120_000,
|
|
intervalMs: 2_000,
|
|
label: "synced synthetic Automation model",
|
|
until: (models) => models.some((model) => model.selectable
|
|
&& (model.id === MODEL_ID || model.id.endsWith(`/${MODEL_ID}`))),
|
|
});
|
|
expect(availableModels.some((model) => model.selectable
|
|
&& (model.id === MODEL_ID || model.id.endsWith(`/${MODEL_ID}`)))).toBe(true);
|
|
}
|
|
|
|
async function triggerManualRun(
|
|
admin: DenSession,
|
|
automationId: string,
|
|
): Promise<Record<string, unknown>> {
|
|
const before = new Set((await listRuns(admin, automationId)).flatMap((run) =>
|
|
typeof run.id === "string" ? [run.id] : []));
|
|
const response = await denFetch(admin, `/v1/automations/${encodeURIComponent(automationId)}/run`, {
|
|
method: "POST",
|
|
headers: auth(admin),
|
|
});
|
|
expect(response.response.status, response.text).toBe(202);
|
|
return waitForNewRun(admin, automationId, before, "manual");
|
|
}
|
|
|
|
test("a Desktop Automation completes through UI, API, schedule, thread, and receipt", { timeout: 20 * 60_000 }, async ({ evidence, place }) => {
|
|
needs({ optIn: ["OPENWORK_EVAL_E2E_TESTS"] });
|
|
const completionBodies: unknown[] = [];
|
|
const providerBaseUrl = await startProviderMock(completionBodies);
|
|
await using den = await server({
|
|
place,
|
|
org: {
|
|
name: `Automation Reliability ${Date.now()}`,
|
|
admin: { name: "Automation Admin" },
|
|
members: { member: { name: "Automation Member" } },
|
|
},
|
|
});
|
|
const orgId = await organizationId(den.admin);
|
|
const providerId = await createProvider(den.admin, orgId, providerBaseUrl, PROVIDER_API_KEY);
|
|
const stamp = Date.now();
|
|
const automationName = `Synthetic lifecycle ${stamp}`;
|
|
const instructions = `Return one concise synthetic reliability result for marker ${stamp}.`;
|
|
|
|
const invalid = await denFetch(den.admin, "/v1/automations", {
|
|
method: "POST",
|
|
headers: { ...auth(den.admin), "x-openwork-org-id": orgId },
|
|
body: JSON.stringify({
|
|
name: `Invalid lifecycle ${stamp}`,
|
|
instructions,
|
|
schedule: { kind: "daily", timezone: "Not/A-Timezone", hour: 9, minute: 0 },
|
|
model: { providerId, modelId: MODEL_ID, variant: null },
|
|
}),
|
|
});
|
|
expect(invalid.response.status).toBe(400);
|
|
expect((await denFetch(den.admin, "/v1/automations", { headers: auth(den.admin) })).text)
|
|
.not.toContain(`Invalid lifecycle ${stamp}`);
|
|
evidence.recordAssertionEvidence(
|
|
"Invalid Automation configuration fails without a runnable record",
|
|
"Den rejected the invalid timezone with HTTP 400 and the invalid name was absent from the owner list.",
|
|
true,
|
|
);
|
|
|
|
const created = await createAutomation(den.admin, orgId, {
|
|
name: automationName,
|
|
instructions,
|
|
providerId,
|
|
});
|
|
const denied = await denFetch(den.members.member, `/v1/automations/${encodeURIComponent(created.automationId)}`, {
|
|
headers: auth(den.members.member),
|
|
});
|
|
expect(denied.response.status).toBe(404);
|
|
evidence.recordAssertionEvidence(
|
|
"Automation ownership remains member-scoped",
|
|
"A different organization member received HTTP 404 for the owner's Automation.",
|
|
true,
|
|
);
|
|
|
|
await using desktop = await app({ den, as: "admin", place });
|
|
await waitForSyntheticModel(desktop);
|
|
|
|
await openAutomation(desktop, created.automationId, automationName);
|
|
const beforeUiRun = new Set((await listRuns(den.admin, created.automationId)).flatMap((run) =>
|
|
typeof run.id === "string" ? [run.id] : []));
|
|
const firstProviderCheckpoint = completionBodies.length;
|
|
await clickButton(desktop, "Run now");
|
|
const uiRun = await waitForNewRun(den.admin, created.automationId, beforeUiRun, "manual");
|
|
const uiRunId = typeof uiRun.id === "string" ? uiRun.id : "";
|
|
const uiReceipt = await waitForTerminalReceipt(den.admin, uiRunId);
|
|
const first = await assertSucceededReceipt(uiReceipt, created);
|
|
const firstRequest = await eventually(() => completionBodies[firstProviderCheckpoint], {
|
|
within: RUN_TIMEOUT_MS,
|
|
intervalMs: 250,
|
|
label: "first synthetic provider completion",
|
|
until: (request) => request !== undefined,
|
|
});
|
|
expect(JSON.stringify(firstRequest)).toContain(String(stamp));
|
|
evidence.recordAssertionEvidence(
|
|
"Run now reaches a real desktop session and durable receipt",
|
|
`UI run ${first.runId} created native thread ${first.sessionId}, returned the deterministic assistant result, and committed four ordered events.`,
|
|
true,
|
|
);
|
|
|
|
const receipts: Record<string, unknown>[] = [uiReceipt];
|
|
for (let index = 0; index < 2; index += 1) {
|
|
const before = new Set((await listRuns(den.admin, created.automationId)).flatMap((run) =>
|
|
typeof run.id === "string" ? [run.id] : []));
|
|
const response = await denFetch(den.admin, `/v1/automations/${encodeURIComponent(created.automationId)}/run`, {
|
|
method: "POST",
|
|
headers: auth(den.admin),
|
|
});
|
|
expect(response.response.status, response.text).toBe(202);
|
|
const run = await waitForNewRun(den.admin, created.automationId, before, "manual");
|
|
const runId = typeof run.id === "string" ? run.id : "";
|
|
const receipt = await waitForTerminalReceipt(den.admin, runId);
|
|
await assertSucceededReceipt(receipt, created);
|
|
receipts.push(receipt);
|
|
}
|
|
|
|
const scheduledAt = Date.now() + 45_000;
|
|
const scheduleResponse = await denFetch(
|
|
den.admin,
|
|
`/v1/automations/${encodeURIComponent(created.automationId)}`,
|
|
{
|
|
method: "PATCH",
|
|
headers: auth(den.admin),
|
|
body: JSON.stringify({ schedule: { kind: "once", timezone: "UTC", at: scheduledAt } }),
|
|
},
|
|
);
|
|
expect(scheduleResponse.response.status, scheduleResponse.text).toBe(200);
|
|
const scheduledRevision = isRecord(scheduleResponse.body) && isRecord(scheduleResponse.body.revision)
|
|
? scheduleResponse.body.revision
|
|
: {};
|
|
expect(scheduledRevision.schedule).toEqual({ kind: "once", timezone: "UTC", at: scheduledAt });
|
|
const beforeScheduled = new Set((await listRuns(den.admin, created.automationId)).flatMap((run) =>
|
|
typeof run.id === "string" ? [run.id] : []));
|
|
const scheduledRun = await waitForNewRun(den.admin, created.automationId, beforeScheduled, "scheduled");
|
|
const scheduledRunId = typeof scheduledRun.id === "string" ? scheduledRun.id : "";
|
|
const scheduledReceipt = await waitForTerminalReceipt(den.admin, scheduledRunId);
|
|
await assertSucceededReceipt(scheduledReceipt, {
|
|
automationId: created.automationId,
|
|
revisionId: typeof scheduledRevision.id === "string" ? scheduledRevision.id : undefined,
|
|
});
|
|
receipts.push(scheduledReceipt);
|
|
|
|
const runIds = receipts.flatMap((receipt) => {
|
|
const run = isRecord(receipt.run) ? receipt.run : {};
|
|
return typeof run.id === "string" ? [run.id] : [];
|
|
});
|
|
const sessionIds = receipts.flatMap((receipt) => {
|
|
const run = isRecord(receipt.run) ? receipt.run : {};
|
|
const thread = isRecord(run.executionThread) ? run.executionThread : {};
|
|
return typeof thread.nativeThreadId === "string" ? [thread.nativeThreadId] : [];
|
|
});
|
|
expect(new Set(runIds).size).toBe(4);
|
|
expect(new Set(sessionIds).size).toBe(4);
|
|
expect(completionBodies.slice(firstProviderCheckpoint)).toHaveLength(4);
|
|
evidence.recordAssertionEvidence(
|
|
"Repeated and scheduled runs remain exactly-once",
|
|
"Three sequential manual runs and one scheduled occurrence produced four unique runs, four unique native sessions, four deterministic model requests, and four terminal receipts.",
|
|
true,
|
|
);
|
|
|
|
const scheduledTerminalRun = isRecord(scheduledReceipt.run) ? scheduledReceipt.run : {};
|
|
const scheduledThread = isRecord(scheduledTerminalRun.executionThread)
|
|
? scheduledTerminalRun.executionThread
|
|
: {};
|
|
const scheduledThreadId = typeof scheduledThread.id === "string" ? scheduledThread.id : "";
|
|
const scheduledWorkspaceId = typeof scheduledThread.workspaceId === "string" ? scheduledThread.workspaceId : "";
|
|
const scheduledSessionId = typeof scheduledThread.nativeThreadId === "string" ? scheduledThread.nativeThreadId : "";
|
|
const receiptQuery = new URLSearchParams({
|
|
automation: created.automationId,
|
|
run: scheduledRunId,
|
|
thread: scheduledThreadId,
|
|
});
|
|
await go(desktop, `/automations?${receiptQuery.toString()}`);
|
|
await clickText(desktop, "Open local thread", {
|
|
selector: `button[data-automation-run-id="${scheduledRunId}"]`,
|
|
timeoutMs: 30_000,
|
|
});
|
|
await eventually(
|
|
() => evalIn(desktop, "window.location.hash"),
|
|
{
|
|
within: 30_000,
|
|
intervalMs: 250,
|
|
label: "native Automation session route",
|
|
until: (hash) => typeof hash === "string"
|
|
&& hash.includes(`/workspace/${encodeURIComponent(scheduledWorkspaceId)}/session/${encodeURIComponent(scheduledSessionId)}`),
|
|
},
|
|
);
|
|
await waitForText(desktop, REPLY, { timeoutMs: 30_000 });
|
|
evidence.recordAssertionEvidence(
|
|
"A receipt opens the actual local execution thread",
|
|
`The scheduled receipt exposed workspace ${scheduledWorkspaceId} and session ${scheduledSessionId}; Open local thread navigated to that session and rendered the deterministic assistant result.`,
|
|
true,
|
|
);
|
|
|
|
await openAutomation(desktop, created.automationId, automationName);
|
|
const detailText = await visibleText(desktop);
|
|
expect(detailText).toContain("succeeded");
|
|
expect(detailText).toContain(MODEL_NAME);
|
|
expect(detailText).not.toMatch(/running|waiting|no assistant result/i);
|
|
evidence.recordAssertionEvidence(
|
|
"The Automation UI reflects the terminal result and configured model",
|
|
`The detail view showed succeeded with ${MODEL_NAME}, without a stale running, waiting, or missing-result message.`,
|
|
true,
|
|
);
|
|
});
|
|
|
|
test("a Desktop Automation recovers across restart before execution and while work is queued", { timeout: 20 * 60_000 }, async ({ evidence, place }) => {
|
|
needs({ optIn: ["OPENWORK_EVAL_E2E_TESTS"] });
|
|
const completionBodies: unknown[] = [];
|
|
const providerBaseUrl = await startProviderMock(completionBodies);
|
|
await using den = await server({
|
|
place,
|
|
org: {
|
|
name: `Automation Recovery ${Date.now()}`,
|
|
admin: { name: "Automation Recovery Admin" },
|
|
},
|
|
});
|
|
const orgId = await organizationId(den.admin);
|
|
const providerId = await createProvider(den.admin, orgId, providerBaseUrl, PROVIDER_API_KEY);
|
|
const stamp = Date.now();
|
|
const created = await createAutomation(den.admin, orgId, {
|
|
name: `Synthetic recovery ${stamp}`,
|
|
instructions: `Return one concise synthetic recovery result for marker ${stamp}.`,
|
|
providerId,
|
|
});
|
|
const profileDir = await mkdtemp(join(tmpdir(), "openwork-automation-recovery-"));
|
|
onTestFinished(() => rm(profileDir, { recursive: true, force: true }));
|
|
let desktop: Awaited<ReturnType<typeof app>> | null = null;
|
|
|
|
try {
|
|
desktop = await app({ den, as: "admin", place, profileDir });
|
|
await waitForSyntheticModel(desktop);
|
|
await desktop.stop();
|
|
desktop = null;
|
|
|
|
desktop = await app({ den, as: "admin", place, profileDir });
|
|
await waitForSyntheticModel(desktop);
|
|
const postRestartRun = await triggerManualRun(den.admin, created.automationId);
|
|
const postRestartRunId = typeof postRestartRun.id === "string" ? postRestartRun.id : "";
|
|
const postRestartReceipt = await waitForTerminalReceipt(den.admin, postRestartRunId);
|
|
const first = await assertSucceededReceipt(postRestartReceipt, created);
|
|
evidence.recordAssertionEvidence(
|
|
"Desktop restart before execution preserves runner readiness",
|
|
`The same isolated profile relaunched before execution, reminted its runner authority, and run ${first.runId} completed in native session ${first.sessionId}.`,
|
|
true,
|
|
);
|
|
|
|
await desktop.stop();
|
|
desktop = null;
|
|
const providerRequestsBeforeConcurrentRuns = completionBodies.length;
|
|
const concurrentResponses = await Promise.all(Array.from({ length: 3 }, () =>
|
|
denFetch(den.admin, `/v1/automations/${encodeURIComponent(created.automationId)}/run`, {
|
|
method: "POST",
|
|
headers: auth(den.admin),
|
|
})));
|
|
for (const response of concurrentResponses) {
|
|
expect(response.response.status, response.text).toBe(202);
|
|
}
|
|
const concurrentRuns = concurrentResponses.map((response) =>
|
|
isRecord(response.body) && isRecord(response.body.run) ? response.body.run : {});
|
|
expect(concurrentRuns.map((run) => run.status).sort()).toEqual(["queued", "skipped", "skipped"]);
|
|
const queuedConcurrentRun = concurrentRuns.find((run) => run.status === "queued");
|
|
const queuedConcurrentRunId = queuedConcurrentRun && typeof queuedConcurrentRun.id === "string"
|
|
? queuedConcurrentRun.id
|
|
: "";
|
|
const cancelResponse = await denFetch(
|
|
den.admin,
|
|
`/v1/automation-runs/${encodeURIComponent(queuedConcurrentRunId)}/cancel`,
|
|
{ method: "POST", headers: auth(den.admin) },
|
|
);
|
|
const cancelledRun = isRecord(cancelResponse.body) && isRecord(cancelResponse.body.run)
|
|
? cancelResponse.body.run
|
|
: {};
|
|
expect(cancelResponse.response.status, cancelResponse.text).toBe(200);
|
|
expect(cancelledRun.status).toBe("cancelled");
|
|
expect(cancelledRun.executionThread).toBeNull();
|
|
expect(completionBodies).toHaveLength(providerRequestsBeforeConcurrentRuns);
|
|
evidence.recordAssertionEvidence(
|
|
"Concurrent manual requests overlap safely and cancellation before claim is terminal",
|
|
"Three simultaneous manual requests with Desktop absent produced one queued run and two durable overlap skips; cancelling the queued run returned cancelled with no execution thread and no provider request.",
|
|
true,
|
|
);
|
|
|
|
const waitingRun = await triggerManualRun(den.admin, created.automationId);
|
|
const waitingRunId = typeof waitingRun.id === "string" ? waitingRun.id : "";
|
|
expect(waitingRun.status).toBe("queued");
|
|
|
|
desktop = await app({ den, as: "admin", place, profileDir });
|
|
const waitingReceipt = await waitForTerminalReceipt(den.admin, waitingRunId);
|
|
const second = await assertSucceededReceipt(waitingReceipt, created);
|
|
expect(second.sessionId).not.toBe(first.sessionId);
|
|
expect(completionBodies).toHaveLength(2);
|
|
evidence.recordAssertionEvidence(
|
|
"Queued work survives Desktop absence and relaunch",
|
|
`Run ${second.runId} remained queued while Desktop was stopped; after relaunch a fresh runner credential claimed that same run and exactly one new native session ${second.sessionId} reached a terminal receipt.`,
|
|
true,
|
|
);
|
|
} finally {
|
|
await desktop?.stop();
|
|
}
|
|
});
|
|
|
|
test("a Desktop Automation records a provider outage and succeeds after recovery", { timeout: 10 * 60_000 }, async ({ evidence, place }) => {
|
|
needs({ optIn: ["OPENWORK_EVAL_E2E_TESTS"] });
|
|
const completionBodies: unknown[] = [];
|
|
const providerControl = { unavailable: true };
|
|
const providerBaseUrl = await startProviderMock(completionBodies, providerControl);
|
|
await using den = await server({
|
|
place,
|
|
org: {
|
|
name: `Automation Provider Recovery ${Date.now()}`,
|
|
admin: { name: "Automation Provider Recovery Admin" },
|
|
},
|
|
});
|
|
const orgId = await organizationId(den.admin);
|
|
const providerId = await createProvider(den.admin, orgId, providerBaseUrl, PROVIDER_API_KEY);
|
|
const created = await createAutomation(den.admin, orgId, {
|
|
name: `Synthetic provider recovery ${Date.now()}`,
|
|
instructions: "Return a result after the synthetic provider recovers.",
|
|
providerId,
|
|
});
|
|
await using desktop = await app({ den, as: "admin", place });
|
|
await waitForSyntheticModel(desktop);
|
|
|
|
const failedRun = await triggerManualRun(den.admin, created.automationId);
|
|
const failedRunId = typeof failedRun.id === "string" ? failedRun.id : "";
|
|
const failedReceipt = await waitForTerminalReceipt(den.admin, failedRunId);
|
|
const failed = isRecord(failedReceipt.run) ? failedReceipt.run : {};
|
|
const failedThread = isRecord(failed.executionThread) ? failed.executionThread : {};
|
|
expect(failed.status).toBe("failed");
|
|
expect(failed.error).toMatchObject({ code: "execution_failed", retryable: false });
|
|
expect(JSON.stringify(failed.error)).toMatch(/temporarily unavailable|503/i);
|
|
expect(typeof failedThread.nativeThreadId).toBe("string");
|
|
expect(typeof failedThread.workspaceId).toBe("string");
|
|
expect(records(failedReceipt.events).map((event) => event.type)).toEqual(["user", "terminal"]);
|
|
evidence.recordAssertionEvidence(
|
|
"A provider outage fails promptly with a linked local execution thread",
|
|
`Run ${failedRunId} received synthetic HTTP 503 responses and reached failed/execution_failed with its native session and workspace preserved; its event sequence ended at terminal instead of remaining running or waiting.`,
|
|
true,
|
|
);
|
|
|
|
providerControl.unavailable = false;
|
|
const recoveredRun = await triggerManualRun(den.admin, created.automationId);
|
|
const recoveredRunId = typeof recoveredRun.id === "string" ? recoveredRun.id : "";
|
|
const recoveredReceipt = await waitForTerminalReceipt(den.admin, recoveredRunId);
|
|
const recovered = await assertSucceededReceipt(recoveredReceipt, created);
|
|
expect(recovered.sessionId).not.toBe(failedThread.nativeThreadId);
|
|
evidence.recordAssertionEvidence(
|
|
"A later run succeeds after the provider recovers",
|
|
`Without editing or recreating the Automation, run ${recovered.runId} completed in a new native session after the provider began serving successful responses.`,
|
|
true,
|
|
);
|
|
});
|
|
|
|
test("Desktop runner claims are idempotent and expired leases recover safely", { timeout: 5 * 60_000 }, async ({ evidence, place }) => {
|
|
needs({ optIn: ["OPENWORK_EVAL_E2E_TESTS"] });
|
|
const completionBodies: unknown[] = [];
|
|
const providerBaseUrl = await startProviderMock(completionBodies);
|
|
await using den = await server({
|
|
place,
|
|
env: {
|
|
DEN_AUTOMATIONS_LEASE_MS: "2500",
|
|
DEN_AUTOMATIONS_POLL_INTERVAL_MS: "1000",
|
|
DEN_AUTOMATIONS_RUNNER_CLAIM_DEADLINE_MS: "30000",
|
|
},
|
|
org: {
|
|
name: `Automation Lease Recovery ${Date.now()}`,
|
|
admin: { name: "Automation Lease Admin" },
|
|
},
|
|
});
|
|
const orgId = await organizationId(den.admin);
|
|
const providerId = await createProvider(den.admin, orgId, providerBaseUrl, PROVIDER_API_KEY);
|
|
const created = await createAutomation(den.admin, orgId, {
|
|
name: `Synthetic lease recovery ${Date.now()}`,
|
|
instructions: "This synthetic run must never reach a provider.",
|
|
providerId,
|
|
});
|
|
const registration = (runnerId: string) => ({
|
|
runnerId,
|
|
protocolVersion: 1,
|
|
supportedExecutionTargets: ["desktop"],
|
|
capabilities: [AUTOMATION_MODEL_ATTENTION_CAPABILITY],
|
|
appVersion: "automation-reliability-eval",
|
|
platform: "darwin",
|
|
concurrency: 1,
|
|
});
|
|
const mintRunner = async (runnerId: string): Promise<string> => {
|
|
const response = await denFetch(den.admin, "/v1/automation-runners/token", {
|
|
method: "POST",
|
|
headers: auth(den.admin),
|
|
body: JSON.stringify(registration(runnerId)),
|
|
});
|
|
const token = isRecord(response.body) && typeof response.body.token === "string"
|
|
? response.body.token
|
|
: "";
|
|
expect(response.response.status, response.text).toBe(200);
|
|
expect(token).not.toBe("");
|
|
return token;
|
|
};
|
|
const runnerRequest = (token: string, path: string, options: RequestInit = {}) =>
|
|
denFetch(den.admin, path, {
|
|
...options,
|
|
headers: { ...options.headers, authorization: `Bearer ${token}` },
|
|
});
|
|
|
|
const firstRunnerToken = await mintRunner(`synthetic-runner-primary-${Date.now()}`);
|
|
const secondRunnerToken = await mintRunner(`synthetic-runner-secondary-${Date.now()}`);
|
|
const run = await triggerManualRun(den.admin, created.automationId);
|
|
const runId = typeof run.id === "string" ? run.id : "";
|
|
const work = await runnerRequest(firstRunnerToken, "/v1/automation-runner/work");
|
|
expect(work.response.status, work.text).toBe(200);
|
|
expect(JSON.stringify(work.body)).toContain(runId);
|
|
|
|
const firstClaim = await runnerRequest(
|
|
firstRunnerToken,
|
|
`/v1/automation-runs/${encodeURIComponent(runId)}/claim`,
|
|
{ method: "POST" },
|
|
);
|
|
const duplicateClaim = await runnerRequest(
|
|
firstRunnerToken,
|
|
`/v1/automation-runs/${encodeURIComponent(runId)}/claim`,
|
|
{ method: "POST" },
|
|
);
|
|
expect(firstClaim.response.status, firstClaim.text).toBe(200);
|
|
expect(duplicateClaim.response.status, duplicateClaim.text).toBe(200);
|
|
const firstAssignment = isRecord(firstClaim.body) && isRecord(firstClaim.body.assignment)
|
|
? firstClaim.body.assignment
|
|
: {};
|
|
const duplicateAssignment = isRecord(duplicateClaim.body) && isRecord(duplicateClaim.body.assignment)
|
|
? duplicateClaim.body.assignment
|
|
: {};
|
|
expect(firstAssignment).toMatchObject({ runId, attempt: 1 });
|
|
expect(duplicateAssignment).toMatchObject({ runId, attempt: 1 });
|
|
|
|
const competingClaim = await runnerRequest(
|
|
secondRunnerToken,
|
|
`/v1/automation-runs/${encodeURIComponent(runId)}/claim`,
|
|
{ method: "POST" },
|
|
);
|
|
expect(competingClaim.response.status, competingClaim.text).toBe(200);
|
|
expect(isRecord(competingClaim.body) ? competingClaim.body.assignment : undefined).toBeNull();
|
|
|
|
await eventually(async () => {
|
|
const receipt = await denFetch(den.admin, `/v1/automation-runs/${encodeURIComponent(runId)}`, {
|
|
headers: auth(den.admin),
|
|
});
|
|
return isRecord(receipt.body) && isRecord(receipt.body.run) ? receipt.body.run : {};
|
|
}, {
|
|
within: 30_000,
|
|
intervalMs: 250,
|
|
label: "first expired lease requeued for recovery",
|
|
until: (recoveredRun) => recoveredRun.status === "queued" && recoveredRun.attemptCount === 1,
|
|
});
|
|
|
|
const recoveryClaim = await runnerRequest(
|
|
firstRunnerToken,
|
|
`/v1/automation-runs/${encodeURIComponent(runId)}/claim`,
|
|
{ method: "POST" },
|
|
);
|
|
const recoveryAssignment = isRecord(recoveryClaim.body) && isRecord(recoveryClaim.body.assignment)
|
|
? recoveryClaim.body.assignment
|
|
: {};
|
|
expect(recoveryClaim.response.status, recoveryClaim.text).toBe(200);
|
|
expect(recoveryAssignment).toMatchObject({ runId, attempt: 2 });
|
|
|
|
const staleCompletion = await runnerRequest(
|
|
firstRunnerToken,
|
|
`/v1/automation-runs/${encodeURIComponent(runId)}/complete`,
|
|
{
|
|
method: "POST",
|
|
body: JSON.stringify({
|
|
attempt: 1,
|
|
status: "succeeded",
|
|
sessionId: "stale-session-must-not-win",
|
|
workspaceId: "stale-workspace-must-not-win",
|
|
resultSummary: "stale completion",
|
|
usage: { inputTokens: 1, outputTokens: 1, costMicros: null },
|
|
error: null,
|
|
}),
|
|
},
|
|
);
|
|
expect(staleCompletion.response.status, staleCompletion.text).toBe(409);
|
|
expect(staleCompletion.body).toEqual({ error: "runner_lease_lost" });
|
|
|
|
const terminalReceipt = await waitForTerminalReceipt(den.admin, runId);
|
|
const terminalRun = isRecord(terminalReceipt.run) ? terminalReceipt.run : {};
|
|
expect(terminalRun).toMatchObject({
|
|
status: "failed",
|
|
attemptCount: 2,
|
|
error: { code: "lease_lost", retryable: false },
|
|
});
|
|
expect(terminalRun.resultSummary).toBeNull();
|
|
expect(completionBodies).toHaveLength(0);
|
|
evidence.recordAssertionEvidence(
|
|
"Duplicate claims and expired leases cannot duplicate execution or accept stale completion",
|
|
`Runner one received the same attempt-1 assignment twice, runner two could not claim it, the first expiry requeued attempt 1, attempt 2 rejected the stale attempt-1 completion with HTTP 409, and the second expiry ended run ${runId} as failed/lease_lost with no provider request.`,
|
|
true,
|
|
);
|
|
});
|