fix: harden workflow lifecycle, concurrency, retries, and error propagation

CHANGELOG.md

@@ -9,7 +9,10 @@
 ### Fixed
 
 - Stop flooding startup with `[Skill conflicts]` when the same skills are installed in both `~/.agents/skills` and `~/.claude/skills`. Byte-identical copies are now skipped silently, and remaining name collisions collapse to one summary line per directory pair; `--verbose` still lists each one.
+- Cancel and drain workflow children before publishing a terminal state, share host admission across concurrent workflows, accept promise arrays in `parallel()`, and distinguish concurrency notices from actual script failures.
+- Clamp workflow concurrency to the host ceiling, back off on temporary 429s with an independent retry budget and adaptive admission cap, and report failed nodes in final tool results instead of silently treating partial workflows as successful.
 - Persist Ultracode on/off state in the current session journal so reopening that session restores its mode without changing global or project settings.
 - Stop appending Ultracode capability tutorials on every turn. Announce session-mode transitions after user input, reuse unchanged state, and restore necessary reminders when context is compacted or replaced.
 - Recognize `/exit` as an alias for `/quit`, including while the agent is running or compacting.
 - Recognize `search_files`, `find_files`, and `list_directory` when selecting default system-prompt file-exploration guidance.
+- Report callback errors (including nested workflow errors raised before any agent starts), recognize explicit counted fan-out requests, and queue ordinary subagent batches independently of their concurrency window.

packages/agent-core/src/agent-loop.ts

@@ -5,6 +5,7 @@
 
 import {
 	type AssistantMessage,
+	type AssistantMessageEvent,
 	type Context,
 	EventStream,
 	type ToolResultMessage,
@@ -47,9 +48,10 @@ export function agentLoop(
 		},
 		signal,
 		streamFn,
-	).then((messages) => {
-		stream.end(messages);
-	});
+	).then(
+		(messages) => stream.end(messages),
+		(error) => stream.fail(error),
+	);
 
 	return stream;
 }
@@ -86,9 +88,10 @@ export function agentLoopContinue(
 		},
 		signal,
 		streamFn,
-	).then((messages) => {
-		stream.end(messages);
-	});
+	).then(
+		(messages) => stream.end(messages),
+		(error) => stream.fail(error),
+	);
 
 	return stream;
 }
@@ -143,8 +146,27 @@ export async function runAgentLoopContinue(
 	return newMessages;
 }
 
