fix(agent-core): end agentLoop/agentLoopContinue streams on rejection

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<any>, 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;
+}

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<AgentEvent, AgentMessage[]> {
 	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<AgentEvent, AgentMessage[]> {
 	);
 }
 
+/**
+ * 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<AgentEvent, AgentMessage[]>,
+	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.
  */

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<void> {
-		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: [] });

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);
+		}
+	});
+});