feat: Send current task as the first event after subscribing to it (#418)

# Description

Sending a `Task` object as the first event after subscribing to it.

Additionally, throwing an `UnsupportedOperationError` error if a `Task`
in terminal state was used for subscribing. This behavior is aligned
with spec requirements:
https://a2a-protocol.org/latest/specification/#1046-subscribetotask

Fixes #323 🦕

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: bartek-gralewicz

src/server/request_handler/default_request_handler.ts

@@ -612,32 +612,38 @@ export class DefaultRequestHandler implements A2ARequestHandler {
     }
 
     const taskId = params.id;
-    const task = await this.taskStore.load(taskId, context);
-    if (!task) {
-      throw new TaskNotFoundError(`Task not found: ${taskId}`);
-    }
 
-    // Yield the current task state first
-    yield { payload: { $case: 'task', value: task } };
+    // Attach to the event bus BEFORE loading the task from the store.
+    // This eliminates the race condition where events published between the store
+    // load and the subscription would be missed. The ExecutionEventQueue constructor
+    // synchronously registers listeners, so all events from this point forward are
+    // buffered in the queue's internal array.
+    const eventBus = this.eventBusManager.getByTaskId(taskId);
+    const eventQueue = eventBus ? new ExecutionEventQueue(eventBus) : undefined;
 
-    // If task is already in a final state, no more events will come.
-    if (TERMINAL_STATE_LIST.includes(task.status!.state)) {
-      return;
-    }
+    try {
+      const task = await this.taskStore.load(taskId, context);
+      if (!task) {
+        throw new TaskNotFoundError(`Task not found: ${taskId}`);
+      }
+      if (task.status?.state !== undefined && TERMINAL_STATE_LIST.includes(task.status.state)) {
+        throw new UnsupportedOperationError(
+          `Task ${taskId} is in a terminal state (${task.status.state}) and cannot be subscribed to.`
+        );
+      }
 
-    const eventBus = this.eventBusManager.getByTaskId(taskId);
-    if (!eventBus) {
-      // No active execution for this task, so no live events.
-      console.warn(`Resubscribe: No active event bus for task ${taskId}.`);
-      return;
-    }
+      if (!eventQueue) {
+        throw new UnsupportedOperationError(`Resubscribe: No active event bus for task ${taskId}.`);
+      }
 
-    // Attach a new queue to the existing bus for this resubscription
-    const eventQueue = new ExecutionEventQueue(eventBus);
-    // Note: The ResultManager part is already handled by the original execution flow.
-    // Resubscribe just listens for new events.
+      // Per spec 3.1.6: "The operation MUST return a Task object as the first event
+      // in the stream, representing the current state of the task at the time of
+      // subscription."
+      yield { payload: { $case: 'task', value: task } };
 
-    try {
+      // Stream live events, filtering by taskId.
+      // The ResultManager is already handled by the original execution flow;
+      // resubscribe only listens for new events.
       for await (const event of eventQueue.events()) {
         switch (event.kind) {
           case 'statusUpdate':
@@ -663,7 +669,7 @@ export class DefaultRequestHandler implements A2ARequestHandler {
         }
       }
     } finally {
-      eventQueue.stop();
+      eventQueue?.stop();
     }
   }
 

test/server/default_request_handler.spec.ts

@@ -53,6 +53,7 @@ import {
 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 { TERMINAL_STATE_LIST } from '../../src/server/utils.js';
 
 describe('DefaultRequestHandler as A2ARequestHandler', () => {
   let handler: A2ARequestHandler;
@@ -1154,14 +1155,8 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
 
   it('sendMessage: should reject if task is in a terminal state', async () => {
     const taskId = 'task-terminal-1';
-    const terminalStates: TaskState[] = [
-      TaskState.TASK_STATE_COMPLETED,
-      TaskState.TASK_STATE_FAILED,
-      TaskState.TASK_STATE_CANCELED,
-      TaskState.TASK_STATE_REJECTED,
-    ];
 
-    for (const state of terminalStates) {
+    for (const state of TERMINAL_STATE_LIST) {
       const fakeTask: Task = {
         id: taskId,
         contextId: 'ctx-terminal',
@@ -1389,6 +1384,102 @@ describe('DefaultRequestHandler as A2ARequestHandler', () => {
     assert.equal(lastSaveCall.status.state, TaskState.TASK_STATE_COMPLETED);
   });
 
+  it('resubscribe: should throw UnsupportedOperationError for terminal-state tasks', async () => {
+    const taskId = 'task-terminal-resub';
+
+    for (const state of TERMINAL_STATE_LIST) {
+      const fakeTask: Task = {
+        id: taskId,
+        contextId: 'ctx-terminal-resub',
+        status: { state: state as TaskState, message: undefined, timestamp: undefined },
+        artifacts: [],
+        history: [],
+        metadata: {},
+      };
+      await mockTaskStore.save(fakeTask, serverCallContext);
+
+      const generator = handler.resubscribe({ id: taskId, tenant: '' }, serverCallContext);
+      try {
+        await generator.next();
+        assert.fail(`Should have thrown for terminal state: ${state}`);
+      } catch (error: unknown) {
+        expect(error).to.be.instanceOf(UnsupportedOperationError);
+        expect((error as Error).message).to.contain(`Task ${taskId} is in a terminal state`);
+      }
+    }
+  });
+
+  it('resubscribe: should throw TaskNotFoundError for non-existent task', async () => {
+    const generator = handler.resubscribe(
+      { id: 'non-existent-task', tenant: '' },
+      serverCallContext
+    );
+    try {
+      await generator.next();
+      assert.fail('Should have thrown TaskNotFoundError');
+    } catch (error: unknown) {
+      expect(error).to.be.instanceOf(TaskNotFoundError);
+    }
+  });
+
+  it('resubscribe: should yield Task as the first event with current state', async () => {
+    const taskId = 'task-resub-first-event';
+    const fakeTask: Task = {
+      id: taskId,
+      contextId: 'ctx-resub-first',
+      status: { state: TaskState.TASK_STATE_WORKING, message: undefined, timestamp: undefined },
+      artifacts: [],
+      history: [],
+      metadata: {},
+    };
+    await mockTaskStore.save(fakeTask, serverCallContext);
+
+    // Create an active event bus
+    const bus = executionEventBusManager.createOrGetByTaskId(taskId);
+
+    const generator = handler.resubscribe({ id: taskId, tenant: '' }, serverCallContext);
+
+    // Advance once to yield the task and create the event queue
+    const firstResult = await generator.next();
+    assert.isFalse(firstResult.done);
+
+    // Now finish the bus to unblock the stream
+    bus.finished();
+
+    const results: StreamResponse[] = [firstResult.value];
+    for await (const event of generator) {
+      results.push(event);
+    }
+
+    assert.lengthOf(results, 1, 'Should yield exactly one event (the initial task snapshot)');
+    assert.equal(results[0].payload?.$case, 'task');
+    assert.deepEqual((results[0].payload as { $case: 'task'; value: Task }).value, fakeTask);
+  });
+
+  it('resubscribe: should throw UnsupportedOperationError when no active event bus exists', async () => {
+    const taskId = 'task-resub-no-bus';
+    const fakeTask: Task = {
+      id: taskId,
+      contextId: 'ctx-resub-no-bus',
+      status: { state: TaskState.TASK_STATE_WORKING, message: undefined, timestamp: undefined },
+      artifacts: [],
+      history: [],
+      metadata: {},
+    };
+    await mockTaskStore.save(fakeTask, serverCallContext);
+
+    const generator = handler.resubscribe({ id: taskId, tenant: '' }, serverCallContext);
+    try {
+      await generator.next();
+      assert.fail('Should have thrown UnsupportedOperationError');
+    } catch (error: unknown) {
+      expect(error).to.be.instanceOf(UnsupportedOperationError);
+      expect((error as Error).message).to.contain(
+        `Resubscribe: No active event bus for task ${taskId}`
+      );
+    }
+  });
+
   it('getTask: should return an existing task from the store', async () => {
     const fakeTask: Task = {
       id: 'task-exist',