fix: propagate executor errors directly to the request handler instead of masking them as failed tasks

Author: JakubWorek

src/server/events/execution_event_bus.ts

@@ -1,21 +1,40 @@
 import { Message, Task, TaskStatusUpdateEvent, TaskArtifactUpdateEvent } from '../../index.js';
 
 /**
- * Discriminant values for {@link AgentExecutionEvent}. Mirror
- * `StreamResponse.payload.$case` values for trivial conversion.
+ * Discriminant values for {@link AgentExecutionEvent}. The wire-mapped
+ * kinds (`message`, `task`, `statusUpdate`, `artifactUpdate`) mirror
+ * `StreamResponse.payload.$case` values for trivial conversion. The
+ * `error` kind is internal to the server pipeline: it is published on
+ * the bus by the framework when the {@link AgentExecutor} rejects, and
+ * consumed by the drain loop to persist a FAILED status and propagate
+ * the exception to the transport. It never crosses the wire.
  */
-export type AgentExecutionEventKind = 'message' | 'task' | 'statusUpdate' | 'artifactUpdate';
+export type AgentExecutionEventKind =
+  | 'message'
+  | 'task'
+  | 'statusUpdate'
+  | 'artifactUpdate'
+  | 'error';
 
 /**
  * Discriminated union wrapper for agent execution events. The `kind`
  * property is the TypeScript discriminant, enabling exhaustive
  * `switch`/`case` narrowing without unsafe casts.
+ *
+ * The `error` variant carries the raw error thrown by the
+ * {@link AgentExecutor}. It is an internal, server-side signalling
+ * event: the framework publishes it in the `.catch` handler of
+ * `AgentExecutor.execute()` and the drain loop consumes it to persist
+ * a FAILED task status and propagate the exception to the transport
+ * layer. Executors MUST NOT publish `error` events directly — use
+ * regular termination (return normally or publish a FAILED status).
  */
 export type AgentExecutionEvent =
   | { kind: 'message'; data: Message }
   | { kind: 'task'; data: Task }
   | { kind: 'statusUpdate'; data: TaskStatusUpdateEvent }
-  | { kind: 'artifactUpdate'; data: TaskArtifactUpdateEvent };
+  | { kind: 'artifactUpdate'; data: TaskArtifactUpdateEvent }
+  | { kind: 'error'; data: unknown };
 
 /**
  * Factory functions for type-safe {@link AgentExecutionEvent} wrappers.
@@ -38,6 +57,13 @@ export const AgentEvent = {
     kind: 'artifactUpdate',
     data,
   }),
+  /**
+   * Wraps a raw executor error for in-band propagation on the event bus.
+   * See the `'error'` variant of {@link AgentExecutionEvent} for the
+   * lifecycle contract — do not publish this from user-facing executor
+   * code; the framework does so automatically.
+   */
+  error: (data: unknown): AgentExecutionEvent => ({ kind: 'error', data }),
 } as const;
 
 /**

src/server/events/execution_event_queue.ts

@@ -38,11 +38,24 @@ export class ExecutionEventQueue {
    * drainable. Blocking callers return a snapshot at AUTH_REQUIRED via a
    * separate code path and a background consumer keeps draining until a
    * terminal state is reached.
+   *
+   * `error` events are handled specially: they are never yielded to the
+   * consumer. Instead the queue closes itself and throws the wrapped
+   * value so the drain loop's try/catch sees the original executor
+   * exception. Executors do not publish these directly — the framework
+   * emits them from the `.catch` on `AgentExecutor.execute()`.
    */
   public async *events(): AsyncGenerator<AgentExecutionEvent, void, undefined> {
     while (!this.stopped || this.eventQueue.length > 0) {
       if (this.eventQueue.length > 0) {
         const event = this.eventQueue.shift()!;
+        if (event.kind === 'error') {
+          // Terminate the queue and rethrow so the consumer's try/catch
+          // observes the executor's exception. Skip yielding — the
+          // `error` variant is an internal signal, not a wire event.
+          this.handleFinished();
+          throw event.data;
+        }
         yield event;
         if (
           event.kind === 'message' ||