fix(agents): wait for terminal status after stream completion

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<typeof fetch>()
+          .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<typeof fetch>()
+          .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<typeof fetch>().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',

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<AgentEvent> {
   // ── 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<AgentResult> {
     // Drain any remaining events if not already done
@@ -388,14 +390,35 @@ export class AgentStream implements AsyncIterable<AgentEvent> {
     // signals stream end, NOT workflow success.
     let serverStatus: Record<string, unknown> | 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<string, unknown>;
+      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<string, unknown>;
+          }
+        } 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}`);
       }
     }