diff --git a/src/agents/__tests__/stream.test.ts b/src/agents/__tests__/stream.test.ts index eac63b88..9232d017 100644 --- a/src/agents/__tests__/stream.test.ts +++ b/src/agents/__tests__/stream.test.ts @@ -297,6 +297,68 @@ describe("AgentStream", () => { }); describe("getResult", () => { + it("waits for terminal status after the SSE stream has ended", async () => { + jest.useFakeTimers(); + try { + const fetchStatus = jest.fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ status: "RUNNING", isComplete: false }))) + .mockResolvedValueOnce(new Response(JSON.stringify({ status: "FAILED", isComplete: true, reasonForIncompletion: "guardrail failed" }))); + global.fetch = fetchStatus; + const stream = new AgentStream("http://localhost/sse", async () => ({}), "wf-1", jest.fn(), "http://localhost"); + stream.events.push({ type: "done", output: "stale output" }); + stream.close(); + + const resultPromise = stream.getResult(); + await jest.advanceTimersByTimeAsync(500); + const result = await resultPromise; + expect(result.status).toBe("FAILED"); + expect(result.error).toBe("guardrail failed"); + expect(fetchStatus).toHaveBeenCalledTimes(2); + } finally { + jest.useRealTimers(); + } + }); + + it("retries a failed status read after observing a running execution", async () => { + jest.useFakeTimers(); + try { + const fetchStatus = jest.fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ status: "RUNNING", isComplete: false }))) + .mockRejectedValueOnce(new Error("temporary connection failure")) + .mockResolvedValueOnce(new Response(JSON.stringify({ status: "COMPLETED", isComplete: true, output: "final output" }))); + global.fetch = fetchStatus; + const stream = new AgentStream("http://localhost/sse", async () => ({}), "wf-1", jest.fn(), "http://localhost"); + stream.events.push({ type: "done", output: "stale output" }); + stream.close(); + const resultPromise = stream.getResult(); + await jest.advanceTimersByTimeAsync(1000); + const result = await resultPromise; + expect(result.status).toBe("COMPLETED"); + expect(result.output).toEqual({ result: "final output" }); + expect(fetchStatus).toHaveBeenCalledTimes(3); + } finally { + jest.useRealTimers(); + } + }); + + it("does not infer success when the server remains non-terminal", async () => { + jest.useFakeTimers(); + try { + global.fetch = jest.fn().mockImplementation(async () => + new Response(JSON.stringify({ status: "RUNNING", isComplete: false })), + ); + const stream = new AgentStream("http://localhost/sse", async () => ({}), "wf-1", jest.fn(), "http://localhost"); + stream.events.push({ type: "done" }); + stream.close(); + const resultPromise = stream.getResult(); + const assertion = expect(resultPromise).rejects.toThrow("terminal status"); + await jest.advanceTimersByTimeAsync(15_000); + await assertion; + } finally { + jest.useRealTimers(); + } + }); + it("builds AgentResult from done event", async () => { const sseChunks = [ 'event:thinking\ndata:{"content":"thinking..."}\n\n', diff --git a/src/agents/stream.ts b/src/agents/stream.ts index 749ae412..329b9006 100644 --- a/src/agents/stream.ts +++ b/src/agents/stream.ts @@ -8,6 +8,7 @@ import { makeAgentResult } from "./result.js"; const SSE_TIMEOUT_MS = 15_000; const MAX_RECONNECT_RETRIES = 5; const POLL_INTERVAL_MS = 500; +const RESULT_STATUS_TIMEOUT_MS = 15_000; // ── AgentStream ───────────────────────────────────────── @@ -370,7 +371,8 @@ export class AgentStream implements AsyncIterable { // ── getResult ──────────────────────────────────────── /** - * Drain all remaining events and build an AgentResult. + * Drain all remaining events and build an AgentResult. After stream closure, + * allow up to 15 seconds for a non-terminal server status to settle. */ async getResult(): Promise { // Drain any remaining events if not already done @@ -388,14 +390,35 @@ export class AgentStream implements AsyncIterable { // signals stream end, NOT workflow success. let serverStatus: Record | null = null; if (this.serverUrl && this.executionId) { - try { - const statusUrl = `${this.serverUrl}/agent/${this.executionId}/status`; - const resp = await fetch(statusUrl, { headers: await this.headerProvider() }); - if (resp.ok) { - serverStatus = (await resp.json()) as Record; + const deadline = Date.now() + RESULT_STATUS_TIMEOUT_MS; + const statusUrl = `${this.serverUrl}/agent/${this.executionId}/status`; + let hasTerminalStatus = false; + while (Date.now() < deadline) { + try { + const resp = await fetch(statusUrl, { + headers: await this.headerProvider(), + signal: AbortSignal.timeout(Math.max(1, deadline - Date.now())), + }); + if (resp.ok) { + serverStatus = (await resp.json()) as Record; + } + } catch { + // Preserve the last observed status across transient read failures. } - } catch { - // Fall back to stream-based inference + // Keep the existing fallback when no server status can be read, but + // never turn an observed RUNNING execution into inferred success. + hasTerminalStatus = serverStatus !== null && ( + serverStatus.isComplete === true || + ["COMPLETED", "FAILED", "TIMED_OUT", "TERMINATED", "CANCELED", "CANCELLED"] + .includes(String(serverStatus.status)) + ); + if (!serverStatus || hasTerminalStatus) { + break; + } + await sleep(Math.min(POLL_INTERVAL_MS, Math.max(0, deadline - Date.now()))); + } + if (serverStatus && !hasTerminalStatus) { + throw new ConductorAgentError(`Timed out waiting for terminal status of ${this.executionId}`); } }