fix: extend gateway chat test timeout on windows
This commit is contained in:
Peter Steinberger
@@ -57,81 +57,193 @@ const withSessionStore = async <T>(
|
|||||||
}
|
}
|
||||||
};
|
};
|
||||||
describe("gateway server chat", () => {
|
describe("gateway server chat", () => {
|
||||||
test("handles history, abort, idempotency, and ordering flows", { timeout: 60_000 }, async () => {
|
const timeoutMs = process.platform === "win32" ? 120_000 : 60_000;
|
||||||
const tempDirs: string[] = [];
|
test(
|
||||||
const { server, ws } = await startServerWithClient();
|
"handles history, abort, idempotency, and ordering flows",
|
||||||
const spy = vi.mocked(agentCommand);
|
{ timeout: timeoutMs },
|
||||||
const resetSpy = () => {
|
async () => {
|
||||||
spy.mockReset();
|
const tempDirs: string[] = [];
|
||||||
spy.mockResolvedValue(undefined);
|
const { server, ws } = await startServerWithClient();
|
||||||
};
|
const spy = vi.mocked(agentCommand);
|
||||||
try {
|
const resetSpy = () => {
|
||||||
await connectOk(ws);
|
spy.mockReset();
|
||||||
await withSessionStore(
|
spy.mockResolvedValue(undefined);
|
||||||
tempDirs,
|
};
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
try {
|
||||||
async (historyDir) => {
|
await connectOk(ws);
|
||||||
const bigText = "x".repeat(200_000);
|
await withSessionStore(
|
||||||
const largeLines: string[] = [];
|
tempDirs,
|
||||||
for (let i = 0; i < 40; i += 1) {
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
largeLines.push(
|
async (historyDir) => {
|
||||||
JSON.stringify({
|
const bigText = "x".repeat(200_000);
|
||||||
message: {
|
const largeLines: string[] = [];
|
||||||
role: "user",
|
for (let i = 0; i < 40; i += 1) {
|
||||||
content: [{ type: "text", text: `${i}:${bigText}` }],
|
largeLines.push(
|
||||||
timestamp: Date.now() + i,
|
JSON.stringify({
|
||||||
},
|
message: {
|
||||||
}),
|
role: "user",
|
||||||
|
content: [{ type: "text", text: `${i}:${bigText}` }],
|
||||||
|
timestamp: Date.now() + i,
|
||||||
|
},
|
||||||
|
}),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
await fs.writeFile(
|
||||||
|
path.join(historyDir, "sess-main.jsonl"),
|
||||||
|
largeLines.join("\n"),
|
||||||
|
"utf-8",
|
||||||
);
|
);
|
||||||
}
|
const cappedRes = await rpcReq<{ messages?: unknown[] }>(ws, "chat.history", {
|
||||||
await fs.writeFile(
|
sessionKey: "main",
|
||||||
path.join(historyDir, "sess-main.jsonl"),
|
limit: 1000,
|
||||||
largeLines.join("\n"),
|
});
|
||||||
"utf-8",
|
expect(cappedRes.ok).toBe(true);
|
||||||
);
|
const cappedMsgs = cappedRes.payload?.messages ?? [];
|
||||||
const cappedRes = await rpcReq<{ messages?: unknown[] }>(ws, "chat.history", {
|
const bytes = Buffer.byteLength(JSON.stringify(cappedMsgs), "utf8");
|
||||||
sessionKey: "main",
|
expect(bytes).toBeLessThanOrEqual(6 * 1024 * 1024);
|
||||||
limit: 1000,
|
expect(cappedMsgs.length).toBeLessThan(60);
|
||||||
});
|
|
||||||
expect(cappedRes.ok).toBe(true);
|
|
||||||
const cappedMsgs = cappedRes.payload?.messages ?? [];
|
|
||||||
const bytes = Buffer.byteLength(JSON.stringify(cappedMsgs), "utf8");
|
|
||||||
expect(bytes).toBeLessThanOrEqual(6 * 1024 * 1024);
|
|
||||||
expect(cappedMsgs.length).toBeLessThan(60);
|
|
||||||
},
|
|
||||||
);
|
|
||||||
await withSessionStore(
|
|
||||||
tempDirs,
|
|
||||||
{
|
|
||||||
main: {
|
|
||||||
sessionId: "sess-main",
|
|
||||||
updatedAt: Date.now(),
|
|
||||||
lastChannel: "whatsapp",
|
|
||||||
lastTo: "+1555",
|
|
||||||
},
|
},
|
||||||
},
|
);
|
||||||
async () => {
|
await withSessionStore(
|
||||||
const routeRes = await rpcReq(ws, "chat.send", {
|
tempDirs,
|
||||||
sessionKey: "main",
|
{
|
||||||
message: "hello",
|
main: {
|
||||||
idempotencyKey: "idem-route",
|
sessionId: "sess-main",
|
||||||
});
|
updatedAt: Date.now(),
|
||||||
expect(routeRes.ok).toBe(true);
|
lastChannel: "whatsapp",
|
||||||
const stored = JSON.parse(
|
lastTo: "+1555",
|
||||||
await fs.readFile(testState.sessionStorePath as string, "utf-8"),
|
},
|
||||||
) as Record<string, { lastChannel?: string; lastTo?: string } | undefined>;
|
},
|
||||||
expect(stored["agent:main:main"]?.lastChannel).toBe("whatsapp");
|
async () => {
|
||||||
expect(stored["agent:main:main"]?.lastTo).toBe("+1555");
|
const routeRes = await rpcReq(ws, "chat.send", {
|
||||||
},
|
sessionKey: "main",
|
||||||
);
|
message: "hello",
|
||||||
await withSessionStore(
|
idempotencyKey: "idem-route",
|
||||||
tempDirs,
|
});
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
expect(routeRes.ok).toBe(true);
|
||||||
async () => {
|
const stored = JSON.parse(
|
||||||
resetSpy();
|
await fs.readFile(testState.sessionStorePath as string, "utf-8"),
|
||||||
let abortInFlight: Promise<unknown> | undefined;
|
) as Record<string, { lastChannel?: string; lastTo?: string } | undefined>;
|
||||||
try {
|
expect(stored["agent:main:main"]?.lastChannel).toBe("whatsapp");
|
||||||
const callsBefore = spy.mock.calls.length;
|
expect(stored["agent:main:main"]?.lastTo).toBe("+1555");
|
||||||
|
},
|
||||||
|
);
|
||||||
|
await withSessionStore(
|
||||||
|
tempDirs,
|
||||||
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
|
async () => {
|
||||||
|
resetSpy();
|
||||||
|
let abortInFlight: Promise<unknown> | undefined;
|
||||||
|
try {
|
||||||
|
const callsBefore = spy.mock.calls.length;
|
||||||
|
spy.mockImplementationOnce(async (opts) => {
|
||||||
|
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
||||||
|
await new Promise<void>((resolve) => {
|
||||||
|
if (!signal) return resolve();
|
||||||
|
if (signal.aborted) return resolve();
|
||||||
|
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||||
|
});
|
||||||
|
});
|
||||||
|
const sendResP = onceMessage(
|
||||||
|
ws,
|
||||||
|
(o) => o.type === "res" && o.id === "send-abort-1",
|
||||||
|
8000,
|
||||||
|
);
|
||||||
|
const abortResP = onceMessage(
|
||||||
|
ws,
|
||||||
|
(o) => o.type === "res" && o.id === "abort-1",
|
||||||
|
8000,
|
||||||
|
);
|
||||||
|
const abortedEventP = onceMessage(
|
||||||
|
ws,
|
||||||
|
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "aborted",
|
||||||
|
8000,
|
||||||
|
);
|
||||||
|
abortInFlight = Promise.allSettled([sendResP, abortResP, abortedEventP]);
|
||||||
|
sendReq(ws, "send-abort-1", "chat.send", {
|
||||||
|
sessionKey: "main",
|
||||||
|
message: "hello",
|
||||||
|
idempotencyKey: "idem-abort-1",
|
||||||
|
timeoutMs: 30_000,
|
||||||
|
});
|
||||||
|
const sendRes = await sendResP;
|
||||||
|
expect(sendRes.ok).toBe(true);
|
||||||
|
await new Promise<void>((resolve, reject) => {
|
||||||
|
const deadline = Date.now() + 1000;
|
||||||
|
const tick = () => {
|
||||||
|
if (spy.mock.calls.length > callsBefore) return resolve();
|
||||||
|
if (Date.now() > deadline)
|
||||||
|
return reject(new Error("timeout waiting for agentCommand"));
|
||||||
|
setTimeout(tick, 5);
|
||||||
|
};
|
||||||
|
tick();
|
||||||
|
});
|
||||||
|
sendReq(ws, "abort-1", "chat.abort", {
|
||||||
|
sessionKey: "main",
|
||||||
|
runId: "idem-abort-1",
|
||||||
|
});
|
||||||
|
const abortRes = await abortResP;
|
||||||
|
expect(abortRes.ok).toBe(true);
|
||||||
|
const evt = await abortedEventP;
|
||||||
|
expect(evt.payload?.runId).toBe("idem-abort-1");
|
||||||
|
expect(evt.payload?.sessionKey).toBe("main");
|
||||||
|
} finally {
|
||||||
|
await abortInFlight;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
);
|
||||||
|
await withSessionStore(
|
||||||
|
tempDirs,
|
||||||
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
|
async () => {
|
||||||
|
sessionStoreSaveDelayMs.value = 120;
|
||||||
|
resetSpy();
|
||||||
|
try {
|
||||||
|
spy.mockImplementationOnce(async (opts) => {
|
||||||
|
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
||||||
|
await new Promise<void>((resolve) => {
|
||||||
|
if (!signal) return resolve();
|
||||||
|
if (signal.aborted) return resolve();
|
||||||
|
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||||
|
});
|
||||||
|
});
|
||||||
|
const abortedEventP = onceMessage(
|
||||||
|
ws,
|
||||||
|
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "aborted",
|
||||||
|
);
|
||||||
|
const sendResP = onceMessage(
|
||||||
|
ws,
|
||||||
|
(o) => o.type === "res" && o.id === "send-abort-save-1",
|
||||||
|
);
|
||||||
|
sendReq(ws, "send-abort-save-1", "chat.send", {
|
||||||
|
sessionKey: "main",
|
||||||
|
message: "hello",
|
||||||
|
idempotencyKey: "idem-abort-save-1",
|
||||||
|
timeoutMs: 30_000,
|
||||||
|
});
|
||||||
|
const abortResP = onceMessage(ws, (o) => o.type === "res" && o.id === "abort-save-1");
|
||||||
|
sendReq(ws, "abort-save-1", "chat.abort", {
|
||||||
|
sessionKey: "main",
|
||||||
|
runId: "idem-abort-save-1",
|
||||||
|
});
|
||||||
|
const abortRes = await abortResP;
|
||||||
|
expect(abortRes.ok).toBe(true);
|
||||||
|
const sendRes = await sendResP;
|
||||||
|
expect(sendRes.ok).toBe(true);
|
||||||
|
const evt = await abortedEventP;
|
||||||
|
expect(evt.payload?.runId).toBe("idem-abort-save-1");
|
||||||
|
expect(evt.payload?.sessionKey).toBe("main");
|
||||||
|
} finally {
|
||||||
|
sessionStoreSaveDelayMs.value = 0;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
);
|
||||||
|
await withSessionStore(
|
||||||
|
tempDirs,
|
||||||
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
|
async () => {
|
||||||
|
resetSpy();
|
||||||
|
const callsBeforeStop = spy.mock.calls.length;
|
||||||
spy.mockImplementationOnce(async (opts) => {
|
spy.mockImplementationOnce(async (opts) => {
|
||||||
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
||||||
await new Promise<void>((resolve) => {
|
await new Promise<void>((resolve) => {
|
||||||
@@ -140,383 +252,284 @@ describe("gateway server chat", () => {
|
|||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
const sendResP = onceMessage(
|
const stopSendResP = onceMessage(
|
||||||
ws,
|
ws,
|
||||||
(o) => o.type === "res" && o.id === "send-abort-1",
|
(o) => o.type === "res" && o.id === "send-stop-1",
|
||||||
8000,
|
8000,
|
||||||
);
|
);
|
||||||
const abortResP = onceMessage(ws, (o) => o.type === "res" && o.id === "abort-1", 8000);
|
sendReq(ws, "send-stop-1", "chat.send", {
|
||||||
const abortedEventP = onceMessage(
|
sessionKey: "main",
|
||||||
|
message: "hello",
|
||||||
|
idempotencyKey: "idem-stop-run",
|
||||||
|
});
|
||||||
|
const stopSendRes = await stopSendResP;
|
||||||
|
expect(stopSendRes.ok).toBe(true);
|
||||||
|
await waitFor(() => spy.mock.calls.length > callsBeforeStop);
|
||||||
|
const abortedStopEventP = onceMessage(
|
||||||
ws,
|
ws,
|
||||||
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "aborted",
|
(o) =>
|
||||||
|
o.type === "event" &&
|
||||||
|
o.event === "chat" &&
|
||||||
|
o.payload?.state === "aborted" &&
|
||||||
|
o.payload?.runId === "idem-stop-run",
|
||||||
8000,
|
8000,
|
||||||
);
|
);
|
||||||
abortInFlight = Promise.allSettled([sendResP, abortResP, abortedEventP]);
|
const stopResP = onceMessage(
|
||||||
sendReq(ws, "send-abort-1", "chat.send", {
|
|
||||||
sessionKey: "main",
|
|
||||||
message: "hello",
|
|
||||||
idempotencyKey: "idem-abort-1",
|
|
||||||
timeoutMs: 30_000,
|
|
||||||
});
|
|
||||||
const sendRes = await sendResP;
|
|
||||||
expect(sendRes.ok).toBe(true);
|
|
||||||
await new Promise<void>((resolve, reject) => {
|
|
||||||
const deadline = Date.now() + 1000;
|
|
||||||
const tick = () => {
|
|
||||||
if (spy.mock.calls.length > callsBefore) return resolve();
|
|
||||||
if (Date.now() > deadline)
|
|
||||||
return reject(new Error("timeout waiting for agentCommand"));
|
|
||||||
setTimeout(tick, 5);
|
|
||||||
};
|
|
||||||
tick();
|
|
||||||
});
|
|
||||||
sendReq(ws, "abort-1", "chat.abort", {
|
|
||||||
sessionKey: "main",
|
|
||||||
runId: "idem-abort-1",
|
|
||||||
});
|
|
||||||
const abortRes = await abortResP;
|
|
||||||
expect(abortRes.ok).toBe(true);
|
|
||||||
const evt = await abortedEventP;
|
|
||||||
expect(evt.payload?.runId).toBe("idem-abort-1");
|
|
||||||
expect(evt.payload?.sessionKey).toBe("main");
|
|
||||||
} finally {
|
|
||||||
await abortInFlight;
|
|
||||||
}
|
|
||||||
},
|
|
||||||
);
|
|
||||||
await withSessionStore(
|
|
||||||
tempDirs,
|
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
|
||||||
async () => {
|
|
||||||
sessionStoreSaveDelayMs.value = 120;
|
|
||||||
resetSpy();
|
|
||||||
try {
|
|
||||||
spy.mockImplementationOnce(async (opts) => {
|
|
||||||
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
|
||||||
await new Promise<void>((resolve) => {
|
|
||||||
if (!signal) return resolve();
|
|
||||||
if (signal.aborted) return resolve();
|
|
||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
|
||||||
});
|
|
||||||
});
|
|
||||||
const abortedEventP = onceMessage(
|
|
||||||
ws,
|
ws,
|
||||||
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "aborted",
|
(o) => o.type === "res" && o.id === "send-stop-2",
|
||||||
|
8000,
|
||||||
);
|
);
|
||||||
const sendResP = onceMessage(
|
sendReq(ws, "send-stop-2", "chat.send", {
|
||||||
ws,
|
|
||||||
(o) => o.type === "res" && o.id === "send-abort-save-1",
|
|
||||||
);
|
|
||||||
sendReq(ws, "send-abort-save-1", "chat.send", {
|
|
||||||
sessionKey: "main",
|
sessionKey: "main",
|
||||||
message: "hello",
|
message: "/stop",
|
||||||
idempotencyKey: "idem-abort-save-1",
|
idempotencyKey: "idem-stop-req",
|
||||||
timeoutMs: 30_000,
|
|
||||||
});
|
});
|
||||||
const abortResP = onceMessage(ws, (o) => o.type === "res" && o.id === "abort-save-1");
|
const stopRes = await stopResP;
|
||||||
sendReq(ws, "abort-save-1", "chat.abort", {
|
expect(stopRes.ok).toBe(true);
|
||||||
sessionKey: "main",
|
const stopEvt = await abortedStopEventP;
|
||||||
runId: "idem-abort-save-1",
|
expect(stopEvt.payload?.sessionKey).toBe("main");
|
||||||
});
|
expect(spy.mock.calls.length).toBe(callsBeforeStop + 1);
|
||||||
const abortRes = await abortResP;
|
},
|
||||||
expect(abortRes.ok).toBe(true);
|
);
|
||||||
const sendRes = await sendResP;
|
resetSpy();
|
||||||
expect(sendRes.ok).toBe(true);
|
let resolveRun: (() => void) | undefined;
|
||||||
const evt = await abortedEventP;
|
const runDone = new Promise<void>((resolve) => {
|
||||||
expect(evt.payload?.runId).toBe("idem-abort-save-1");
|
resolveRun = resolve;
|
||||||
expect(evt.payload?.sessionKey).toBe("main");
|
});
|
||||||
} finally {
|
spy.mockImplementationOnce(async () => {
|
||||||
sessionStoreSaveDelayMs.value = 0;
|
await runDone;
|
||||||
}
|
});
|
||||||
},
|
const started = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
||||||
);
|
|
||||||
await withSessionStore(
|
|
||||||
tempDirs,
|
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
|
||||||
async () => {
|
|
||||||
resetSpy();
|
|
||||||
const callsBeforeStop = spy.mock.calls.length;
|
|
||||||
spy.mockImplementationOnce(async (opts) => {
|
|
||||||
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
|
||||||
await new Promise<void>((resolve) => {
|
|
||||||
if (!signal) return resolve();
|
|
||||||
if (signal.aborted) return resolve();
|
|
||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
|
||||||
});
|
|
||||||
});
|
|
||||||
const stopSendResP = onceMessage(
|
|
||||||
ws,
|
|
||||||
(o) => o.type === "res" && o.id === "send-stop-1",
|
|
||||||
8000,
|
|
||||||
);
|
|
||||||
sendReq(ws, "send-stop-1", "chat.send", {
|
|
||||||
sessionKey: "main",
|
|
||||||
message: "hello",
|
|
||||||
idempotencyKey: "idem-stop-run",
|
|
||||||
});
|
|
||||||
const stopSendRes = await stopSendResP;
|
|
||||||
expect(stopSendRes.ok).toBe(true);
|
|
||||||
await waitFor(() => spy.mock.calls.length > callsBeforeStop);
|
|
||||||
const abortedStopEventP = onceMessage(
|
|
||||||
ws,
|
|
||||||
(o) =>
|
|
||||||
o.type === "event" &&
|
|
||||||
o.event === "chat" &&
|
|
||||||
o.payload?.state === "aborted" &&
|
|
||||||
o.payload?.runId === "idem-stop-run",
|
|
||||||
8000,
|
|
||||||
);
|
|
||||||
const stopResP = onceMessage(ws, (o) => o.type === "res" && o.id === "send-stop-2", 8000);
|
|
||||||
sendReq(ws, "send-stop-2", "chat.send", {
|
|
||||||
sessionKey: "main",
|
|
||||||
message: "/stop",
|
|
||||||
idempotencyKey: "idem-stop-req",
|
|
||||||
});
|
|
||||||
const stopRes = await stopResP;
|
|
||||||
expect(stopRes.ok).toBe(true);
|
|
||||||
const stopEvt = await abortedStopEventP;
|
|
||||||
expect(stopEvt.payload?.sessionKey).toBe("main");
|
|
||||||
expect(spy.mock.calls.length).toBe(callsBeforeStop + 1);
|
|
||||||
},
|
|
||||||
);
|
|
||||||
resetSpy();
|
|
||||||
let resolveRun: (() => void) | undefined;
|
|
||||||
const runDone = new Promise<void>((resolve) => {
|
|
||||||
resolveRun = resolve;
|
|
||||||
});
|
|
||||||
spy.mockImplementationOnce(async () => {
|
|
||||||
await runDone;
|
|
||||||
});
|
|
||||||
const started = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
|
||||||
sessionKey: "main",
|
|
||||||
message: "hello",
|
|
||||||
idempotencyKey: "idem-status-1",
|
|
||||||
});
|
|
||||||
expect(started.ok).toBe(true);
|
|
||||||
expect(started.payload?.status).toBe("started");
|
|
||||||
const inFlightRes = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
|
||||||
sessionKey: "main",
|
|
||||||
message: "hello",
|
|
||||||
idempotencyKey: "idem-status-1",
|
|
||||||
});
|
|
||||||
expect(inFlightRes.ok).toBe(true);
|
|
||||||
expect(inFlightRes.payload?.status).toBe("in_flight");
|
|
||||||
resolveRun?.();
|
|
||||||
let completed = false;
|
|
||||||
for (let i = 0; i < 50; i++) {
|
|
||||||
const again = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
|
||||||
sessionKey: "main",
|
sessionKey: "main",
|
||||||
message: "hello",
|
message: "hello",
|
||||||
idempotencyKey: "idem-status-1",
|
idempotencyKey: "idem-status-1",
|
||||||
});
|
});
|
||||||
if (again.ok && again.payload?.status === "ok") {
|
expect(started.ok).toBe(true);
|
||||||
completed = true;
|
expect(started.payload?.status).toBe("started");
|
||||||
break;
|
const inFlightRes = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
||||||
}
|
sessionKey: "main",
|
||||||
await new Promise((r) => setTimeout(r, 10));
|
message: "hello",
|
||||||
}
|
idempotencyKey: "idem-status-1",
|
||||||
expect(completed).toBe(true);
|
|
||||||
resetSpy();
|
|
||||||
spy.mockImplementationOnce(async (opts) => {
|
|
||||||
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
|
||||||
await new Promise<void>((resolve) => {
|
|
||||||
if (!signal) return resolve();
|
|
||||||
if (signal.aborted) return resolve();
|
|
||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
|
||||||
});
|
});
|
||||||
});
|
expect(inFlightRes.ok).toBe(true);
|
||||||
const abortedEventP = onceMessage(
|
expect(inFlightRes.payload?.status).toBe("in_flight");
|
||||||
ws,
|
resolveRun?.();
|
||||||
(o) =>
|
let completed = false;
|
||||||
o.type === "event" &&
|
for (let i = 0; i < 50; i++) {
|
||||||
o.event === "chat" &&
|
const again = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
||||||
o.payload?.state === "aborted" &&
|
sessionKey: "main",
|
||||||
o.payload?.runId === "idem-abort-all-1",
|
message: "hello",
|
||||||
);
|
idempotencyKey: "idem-status-1",
|
||||||
const startedAbortAll = await rpcReq(ws, "chat.send", {
|
});
|
||||||
sessionKey: "main",
|
if (again.ok && again.payload?.status === "ok") {
|
||||||
message: "hello",
|
completed = true;
|
||||||
idempotencyKey: "idem-abort-all-1",
|
break;
|
||||||
});
|
}
|
||||||
expect(startedAbortAll.ok).toBe(true);
|
await new Promise((r) => setTimeout(r, 10));
|
||||||
const abortRes = await rpcReq<{
|
}
|
||||||
ok?: boolean;
|
expect(completed).toBe(true);
|
||||||
aborted?: boolean;
|
resetSpy();
|
||||||
runIds?: string[];
|
spy.mockImplementationOnce(async (opts) => {
|
||||||
}>(ws, "chat.abort", { sessionKey: "main" });
|
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
||||||
expect(abortRes.ok).toBe(true);
|
await new Promise<void>((resolve) => {
|
||||||
expect(abortRes.payload?.aborted).toBe(true);
|
if (!signal) return resolve();
|
||||||
expect(abortRes.payload?.runIds ?? []).toContain("idem-abort-all-1");
|
if (signal.aborted) return resolve();
|
||||||
await abortedEventP;
|
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||||
const noDeltaP = onceMessage(
|
});
|
||||||
ws,
|
});
|
||||||
(o) =>
|
const abortedEventP = onceMessage(
|
||||||
o.type === "event" &&
|
ws,
|
||||||
o.event === "chat" &&
|
(o) =>
|
||||||
(o.payload?.state === "delta" || o.payload?.state === "final") &&
|
o.type === "event" &&
|
||||||
o.payload?.runId === "idem-abort-all-1",
|
o.event === "chat" &&
|
||||||
250,
|
o.payload?.state === "aborted" &&
|
||||||
);
|
o.payload?.runId === "idem-abort-all-1",
|
||||||
emitAgentEvent({
|
);
|
||||||
runId: "idem-abort-all-1",
|
const startedAbortAll = await rpcReq(ws, "chat.send", {
|
||||||
stream: "assistant",
|
sessionKey: "main",
|
||||||
data: { text: "should be suppressed" },
|
message: "hello",
|
||||||
});
|
idempotencyKey: "idem-abort-all-1",
|
||||||
emitAgentEvent({
|
});
|
||||||
runId: "idem-abort-all-1",
|
expect(startedAbortAll.ok).toBe(true);
|
||||||
stream: "lifecycle",
|
const abortRes = await rpcReq<{
|
||||||
data: { phase: "end" },
|
|
||||||
});
|
|
||||||
await expect(noDeltaP).rejects.toThrow(/timeout/i);
|
|
||||||
await withSessionStore(tempDirs, {}, async () => {
|
|
||||||
const abortUnknown = await rpcReq<{
|
|
||||||
ok?: boolean;
|
ok?: boolean;
|
||||||
aborted?: boolean;
|
aborted?: boolean;
|
||||||
}>(ws, "chat.abort", { sessionKey: "main", runId: "missing-run" });
|
runIds?: string[];
|
||||||
expect(abortUnknown.ok).toBe(true);
|
}>(ws, "chat.abort", { sessionKey: "main" });
|
||||||
expect(abortUnknown.payload?.aborted).toBe(false);
|
expect(abortRes.ok).toBe(true);
|
||||||
});
|
expect(abortRes.payload?.aborted).toBe(true);
|
||||||
await withSessionStore(
|
expect(abortRes.payload?.runIds ?? []).toContain("idem-abort-all-1");
|
||||||
tempDirs,
|
await abortedEventP;
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
const noDeltaP = onceMessage(
|
||||||
async () => {
|
ws,
|
||||||
resetSpy();
|
(o) =>
|
||||||
let agentStartedResolve: (() => void) | undefined;
|
o.type === "event" &&
|
||||||
const agentStartedP = new Promise<void>((resolve) => {
|
o.event === "chat" &&
|
||||||
agentStartedResolve = resolve;
|
(o.payload?.state === "delta" || o.payload?.state === "final") &&
|
||||||
});
|
o.payload?.runId === "idem-abort-all-1",
|
||||||
spy.mockImplementationOnce(async (opts) => {
|
250,
|
||||||
agentStartedResolve?.();
|
);
|
||||||
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
emitAgentEvent({
|
||||||
await new Promise<void>((resolve) => {
|
runId: "idem-abort-all-1",
|
||||||
if (!signal) return resolve();
|
stream: "assistant",
|
||||||
if (signal.aborted) return resolve();
|
data: { text: "should be suppressed" },
|
||||||
signal.addEventListener("abort", () => resolve(), { once: true });
|
});
|
||||||
|
emitAgentEvent({
|
||||||
|
runId: "idem-abort-all-1",
|
||||||
|
stream: "lifecycle",
|
||||||
|
data: { phase: "end" },
|
||||||
|
});
|
||||||
|
await expect(noDeltaP).rejects.toThrow(/timeout/i);
|
||||||
|
await withSessionStore(tempDirs, {}, async () => {
|
||||||
|
const abortUnknown = await rpcReq<{
|
||||||
|
ok?: boolean;
|
||||||
|
aborted?: boolean;
|
||||||
|
}>(ws, "chat.abort", { sessionKey: "main", runId: "missing-run" });
|
||||||
|
expect(abortUnknown.ok).toBe(true);
|
||||||
|
expect(abortUnknown.payload?.aborted).toBe(false);
|
||||||
|
});
|
||||||
|
await withSessionStore(
|
||||||
|
tempDirs,
|
||||||
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
|
async () => {
|
||||||
|
resetSpy();
|
||||||
|
let agentStartedResolve: (() => void) | undefined;
|
||||||
|
const agentStartedP = new Promise<void>((resolve) => {
|
||||||
|
agentStartedResolve = resolve;
|
||||||
});
|
});
|
||||||
});
|
spy.mockImplementationOnce(async (opts) => {
|
||||||
const sendResP = onceMessage(
|
agentStartedResolve?.();
|
||||||
ws,
|
const signal = (opts as { abortSignal?: AbortSignal }).abortSignal;
|
||||||
(o) => o.type === "res" && o.id === "send-mismatch-1",
|
await new Promise<void>((resolve) => {
|
||||||
10_000,
|
if (!signal) return resolve();
|
||||||
);
|
if (signal.aborted) return resolve();
|
||||||
sendReq(ws, "send-mismatch-1", "chat.send", {
|
signal.addEventListener("abort", () => resolve(), { once: true });
|
||||||
sessionKey: "main",
|
});
|
||||||
message: "hello",
|
});
|
||||||
idempotencyKey: "idem-mismatch-1",
|
const sendResP = onceMessage(
|
||||||
timeoutMs: 30_000,
|
ws,
|
||||||
});
|
(o) => o.type === "res" && o.id === "send-mismatch-1",
|
||||||
await agentStartedP;
|
10_000,
|
||||||
const abortMismatch = await rpcReq(ws, "chat.abort", {
|
);
|
||||||
sessionKey: "other",
|
sendReq(ws, "send-mismatch-1", "chat.send", {
|
||||||
runId: "idem-mismatch-1",
|
sessionKey: "main",
|
||||||
});
|
message: "hello",
|
||||||
expect(abortMismatch.ok).toBe(false);
|
idempotencyKey: "idem-mismatch-1",
|
||||||
expect(abortMismatch.error?.code).toBe("INVALID_REQUEST");
|
timeoutMs: 30_000,
|
||||||
const abortMismatch2 = await rpcReq(ws, "chat.abort", {
|
});
|
||||||
sessionKey: "main",
|
await agentStartedP;
|
||||||
runId: "idem-mismatch-1",
|
const abortMismatch = await rpcReq(ws, "chat.abort", {
|
||||||
});
|
sessionKey: "other",
|
||||||
expect(abortMismatch2.ok).toBe(true);
|
runId: "idem-mismatch-1",
|
||||||
const sendRes = await sendResP;
|
});
|
||||||
expect(sendRes.ok).toBe(true);
|
expect(abortMismatch.ok).toBe(false);
|
||||||
},
|
expect(abortMismatch.error?.code).toBe("INVALID_REQUEST");
|
||||||
);
|
const abortMismatch2 = await rpcReq(ws, "chat.abort", {
|
||||||
await withSessionStore(
|
sessionKey: "main",
|
||||||
tempDirs,
|
runId: "idem-mismatch-1",
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
});
|
||||||
async () => {
|
expect(abortMismatch2.ok).toBe(true);
|
||||||
resetSpy();
|
const sendRes = await sendResP;
|
||||||
spy.mockResolvedValueOnce(undefined);
|
expect(sendRes.ok).toBe(true);
|
||||||
sendReq(ws, "send-complete-1", "chat.send", {
|
},
|
||||||
sessionKey: "main",
|
);
|
||||||
message: "hello",
|
await withSessionStore(
|
||||||
idempotencyKey: "idem-complete-1",
|
tempDirs,
|
||||||
timeoutMs: 30_000,
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
});
|
async () => {
|
||||||
const sendCompleteRes = await onceMessage(
|
resetSpy();
|
||||||
ws,
|
spy.mockResolvedValueOnce(undefined);
|
||||||
(o) => o.type === "res" && o.id === "send-complete-1",
|
sendReq(ws, "send-complete-1", "chat.send", {
|
||||||
);
|
|
||||||
expect(sendCompleteRes.ok).toBe(true);
|
|
||||||
let completedRun = false;
|
|
||||||
for (let i = 0; i < 50; i++) {
|
|
||||||
const again = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
|
||||||
sessionKey: "main",
|
sessionKey: "main",
|
||||||
message: "hello",
|
message: "hello",
|
||||||
idempotencyKey: "idem-complete-1",
|
idempotencyKey: "idem-complete-1",
|
||||||
timeoutMs: 30_000,
|
timeoutMs: 30_000,
|
||||||
});
|
});
|
||||||
if (again.ok && again.payload?.status === "ok") {
|
const sendCompleteRes = await onceMessage(
|
||||||
completedRun = true;
|
ws,
|
||||||
break;
|
(o) => o.type === "res" && o.id === "send-complete-1",
|
||||||
|
);
|
||||||
|
expect(sendCompleteRes.ok).toBe(true);
|
||||||
|
let completedRun = false;
|
||||||
|
for (let i = 0; i < 50; i++) {
|
||||||
|
const again = await rpcReq<{ runId?: string; status?: string }>(ws, "chat.send", {
|
||||||
|
sessionKey: "main",
|
||||||
|
message: "hello",
|
||||||
|
idempotencyKey: "idem-complete-1",
|
||||||
|
timeoutMs: 30_000,
|
||||||
|
});
|
||||||
|
if (again.ok && again.payload?.status === "ok") {
|
||||||
|
completedRun = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
await new Promise((r) => setTimeout(r, 10));
|
||||||
}
|
}
|
||||||
await new Promise((r) => setTimeout(r, 10));
|
expect(completedRun).toBe(true);
|
||||||
}
|
const abortCompleteRes = await rpcReq(ws, "chat.abort", {
|
||||||
expect(completedRun).toBe(true);
|
sessionKey: "main",
|
||||||
const abortCompleteRes = await rpcReq(ws, "chat.abort", {
|
runId: "idem-complete-1",
|
||||||
sessionKey: "main",
|
});
|
||||||
runId: "idem-complete-1",
|
expect(abortCompleteRes.ok).toBe(true);
|
||||||
});
|
expect(abortCompleteRes.payload?.aborted).toBe(false);
|
||||||
expect(abortCompleteRes.ok).toBe(true);
|
},
|
||||||
expect(abortCompleteRes.payload?.aborted).toBe(false);
|
);
|
||||||
},
|
await withSessionStore(
|
||||||
);
|
tempDirs,
|
||||||
await withSessionStore(
|
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
||||||
tempDirs,
|
async () => {
|
||||||
{ main: { sessionId: "sess-main", updatedAt: Date.now() } },
|
const res1 = await rpcReq(ws, "chat.send", {
|
||||||
async () => {
|
sessionKey: "main",
|
||||||
const res1 = await rpcReq(ws, "chat.send", {
|
message: "first",
|
||||||
sessionKey: "main",
|
idempotencyKey: "idem-1",
|
||||||
message: "first",
|
});
|
||||||
idempotencyKey: "idem-1",
|
expect(res1.ok).toBe(true);
|
||||||
});
|
const res2 = await rpcReq(ws, "chat.send", {
|
||||||
expect(res1.ok).toBe(true);
|
sessionKey: "main",
|
||||||
const res2 = await rpcReq(ws, "chat.send", {
|
message: "second",
|
||||||
sessionKey: "main",
|
idempotencyKey: "idem-2",
|
||||||
message: "second",
|
});
|
||||||
idempotencyKey: "idem-2",
|
expect(res2.ok).toBe(true);
|
||||||
});
|
const final1P = onceMessage(
|
||||||
expect(res2.ok).toBe(true);
|
ws,
|
||||||
const final1P = onceMessage(
|
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "final",
|
||||||
ws,
|
8000,
|
||||||
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "final",
|
);
|
||||||
8000,
|
emitAgentEvent({
|
||||||
);
|
runId: "idem-1",
|
||||||
emitAgentEvent({
|
stream: "lifecycle",
|
||||||
runId: "idem-1",
|
data: { phase: "end" },
|
||||||
stream: "lifecycle",
|
});
|
||||||
data: { phase: "end" },
|
const final1 = await final1P;
|
||||||
});
|
const run1 =
|
||||||
const final1 = await final1P;
|
final1.payload && typeof final1.payload === "object"
|
||||||
const run1 =
|
? (final1.payload as { runId?: string }).runId
|
||||||
final1.payload && typeof final1.payload === "object"
|
: undefined;
|
||||||
? (final1.payload as { runId?: string }).runId
|
expect(run1).toBe("idem-1");
|
||||||
: undefined;
|
const final2P = onceMessage(
|
||||||
expect(run1).toBe("idem-1");
|
ws,
|
||||||
const final2P = onceMessage(
|
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "final",
|
||||||
ws,
|
8000,
|
||||||
(o) => o.type === "event" && o.event === "chat" && o.payload?.state === "final",
|
);
|
||||||
8000,
|
emitAgentEvent({
|
||||||
);
|
runId: "idem-2",
|
||||||
emitAgentEvent({
|
stream: "lifecycle",
|
||||||
runId: "idem-2",
|
data: { phase: "end" },
|
||||||
stream: "lifecycle",
|
});
|
||||||
data: { phase: "end" },
|
const final2 = await final2P;
|
||||||
});
|
const run2 =
|
||||||
const final2 = await final2P;
|
final2.payload && typeof final2.payload === "object"
|
||||||
const run2 =
|
? (final2.payload as { runId?: string }).runId
|
||||||
final2.payload && typeof final2.payload === "object"
|
: undefined;
|
||||||
? (final2.payload as { runId?: string }).runId
|
expect(run2).toBe("idem-2");
|
||||||
: undefined;
|
},
|
||||||
expect(run2).toBe("idem-2");
|
);
|
||||||
},
|
} finally {
|
||||||
);
|
testState.sessionStorePath = undefined;
|
||||||
} finally {
|
sessionStoreSaveDelayMs.value = 0;
|
||||||
testState.sessionStorePath = undefined;
|
ws.close();
|
||||||
sessionStoreSaveDelayMs.value = 0;
|
await server.close();
|
||||||
ws.close();
|
await Promise.all(tempDirs.map((dir) => fs.rm(dir, { recursive: true, force: true })));
|
||||||
await server.close();
|
}
|
||||||
await Promise.all(tempDirs.map((dir) => fs.rm(dir, { recursive: true, force: true })));
|
},
|
||||||
}
|
);
|
||||||
});
|
|
||||||
});
|
});
|
||||||
|
|||||||
Reference in New Issue
Block a user