fix(studio): timeout stalled run_sse streams (#986)

Co-authored-by: shanchunhua <shanchunhua@bytedance.com>

Author: xgtcode

frontend/src/adk/client.ts

@@ -1642,6 +1642,60 @@ export const RUN_SSE_EMPTY_RESPONSE_ERROR =
 export const RUN_SSE_INCOMPLETE_RESPONSE_ERROR =
   formatRunSseError("HTTP 200，SSE 响应中没有可展示的模型回复。");
 
+const RUN_SSE_FIRST_EVENT_TIMEOUT_MS = 30_000;
+export const RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR =
+  formatRunSseError("30 秒内未收到首个 SSE 事件。");
+
+interface RunSseFirstEventDeadline {
+  signal?: AbortSignal;
+  clearDeadline: () => void;
+  cleanup: () => void;
+  timedOut: () => boolean;
+}
+
+function runSseFirstEventDeadline(
+  signal: AbortSignal | undefined,
+): RunSseFirstEventDeadline {
+  if (signal?.aborted) {
+    return {
+      signal,
+      clearDeadline: () => {},
+      cleanup: () => {},
+      timedOut: () => false,
+    };
+  }
+
+  const controller = new AbortController();
+  let didTimeout = false;
+  let deadlineCleared = false;
+  const clearDeadline = () => {
+    if (deadlineCleared) return;
+    deadlineCleared = true;
+    clearTimeout(timer);
+  };
+  const onAbort = () => {
+    if (controller.signal.aborted) return;
+    controller.abort(signal?.reason ?? new DOMException("Aborted", "AbortError"));
+  };
+  const timer = setTimeout(() => {
+    if (deadlineCleared || controller.signal.aborted) return;
+    didTimeout = true;
+    deadlineCleared = true;
+    controller.abort(new Error(RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR));
+  }, RUN_SSE_FIRST_EVENT_TIMEOUT_MS);
+  signal?.addEventListener("abort", onAbort, { once: true });
+
+  return {
+    signal: controller.signal,
+    clearDeadline,
+    cleanup: () => {
+      clearDeadline();
+      signal?.removeEventListener("abort", onAbort);
+    },
+    timedOut: () => didTimeout,
+  };
+}
+
 /** Stream agent events for one user turn. */
 export async function* runSSE({
   appName,
@@ -1698,6 +1752,7 @@ export async function* runSSE({
     };
   }
   let res: Response;
+  const firstEventDeadline = runSseFirstEventDeadline(signal);
   try {
     res = await apiFetch(
       "/run_sse",
@@ -1717,16 +1772,19 @@ export async function* runSSE({
             ? { veadkInvocation: invocationMetadata }
             : undefined,
         }),
-        signal,
+        signal: firstEventDeadline.signal,
       },
       ep,
       0,
     );
   } catch (error) {
+    firstEventDeadline.cleanup();
+    if (firstEventDeadline.timedOut()) throw new Error(RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR);
     if (signal?.aborted || (error as Error)?.name === "AbortError") throw error;
     throw new Error(formatRunSseError(error));
   }
   if (!res.ok) {
+    firstEventDeadline.cleanup();
     const detail = await httpErrorMessage(res, "运行会话失败");
     throw new Error(
       formatRunSseError(`run_sse failed: ${res.status}：${detail}`),
@@ -1736,6 +1794,7 @@ export async function* runSSE({
   try {
     for await (const evt of parseSSE(res)) {
       receivedEvent = true;
+      firstEventDeadline.clearDeadline();
       const event = evt as AdkEvent;
       if (typeof event.error === "string") event.error = formatRunSseError(event.error);
       if (typeof event.errorMessage === "string") {
@@ -1747,8 +1806,11 @@ export async function* runSSE({
       yield event;
     }
   } catch (error) {
+    if (firstEventDeadline.timedOut()) throw new Error(RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR);
     if (signal?.aborted || (error as Error)?.name === "AbortError") throw error;
     throw new Error(formatRunSseError(error));
+  } finally {
+    firstEventDeadline.cleanup();
   }
   if (!receivedEvent) throw new Error(RUN_SSE_EMPTY_RESPONSE_ERROR);
 }
@@ -3678,25 +3740,44 @@ export async function* runGeneratedAgentTestSSE({
   signal?: AbortSignal;
 }): AsyncGenerator<AdkEvent, void, unknown> {
   const parts: Record<string, unknown>[] = text.trim() ? [{ text }] : [];
-  const res = await apiFetch(
-    `/web/generated-agent-test-runs/${runId}/run_sse`,
-    {
-      method: "POST",
-      headers: { "Content-Type": "application/json" },
-      body: JSON.stringify({
-        user_id: userId,
-        session_id: sessionId,
-        new_message: { role: "user", parts },
-        streaming: true,
-      }),
-      signal,
-    },
-    {},
-    0,
-  );
-  if (!res.ok) throw new Error(await httpErrorMessage(res, "调试运行失败"));
-  for await (const evt of parseSSE(res)) {
-    yield evt as AdkEvent;
+  const firstEventDeadline = runSseFirstEventDeadline(signal);
+  let res: Response;
+  try {
+    res = await apiFetch(
+      `/web/generated-agent-test-runs/${runId}/run_sse`,
+      {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({
+          user_id: userId,
+          session_id: sessionId,
+          new_message: { role: "user", parts },
+          streaming: true,
+        }),
+        signal: firstEventDeadline.signal,
+      },
+      {},
+      0,
+    );
+  } catch (error) {
+    firstEventDeadline.cleanup();
+    if (firstEventDeadline.timedOut()) throw new Error(RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR);
+    throw error;
+  }
+  if (!res.ok) {
+    firstEventDeadline.cleanup();
+    throw new Error(await httpErrorMessage(res, "调试运行失败"));
+  }
+  try {
+    for await (const evt of parseSSE(res)) {
+      firstEventDeadline.clearDeadline();
+      yield evt as AdkEvent;
+    }
+  } catch (error) {
+    if (firstEventDeadline.timedOut()) throw new Error(RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR);
+    throw error;
+  } finally {
+    firstEventDeadline.cleanup();
   }
 }
 

frontend/tests/runSseAbort.test.mjs

@@ -39,7 +39,7 @@ const result = await build({
 const moduleUrl = `data:text/javascript;base64,${Buffer.from(
   result.outputFiles[0].contents,
 ).toString("base64")}`;
-const { runSSE } = await import(moduleUrl);
+const { RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR, runSSE } = await import(moduleUrl);
 
 test("runSSE forwards cancellation after yielding partial output", async (t) => {
   const previousFetch = globalThis.fetch;
@@ -87,7 +87,8 @@ test("runSSE forwards cancellation after yielding partial output", async (t) =>
   assert.equal(first.done, false);
   assert.equal(first.value.partial, true);
   assert.equal(first.value.content.parts[0].text, "part");
-  assert.equal(requestSignal, abortController.signal);
+  assert.ok(requestSignal instanceof AbortSignal);
+  assert.equal(requestSignal.aborted, false);
 
   const next = events.next();
   abortController.abort(new DOMException("Stopped by user", "AbortError"));
@@ -98,6 +99,102 @@ test("runSSE forwards cancellation after yielding partial output", async (t) =>
   });
 });
 
+test("runSSE aborts when no first event arrives before the deadline", async (t) => {
+  const previousFetch = globalThis.fetch;
+  const previousSetTimeout = globalThis.setTimeout;
+  const previousClearTimeout = globalThis.clearTimeout;
+  t.after(() => {
+    globalThis.fetch = previousFetch;
+    globalThis.setTimeout = previousSetTimeout;
+    globalThis.clearTimeout = previousClearTimeout;
+  });
+
+  let timeoutCallback;
+  let timeoutMs;
+  globalThis.setTimeout = (callback, ms, ...args) => {
+    timeoutMs = ms;
+    timeoutCallback = () => callback(...args);
+    return 1;
+  };
+  globalThis.clearTimeout = () => {};
+  globalThis.fetch = async (_url, init) => {
+    return new Promise((_resolve, reject) => {
+      init.signal.addEventListener(
+        "abort",
+        () => reject(init.signal.reason ?? new DOMException("Aborted", "AbortError")),
+        { once: true },
+      );
+    });
+  };
+
+  const events = runSSE({
+    appName: "agent",
+    userId: "user",
+    sessionId: "session",
+    text: "hello",
+  });
+
+  const next = events.next();
+  assert.equal(timeoutMs, 30_000);
+  timeoutCallback();
+
+  await assert.rejects(next, (error) => {
+    assert.equal(error.message, RUN_SSE_FIRST_EVENT_TIMEOUT_ERROR);
+    assert.match(error.message, /30 秒内未收到首个 SSE 事件/);
+    assert.match(error.message, /请检查共享公网出口等网络配置，然后重试/);
+    return true;
+  });
+});
+
+test("runSSE clears the first-event deadline after yielding the first event", async (t) => {
+  const previousFetch = globalThis.fetch;
+  const previousSetTimeout = globalThis.setTimeout;
+  const previousClearTimeout = globalThis.clearTimeout;
+  t.after(() => {
+    globalThis.fetch = previousFetch;
+    globalThis.setTimeout = previousSetTimeout;
+    globalThis.clearTimeout = previousClearTimeout;
+  });
+
+  let requestSignal;
+  let timeoutCallback;
+  let clearedTimer;
+  globalThis.setTimeout = (callback, ms, ...args) => {
+    assert.equal(ms, 30_000);
+    timeoutCallback = () => callback(...args);
+    return 7;
+  };
+  globalThis.clearTimeout = (timer) => {
+    clearedTimer = timer;
+  };
+  globalThis.fetch = async (_url, init) => {
+    requestSignal = init.signal;
+    return new Response(
+      'data: {"partial":true,"content":{"parts":[{"text":"part"}]}}\n\n',
+      {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      },
+    );
+  };
+
+  const events = runSSE({
+    appName: "agent",
+    userId: "user",
+    sessionId: "session",
+    text: "hello",
+  });
+
+  const first = await events.next();
+  assert.equal(first.done, false);
+  assert.equal(first.value.content.parts[0].text, "part");
+  assert.equal(clearedTimer, 7);
+  timeoutCallback();
+  assert.equal(requestSignal.aborted, false);
+
+  await events.return();
+});
+
 test("runSSE formats a fetch rejection before any response arrives", async (t) => {
   const previousFetch = globalThis.fetch;
   t.after(() => {