fix(server): cast finished listener call, remove stale TODO, document execute contract (#553)

# Description

Three small SDK quality fixes surfaced while reviewing the codebase:

## SDK changes
- **`execution_event_bus`**: Cast `listener.call(this)` for `'finished'`
listeners (which are invoked with no args despite the typed `Listener`
signature). Clears 2 TS strict-mode issues; `.betterer.results`
regression baseline shrinks accordingly.
- **`jsonrpc_transport_handler`**: Removed stale TODO about mid-stream
SSE errors; the Express layer already catches and writes
`formatSSEErrorEvent` (covered by `express_app.spec.ts`).
- **`AgentExecutor.execute` JSDoc**: Documented the contract that every
streaming turn must begin with a `task` or `message` event (including
follow-ups). The server already enforces this; the JSDoc just makes it
explicit to implementors.

Author: JakubWorek

.betterer.results

@@ -4,10 +4,5 @@
 // https://phenomnomnominal.github.io/betterer/docs/results-file/#merge
 //
 exports[`TypeScript Strict Mode`] = {
-  value: `{
-    "src/server/events/execution_event_bus.ts:1102247656": [
-      [233, 15, 4, "tsc: Expected 2 arguments, but got 1.", "2087764327"],
-      [254, 15, 4, "tsc: Expected 2 arguments, but got 1.", "2087764327"]
-    ]
-  }`
+  value: `{}`
 };

src/compat/v0_3/server/transports/jsonrpc/jsonrpc_transport_handler.ts

@@ -111,21 +111,17 @@ export class LegacyJsonRpcTransportHandler {
                 context
               );
 
+        // Errors thrown by `agentEventStream` propagate out of the
+        // generator; the Express layer catches them, logs the failure,
+        // and writes a final SSE `event: error` frame carrying the
+        // JSON-RPC error envelope before closing the stream.
         return (async function* legacyJsonRpcEventStream(): AsyncGenerator<
           LegacyJSONRPCResponse,
           void,
           undefined
         > {
-          try {
-            for await (const event of agentEventStream) {
-              yield toCompatStreamResponse(event, requestId);
-            }
-          } catch (streamError) {
-            console.error(
-              `Error in agent event stream for ${method} (request ${requestId}):`,
-              streamError
-            );
-            throw streamError;
+          for await (const event of agentEventStream) {
+            yield toCompatStreamResponse(event, requestId);
           }
         })();
       }

src/samples/extensions/extensions.ts

@@ -3,6 +3,8 @@ import {
   RequestContext,
   ExecutionEventBus,
   AgentExecutionEvent,
+  EventListener,
+  FinishedListener,
 } from '../../server/index.js';
 
 export const EXTENSION_URI = 'https://github.com/a2aproject/a2a-js/src/samples/extensions/v1';
@@ -72,18 +74,36 @@ class TimestampingEventQueue implements ExecutionEventBus {
     this._delegate.finished();
   }
 
-  on(eventName: 'event' | 'finished', listener: (event: AgentExecutionEvent) => void): this {
-    this._delegate.on(eventName, listener);
+  on(eventName: 'event', listener: EventListener): this;
+  on(eventName: 'finished', listener: FinishedListener): this;
+  on(eventName: 'event' | 'finished', listener: EventListener | FinishedListener): this {
+    if (eventName === 'event') {
+      this._delegate.on('event', listener as EventListener);
+    } else {
+      this._delegate.on('finished', listener as FinishedListener);
+    }
     return this;
   }
 
-  off(eventName: 'event' | 'finished', listener: (event: AgentExecutionEvent) => void): this {
-    this._delegate.off(eventName, listener);
+  off(eventName: 'event', listener: EventListener): this;
+  off(eventName: 'finished', listener: FinishedListener): this;
+  off(eventName: 'event' | 'finished', listener: EventListener | FinishedListener): this {
+    if (eventName === 'event') {
+      this._delegate.off('event', listener as EventListener);
+    } else {
+      this._delegate.off('finished', listener as FinishedListener);
+    }
     return this;
   }
 
-  once(eventName: 'event' | 'finished', listener: (event: AgentExecutionEvent) => void): this {
-    this._delegate.once(eventName, listener);
+  once(eventName: 'event', listener: EventListener): this;
+  once(eventName: 'finished', listener: FinishedListener): this;
+  once(eventName: 'event' | 'finished', listener: EventListener | FinishedListener): this {
+    if (eventName === 'event') {
+      this._delegate.once('event', listener as EventListener);
+    } else {
+      this._delegate.once('finished', listener as FinishedListener);
+    }
     return this;
   }
 

src/server/agent_execution/agent_executor.ts

@@ -5,6 +5,11 @@ export interface AgentExecutor {
   /**
    * Executes the agent logic and publishes events to the bus.
    *
+   * Every call MUST publish either a `task` or a `message` event as its
+   * first event — including follow-up turns where `requestContext.task`
+   * is already set. The server enforces this ordering and rejects
+   * streams that begin with a `statusUpdate` or `artifactUpdate`.
+   *
    * Multi-tenant implementations can read the tenant identifier from
    * `requestContext.context.tenant`.
    */

src/server/events/execution_event_bus.ts

@@ -52,11 +52,20 @@ export function assertUnreachableEvent(event: never): never {
 /** Event names supported by {@link ExecutionEventBus}. */
 export type ExecutionEventName = 'event' | 'finished';
 
+/** Listener for `'event'` notifications, invoked with the published event. */
+export type EventListener = (event: AgentExecutionEvent) => void;
+
+/** Listener for `'finished'` notifications, invoked with no arguments. */
+export type FinishedListener = () => void;
+
 export interface ExecutionEventBus {
   publish(event: AgentExecutionEvent): void;
-  on(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this;
-  off(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this;
-  once(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this;
+  on(eventName: 'event', listener: EventListener): this;
+  on(eventName: 'finished', listener: FinishedListener): this;
+  off(eventName: 'event', listener: EventListener): this;
+  off(eventName: 'finished', listener: FinishedListener): this;
+  once(eventName: 'event', listener: EventListener): this;
+  once(eventName: 'finished', listener: FinishedListener): this;
   removeAllListeners(eventName?: ExecutionEventName): this;
   finished(): void;
 }
@@ -74,7 +83,6 @@ const CustomEventImpl: typeof CustomEvent =
         }
       } as typeof CustomEvent);
 
-type Listener = (event: AgentExecutionEvent) => void;
 type WrappedListener = (e: Event) => void;
 
 // Should always pass for 'event' type events since we control the dispatch
@@ -92,11 +100,11 @@ function isAgentExecutionCustomEvent(e: Event): e is CustomEvent<AgentExecutionE
  * `listenerCount`, `rawListeners`, etc. are not available.
  */
 export class DefaultExecutionEventBus extends EventTarget implements ExecutionEventBus {
-  // Separate storage for each event type — both use the interface's
-  // Listener type but are invoked differently (with event payload vs. no
-  // arguments).
-  private readonly eventListeners: Map<Listener, WrappedListener[]> = new Map();
-  private readonly finishedListeners: Map<Listener, WrappedListener[]> = new Map();
+  // Separate storage so each event type can hold listeners of its own
+  // signature: 'event' listeners receive a payload, 'finished' listeners
+  // are invoked with no arguments.
+  private readonly eventListeners: Map<EventListener, WrappedListener[]> = new Map();
+  private readonly finishedListeners: Map<FinishedListener, WrappedListener[]> = new Map();
 
   publish(event: AgentExecutionEvent): void {
     this.dispatchEvent(new CustomEventImpl('event', { detail: event }));
@@ -106,29 +114,35 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     this.dispatchEvent(new Event('finished'));
   }
 
-  on(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this {
+  on(eventName: 'event', listener: EventListener): this;
+  on(eventName: 'finished', listener: FinishedListener): this;
+  on(eventName: ExecutionEventName, listener: EventListener | FinishedListener): this {
     if (eventName === 'event') {
-      this.addEventListenerInternal(listener);
+      this.addEventListenerInternal(listener as EventListener);
     } else {
-      this.addFinishedListenerInternal(listener);
+      this.addFinishedListenerInternal(listener as FinishedListener);
     }
     return this;
   }
 
-  off(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this {
+  off(eventName: 'event', listener: EventListener): this;
+  off(eventName: 'finished', listener: FinishedListener): this;
+  off(eventName: ExecutionEventName, listener: EventListener | FinishedListener): this {
     if (eventName === 'event') {
-      this.removeEventListenerInternal(listener);
+      this.removeEventListenerInternal(listener as EventListener);
     } else {
-      this.removeFinishedListenerInternal(listener);
+      this.removeFinishedListenerInternal(listener as FinishedListener);
     }
     return this;
   }
 
-  once(eventName: ExecutionEventName, listener: (event: AgentExecutionEvent) => void): this {
+  once(eventName: 'event', listener: EventListener): this;
+  once(eventName: 'finished', listener: FinishedListener): this;
+  once(eventName: ExecutionEventName, listener: EventListener | FinishedListener): this {
     if (eventName === 'event') {
-      this.addEventListenerOnceInternal(listener);
+      this.addEventListenerOnceInternal(listener as EventListener);
     } else {
-      this.addFinishedListenerOnceInternal(listener);
+      this.addFinishedListenerOnceInternal(listener as FinishedListener);
     }
     return this;
   }
@@ -157,9 +171,9 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
 
   // Listener tracking helpers.
 
-  private trackListener(
-    listenerMap: Map<Listener, WrappedListener[]>,
-    listener: Listener,
+  private trackListener<L>(
+    listenerMap: Map<L, WrappedListener[]>,
+    listener: L,
     wrapped: WrappedListener
   ): void {
     const existing = listenerMap.get(listener);
@@ -170,9 +184,9 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     }
   }
 
-  private untrackWrappedListener(
-    listenerMap: Map<Listener, WrappedListener[]>,
-    listener: Listener,
+  private untrackWrappedListener<L>(
+    listenerMap: Map<L, WrappedListener[]>,
+    listener: L,
     wrapped: WrappedListener
   ): void {
     const wrappedList = listenerMap.get(listener);
@@ -189,7 +203,7 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
 
   // 'event' listeners.
 
-  private addEventListenerInternal(listener: Listener): void {
+  private addEventListenerInternal(listener: EventListener): void {
     const wrapped: WrappedListener = (e: Event) => {
       if (!isAgentExecutionCustomEvent(e)) {
         throw new Error('Internal error: expected CustomEvent for "event" type');
@@ -201,7 +215,7 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     this.addEventListener('event', wrapped);
   }
 
-  private removeEventListenerInternal(listener: Listener): void {
+  private removeEventListenerInternal(listener: EventListener): void {
     const wrappedList = this.eventListeners.get(listener);
     if (wrappedList && wrappedList.length > 0) {
       const wrapped = wrappedList.pop()!;
@@ -212,7 +226,7 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     }
   }
 
-  private addEventListenerOnceInternal(listener: Listener): void {
+  private addEventListenerOnceInternal(listener: EventListener): void {
     const wrapped: WrappedListener = (e: Event) => {
       if (!isAgentExecutionCustomEvent(e)) {
         throw new Error('Internal error: expected CustomEvent for "event" type');
@@ -225,11 +239,10 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     this.addEventListener('event', wrapped, { once: true });
   }
 
-  // 'finished' listeners. The interface declares listeners as taking an
-  // `AgentExecutionEvent`, but for 'finished' they're invoked with no
-  // arguments (matching EventEmitter behaviour).
+  // 'finished' listeners. Invoked with no arguments; the interface
+  // declares them as `FinishedListener` so callers get a precise type.
 
-  private addFinishedListenerInternal(listener: Listener): void {
+  private addFinishedListenerInternal(listener: FinishedListener): void {
     const wrapped: WrappedListener = () => {
       listener.call(this);
     };
@@ -238,7 +251,7 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     this.addEventListener('finished', wrapped);
   }
 
-  private removeFinishedListenerInternal(listener: Listener): void {
+  private removeFinishedListenerInternal(listener: FinishedListener): void {
     const wrappedList = this.finishedListeners.get(listener);
     if (wrappedList && wrappedList.length > 0) {
       const wrapped = wrappedList.pop()!;
@@ -249,7 +262,7 @@ export class DefaultExecutionEventBus extends EventTarget implements ExecutionEv
     }
   }
 
-  private addFinishedListenerOnceInternal(listener: Listener): void {
+  private addFinishedListenerOnceInternal(listener: FinishedListener): void {
     const wrapped: WrappedListener = () => {
       this.untrackWrappedListener(this.finishedListeners, listener, wrapped);
       listener.call(this);

src/server/index.ts

@@ -8,8 +8,10 @@ export { RequestContext } from './agent_execution/request_context.js';
 export type {
   AgentExecutionEvent,
   AgentExecutionEventKind,
+  EventListener,
   ExecutionEventBus,
   ExecutionEventName,
+  FinishedListener,
 } from './events/execution_event_bus.js';
 export {
   AgentEvent,

src/server/transports/jsonrpc/jsonrpc_transport_handler.ts

@@ -117,32 +117,21 @@ export class JsonRpcTransportHandler {
             : this.requestHandler.resubscribe(SubscribeToTaskRequest.fromJSON(params), context);
 
         // Wrap the agent event stream into a JSON-RPC result stream.
+        // Errors thrown by `agentEventStream` propagate out of the
+        // generator; the Express layer catches them, logs the failure,
+        // and writes a final SSE `event: error` frame carrying the
+        // JSON-RPC error envelope before closing the stream.
         return (async function* jsonRpcEventStream(): AsyncGenerator<
           JSONRPCResponse,
           void,
           undefined
         > {
-          try {
-            for await (const event of agentEventStream) {
-              yield {
-                jsonrpc: '2.0',
-                id: requestId,
-                result: StreamResponse.toJSON(event),
-              };
-            }
-          } catch (streamError) {
-            // If the underlying agent stream throws an error, we need to yield a JSONRPCErrorResponse.
-            // However, an AsyncGenerator is expected to yield JSONRPCResult.
-            // This indicates an issue with how errors from the agent's stream are propagated.
-            // For now, log it. The Express layer will handle the generator ending.
-            console.error(
-              `Error in agent event stream for ${method} (request ${requestId}):`,
-              streamError
-            );
-            // Ideally, the Express layer should catch this and send a final error to the client if the stream breaks.
-            // Or, the agentEventStream itself should yield a final error event that gets wrapped.
-            // For now, we re-throw so it can be caught by the Express layer streaming support.
-            throw streamError;
+          for await (const event of agentEventStream) {
+            yield {
+              jsonrpc: '2.0',
+              id: requestId,
+              result: StreamResponse.toJSON(event),
+            };
           }
         })();
       } else {