Feat: port TaskResultAggregator from adk-python to the A2A agent executor

core/src/a2a/a2a_event.ts

@@ -32,6 +32,7 @@ export enum TaskState {
   CANCELED = 'canceled',
   REJECTED = 'rejected',
   INPUT_REQUIRED = 'input-required',
+  AUTH_REQUIRED = 'auth-required',
 }
 
 /**

core/src/a2a/agent_executor.ts

@@ -33,6 +33,7 @@ import {
   getA2ASessionMetadata,
 } from './metadata_converter_utils.js';
 import {toA2AParts, toGenAIContent} from './part_converter_utils.js';
+import {TaskResultAggregator} from './task_result_aggregator.js';
 
 /**
  * Represents a runner or a configuration for a runner.
@@ -149,6 +150,7 @@ export class A2AAgentExecutor implements AgentExecutor {
       );
 
       const adkEvents: AdkEvent[] = [];
+      const taskResultAggregator = new TaskResultAggregator();
       for await (const adkEvent of adkRunner.runAsync({
         userId,
         sessionId,
@@ -171,13 +173,19 @@ export class A2AAgentExecutor implements AgentExecutor {
           a2aEvent,
         );
 
+        // simplicity: convertAdkEventToA2AEvent emits only artifact updates, so
+        // the aggregator records nothing until that converter also emits the
+        // intermediate auth-required / input-required status updates.
+        taskResultAggregator.processEvent(a2aEvent);
         eventBus.publish(a2aEvent);
       }
 
       await this.publishFinalTaskStatus({
         executorContext,
         eventBus,
-        event: getFinalTaskStatusUpdate(adkEvents, executorContext),
+        event: taskResultAggregator.resolveFinalStatus(
+          getFinalTaskStatusUpdate(adkEvents, executorContext),
+        ),
       });
     } catch (e: unknown) {
       const error = e as Error;

core/src/a2a/task_result_aggregator.ts

@@ -0,0 +1,98 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import {
+  TaskState as A2ATaskState,
+  Message,
+  TaskStatusUpdateEvent,
+} from '@a2a-js/sdk';
+import {A2AEvent, isTaskStatusUpdateEvent, TaskState} from './a2a_event.js';
+
+/**
+ * Folds the task status updates emitted during one agent run into a single
+ * final task state.
+ *
+ * Precedence, highest first: `failed`, `auth-required`, `input-required`,
+ * `working`; a lower-priority update never overwrites a higher-priority one
+ * that has already been recorded.
+ *
+ * One instance per run: the aggregator carries per-request state.
+ */
+export class TaskResultAggregator {
+  private state: A2ATaskState = TaskState.WORKING;
+  private message?: Message;
+
+  /** The aggregated final task state. */
+  get taskState(): A2ATaskState {
+    return this.state;
+  }
+
+  /** The status message recorded alongside {@link taskState}. */
+  get taskStatusMessage(): Message | undefined {
+    return this.message;
+  }
+
+  /**
+   * Records the task-state signal carried by `event`, then rewrites the event
+   * in place so it is forwarded as `working`.
+   *
+   * The aggregated state is tracked here instead, because a terminal state
+   * reaching the A2A request handler mid-stream ends event aggregation before
+   * the run is over. `event.final` and every other field are left untouched.
+   */
+  processEvent(event: A2AEvent): void {
+    if (!isTaskStatusUpdateEvent(event)) {
+      return;
+    }
+
+    const {state, message} = event.status;
+
+    if (state === TaskState.FAILED) {
+      this.state = TaskState.FAILED;
+      this.message = message;
+    } else if (
+      state === TaskState.AUTH_REQUIRED &&
+      this.state !== TaskState.FAILED
+    ) {
+      this.state = TaskState.AUTH_REQUIRED;
+      this.message = message;
+    } else if (
+      state === TaskState.INPUT_REQUIRED &&
+      this.state !== TaskState.FAILED &&
+      this.state !== TaskState.AUTH_REQUIRED
+    ) {
+      this.state = TaskState.INPUT_REQUIRED;
+      this.message = message;
+    } else if (this.state === TaskState.WORKING) {
+      this.message = message;
+    }
+
+    event.status.state = TaskState.WORKING;
+  }
+
+  /**
+   * Returns the final status event to publish for the run.
+   *
+   * `fallback` is the status derived from the ADK event stream. It is returned
+   * unchanged unless a signal of higher priority than `working` was observed,
+   * in which case the aggregated state and status message win — the same
+   * resolution `a2a_agent_executor.py` performs when it closes out a task.
+   */
+  resolveFinalStatus(fallback: TaskStatusUpdateEvent): TaskStatusUpdateEvent {
+    if (this.state === TaskState.WORKING) {
+      return fallback;
+    }
+
+    return {
+      ...fallback,
+      status: {
+        state: this.state,
+        message: this.message,
+        timestamp: fallback.status.timestamp,
+      },
+    };
+  }
+}

