fix: handle errors occurred in non-blocking sendMessage (#187)

# Description

This PR fixes the error of server crash in case of error in the event
loop for non blocking event, if the first reply to the client has been
already sent.
Two different error handling strategies are implemented based on the
origin of the error:

1. If the error originates from the pushNotification service, the loop
is not interrupted, and the error is simply printed on the console error
output
2. If the error is originating from the ExecutionEventQueue or from a
taskStore operation (resultManager interaction), the error is
interrupting the loop.
In the catch clause a TaskStatusUpdateEvent is generated, with
taskStatus equal to "failed", if an already existing Task has been
processed by the ResultManager. The new TaskStatusUpdateEvent is saved
in the taskStore. If a new error occurs within this operation, it is
simply printed on the console error output.

Fixes #186  🦕

Author: guglielmo-san

src/server/request_handler/default_request_handler.ts

@@ -167,7 +167,11 @@ export class DefaultRequestHandler implements A2ARequestHandler {
       for await (const event of eventQueue.events()) {
         await resultManager.processEvent(event);
 
-        await this._sendPushNotificationIfNeeded(event);
+        try {
+          await this._sendPushNotificationIfNeeded(event);
+        } catch (error) {
+          console.error(`Error sending push notification: ${error}`);
+        }
 
         if (options?.firstResultResolver && !firstResultSent) {
           let firstResult: Message | Task | undefined;
@@ -187,13 +191,15 @@ export class DefaultRequestHandler implements A2ARequestHandler {
           A2AError.internalError('Execution finished before a message or task was produced.')
         );
       }
-    } catch (error) {
+    } catch (error: any) {
       console.error(`Event processing loop failed for task ${taskId}:`, error);
-      if (options?.firstResultRejector && !firstResultSent) {
-        options.firstResultRejector(error);
-      }
-      // re-throw error for blocking case to catch
-      throw error;
+      this._handleProcessingError(
+        error,
+        resultManager,
+        firstResultSent,
+        taskId,
+        options?.firstResultRejector
+      );
     } finally {
       this.eventBusManager.cleanupByTaskId(taskId);
     }
@@ -621,4 +627,54 @@ export class DefaultRequestHandler implements A2ARequestHandler {
     // Send push notification in the background.
     this.pushNotificationSender?.send(task);
   }
+
+  private async _handleProcessingError(
+    error: any,
+    resultManager: ResultManager,
+    firstResultSent: boolean,
+    taskId: string,
+    firstResultRejector?: (reason: any) => void
+  ): Promise<void> {
+    // Non-blocking case with with first result not sent
+    if (firstResultRejector && !firstResultSent) {
+      firstResultRejector(error);
+      return;
+    }
+
+    // re-throw error for blocking case to catch
+    if (!firstResultRejector) {
+      throw error;
+    }
+
+    // Non-blocking case with first result already sent
+    const currentTask = resultManager.getCurrentTask();
+    if (currentTask) {
+      const statusUpdateFailed: TaskStatusUpdateEvent = {
+        taskId: currentTask.id,
+        contextId: currentTask.contextId,
+        status: {
+          state: 'failed',
+          message: {
+            kind: 'message',
+            role: 'agent',
+            messageId: uuidv4(),
+            parts: [{ kind: 'text', text: `Event processing loop failed: ${error.message}` }],
+            taskId: currentTask.id,
+            contextId: currentTask.contextId,
+          },
+          timestamp: new Date().toISOString(),
+        },
+        kind: 'status-update',
+        final: true,
+      };
+
+      try {
+        await resultManager.processEvent(statusUpdateFailed);
+      } catch (error) {
+        console.error(`Event processing loop failed for task ${taskId}: ${error.message}`);
+      }
+    } else {
+      console.error(`Event processing loop failed for task ${taskId}: ${error.message}`);
+    }
+  }
 }

test/server/default_request_handler.spec.ts

@@ -40,6 +40,8 @@ import {
 } from './mocks/agent-executor.mock.js';
 import { MockPushNotificationSender } from './mocks/push_notification_sender.mock.js';
 import { ServerCallContext } from '../../src/server/context.js';
+import { MockTaskStore } from './mocks/task_store.mock.js';
+import { TextPart } from 'genkit/model';
 
 describe('DefaultRequestHandler as A2ARequestHandler', () => {
   let handler: A2ARequestHandler;
@@ -277,6 +279,87 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     assert.equal(saveSpy.secondCall.args[0].status.state, 'completed');
   });
 
+  it('sendMessage: (non-blocking) should handle failure in event loop after successfull task event', async () => {
+    clock = sinon.useFakeTimers();
+
+    const mockTaskStore = new MockTaskStore();
+    const handler = new DefaultRequestHandler(
+      testAgentCard,
+      mockTaskStore,
+      mockAgentExecutor,
+      executionEventBusManager
+    );
+
+    const params: MessageSendParams = {
+      message: createTestMessage('msg-nonblock', 'Do a long task'),
+      configuration: {
+        blocking: false,
+        acceptedOutputModes: [],
+      },
+    };
+
+    const taskId = 'task-nonblock-123';
+    const contextId = 'ctx-nonblock-abc';
+    (mockAgentExecutor as MockAgentExecutor).execute.callsFake(async (ctx, bus) => {
+      // First event is the task creation, which should be returned immediately
+      bus.publish({
+        id: taskId,
+        contextId,
+        status: { state: 'submitted' },
+        kind: 'task',
+      });
+
+      // Simulate work before publishing more events
+      await clock.tickAsync(500);
+
+      bus.publish({
+        taskId,
+        contextId,
+        kind: 'status-update',
+        status: { state: 'completed' },
+        final: true,
+      });
+      bus.finished();
+    });
+
+    let finalTaskSaved: Task | undefined;
+    const errorMessage = 'Error thrown on saving completed task notification';
+    (mockTaskStore as MockTaskStore).save.callsFake(async (task) => {
+      if (task.status.state == 'completed') {
+        throw new Error(errorMessage);
+      }
+
+      if (task.status.state == 'failed') {
+        finalTaskSaved = task;
+      }
+    });
+
+    // This call should return as soon as the first 'task' event is published
+    const immediateResult = await handler.sendMessage(params);
+
+    // Assert that we got the initial task object back right away
+    const taskResult = immediateResult as Task;
+    assert.equal(taskResult.kind, 'task');
+    assert.equal(taskResult.id, taskId);
+    assert.equal(
+      taskResult.status.state,
+      'submitted',
+      "Should return immediately with 'submitted' state"
+    );
+
+    // Allow the background processing to complete
+    await clock.runAllAsync();
+
+    assert.equal(finalTaskSaved!.status.state, 'failed');
+    assert.equal(finalTaskSaved!.id, taskId);
+    assert.equal(finalTaskSaved!.contextId, contextId);
+    assert.equal(finalTaskSaved!.status.message!.role, 'agent');
+    assert.equal(
+      (finalTaskSaved!.status.message!.parts[0] as TextPart).text,
+      `Event processing loop failed: ${errorMessage}`
+    );
+  });
+
   it('sendMessage: should handle agent execution failure for non-blocking calls', async () => {
     const errorMessage = 'Agent failed!';
     (mockAgentExecutor as MockAgentExecutor).execute.rejects(new Error(errorMessage));

test/server/mocks/task_store.mock.ts

@@ -0,0 +1,8 @@
+import sinon, { SinonStub } from 'sinon';
+import { Task } from '../../../src/index.js';
+import { TaskStore } from '../../../src/server/store.js';
+
+export class MockTaskStore implements TaskStore {
+  public save: SinonStub<[Task], Promise<void>> = sinon.stub();
+  public load: SinonStub<[string], Promise<Task | undefined>> = sinon.stub();
+}