-function createAgentStream(): EventStream<AgentEvent, AgentMessage[]> {
-	return new EventStream<AgentEvent, AgentMessage[]>(
+class AgentLoopStream extends EventStream<AgentEvent, AgentMessage[]> {
+	private failed = false;
+	private failure: unknown;
+	fail(error: unknown): void {
+		this.failed = true;
+		this.failure = error;
+		this.end([]);
+	}
+	override async *[Symbol.asyncIterator](): AsyncGenerator<AgentEvent> {
+		yield* super[Symbol.asyncIterator]();
+		if (this.failed) throw this.failure;
+	}
+	override async result(): Promise<AgentMessage[]> {
+		const result = await super.result();
+		if (this.failed) throw this.failure;
+		return result;
+	}
+}
+
+function createAgentStream(): AgentLoopStream {
+	return new AgentLoopStream(
 		(event: AgentEvent) => event.type === "agent_end",
 		(event: AgentEvent) => (event.type === "agent_end" ? event.messages : []),
 	);
@@ -338,19 +360,21 @@ async function streamAssistantResponse(
 		tools: context.tools,
 	};
 
-	// Resolve API key (important for expiring tokens)
-	const resolvedApiKey =
-		(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;
-
-	const response = await streamFunction(config.model, llmContext, {
-		...config,
-		apiKey: resolvedApiKey,
+	const response = normalizeProviderStream(
+		async () => {
+			// Auth refresh and stream construction can throw before the first event.
+			const resolvedApiKey =
+				(config.getApiKey ? await config.getApiKey(config.model.provider) : undefined) || config.apiKey;
+			return streamFunction(config.model, llmContext, { ...config, apiKey: resolvedApiKey, signal });
+		},
+		config.model,
 		signal,
-	});
+	);
 
 	let partialMessage: AssistantMessage | null = null;
 	let addedPartial = false;
-
+	// Consumer/persistence errors must escape. Only the producer normalizes
+	// provider exceptions, so a failed message_end cannot emit a second terminal.
 	for await (const event of response) {
 		switch (event.type) {
 			case "start":
@@ -382,7 +406,10 @@ async function streamAssistantResponse(
 
 			case "done":
 			case "error": {
-				const finalMessage = await response.result();
+				// The terminal event already carries the final message. Do not wait on
+				// result() here: a provider that closes or rejects immediately after
+				// emitting the terminal event may leave that promise unresolved.
+				const finalMessage = event.type === "done" ? event.message : event.error;
 				if (addedPartial) {
 					context.messages[context.messages.length - 1] = finalMessage;
 				} else {
@@ -396,16 +423,58 @@ async function streamAssistantResponse(
 			}
 		}
 	}
+	throw new Error("Provider stream ended without a terminal event");
+}
 
-	const finalMessage = await response.result();
-	if (addedPartial) {
-		context.messages[context.messages.length - 1] = finalMessage;
-	} else {
-		context.messages.push(finalMessage);
-		await emit({ type: "message_start", message: { ...finalMessage } });
+async function* normalizeProviderStream(
+	createStream: () => ReturnType<StreamFn>,
+	model: AgentLoopConfig["model"],
+	signal: AbortSignal | undefined,
+): AsyncGenerator<AssistantMessageEvent> {
+	let partial: AssistantMessage | null = null;
+	let terminal = false;
+	try {
+		const response = await createStream();
+		for await (const event of response) {
+			if ("partial" in event) partial = event.partial;
+			terminal = event.type === "done" || event.type === "error";
+			yield event;
+			if (terminal) return;
+		}
+		throw new Error("The model response ended before completion");
+	} catch (error) {
+		if (terminal) throw error;
+		const message = interruptedResponse(partial, model, signal, error);
+		yield { type: "error", reason: signal?.aborted ? "aborted" : "error", error: message };
 	}
-	await emit({ type: "message_end", message: finalMessage });
-	return finalMessage;
+}
+
+function interruptedResponse(
+	partialMessage: AssistantMessage | null,
+	model: AgentLoopConfig["model"],
+	signal: AbortSignal | undefined,
+	error: unknown,
+): AssistantMessage {
+	return {
+		...(partialMessage ?? {
+			role: "assistant",
+			content: [],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			timestamp: Date.now(),
+		}),
+		stopReason: signal?.aborted ? "aborted" : "error",
+		errorMessage: error instanceof Error ? error.message : String(error),
+	};
 }
 
 /**
@@ -736,7 +805,7 @@ async function executePreparedToolCall(
 		);
 		acceptingUpdates = false;
 		await Promise.all(updateEvents);
-		return { result, isError: false };
+		return { result, isError: result.isError === true };
 	} catch (error) {
 		acceptingUpdates = false;
 		await Promise.all(updateEvents);

packages/agent-core/src/types.ts

@@ -383,6 +383,8 @@ export interface AgentState {
 
 /** Final or partial result produced by a tool. */
 export interface AgentToolResult<T> {
+	/** A failed tool result may retain successful partial output and structured details. */
+	isError?: boolean;
 	/** Text or image content returned to the model. */
 	content: (TextContent | ImageContent)[];
 	/** Arbitrary structured details for logs or UI rendering. */

packages/agent-core/test/agent-loop.test.ts

@@ -1,14 +1,15 @@
 import {
 	type AssistantMessage,
 	type AssistantMessageEvent,
+	type AssistantMessageEventStream,
 	EventStream,
 	type Message,
 	type Model,
 	type UserMessage,
 } from "@step-harness/providers";
 import { Type } from "typebox";
 import { describe, expect, it } from "vitest";
-import { agentLoop, agentLoopContinue } from "../src/agent-loop.ts";
+import { agentLoop, agentLoopContinue, runAgentLoop } from "../src/agent-loop.ts";
 import { setDefaultStreamFn } from "../src/index.ts";
 import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool } from "../src/types.ts";
 
@@ -163,6 +164,117 @@ describe("agentLoop with AgentMessage", () => {
 		expect(eventTypes).toContain("agent_end");
 	});
 
+	it("finalizes a partial assistant when the provider stream throws", async () => {
+		const partial = createAssistantMessage([{ type: "text", text: "partial response" }], "pending");
+		const streamFn = () =>
+			({
+				async *[Symbol.asyncIterator]() {
+					yield { type: "start", partial } satisfies AssistantMessageEvent;
+					throw new Error("connection terminated");
+				},
+				result: () => new Promise<AssistantMessage>(() => {}),
+			}) as unknown as AssistantMessageEventStream;
+
+		const events: AgentEvent[] = [];
+		const stream = agentLoop(
+			[createUserMessage("Hello")],
+			{ systemPrompt: "", messages: [], tools: [] },
+			{ model: createModel(), convertToLlm: identityConverter },
+			undefined,
+			streamFn,
+		);
+		for await (const event of stream) events.push(event);
+
+		const assistantEnd = events.find((event) => event.type === "message_end" && event.message.role === "assistant");
+		expect(assistantEnd).toMatchObject({
+			type: "message_end",
+			message: { role: "assistant", stopReason: "error", errorMessage: "connection terminated" },
+		});
+		expect(events.at(-1)?.type).toBe("agent_end");
+	});
+
+	it("finalizes synchronous provider construction failures", async () => {
+		const stream = agentLoop(
+			[createUserMessage("Hello")],
+			{ systemPrompt: "", messages: [], tools: [] },
+			{ model: createModel(), convertToLlm: identityConverter },
+			undefined,
+			() => {
+				throw new Error("provider construction failed");
+			},
+		);
+		const events: AgentEvent[] = [];
+		for await (const event of stream) events.push(event);
+		expect(
+			events.filter((event) => event.type === "message_end" && event.message.role === "assistant"),
+		).toMatchObject([{ message: { stopReason: "error", errorMessage: "provider construction failed" } }]);
+		expect(events.at(-1)?.type).toBe("agent_end");
+		expect((await stream.result()).at(-1)).toMatchObject({ stopReason: "error" });
+	});
+
+	it.each(["new", "continue"])(
+		"%s loop surfaces conversion failures instead of leaving consumers hanging",
+		async (mode) => {
+			const failure = new Error("conversion failed");
+			const context = { systemPrompt: "", messages: [createUserMessage("Hello")], tools: [] };
+			const config: AgentLoopConfig = {
+				model: createModel(),
+				convertToLlm: () => {
+					throw failure;
+				},
+			};
+			const streamFn = () => new MockAssistantStream();
+			const stream =
+				mode === "new"
+					? agentLoop([], context, config, undefined, streamFn)
+					: agentLoopContinue(context, config, undefined, streamFn);
+			await expect(
+				(async () => {
+					for await (const _event of stream) {
+						/* consume */
+					}
+				})(),
+			).rejects.toBe(failure);
+			await expect(stream.result()).rejects.toBe(failure);
+		},
+	);
+
+	it.each(["message_update", "message_end"])(
+		"does not turn %s listener failures into provider errors",
+		async (failAt) => {
+			const failure = new Error("listener failed: 503");
+			const partial = createAssistantMessage([{ type: "text", text: "partial" }], "pending");
+			const final = createAssistantMessage([{ type: "text", text: "complete final content" }]);
+			const events: AgentEvent[] = [];
+			await expect(
+				runAgentLoop(
+					[createUserMessage("Hello")],
+					{ systemPrompt: "", messages: [], tools: [] },
+					{ model: createModel(), convertToLlm: identityConverter },
+					(event) => {
+						events.push(event);
+						if (
+							(event.type === "message_update" && failAt === "message_update") ||
+							(event.type === "message_end" && failAt === "message_end" && event.message.role === "assistant")
+						)
+							throw failure;
+					},
+					undefined,
+					() => {
+						const response = new MockAssistantStream();
+						response.push({ type: "start", partial });
+						response.push({ type: "text_delta", contentIndex: 0, delta: "partial", partial });
+						response.push({ type: "done", reason: "stop", message: final });
+						return response;
+					},
+				),
+			).rejects.toBe(failure);
+			const ends = events.filter((event) => event.type === "message_end" && event.message.role === "assistant");
+			expect(ends).toHaveLength(failAt === "message_end" ? 1 : 0);
+			if (ends.length) expect(ends[0]).toMatchObject({ message: final });
+		},
+	);
+
 	it("should handle custom message types via convertToLlm", async () => {
 		// Create a custom message type
 		interface CustomNotification {
@@ -368,6 +480,59 @@ describe("agentLoop with AgentMessage", () => {
 		expect(toolResult?.role === "toolResult" ? toolResult.usage : undefined).toEqual(patchedToolUsage);
 	});
 
+	it("preserves explicit tool failure details in events and the next model context", async () => {
+		const details = { status: "partial_failed", completed: 12, failed: 10 };
+		const tool: AgentTool = {
+			name: "workflow",
+			label: "Workflow",
+			description: "Run nodes",
+			parameters: Type.Object({}),
+			execute: async () => ({ isError: true, details, content: [{ type: "text", text: "10 nodes failed: 429" }] }),
+		};
+		let calls = 0;
+		const stream = agentLoop(
+			[createUserMessage("run")],
+			{ systemPrompt: "", messages: [], tools: [tool] },
+			{ model: createModel(), convertToLlm: identityConverter },
+			undefined,
+			(_model, context) => {
+				const response = new MockAssistantStream();
+				if (calls++ === 0) {
+					response.push({
+						type: "done",
+						reason: "toolUse",
+						message: createAssistantMessage(
+							[{ type: "toolCall", id: "wf", name: "workflow", arguments: {} }],
+							"toolUse",
+						),
+					});
+				} else {
+					expect(context.messages.at(-1)).toMatchObject({
+						role: "toolResult",
+						isError: true,
+						content: [{ text: "10 nodes failed: 429" }],
+					});
+					response.push({
+						type: "done",
+						reason: "stop",
+						message: createAssistantMessage([{ type: "text", text: "partial failure" }]),
+					});
+				}
+				return response;
+			},
+		);
+		const events: AgentEvent[] = [];
+		for await (const event of stream) events.push(event);
+		expect(events.find((event) => event.type === "tool_execution_end")).toMatchObject({
+			isError: true,
+			result: { details },
+		});
+		expect((await stream.result()).find((message) => message.role === "toolResult")).toMatchObject({
+			isError: true,
+			details,
+		});
+	});
+
 	it("should not execute tool calls from a length-truncated assistant message", async () => {
 		const toolSchema = Type.Object({ value: Type.String() });
 		const executed: string[] = [];