203 lines
12 KiB
TypeScript
203 lines
12 KiB
TypeScript
import { afterAll, afterEach, beforeEach, describe, expect, it } from "bun:test";
|
|
import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import type { AgentEvent, CreateAgentRunInput } from "../src/services/api/agent";
|
|
|
|
// Exercise the production protocol, isolating only transport/storage dependencies.
|
|
// Temporary modules avoid mock.module leaking into unrelated Bun test suites.
|
|
const dir = mkdtempSync(join(tmpdir(), "agent-reliability-"));
|
|
const root = new URL("../src/", import.meta.url);
|
|
const requestPath = join(dir, "request.ts");
|
|
writeFileSync(requestPath, 'export const apiBaseURL = "https://agent.invalid/api"; export const http = { post: async () => { throw new Error("unexpected POST"); } };');
|
|
writeFileSync(join(dir, "agent.ts"), readFileSync(new URL("services/api/agent.ts", root), "utf8")
|
|
.replace('"@/services/api/request"', JSON.stringify(requestPath))
|
|
.replace('"@/services/api/task-text-stream"', JSON.stringify(new URL("services/api/task-text-stream.ts", root).pathname)));
|
|
const api: typeof import("../src/services/api/agent") = await import(join(dir, "agent.ts"));
|
|
const transport = await import(requestPath);
|
|
const storagePath = join(dir, "storage.ts");
|
|
writeFileSync(storagePath, 'export const data = new Map(); export const localForageStorageForScope = () => ({ getItem: async (k) => data.get(k) ?? null, setItem: async (k,v) => {data.set(k,v);}, removeItem: async (k) => {data.delete(k);} });');
|
|
writeFileSync(join(dir, "scope.ts"), 'export const getActiveUserScope = () => "test-user";');
|
|
writeFileSync(join(dir, "conversations.ts"), readFileSync(new URL("services/cloud-agent-conversations.ts", root), "utf8")
|
|
.replace('"@/lib/localforage-storage"', JSON.stringify(storagePath))
|
|
.replace('"@/lib/user-scope"', JSON.stringify(join(dir, "scope.ts")))
|
|
.replace('"@/lib/markdown-plain-text"', JSON.stringify(new URL("lib/markdown-plain-text.ts", root).pathname)));
|
|
const conversations: typeof import("../src/services/cloud-agent-conversations") = await import(join(dir, "conversations.ts"));
|
|
const storage = await import(storagePath);
|
|
const nativeFetch = globalThis.fetch;
|
|
const nativeTimer = globalThis.setTimeout;
|
|
let delays: number[] = [];
|
|
let stops: Array<() => void> = [];
|
|
beforeEach(() => {
|
|
delays = [];
|
|
storage.data.clear();
|
|
// Retain watchdog semantics; accelerate only retry waits, recording real delays.
|
|
globalThis.setTimeout = ((fn: TimerHandler, ms?: number, ...args: unknown[]) => {
|
|
delays.push(ms ?? 0);
|
|
return nativeTimer(fn, ms === 45_000 ? ms : Math.min(ms ?? 0, 1), ...args);
|
|
}) as typeof setTimeout;
|
|
});
|
|
afterEach(() => {
|
|
stops.forEach((stop) => stop());
|
|
stops = [];
|
|
globalThis.fetch = nativeFetch;
|
|
globalThis.setTimeout = nativeTimer;
|
|
});
|
|
afterAll(() => rmSync(dir, { recursive: true, force: true }));
|
|
const frame = (event: string, data: unknown) => `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
|
|
const snapshot = (status: string, extra = {}) => frame("run_snapshot", { id: "run", status, ...extra });
|
|
const event = (seq: number) => frame("agent_event", { runId: "run", eventId: `run:${seq}`, seq, type: "tool_completed", payload: { nodeId: "node" } });
|
|
const stream = (body: string) => new Response(body, { headers: { "content-type": "text/event-stream" } });
|
|
function observe(options: { timeoutMs?: number; after?: number } = {}) {
|
|
const events: AgentEvent[] = [];
|
|
const statuses: string[] = [];
|
|
let finish!: (error?: unknown) => void;
|
|
const done = new Promise<unknown>((resolve, reject) => {
|
|
const deadline = nativeTimer(() => reject(new Error("subscription did not settle")), 3000);
|
|
finish = (error) => { clearTimeout(deadline); resolve(error); };
|
|
});
|
|
const stop = api.subscribeAgentEvents("run", (e) => {
|
|
events.push(e);
|
|
if (e.type === "run_status" && !e.payload.cleanupPending && ["completed", "failed", "cancelled"].includes(String(e.payload.status))) finish();
|
|
}, { ...options, onError: finish, onConnectionChange: (s) => statuses.push(s) });
|
|
stops.push(stop);
|
|
return { events, statuses, done, stop, finish };
|
|
}
|
|
|
|
describe("Agent SSE recovery", () => {
|
|
it("sends the user's final image settings with the approval and keeps cancellation", async () => {
|
|
const requests: unknown[][] = [];
|
|
transport.http.post = async (...args: unknown[]) => { requests.push(args); return { accepted: true }; };
|
|
const signal = new AbortController().signal;
|
|
const mediaSettings = { logicalModelId: "chosen-image", size: "1536x1024", quality: "high" };
|
|
await api.decideAgentApproval("run/id", "approval/id", "approve", "", signal, mediaSettings);
|
|
expect(requests).toEqual([["/agent/runs/run%2Fid/approvals/approval%2Fid/decision", { decision: "approve", reason: undefined, mediaSettings }, { signal }]]);
|
|
});
|
|
it.each([401, 403, 404, 422])("stops immediately on definitive %s", async (status) => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async () => { requests++; return new Response("", { status }); }) as typeof fetch;
|
|
const result = observe();
|
|
expect(await result.done).toMatchObject({ status });
|
|
expect(requests).toBe(1);
|
|
expect(result.statuses.at(-1)).toBe("disconnected");
|
|
});
|
|
|
|
it.each(["", snapshot("running")])("bounds clean EOF retries even with repeated snapshots", async (body) => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async () => { requests++; return stream(body); }) as typeof fetch;
|
|
const result = observe();
|
|
expect(await result.done).toBeInstanceOf(Error);
|
|
expect(requests).toBe(5);
|
|
expect(delays.filter((ms) => ms !== 45_000)).toHaveLength(4);
|
|
});
|
|
|
|
it("resumes the persisted cursor and deduplicates replayed events", async () => {
|
|
const cursors: string[] = [];
|
|
globalThis.fetch = (async (url, init) => {
|
|
const cursor = new URL(String(url)).searchParams.get("after")!;
|
|
cursors.push(cursor);
|
|
expect((init?.headers as Record<string, string>)["Last-Event-ID"]).toBe(cursor);
|
|
if (cursors.length === 2) throw new TypeError("connection reset");
|
|
return stream(cursors.length === 1 ? event(1) + snapshot("running") : event(1) + event(2) + snapshot("completed"));
|
|
}) as typeof fetch;
|
|
const result = observe();
|
|
expect(await result.done).toBeUndefined();
|
|
expect(cursors).toEqual(["0", "1", "1"]);
|
|
expect(result.events.filter((e) => e.seq > 0).map((e) => e.seq)).toEqual([1, 2]);
|
|
});
|
|
|
|
it("waits for durable cleanup after a terminal status", async () => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async () => stream(snapshot("cancelled", { cleanupPending: ++requests === 1 }))) as typeof fetch;
|
|
const result = observe();
|
|
expect(await result.done).toBeUndefined();
|
|
expect(requests).toBe(2);
|
|
expect(result.events.filter((e) => e.type === "run_status").map((e) => e.payload.cleanupPending)).toEqual([true, false]);
|
|
});
|
|
|
|
it("honors Retry-After rather than immediately retrying 429", async () => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async () => ++requests === 1 ? new Response("", { status: 429, headers: { "retry-after": "60" } }) : stream(snapshot("completed"))) as typeof fetch;
|
|
const result = observe();
|
|
await result.done;
|
|
expect(requests).toBe(2);
|
|
expect(delays).toContain(60_000);
|
|
});
|
|
|
|
it("returns long Retry-After waits to the user without retrying early", async () => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async () => { requests++; return new Response("", { status: 429, headers: { "retry-after": "600" } }); }) as typeof fetch;
|
|
const result = observe();
|
|
expect(await result.done).toMatchObject({ status: 429 });
|
|
expect(requests).toBe(1);
|
|
});
|
|
|
|
it("times out a stalled connection with a bounded retry budget", async () => {
|
|
let requests = 0;
|
|
globalThis.fetch = (async (_url, init) => {
|
|
requests++;
|
|
return new Promise<Response>((_resolve, reject) => init?.signal?.addEventListener("abort", () => reject(new Error("timeout")), { once: true }));
|
|
}) as typeof fetch;
|
|
expect(await observe({ timeoutMs: 10 }).done).toBeInstanceOf(Error);
|
|
expect(requests).toBe(5);
|
|
});
|
|
|
|
it("unsubscribe aborts the live reader without reporting a run failure", async () => {
|
|
let cancelled = false;
|
|
globalThis.fetch = (async () => new Response(new ReadableStream({ cancel() { cancelled = true; } }), { headers: { "content-type": "text/event-stream" } })) as typeof fetch;
|
|
const result = observe();
|
|
await new Promise((resolve) => nativeTimer(resolve, 5));
|
|
result.stop();
|
|
await new Promise((resolve) => nativeTimer(resolve, 5));
|
|
expect(cancelled).toBe(true);
|
|
expect(result.statuses).not.toContain("disconnected");
|
|
result.finish();
|
|
await result.done;
|
|
});
|
|
});
|
|
|
|
describe("Agent admission replay and durable pending records", () => {
|
|
const input: CreateAgentRunInput = { canvasId: "canvas", prompt: "generate", model: "original", idempotencyKey: "stable-key-123" };
|
|
it("retries with a frozen original body/key and a bounded HTTP timeout", async () => {
|
|
const bodies: unknown[] = [];
|
|
const mutable = { ...input };
|
|
transport.http.post = async (_path: string, body: unknown, options: unknown) => {
|
|
bodies.push(structuredClone(body));
|
|
expect(options).toEqual({ timeout: 15_000 });
|
|
mutable.model = "changed";
|
|
if (bodies.length < 3) throw { retryable: true };
|
|
return { run: { id: "accepted" } };
|
|
};
|
|
expect(await api.createAgentRun(mutable)).toMatchObject({ run: { id: "accepted" } });
|
|
expect(bodies).toEqual([input, input, input]);
|
|
});
|
|
|
|
it("limits transport failures to three attempts, but never retries admission rejection", async () => {
|
|
let attempts = 0;
|
|
transport.http.post = async () => { attempts++; throw { retryable: true }; };
|
|
await expect(api.sendAgentMessage("parent", input)).rejects.toMatchObject({ retryable: true });
|
|
expect(attempts).toBe(3);
|
|
attempts = 0;
|
|
transport.http.post = async () => { attempts++; throw { status: 400, retryable: false }; };
|
|
await expect(api.createAgentRun(input)).rejects.toMatchObject({ status: 400 });
|
|
expect(attempts).toBe(1);
|
|
});
|
|
|
|
it("restores the full original request and message identity, isolated by conversation", async () => {
|
|
const pending = { fingerprint: "fingerprint", key: input.idempotencyKey, request: input, parentRunId: "parent", messageId: "bubble" };
|
|
await conversations.saveCloudAgentPendingSubmission("canvas", "conversation", pending);
|
|
expect(await conversations.loadCloudAgentPendingSubmission("canvas", "conversation")).toEqual(pending);
|
|
expect(await conversations.loadCloudAgentPendingSubmission("canvas", "other")).toBeNull();
|
|
await conversations.clearCloudAgentPendingSubmission("canvas", "conversation");
|
|
expect(await conversations.loadCloudAgentPendingSubmission("canvas", "conversation")).toBeNull();
|
|
});
|
|
|
|
it.each(["{broken", "null", "{}", JSON.stringify({ fingerprint: "f", key: input.idempotencyKey, request: { ...input, idempotencyKey: "wrong-key" } })])("fails closed on corrupt pending recovery data", async (raw) => {
|
|
await conversations.saveCloudAgentPendingSubmission("canvas", "conversation", { fingerprint: "f", key: input.idempotencyKey });
|
|
const key = [...storage.data.keys()][0];
|
|
storage.data.set(key, raw);
|
|
await expect(conversations.loadCloudAgentPendingSubmission("canvas", "conversation")).rejects.toThrow("已暂停本对话发送");
|
|
expect(storage.data.get(key)).toBe(raw);
|
|
});
|
|
});
|