fix: propagate executor errors directly to the caller instead of emitting synthetic failure tasks

src/server/request_handler/default_request_handler.ts

@@ -345,15 +345,14 @@ export class DefaultRequestHandler implements A2ARequestHandler {
   /**
    * Runs the executor for a blocking `sendMessage` call and ties the
    * event bus lifecycle to the executor's settlement. On rejection,
-   * publishes a synthetic Task + statusUpdate(FAILED) so the consumer's
-   * event loop terminates with a usable final result and any concurrent
-   * resubscribers see the failure on the same wire.
+   * lets the error propagate up to the caller's Promise.
    */
   private _runExecutor(
     taskId: string,
     eventBus: ExecutionEventBus,
     requestContext: RequestContext,
-    finalMessageForAgent: Message
+    finalMessageForAgent: Message,
+    onExecutorError?: (err: unknown) => void
   ): void {
     // Track the last task state on the bus directly: the consumer loop
     // that drains into `ResultManager` runs in a separate microtask, so
@@ -362,57 +361,10 @@ export class DefaultRequestHandler implements A2ARequestHandler {
     this.agentExecutor
       .execute(requestContext, eventBus)
       .catch((err: unknown) => {
-        // Promises can reject with any value, so coerce defensively
-        // before reading `.message`.
-        const errorMessage = extractErrorMessage(err);
         console.error(`Agent execution failed for message ${finalMessageForAgent.messageId}:`, err);
-        // The synthetic Task id MUST be `requestContext.taskId` — the
-        // id the bus is registered under and the id we hand back to
-        // the client. A fresh uuid would make the returned Task
-        // unreachable via `getTask`.
-        const errorTask: Task = {
-          id: requestContext.taskId,
-          contextId: finalMessageForAgent.contextId!,
-          status: {
-            state: TaskState.TASK_STATE_FAILED,
-            message: {
-              role: Role.ROLE_AGENT,
-              messageId: uuidv4(),
-              taskId: requestContext.taskId,
-              contextId: finalMessageForAgent.contextId!,
-              parts: [
-                {
-                  content: { $case: 'text', value: `Agent execution error: ${errorMessage}` },
-                  mediaType: 'text/plain',
-                  filename: '',
-                  metadata: {},
-                },
-              ],
-              metadata: {},
-              extensions: [],
-              referenceTaskIds: [],
-            },
-            timestamp: new Date().toISOString(),
-          },
-          artifacts: [],
-          history: requestContext.task?.history ? [...requestContext.task.history] : [],
-          metadata: {},
-        };
-        if (
-          finalMessageForAgent &&
-          !errorTask.history?.find((m) => m.messageId === finalMessageForAgent.messageId)
-        ) {
-          errorTask.history?.push(finalMessageForAgent);
+        if (onExecutorError) {
+          onExecutorError(err);
         }
-        eventBus.publish(AgentEvent.task(errorTask));
-        eventBus.publish(
-          AgentEvent.statusUpdate({
-            taskId: errorTask.id,
-            contextId: errorTask.contextId,
-            status: errorTask.status,
-            metadata: {},
-          })
-        );
       })
       .finally(() => {
         // Close the bus for terminal tasks; keep it alive for
@@ -577,12 +529,6 @@ export class DefaultRequestHandler implements A2ARequestHandler {
     // Attach the queue before kicking off the executor so no events are missed.
     const eventQueue = new ExecutionEventQueue(eventBus);
 
-    // Run the executor in the background. Bus cleanup is tied to the
-    // executor's lifecycle, not the consumer's, so a `tasks/resubscribe`
-    // arriving after the consumer settles can still find the bus while
-    // the executor is still publishing.
-    this._runExecutor(taskId, eventBus, requestContext, finalMessageForAgent);
-
     const historyLengthConfig = params.configuration;
 
     if (isBlocking) {
@@ -592,6 +538,13 @@ export class DefaultRequestHandler implements A2ARequestHandler {
       // observed, and the drain detaches into the background so the
       // executor can keep publishing on the same bus.
       return new Promise<Message | Task>((resolve, reject) => {
+        // Run the executor in the background. Bus cleanup is tied to
+        // the executor's lifecycle, not the consumer's, so a
+        // `tasks/resubscribe` arriving after the consumer settles can
+        // still find the bus while the executor is still publishing.
+        // Executor rejection short-circuits the outer promise. `reject()`
+        // is a no-op if the promise was already settled by the drain path.
+        this._runExecutor(taskId, eventBus, requestContext, finalMessageForAgent, reject);
         const pending = this._processEvents(taskId, resultManager, eventQueue, context, {
           authRequiredSnapshotResolver: (snapshot) => {
             this._applyHistoryLengthSemantics(snapshot, historyLengthConfig ?? {});
@@ -627,6 +580,7 @@ export class DefaultRequestHandler implements A2ARequestHandler {
     } else {
       // Non-blocking mode resolves with the first task/message event.
       return new Promise<Message | Task>((resolve, reject) => {
+        this._runExecutor(taskId, eventBus, requestContext, finalMessageForAgent, reject);
         this._processEvents(taskId, resultManager, eventQueue, context, {
           firstResultResolver: (result) => {
             if (isTask(result)) {

test/server/default_request_handler.spec.ts

@@ -312,7 +312,7 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     assert.deepEqual(taskResult.artifacts![0], testArtifact);
   });
 
-  it('sendMessage: should handle agent execution failure for blocking calls', async () => {
+  it('sendMessage: should surface executor rejection as a thrown error on blocking calls', async () => {
     const errorMessage = 'Agent failed!';
     (mockAgentExecutor as MockAgentExecutor).execute.mockRejectedValue(new Error(errorMessage));
 
@@ -328,18 +328,8 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
       metadata: {},
     };
 
-    const blockingResult = await handler.sendMessage(blockingParams, serverCallContext);
-    const blockingTask = blockingResult as Task;
-
-    assert.equal(
-      blockingTask.status.state,
-      TaskState.TASK_STATE_FAILED,
-      'Task status should be failed'
-    );
-    assert.include(
-      (blockingTask.status.message?.parts[0].content as { $case: 'text'; value: string }).value,
-      errorMessage,
-      'Error message should be in the status'
+    await expect(handler.sendMessage(blockingParams, serverCallContext)).rejects.toThrow(
+      errorMessage
     );
   });
 
@@ -428,6 +418,24 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     assert.equal(saveSpy.mock.calls[1][0].status.state, TaskState.TASK_STATE_COMPLETED);
   });
 
+  it('sendMessage: (non-blocking) executor rejection before first event surfaces as thrown error', async () => {
+    const errorMessage = 'boom before first event';
+    (mockAgentExecutor as MockAgentExecutor).execute.mockRejectedValue(new Error(errorMessage));
+
+    const params: SendMessageRequest = {
+      message: createTestMessage('msg-370-nonblocking', 'Do work'),
+      tenant: '',
+      configuration: {
+        acceptedOutputModes: [],
+        taskPushNotificationConfig: undefined,
+        returnImmediately: true,
+      },
+      metadata: {},
+    };
+
+    await expect(handler.sendMessage(params, serverCallContext)).rejects.toThrow(errorMessage);
+  });
+
   it('sendMessage: (non-blocking) should handle failure in event loop after successfull task event', async () => {
     vi.useFakeTimers();
 
@@ -533,7 +541,7 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     );
   });
 
-  it('sendMessage: should handle agent execution failure for non-blocking calls', async () => {
+  it('sendMessage: should surface executor rejection as a thrown error on non-blocking calls', async () => {
     const errorMessage = 'Agent failed!';
     (mockAgentExecutor as MockAgentExecutor).execute.mockRejectedValue(new Error(errorMessage));
 
@@ -549,18 +557,8 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
       metadata: {},
     };
 
-    const nonBlockingResult = await handler.sendMessage(nonBlockingParams, serverCallContext);
-    const nonBlockingTask = nonBlockingResult as Task;
-
-    assert.equal(
-      nonBlockingTask.status.state,
-      TaskState.TASK_STATE_FAILED,
-      'Task status should be failed'
-    );
-    assert.include(
-      (nonBlockingTask.status.message?.parts[0].content as { $case: 'text'; value: string }).value,
-      errorMessage,
-      'Error message should be in the status'
+    await expect(handler.sendMessage(nonBlockingParams, serverCallContext)).rejects.toThrow(
+      errorMessage
     );
   });