fix: harden workflow lifecycle, concurrency, and retries with structured errors

CHANGELOG.md

@@ -9,7 +9,13 @@
 ### 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), and queue ordinary subagent batches beyond their concurrency window in the same process-wide admission window as workflow nodes. Subagent batches also honor `STEP_WORKFLOW_CONCURRENCY_LIMIT` and hold each slot until the child process exits.
+- Let a tool return `isError: true` while keeping its partial output and details, so failed subagent and workflow results reach the model as errors. Finalize a partial assistant message when a provider stream throws, ends without a terminal event, or fails during construction, instead of leaving the turn unfinished.
+- Classify provider failures structurally (HTTP status, provider error code, transport error) as rate limit, quota, auth, transient, or invalid request. Session retries, the separate 429 retry budget, and workflow congestion control use this classification instead of matching error text, so a 400 that mentions "429" is no longer retried and an exhausted quota returned as 429 is not treated as throttling.
+- Resume workflows by matching each call's prompt and options instead of the journal's sequence prefix, so a failed or reordered node no longer forces every later call to re-run. Once a live call writes files, runs a shell command, or uses an unrecognized tool, cached calls that read the environment re-run as well; cached results are re-validated against the call's schema, and calls with a runtime (safeParse) schema always run live.

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

@@ -5,7 +5,9 @@
 
 import {
 	type AssistantMessage,
+	type AssistantMessageEvent,
 	type Context,
+	classifyProviderError,
 	EventStream,
 	type ToolResultMessage,
 	validateToolArguments,
@@ -47,9 +49,10 @@ export function agentLoop(
 		},
 		signal,
 		streamFn,
-	).then((messages) => {
-		stream.end(messages);
-	});
+	).then(
+		(messages) => stream.end(messages),
+		(error) => stream.fail(error),
+	);
 
 	return stream;
 }
@@ -86,9 +89,10 @@ export function agentLoopContinue(
 		},
 		signal,
 		streamFn,
-	).then((messages) => {
-		stream.end(messages);
-	});
+	).then(
+		(messages) => stream.end(messages),
+		(error) => stream.fail(error),
+	);
 
 	return stream;
 }
@@ -143,8 +147,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 +361,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 +407,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 +424,63 @@ 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;
+	let truncated = 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;
+		}
+		truncated = true;
+		throw new Error("The model response ended before completion");
+	} catch (error) {
+		if (terminal) throw error;
+		const message = interruptedResponse(partial, model, signal, error);
+		// A stream that closed without a terminal event is a dropped connection.
+		const errorKind = signal?.aborted ? undefined : truncated ? "transient" : classifyProviderError(error);
+		if (errorKind) message.errorKind = errorKind;
+		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 +811,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. */