core/test/a2a/agent_executor_test.ts

@@ -12,6 +12,7 @@ import {
   BaseSessionService,
   createEvent,
   createEventActions,
+  createSession,
   Runner,
   RunnerConfig,
   Session,
@@ -32,6 +33,22 @@ vi.mock('../../src/runner/runner.js', async (importOriginal) => {
   };
 });
 
+/**
+ * Points the mocked Runner class at `runAsync`.
+ *
+ * The module-level `vi.mock` above replaces Runner with a structurally partial
+ * double — the executor only reads `appName`, `sessionService` and `runAsync`
+ * from it, not `agent`, `pluginManager` or the runner brand symbol — so the
+ * cast to the full class is unavoidable and is confined to this one site.
+ */
+function stubRunner(runAsync: Runner['runAsync']): void {
+  vi.mocked(Runner).mockImplementation(((config: RunnerConfig) => ({
+    appName: config.appName,
+    sessionService: config.sessionService,
+    runAsync,
+  })) as unknown as () => Runner);
+}
+
 describe('A2AAgentExecutor', () => {
   let mockSessionService: Mocked<BaseSessionService>;
   let mockEventBus: Mocked<ExecutionEventBus>;
@@ -109,13 +126,7 @@ describe('A2AAgentExecutor', () => {
       }
     }
 
-    vi.mocked(Runner).mockImplementation(((config: RunnerConfig) => {
-      return {
-        appName: config?.appName,
-        sessionService: config?.sessionService,
-        runAsync: mockRunAsync,
-      } as unknown as Runner;
-    }) as unknown as () => Runner);
+    stubRunner(mockRunAsync);
 
     let beforeExecutedCalled = false;
     let afterEventCount = 0;
@@ -164,6 +175,39 @@ describe('A2AAgentExecutor', () => {
     );
   });
 
+  it('publishes a completed final status when the run only produces artifact updates', async () => {
+    mockSessionService.getSession.mockResolvedValue(
+      createSession({
+        id: 'session-id',
+        userId: 'test-user',
+        appName: 'test-app',
+      }),
+    );
+
+    async function* mockRunAsync() {
+      yield createEvent({
+        author: 'model',
+        content: {role: 'model', parts: [{text: 'the answer'}]},
+        partial: false,
+        actions: createEventActions(),
+      });
+    }
+
+    stubRunner(mockRunAsync);
+
+    const executor = new A2AAgentExecutor({
+      runner: {appName: 'test-app', sessionService: mockSessionService},
+    });
+
+    await executor.execute(createRequestContext(), mockEventBus);
+
+    const calls = mockEventBus.publish.mock.calls;
+    const finalEvent = calls[calls.length - 1][0] as TaskStatusUpdateEvent;
+    expect(finalEvent.kind).toBe('status-update');
+    expect(finalEvent.status.state).toBe('completed');
+    expect(finalEvent.final).toBe(true);
+  });
+
   it('should return early with input required event if task needs input', async () => {
     const mockSession = {
       id: 'session-id',
@@ -232,13 +276,7 @@ describe('A2AAgentExecutor', () => {
       throw new Error('LLM failed');
     }
 
-    vi.mocked(Runner).mockImplementation(((config: RunnerConfig) => {
-      return {
-        appName: config?.appName,
-        sessionService: config?.sessionService,
-        runAsync: mockRunAsyncWithError,
-      } as unknown as Runner;
-    }) as unknown as () => Runner);
+    stubRunner(mockRunAsyncWithError);
 
     const executor = new A2AAgentExecutor({
       runner: {