fix(server): await _handleProcessingError so blocking drains surface errors (#579)

# Description

`_processEvents` catches errors from the event-processing loop and
delegates to
`_handleProcessingError`, which — per its own comment — re-throws on the
blocking
path (no `firstResultRejector`) *"so the caller's await catches it"*:

```ts
} catch (error) {
  console.error(`Event processing loop failed for task ${taskId}:`, error);
  this._handleProcessingError(          // <-- not awaited
    error, resultManager, firstResultSent, taskId, options?.firstResultRejector,
  );
} finally {
  eventQueue.stop();
}
```

```ts
// Blocking: re-throw so the caller's await catches it.
if (!firstResultRejector) {
  throw error;
}
```

The call is **not awaited**. `_handleProcessingError` is `async`, so its
`throw`
becomes a floating rejection instead of propagating out of
`_processEvents`. The
generator resolves as if the drain had succeeded.

The one blocking caller that passes no `firstResultRejector` is
`cancelTask`'s
drain:

```ts
await this._processEvents(
  taskId, new ResultManager(this.taskStore, context), eventQueue, context,
);
```

So when the drain throws (e.g. the task store fails while persisting the
cancellation), `cancelTask` silently swallows the real error, reloads
the task,
sees a non-canceled state, and throws a misleading
`TaskNotCancelableError` —
while the true error surfaces as an unhandled promise rejection.

## Fix

Await `_handleProcessingError`. The blocking re-throw now propagates
through the
`finally` (which still detaches the queue) and out to the awaiting
caller,
matching the documented intent.

The non-blocking paths are unaffected: `sendMessage`/`sendMessageStream`
pass a
`firstResultRejector`, so `_handleProcessingError` rejects or persists a
FAILED
status rather than throwing.

## Tests

Added a regression test asserting `cancelTask` surfaces a drain-time
persistence
error instead of masking it with `TaskNotCancelableError`. Verified it
fails
without the fix (throws `TaskNotCancelableError`) and passes with it.

Full suite green (1368 tests).

- [x] Follows the `CONTRIBUTING` guide
- [x] PR title uses Conventional Commits (`fix:`)
- [x] Tests and linter pass
- [ ] Docs updated (not necessary)

Author: RenKoya1

src/server/request_handler/default_request_handler.ts

@@ -322,7 +322,11 @@ export class DefaultRequestHandler implements A2ARequestHandler {
       }
     } catch (error) {
       console.error(`Event processing loop failed for task ${taskId}:`, error);
-      this._handleProcessingError(
+      // Must be awaited: `_handleProcessingError` re-throws in the blocking
+      // (no `firstResultRejector`) case so the caller's `await` catches it.
+      // Without `await`, that throw escapes as a floating rejection and the
+      // caller (e.g. `cancelTask`'s drain) resolves as if nothing failed.
+      await this._handleProcessingError(
         error,
         resultManager,
         firstResultSent,

test/server/default_request_handler.spec.ts

@@ -3277,6 +3277,65 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     }
   });
 
+  it('cancelTask: should surface an error thrown while draining the cancellation', async () => {
+    // Regression: `_processEvents` re-throws (via `_handleProcessingError`)
+    // on the blocking drain path used by `cancelTask`. That handler must be
+    // awaited, otherwise the throw escapes as a floating rejection, the drain
+    // resolves as if it succeeded, and `cancelTask` masks the real failure
+    // with a misleading `TaskNotCancelableError`.
+    vi.useFakeTimers();
+    const cancellableExecutor = new CancellableMockAgentExecutor();
+    handler = new DefaultRequestHandler(
+      testAgentCard,
+      mockTaskStore,
+      cancellableExecutor,
+      executionEventBusManager
+    );
+
+    const streamParams: SendMessageRequest = {
+      message: createTestMessage('msg-cancel-drain', 'Start and cancel'),
+    } as SendMessageRequest;
+    const streamGenerator = handler.sendMessageStream(streamParams, serverCallContext);
+
+    const streamEvents: any[] = [];
+    (async () => {
+      for await (const event of streamGenerator) {
+        streamEvents.push(event);
+      }
+    })().catch(() => {
+      // The injected persistence failure also surfaces on the stream; ignore.
+    });
+
+    // Allow the task to be created and reach TASK_STATE_WORKING.
+    await vi.advanceTimersByTimeAsync(25);
+
+    const createdTaskEvent = streamEvents.find((e) => e.payload?.$case === 'task');
+    assert.isDefined(createdTaskEvent, 'Task creation event should have been received');
+    const taskId = createdTaskEvent.payload.value.id;
+
+    // Inject a failure when the cancellation (CANCELED) state is persisted
+    // during the drain. Earlier SUBMITTED/WORKING saves already succeeded.
+    const drainError = new Error('drain persistence failed');
+    const realSave = mockTaskStore.save.bind(mockTaskStore);
+    vi.spyOn(mockTaskStore, 'save').mockImplementation(async (task, ctx) => {
+      if (task.status?.state === TaskState.TASK_STATE_CANCELED) {
+        throw drainError;
+      }
+      return realSave(task, ctx);
+    });
+
+    // The real drain failure must surface, not a masking TaskNotCancelableError.
+    // Build the rejection assertion before running timers so the rejection is
+    // observed as soon as it happens (no floating unhandled rejection).
+    const cancelPromise = handler.cancelTask(
+      { id: taskId, tenant: '', metadata: {} },
+      serverCallContext
+    );
+    const rejectsWithDrainError = expect(cancelPromise).rejects.toBe(drainError);
+    await vi.runAllTimersAsync();
+    await rejectsWithDrainError;
+  });
+
   it('cancelTask: should fail for tasks in a terminal state', async () => {
     const taskId = 'task-terminal';
     const fakeTask: Task = {