diff --git a/.changeset/clever-otters-show.md b/.changeset/clever-otters-show.md new file mode 100644 index 000000000..359ff9d7e --- /dev/null +++ b/.changeset/clever-otters-show.md @@ -0,0 +1,71 @@ +--- +"@voltagent/core": patch +--- + +fix: sub-agent stream error handling and propagation - #521 + +## What Changed + +Fixed a critical issue where sub-agent stream errors were incorrectly reported as successful operations with empty responses. Supervisors now properly detect and handle sub-agent failures with configurable error handling behavior. + +## The Problem (Before) + +When a sub-agent's `streamText` encountered an error event: + +- ❌ Supervisor received `status: "success"` with empty response +- ❌ No way to distinguish between empty success and failure +- ❌ Error details were lost, making debugging difficult +- ❌ Supervisors would continue as if the operation succeeded + +```typescript +// Before: Sub-agent fails but supervisor doesn't know +const result = await subAgentManager.handoffTask({ + task: "Process data", + targetAgent: failingAgent, +}); + +// result.status === "success" (WRONG!) +// result.result === "" (No error info) +``` + +## The Solution (After) + +Sub-agent errors are now properly detected and reported: + +- ✅ Stream errors return `status: "error"` with error details +- ✅ Error messages included in responses (configurable) +- ✅ Partial content preserved when errors occur after text generation +- ✅ New configuration options for flexible error handling + +```typescript +// After: Proper error detection and handling +const result = await subAgentManager.handoffTask({ + task: "Process data", + targetAgent: failingAgent, +}); + +// result.status === "error" (CORRECT!) +// result.result === "Error in FailingAgent: Stream processing failed" +// result.error === Error object with full details +``` + +## New Configuration Options + +Added `SupervisorConfig` options for customizable error handling: + +```typescript +const supervisor = new Agent({ + name: "Supervisor", + subAgents: [agent1, agent2], + supervisorConfig: { + // Throw exceptions on stream errors instead of returning error results + throwOnStreamError: false, // default: false + + // Include error messages in empty responses + includeErrorInEmptyResponse: true, // default: true + + // Custom guidelines for error handling + customGuidelines: ["When a sub-agent fails, provide alternative solutions"], + }, +}); +``` diff --git a/packages/core/src/agent/index.ts b/packages/core/src/agent/index.ts index a2a89937f..b7c9229bd 100644 --- a/packages/core/src/agent/index.ts +++ b/packages/core/src/agent/index.ts @@ -1 +1,2 @@ export { Agent } from "./agent"; +export type { SupervisorConfig } from "./types"; diff --git a/packages/core/src/agent/subagent/index.spec.ts b/packages/core/src/agent/subagent/index.spec.ts index 4c0f1edc0..1a829a24d 100644 --- a/packages/core/src/agent/subagent/index.spec.ts +++ b/packages/core/src/agent/subagent/index.spec.ts @@ -885,7 +885,7 @@ describe("SubAgentManager", () => { // Create subAgentManager with error type in configuration const supervisorConfig = { fullStreamEventForwarding: { - types: ["tool-call", "tool-result", "error"], + types: ["tool-call", "tool-result", "error"] as ("tool-call" | "tool-result" | "error")[], }, }; const localSubAgentManager = new SubAgentManager("Main Agent", [], supervisorConfig); @@ -943,6 +943,127 @@ describe("SubAgentManager", () => { expect(result.status).toBe("success"); }); + it("should return error status when stream error occurs with no text content", async () => { + const mockAgent = new MockAgent("error-only-agent", "Error Only Agent"); + + // Mock streamText to only emit an error (no text content) + mockAgent.streamText = vi.fn().mockReturnValue({ + fullStream: (async function* () { + yield { type: "error", error: new Error("Critical stream error") }; + })(), + textStream: (async function* () { + // No text emitted + })(), + }); + + const options: AgentHandoffOptions = { + task: "Task that will fail immediately", + targetAgent: mockAgent, + context: {}, + sharedContext: [], + forwardEvent: vi.fn(), + }; + + const result = await subAgentManager.handoffTask(options); + + // Should return error status + expect(result.status).toBe("error"); + expect(result.error).toBeInstanceOf(Error); + expect((result.error as Error).message).toBe("Critical stream error"); + + // By default, includeErrorInEmptyResponse is true, so error message should be in result + expect(result.result).toContain("Error in Error Only Agent"); + expect(result.result).toContain("Critical stream error"); + }); + + it("should return success with partial content when error occurs after text", async () => { + const mockAgent = new MockAgent("partial-content-agent", "Partial Content Agent"); + + // Mock streamText to emit some text then error + mockAgent.streamText = vi.fn().mockReturnValue({ + fullStream: (async function* () { + yield { type: "text-delta", textDelta: "Partial response" }; + yield { type: "error", error: new Error("Stream interrupted") }; + })(), + }); + + const options: AgentHandoffOptions = { + task: "Task with partial completion", + targetAgent: mockAgent, + context: {}, + sharedContext: [], + forwardEvent: vi.fn(), + }; + + const result = await subAgentManager.handoffTask(options); + + // Should return success with the partial content + expect(result.status).toBe("success"); + expect(result.result).toBe("Partial response"); + expect(result.error).toBeUndefined(); + }); + + it("should throw error when throwOnStreamError is true", async () => { + const supervisorConfig = { + throwOnStreamError: true, + }; + const localSubAgentManager = new SubAgentManager("Main Agent", [], supervisorConfig); + + const mockAgent = new MockAgent("throw-error-agent", "Throw Error Agent"); + + // Mock streamText to only emit an error + mockAgent.streamText = vi.fn().mockReturnValue({ + fullStream: (async function* () { + yield { type: "error", error: new Error("Stream error to throw") }; + })(), + }); + + const options: AgentHandoffOptions = { + task: "Task that should throw", + targetAgent: mockAgent, + context: {}, + sharedContext: [], + forwardEvent: vi.fn(), + }; + + // Should throw the error + await expect(localSubAgentManager.handoffTask(options)).rejects.toThrow( + "Stream error in Throw Error Agent: Stream error to throw", + ); + }); + + it("should not include error text when includeErrorInEmptyResponse is false", async () => { + const supervisorConfig = { + includeErrorInEmptyResponse: false, + }; + const localSubAgentManager = new SubAgentManager("Main Agent", [], supervisorConfig); + + const mockAgent = new MockAgent("no-error-text-agent", "No Error Text Agent"); + + // Mock streamText to only emit an error + mockAgent.streamText = vi.fn().mockReturnValue({ + fullStream: (async function* () { + yield { type: "error", error: new Error("Hidden error message") }; + })(), + }); + + const options: AgentHandoffOptions = { + task: "Task with hidden error", + targetAgent: mockAgent, + context: {}, + sharedContext: [], + forwardEvent: vi.fn(), + }; + + const result = await localSubAgentManager.handoffTask(options); + + // Should return error status but empty result + expect(result.status).toBe("error"); + expect(result.result).toBe(""); + expect(result.error).toBeInstanceOf(Error); + expect((result.error as Error).message).toBe("Hidden error message"); + }); + it("should forward events through delegate tool", async () => { const forwardEventSpy = vi.fn(); const mockAgent = new MockAgent("delegate-agent", "Delegate Agent"); @@ -979,6 +1100,45 @@ describe("SubAgentManager", () => { } }); + it("should propagate stream errors through delegate_task tool", async () => { + const forwardEventSpy = vi.fn(); + const mockAgent = new MockAgent("error-delegate-agent", "Error Delegate Agent"); + + // Mock streamText to only emit an error + mockAgent.streamText = vi.fn().mockReturnValue({ + fullStream: (async function* () { + yield { type: "error", error: new Error("Delegation error") }; + })(), + }); + + subAgentManager.addSubAgent(mockAgent as any); + + const tool = subAgentManager.createDelegateTool({ + sourceAgent: { id: "supervisor-agent" } as any, + operationContext: { userContext: new Map(), systemContext: new Map() } as any, + currentHistoryEntryId: "history-error", + forwardEvent: forwardEventSpy, + }); + + const result = await tool.execute({ + task: "Task that will fail in delegation", + targetAgents: ["Error Delegate Agent"], + context: {}, + }); + + // Should return structured results with error status + expect(Array.isArray(result)).toBe(true); + expect(result[0]).toMatchObject({ + agentName: "Error Delegate Agent", + status: "error", + error: "Delegation error", + }); + + // Error message should be included in response by default + expect(result[0].response).toContain("Error in Error Delegate Agent"); + expect(result[0].response).toContain("Delegation error"); + }); + it("should handle multiple agents with event forwarding", async () => { const forwardEventSpy = vi.fn(); const mockAgent1 = new MockAgent("multi-agent-1", "Multi Agent 1"); @@ -1089,7 +1249,12 @@ describe("SubAgentManager", () => { it("should use custom event types from supervisor configuration", async () => { const supervisorConfig = { fullStreamEventForwarding: { - types: ["tool-call", "tool-result", "text-delta", "reasoning"], + types: ["tool-call", "tool-result", "text-delta", "reasoning"] as ( + | "tool-call" + | "tool-result" + | "text-delta" + | "reasoning" + )[], }, }; subAgentManager = new SubAgentManager("Main Agent", [], supervisorConfig); @@ -1115,7 +1280,7 @@ describe("SubAgentManager", () => { it("should respect addSubAgentPrefix configuration", async () => { const supervisorConfig = { fullStreamEventForwarding: { - types: ["tool-call"], + types: ["tool-call"] as "tool-call"[], addSubAgentPrefix: false, }, }; diff --git a/packages/core/src/agent/subagent/index.ts b/packages/core/src/agent/subagent/index.ts index 48f3664f5..8b5b17fe6 100644 --- a/packages/core/src/agent/subagent/index.ts +++ b/packages/core/src/agent/subagent/index.ts @@ -295,6 +295,9 @@ ${guidelinesText} // Use the provided conversationId or generate a new one const handoffConversationId = conversationId || crypto.randomUUID(); + // Track if we should rethrow stream errors + let streamErrorToThrow: Error | null = null; + try { // Call onHandoff hook if source agent is provided if (sourceAgent && targetAgent.hooks) { @@ -391,6 +394,10 @@ ${task}\n\nContext: ${safeStringify(context, { indentation: 2 })}`; // Collect all stream chunks for final result finalResult = ""; + // Track stream errors and whether we received any text content + let streamError: Error | null = null; + let hasTextContent = false; + if (streamResponse.fullStream && forwardEvent) { // Get event forwarding configuration const eventForwardingConfig = { @@ -410,6 +417,7 @@ ${task}\n\nContext: ${safeStringify(context, { indentation: 2 })}`; switch (part.type) { case "text-delta": { finalResult += part.textDelta; + hasTextContent = true; const eventData = { type: "text-delta", @@ -491,6 +499,9 @@ ${task}\n\nContext: ${safeStringify(context, { indentation: 2 })}`; } case "error": { + // Capture the error for proper handling + streamError = part.error; + const eventData = { type: "error", data: { @@ -513,9 +524,52 @@ ${task}\n\nContext: ${safeStringify(context, { indentation: 2 })}`; } else { for await (const part of streamResponse.textStream) { finalResult += part; + hasTextContent = true; } } + // Handle stream errors based on configuration + if (streamError && !hasTextContent) { + const errorMessage = + streamError instanceof Error ? streamError.message : String(streamError); + + // Check if we should throw the error + if (this.supervisorConfig?.throwOnStreamError) { + // Store the error to throw after the try-catch + streamErrorToThrow = new Error(`Stream error in ${targetAgent.name}: ${errorMessage}`); + // Still throw here to exit the try block + throw streamErrorToThrow; + } + + // Check if we should include error message in empty response + const includeErrorInResponse = this.supervisorConfig?.includeErrorInEmptyResponse ?? true; + + return { + result: includeErrorInResponse ? `Error in ${targetAgent.name}: ${errorMessage}` : "", + conversationId: handoffConversationId, + messages: [ + taskMessage, + { + role: "system" as const, + content: `Stream error occurred: ${errorMessage}`, + }, + ], + status: "error", + error: streamError, + }; + } + + // If we have partial content despite an error, log warning but return the content + if (streamError && hasTextContent) { + const logger = + options.parentOperationContext?.logger || + getGlobalLogger().child({ component: "subagent-manager" }); + logger.warn(`Stream error occurred after partial content in ${targetAgent.name}`, { + error: streamError, + partialContent: finalResult, + }); + } + finalMessages = [taskMessage, { role: "assistant", content: finalResult }]; } @@ -526,6 +580,11 @@ ${task}\n\nContext: ${safeStringify(context, { indentation: 2 })}`; status: "success", }; } catch (error) { + // If this is the stream error we marked for rethrowing, rethrow it + if (streamErrorToThrow && error === streamErrorToThrow) { + throw error; + } + const logger = options.parentOperationContext?.logger || getGlobalLogger().child({ component: "subagent-manager" }); diff --git a/packages/core/src/agent/types.ts b/packages/core/src/agent/types.ts index d2db04e48..d0dfa9388 100644 --- a/packages/core/src/agent/types.ts +++ b/packages/core/src/agent/types.ts @@ -131,6 +131,23 @@ export type SupervisorConfig = { * @default { types: ['tool-call', 'tool-result'], addSubAgentPrefix: true } */ fullStreamEventForwarding?: FullStreamEventForwardingConfig; + + /** + * Whether to throw an exception when a subagent stream encounters an error + * If true, stream errors will cause the handoff to throw an exception + * If false, errors will be captured and returned in the result + * @default false + */ + throwOnStreamError?: boolean; + + /** + * Whether to include error message in the result when no text content was produced + * Only applies when throwOnStreamError is false + * If true, the error message will be included in the result field + * If false, the result will be empty but status will still be 'error' + * @default true + */ + includeErrorInEmptyResponse?: boolean; }; /** diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 9a0ca8f47..9210a21d5 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -27,6 +27,7 @@ export type { TextSubAgentConfig, ObjectSubAgentConfig, } from "./agent/subagent/types"; +export type { SupervisorConfig } from "./agent/types"; export * from "./tool"; export * from "./tool/reasoning/index"; export * from "./memory"; diff --git a/website/docs/agents/subagents.md b/website/docs/agents/subagents.md index f266c8580..afa6a80a3 100644 --- a/website/docs/agents/subagents.md +++ b/website/docs/agents/subagents.md @@ -167,6 +167,142 @@ fullStreamEventForwarding: { This configuration helps you balance between stream performance and information richness, allowing you to see exactly what you need from subagent interactions. +### Error Handling Configuration + +Control how your supervisor handles subagent failures with configurable error handling behavior. This is especially useful when subagents encounter stream errors or fail to generate responses. + +```ts +const supervisorAgent = new Agent({ + name: "Supervisor", + instructions: "Coordinate between agents", + llm: new VercelAIProvider(), + model: openai("gpt-4o-mini"), + subAgents: [dataProcessor, analyzer], + + supervisorConfig: { + // Control whether stream errors throw exceptions + throwOnStreamError: false, // default: false + + // Control whether error messages appear in empty responses + includeErrorInEmptyResponse: true, // default: true + }, +}); +``` + +#### Configuration Options + +**`throwOnStreamError`** (boolean, default: `false`) + +- When `false`: Stream errors are caught and returned as error results with `status: "error"` +- When `true`: Stream errors throw exceptions that must be caught with try/catch +- Use `true` when you want to handle errors at a higher level or trigger retry logic + +**`includeErrorInEmptyResponse`** (boolean, default: `true`) + +- When `true`: Error messages are included in the response when no content was generated +- When `false`: Returns empty string in result, but still marks status as "error" +- Use `false` when you want to handle error messaging yourself + +#### Common Error Handling Patterns + +**Default - Graceful Error Handling:** + +```ts +// Errors are returned as results with helpful messages +supervisorConfig: { + throwOnStreamError: false, + includeErrorInEmptyResponse: true, +} + +// Usage: +const result = await supervisor.streamText("Process data"); +// If subagent fails: +// result contains error message like "Error in DataProcessor: Stream failed" +``` + +**Exception-Based - For Retry Logic:** + +```ts +// Errors throw exceptions for custom handling +supervisorConfig: { + throwOnStreamError: true, +} + +// Usage with retry: +let retries = 3; +while (retries > 0) { + try { + const result = await supervisor.streamText("Process data"); + break; + } catch (error) { + console.error(`Attempt failed: ${error.message}`); + retries--; + if (retries === 0) throw error; + } +} +``` + +**Silent Errors - Custom Messaging:** + +```ts +// Errors don't include automatic messages +supervisorConfig: { + includeErrorInEmptyResponse: false, +} + +// Usage with custom error handling: +const result = await supervisor.streamText("Process data"); +for await (const event of result.fullStream) { + if (event.type === "error") { + // Provide custom user-friendly error message + console.log("We're having trouble processing your request. Please try again."); + } +} +``` + +**Production Setup - Detailed Error Tracking:** + +```ts +supervisorConfig: { + throwOnStreamError: false, // Don't crash the app + includeErrorInEmptyResponse: true, // Help with debugging + + // Also capture all error events for monitoring + fullStreamEventForwarding: { + types: ['tool-call', 'tool-result', 'error'], + }, +} +``` + +#### Error Handling in Practice + +When a subagent encounters an error, the supervisor's behavior depends on your configuration: + +```ts +// Example: Subagent fails during stream +const supervisor = new Agent({ + name: "Supervisor", + subAgents: [unreliableAgent], + supervisorConfig: { + throwOnStreamError: false, + includeErrorInEmptyResponse: true, + }, +}); + +// The supervisor handles the failure gracefully +const response = await supervisor.streamText("Do something risky"); + +// Check the response +if (response.status === "error") { + console.log("Subagent failed:", response.error); + // response.result contains: "Error in UnreliableAgent: [error details]" +} else { + // Process successful response +} +``` + +This configuration ensures your supervisor agents can handle subagent failures gracefully, providing better reliability and debugging capabilities in production environments. + #### Using with fullStream When using `fullStream` to get detailed streaming events, the configuration controls what you receive from subagents: @@ -305,6 +441,23 @@ supervisorConfig: { } ``` +**Handle errors gracefully:** + +```ts +supervisorConfig: { + throwOnStreamError: false, // Return errors as results (default) + includeErrorInEmptyResponse: true // Include error details in response (default) +} +``` + +**Throw exceptions for retry logic:** + +```ts +supervisorConfig: { + throwOnStreamError: true; // Throw exceptions on subagent failures +} +``` + ## How Subagents Work The core mechanism involves the supervisor agent delegating tasks to its subagents using the automatically provided `delegate_task` tool.