From 64b0593a9e65dc3c339afa15a69e121005e0827f Mon Sep 17 00:00:00 2001 From: Jason Kneen Date: Tue, 22 Sep 2026 20:38:40 +0100 Subject: [PATCH] fix(agent-core): end agentLoop/agentLoopContinue streams on rejection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Invariant: a failure is its own terminal outcome — no consumer of the event stream may hang. agentLoop and agentLoopContinue did `void runAgentLoop(...).then((messages) => stream.end(messages))` with no rejection handler. If anything inside the loop threw synchronously or rejected (convertToLlm, transformContext, getApiKey, streamFn, a tool hook, prepareNextTurn), the returned EventStream never completed: `for await` consumers and `await stream.result()` hung forever, plus an unhandled promise rejection. Fix: attach a rejection handler to both functions that synthesizes an assistant failure message (stopReason "aborted" if the signal was aborted, else "error", errorMessage from the caught error, EMPTY_USAGE, model/api/provider from config.model) and pushes message_start, message_end, turn_end, and agent_end for it — mirroring Agent.handleRunFailure's semantics exactly. Since EventStream marks itself done as soon as an agent_end event is pushed, no separate stream.end(...) call follows the failure push. The final messages array is the messages already observed via message_end plus the synthesized failure message, tracked by tapping the emit callback (no change to runAgentLoop/runAgentLoopContinue's own signatures). Extracted the failure-message construction into a small shared helper (agent-failure.ts) and had Agent.handleRunFailure reuse it instead of duplicating the shape. --- packages/agent-core/src/agent-failure.ts | 36 ++++++ packages/agent-core/src/agent-loop.ts | 56 ++++++++- packages/agent-core/src/agent.ts | 22 +--- packages/agent-core/test/agent-loop.test.ts | 130 ++++++++++++++++++++ 4 files changed, 218 insertions(+), 26 deletions(-) create mode 100644 packages/agent-core/src/agent-failure.ts diff --git a/packages/agent-core/src/agent-failure.ts b/packages/agent-core/src/agent-failure.ts new file mode 100644 index 00000000..dfcb426c --- /dev/null +++ b/packages/agent-core/src/agent-failure.ts @@ -0,0 +1,36 @@ +/** + * Shared construction of the synthetic assistant message used to report an + * agent run that failed outside the normal turn loop (a rejected promise, an + * uncaught throw). Keeps the stateful `Agent` class and the standalone + * `agentLoop`/`agentLoopContinue` functions reporting failures the same way: + * a failure is its own terminal outcome, never a hang. + */ +import type { Model } from "@step-harness/providers"; +import type { AgentMessage } from "./types.ts"; + +export const EMPTY_USAGE = { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + totalTokens: 0, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, +}; + +/** + * Build an assistant message representing a run failure, mirroring + * `Agent.handleRunFailure`'s semantics. + */ +export function createFailureMessage(model: Model, aborted: boolean, error: unknown): AgentMessage { + return { + role: "assistant", + content: [{ type: "text", text: "" }], + api: model.api, + provider: model.provider, + model: model.id, + usage: EMPTY_USAGE, + stopReason: aborted ? "aborted" : "error", + errorMessage: error instanceof Error ? error.message : String(error), + timestamp: Date.now(), + } satisfies AgentMessage; +} diff --git a/packages/agent-core/src/agent-loop.ts b/packages/agent-core/src/agent-loop.ts index 0de9ec85..617cf2dc 100644 --- a/packages/agent-core/src/agent-loop.ts +++ b/packages/agent-core/src/agent-loop.ts @@ -10,6 +10,7 @@ import { type ToolResultMessage, validateToolArguments, } from "@step-harness/providers"; +import { createFailureMessage } from "./agent-failure.ts"; import { getDefaultStreamFn } from "./stream-fn.ts"; import type { AgentContext, @@ -37,19 +38,28 @@ export function agentLoop( streamFn: StreamFn, ): EventStream { const stream = createAgentStream(); + const messagesSoFar: AgentMessage[] = []; void runAgentLoop( prompts, context, config, async (event) => { + if (event.type === "message_end") { + messagesSoFar.push(event.message); + } stream.push(event); }, signal, streamFn, - ).then((messages) => { - stream.end(messages); - }); + ).then( + (messages) => { + stream.end(messages); + }, + (error) => { + emitLoopFailure(stream, messagesSoFar, config, signal, error); + }, + ); return stream; } @@ -77,18 +87,27 @@ export function agentLoopContinue( } const stream = createAgentStream(); + const messagesSoFar: AgentMessage[] = []; void runAgentLoopContinue( context, config, async (event) => { + if (event.type === "message_end") { + messagesSoFar.push(event.message); + } stream.push(event); }, signal, streamFn, - ).then((messages) => { - stream.end(messages); - }); + ).then( + (messages) => { + stream.end(messages); + }, + (error) => { + emitLoopFailure(stream, messagesSoFar, config, signal, error); + }, + ); return stream; } @@ -150,6 +169,31 @@ function createAgentStream(): EventStream { ); } +/** + * Invariant: a failure is its own terminal outcome — consumers of the stream + * (`for await`, `stream.result()`) must never hang. If `runAgentLoop`/ + * `runAgentLoopContinue` rejects (e.g. `convertToLlm`, `transformContext`, + * `getApiKey`, or `streamFn` throws), synthesize a failure message mirroring + * `Agent.handleRunFailure` and end the stream with it instead of leaving the + * rejection unhandled. + */ +function emitLoopFailure( + stream: EventStream, + messagesSoFar: AgentMessage[], + config: AgentLoopConfig, + signal: AbortSignal | undefined, + error: unknown, +): void { + const failureMessage = createFailureMessage(config.model, signal?.aborted === true, error); + const messages = [...messagesSoFar, failureMessage]; + stream.push({ type: "message_start", message: failureMessage }); + stream.push({ type: "message_end", message: failureMessage }); + stream.push({ type: "turn_end", message: failureMessage, toolResults: [] }); + // Pushing `agent_end` already completes the stream (see EventStream.push); + // no separate `stream.end(...)` call is needed or safe to add here. + stream.push({ type: "agent_end", messages }); +} + /** * Main loop logic shared by agentLoop and agentLoopContinue. */ diff --git a/packages/agent-core/src/agent.ts b/packages/agent-core/src/agent.ts index 68365590..2321fb2d 100644 --- a/packages/agent-core/src/agent.ts +++ b/packages/agent-core/src/agent.ts @@ -7,6 +7,7 @@ import type { ThinkingBudgets, Transport, } from "@step-harness/providers"; +import { createFailureMessage } from "./agent-failure.ts"; import { runAgentLoop, runAgentLoopContinue } from "./agent-loop.ts"; import { getDefaultStreamFn } from "./stream-fn.ts"; import type { @@ -36,15 +37,6 @@ function defaultConvertToLlm(messages: AgentMessage[]): Message[] { ); } -const EMPTY_USAGE = { - input: 0, - output: 0, - cacheRead: 0, - cacheWrite: 0, - totalTokens: 0, - cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, -}; - const DEFAULT_MODEL = { id: "unknown", name: "unknown", @@ -509,17 +501,7 @@ export class Agent { } private async handleRunFailure(error: unknown, aborted: boolean): Promise { - const failureMessage = { - role: "assistant", - content: [{ type: "text", text: "" }], - api: this._state.model.api, - provider: this._state.model.provider, - model: this._state.model.id, - usage: EMPTY_USAGE, - stopReason: aborted ? "aborted" : "error", - errorMessage: error instanceof Error ? error.message : String(error), - timestamp: Date.now(), - } satisfies AgentMessage; + const failureMessage = createFailureMessage(this._state.model, aborted, error); await this.processEvents({ type: "message_start", message: failureMessage }); await this.processEvents({ type: "message_end", message: failureMessage }); await this.processEvents({ type: "turn_end", message: failureMessage, toolResults: [] }); diff --git a/packages/agent-core/test/agent-loop.test.ts b/packages/agent-core/test/agent-loop.test.ts index 79957ef1..fa65b333 100644 --- a/packages/agent-core/test/agent-loop.test.ts +++ b/packages/agent-core/test/agent-loop.test.ts @@ -1730,3 +1730,133 @@ describe("tool-call markup leak retry", () => { expect(calls()).toBe(2); }); }); + +describe("agentLoop / agentLoopContinue rejection handling", () => { + // Invariant under test: a rejection from anywhere inside the loop (a + // throwing convertToLlm/transformContext/getApiKey/streamFn, etc.) is its + // own terminal outcome. The stream must still end, `stream.result()` must + // still resolve (never hang), and no unhandled rejection should surface + // (vitest fails the run on those). + + it("agentLoop: a throwing convertToLlm terminates the stream with a failure message", async () => { + const context: AgentContext = { systemPrompt: "", messages: [], tools: [] }; + const config: AgentLoopConfig = { + model: createModel(), + convertToLlm: () => { + throw new Error("convertToLlm boom"); + }, + }; + + const streamFn = () => { + throw new Error("streamFn should never be called"); + }; + + const stream = agentLoop([createUserMessage("go")], context, config, undefined, streamFn); + + const events: AgentEvent[] = []; + for await (const event of stream) { + events.push(event); + } + + expect(events.at(-1)?.type).toBe("agent_end"); + + const messages = await stream.result(); + const failure = messages.at(-1) as AssistantMessage; + expect(failure.role).toBe("assistant"); + expect(failure.stopReason).toBe("error"); + expect(failure.errorMessage).toBe("convertToLlm boom"); + expect(failure.usage).toEqual(createUsage()); + expect(failure.model).toBe("mock"); + expect(failure.api).toBe("openai-responses"); + expect(failure.provider).toBe("openai"); + }); + + it("agentLoopContinue: a throwing convertToLlm terminates the stream with a failure message", async () => { + const context: AgentContext = { + systemPrompt: "", + messages: [createUserMessage("Hello")], + tools: [], + }; + const config: AgentLoopConfig = { + model: createModel(), + convertToLlm: () => { + throw new Error("convertToLlm boom (continue)"); + }, + }; + + const streamFn = () => { + throw new Error("streamFn should never be called"); + }; + + const stream = agentLoopContinue(context, config, undefined, streamFn); + + const events: AgentEvent[] = []; + for await (const event of stream) { + events.push(event); + } + + expect(events.at(-1)?.type).toBe("agent_end"); + + const messages = await stream.result(); + const failure = messages.at(-1) as AssistantMessage; + expect(failure.stopReason).toBe("error"); + expect(failure.errorMessage).toBe("convertToLlm boom (continue)"); + }); + + it("agentLoop: reports stopReason 'aborted' when the signal is already aborted", async () => { + const context: AgentContext = { systemPrompt: "", messages: [], tools: [] }; + const config: AgentLoopConfig = { + model: createModel(), + convertToLlm: () => { + throw new Error("boom while aborted"); + }, + }; + + const controller = new AbortController(); + controller.abort(); + + const streamFn = () => { + throw new Error("streamFn should never be called"); + }; + + const stream = agentLoop([createUserMessage("go")], context, config, controller.signal, streamFn); + + for await (const _event of stream) { + // drain + } + + const messages = await stream.result(); + const failure = messages.at(-1) as AssistantMessage; + expect(failure.stopReason).toBe("aborted"); + expect(failure.errorMessage).toBe("boom while aborted"); + }); + + it("agentLoop: does not raise an unhandled rejection when the loop rejects", async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => unhandled.push(reason); + process.on("unhandledRejection", onUnhandled); + + try { + const context: AgentContext = { systemPrompt: "", messages: [], tools: [] }; + const config: AgentLoopConfig = { + model: createModel(), + convertToLlm: () => { + throw new Error("unhandled check boom"); + }, + }; + const streamFn = () => { + throw new Error("streamFn should never be called"); + }; + + const stream = agentLoop([createUserMessage("go")], context, config, undefined, streamFn); + await stream.result(); + + // Give any (incorrectly) unhandled rejection a microtask/macrotask to surface. + await new Promise((resolve) => setTimeout(resolve, 0)); + + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } + }); +});