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

`_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".

The call was not awaited. `_handleProcessingError` is async, so its throw
became a floating rejection instead of propagating out of `_processEvents`.
The generator resolved as if the drain had succeeded, and the one blocking
caller with no `firstResultRejector` — `cancelTask`'s drain — silently
swallowed the failure. It then reloaded the task, saw a non-canceled state,
and threw a misleading `TaskNotCancelableError` (masking the real error),
while the true error surfaced as an unhandled promise rejection.

Await the call. 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: they pass a
`firstResultRejector`, so `_handleProcessingError` rejects/persists rather
than throwing.

Adds a regression test asserting `cancelTask` surfaces a drain-time
persistence error instead of masking it with `TaskNotCancelableError`.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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 = {