From acd85dfa5c842c299512faa4e508c9e8ec505f30 Mon Sep 17 00:00:00 2001 From: Ling-Sen Peng Date: Fri, 31 Jul 2026 19:04:32 -0700 Subject: [PATCH] fix(agents): wait for a terminal status in getResult() (#155) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AgentRuntime.run() could return status "RUNNING" for an execution that had already finished on the server — wrong, not late: the failing calls returned in 2-4s, nowhere near any timeout. run() takes its result from the SSE stream, and getResult() read the status endpoint once. The stream ending does not mean the workflow ended, so that single read can land mid-flight and the non-terminal status is returned verbatim. The code comment already said "poll the server for the real terminal status" — it just never polled. Most visible with guardrails: server-side the workflow reaches FAILED with reasonForIncompletion set, after the expected retry iterations and a TERMINATE task, while the SDK reports RUNNING — a guardrail that worked perfectly looks like it never fired, and callers branching on result.status take the wrong branch. Read until the status is terminal, bounded by RESULT_SETTLE_TIMEOUT_MS (30s) so a long-running execution still returns rather than hanging. Reuses the existing TERMINAL_STATUSES set, so this path now agrees with wait(), liveness, and schedules. Only a *successful but non-terminal* read is retried. An unreachable or erroring endpoint returns immediately with what was seen last — preserving the previous behaviour on that path, so a broken endpoint cannot burn the settle budget. The non-streaming path never had this bug: _pollForCompletion() already loops until isComplete. That asymmetry was the defect. Tests: three regression cases — a RUNNING/RUNNING/FAILED sequence resolving to FAILED (the reported bug), a terminal-first status doing exactly one read, and an erroring endpoint falling back to stream inference in one read. Verified end to end against the 4.0.0-rc4 e2e bundle on a live server, with streaming enabled and no server change: Suite 8 goes from 3 failed / 4 passed to 7 passed / 7. Co-Authored-By: Claude Opus 5 (1M context) --- src/agents/__tests__/stream.test.ts | 116 ++++++++++++++++++++++++++++ src/agents/stream.ts | 69 +++++++++++++---- 2 files changed, 171 insertions(+), 14 deletions(-) diff --git a/src/agents/__tests__/stream.test.ts b/src/agents/__tests__/stream.test.ts index eac63b88..cd5811a2 100644 --- a/src/agents/__tests__/stream.test.ts +++ b/src/agents/__tests__/stream.test.ts @@ -324,6 +324,122 @@ describe("AgentStream", () => { expect(result.status).toBe("FAILED"); expect(result.error).toBe("something broke"); }); + + // ── Regression: #155 ─────────────────────────────── + // + // The stream ending does not mean the workflow ended. A single status read + // can land while the execution is still RUNNING and report that as the + // final answer — e.g. a guardrail that is about to escalate to FAILED. + + it("waits for a non-terminal status to settle before building the result", async () => { + const sseChunks = ['event:done\ndata:{"output":{"partial":true}}\n\n']; + const statuses = [ + { status: "RUNNING", output: {} }, + { status: "RUNNING", output: {} }, + { status: "FAILED", output: { blocked: true }, reasonForIncompletion: "Do not include secrets." }, + ]; + + let statusCall = 0; + global.fetch = jest.fn(async (url: unknown) => { + if (String(url).includes("/status")) { + const body = statuses[Math.min(statusCall++, statuses.length - 1)]; + return { ok: true, status: 200, json: async () => body, headers: new Headers() }; + } + return { + ok: true, + status: 200, + body: createSSEStream(sseChunks), + text: async () => "", + headers: new Headers(), + }; + }) as unknown as typeof fetch; + + const stream = new AgentStream( + "http://localhost/sse", + async () => ({}), + "wf-1", + jest.fn(), + "http://localhost/api", + ); + + const result = await stream.getResult(); + + // Before the fix this was "RUNNING" — the first read, taken as final. + expect(result.status).toBe("FAILED"); + expect(result.error).toBe("Do not include secrets."); + expect(statusCall).toBe(3); + }); + + it("returns immediately on a terminal status without extra polling", async () => { + const sseChunks = ['event:done\ndata:{"output":{}}\n\n']; + + let statusCall = 0; + global.fetch = jest.fn(async (url: unknown) => { + if (String(url).includes("/status")) { + statusCall++; + return { + ok: true, + status: 200, + json: async () => ({ status: "COMPLETED", output: { answer: 42 } }), + headers: new Headers(), + }; + } + return { + ok: true, + status: 200, + body: createSSEStream(sseChunks), + text: async () => "", + headers: new Headers(), + }; + }) as unknown as typeof fetch; + + const stream = new AgentStream( + "http://localhost/sse", + async () => ({}), + "wf-1", + jest.fn(), + "http://localhost/api", + ); + + const result = await stream.getResult(); + expect(result.status).toBe("COMPLETED"); + expect(result.output).toEqual({ answer: 42 }); + expect(statusCall).toBe(1); + }); + + it("falls back to stream inference when the status endpoint errors", async () => { + const sseChunks = ['event:done\ndata:{"output":{"answer":42}}\n\n']; + + let statusCall = 0; + global.fetch = jest.fn(async (url: unknown) => { + if (String(url).includes("/status")) { + statusCall++; + return { ok: false, status: 500, json: async () => ({}), headers: new Headers() }; + } + return { + ok: true, + status: 200, + body: createSSEStream(sseChunks), + text: async () => "", + headers: new Headers(), + }; + }) as unknown as typeof fetch; + + const stream = new AgentStream( + "http://localhost/sse", + async () => ({}), + "wf-1", + jest.fn(), + "http://localhost/api", + ); + + const result = await stream.getResult(); + + // An endpoint that is not answering must not burn the settle budget — + // give up after one attempt, exactly as before the fix. + expect(result.status).toBe("COMPLETED"); + expect(statusCall).toBe(1); + }); }); describe("executionId", () => { diff --git a/src/agents/stream.ts b/src/agents/stream.ts index 749ae412..8fb1f8f7 100644 --- a/src/agents/stream.ts +++ b/src/agents/stream.ts @@ -1,13 +1,22 @@ import type { AgentEvent, AgentResult, AgentStatus } from "./types.js"; import { stripInternalEventKeys } from "./types.js"; import { SSETimeoutError, SSEUnavailableError, ConductorAgentError } from "./errors.js"; -import { makeAgentResult } from "./result.js"; +import { makeAgentResult, TERMINAL_STATUSES } from "./result.js"; // ── Constants ─────────────────────────────────────────── const SSE_TIMEOUT_MS = 15_000; const MAX_RECONNECT_RETRIES = 5; const POLL_INTERVAL_MS = 500; +/** + * How long `getResult()` waits for a non-terminal execution to settle. + * + * The stream ending does not mean the workflow ended, so the status read can + * land mid-flight. This bounds that reconciliation only — it is not a run + * timeout, and a still-running execution after this simply reports its last + * observed status rather than throwing. + */ +const RESULT_SETTLE_TIMEOUT_MS = 30_000; // ── AgentStream ───────────────────────────────────────── @@ -385,19 +394,12 @@ export class AgentStream implements AsyncIterable { const errorEvent = this.events.findLast((e) => e.type === "error"); // Poll the server for the real terminal status — the done SSE event - // 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; - } - } catch { - // Fall back to stream-based inference - } - } + // signals stream end, NOT workflow success. The stream can close before + // the workflow's terminal transition, so a single read here can catch the + // execution mid-flight and report RUNNING for what is about to be FAILED. + // Keep reading until the status is terminal, bounded so a genuinely + // long-running execution still returns rather than hanging. + const serverStatus = await this._fetchTerminalStatus(); const status = (serverStatus?.status as string) ?? @@ -413,6 +415,45 @@ export class AgentStream implements AsyncIterable { events: [...this.events], }); } + + /** + * Read the execution status, waiting for it to become terminal. + * + * Returns as soon as the status is one of {@link TERMINAL_STATUSES}. If the + * execution is still non-terminal when the budget expires, returns the last + * status seen — callers get the best available answer, never a hang. + * + * Only a *successful but non-terminal* read is retried. An unreachable or + * erroring endpoint returns immediately with whatever was seen last (`null` + * on the first attempt, so the caller falls back to stream-based inference) + * rather than spending the settle budget on an endpoint that is not + * answering — which preserves the previous behaviour on that path. + */ + private async _fetchTerminalStatus(): Promise | null> { + if (!this.serverUrl || !this.executionId) return null; + + const statusUrl = `${this.serverUrl}/agent/${this.executionId}/status`; + const deadline = Date.now() + RESULT_SETTLE_TIMEOUT_MS; + let last: Record | null = null; + + for (;;) { + let current: Record | null = null; + try { + const resp = await fetch(statusUrl, { headers: await this.headerProvider() }); + if (resp.ok) current = (await resp.json()) as Record; + } catch { + // Treated the same as a non-ok response: stop and use what we have. + } + + if (!current) return last; + + last = current; + if (TERMINAL_STATUSES.has(current.status as string)) return current; + if (Date.now() >= deadline) return last; + + await sleep(POLL_INTERVAL_MS); + } + } } // ── Helpers ─────────────────────────────────────────────