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 ─────────────────────────────────────────────