github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(frontend): cover replay gap run identity and cancellation (#6637)

Browse Source 
Co-authored-by: TYY-2005 <334029581+TYY-2005@users.noreply.github.com>
This commit is contained in:
TYY-2005 2026-10-11 12:00:49 +08:00 committed by GitHub
parent 4beb6c4ea4
commit cb4393879d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 115 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
frontend/tests/unit/core/api/api-client.test.ts
View File

@ -1360,3 +1360,118 @@ test("short-circuits reconnect to an interrupted (user-cancelled) run", async ()
expect(fetchFn).toHaveBeenCalledTimes(1);
expect(sessionStorage.removeItem).toHaveBeenCalledWith("lg:stream:thread-1");
});
test.each(["initial", "join"] as const)(
"rejects replay gaps for another run on the %s stream",
async (source) => {
const sessionStorage = makeSessionStorage();
sessionStorage.setItem("lg:stream:thread-guard", "run-newer");
sessionStorage.removeItem.mockClear();
sessionStorage.setItem.mockClear();
const gap = {
code: "stream_replay_gap",
run_id: "run-other",
requested_event_id: "1-0",
earliest_available_event_id: "2-0",
latest_available_event_id: "3-0",
recovery: "reload_durable_state",
};
const fetchFn = rs.fn(async (url: string | URL) => {
const path = new URL(url.toString()).pathname;
if (path.endsWith("/runs/run-guard")) {
return new Response(JSON.stringify({ status: "running" }));
}
if (
path.endsWith("/runs/stream") ||
path.endsWith("/runs/run-guard/stream")
) {
return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`, {
"Content-Location": "/threads/thread-guard/runs/run-guard",
});
}
return new Response(JSON.stringify({ values: { messages: [] } }));
});
rs.stubGlobal("window", {
location: { origin: "http://localhost:2026" },
sessionStorage,
});
rs.stubGlobal("fetch", fetchFn);
const client = getAPIClient(true);
const stream =
source === "initial"
? client.runs.stream("thread-guard", "lead_agent", {})
: client.runs.joinStream("thread-guard", "run-guard");
await expect(stream.next()).rejects.toThrow(
"Stream replay gap does not match the active thread run.",
);
expect(fetchFn).toHaveBeenCalledTimes(source === "initial" ? 1 : 2);
expect(sessionStorage.getItem("lg:stream:thread-guard")).toBe("run-newer");
expect(sessionStorage.removeItem).not.toHaveBeenCalled();
expect(sessionStorage.setItem).not.toHaveBeenCalled();
},
);
test.each(["initial", "join"] as const)(
"propagates cancellation during %s stream gap recovery without rejoining",
async (source) => {
const sessionStorage = makeSessionStorage();
sessionStorage.setItem("lg:stream:thread-abort", "run-abort");
sessionStorage.setItem.mockClear();
const controller = new AbortController();
const abortError = new DOMException(
"AbortError: Recovery cancelled",
"AbortError",
);
const gap = {
code: "stream_replay_gap",
run_id: "run-abort",
requested_event_id: "1-0",
earliest_available_event_id: "2-0",
latest_available_event_id: "3-0",
recovery: "reload_durable_state",
};
const fetchFn = rs.fn(async (url: string | URL, init?: RequestInit) => {
const path = new URL(url.toString()).pathname;
if (path.endsWith("/runs/run-abort")) {
return new Response(JSON.stringify({ status: "running" }));
}
if (
path.endsWith("/runs/stream") ||
path.endsWith("/runs/run-abort/stream")
) {
return makeSSEResponse(`event: gap\ndata: ${JSON.stringify(gap)}\n\n`, {
"Content-Location": "/threads/thread-abort/runs/run-abort",
});
}
if (path.endsWith("/threads/thread-abort/state")) {
expect(init?.signal).toBe(controller.signal);
controller.abort();
throw abortError;
}
throw new Error(`Unexpected request: ${path}`);
});
rs.stubGlobal("window", {
location: { origin: "http://localhost:2026" },
sessionStorage,
});
rs.stubGlobal("fetch", fetchFn);
const client = getAPIClient(true);
const options = { signal: controller.signal };
const stream =
source === "initial"
? client.runs.stream("thread-abort", "lead_agent", options)
: client.runs.joinStream("thread-abort", "run-abort", options);
await expect(stream.next()).resolves.toMatchObject({
done: false,
value: { event: "custom", data: { type: "stream_replay_gap", ...gap } },
});
await expect(stream.next()).rejects.toBe(abortError);
expect(fetchFn).toHaveBeenCalledTimes(source === "initial" ? 2 : 3);
expect(sessionStorage.getItem("lg:stream:thread-abort")).toBeNull();
expect(sessionStorage.setItem).not.toHaveBeenCalled();
},
);