feat: update push notification system to support message payloads and refine v0.3 compatibility wire version resolution

Author: JakubWorek

src/compat/v0_3/README.md

@@ -52,9 +52,9 @@ Webhooks registered over v0.3 transports must receive the v0.3-shaped HTTP body,
 
 This is implemented by two pieces working together:
 
-1. **`PushNotificationStore` captures the wire version.** The `InMemoryPushNotificationStore` (and any conforming implementation) reads `context.requestedVersion` on `save()` and persists it alongside the config as a `StoredPushNotificationConfig { config, wireVersion }`. When the same task later emits events, `load()` returns the wire version next to each config.
+1. **`PushNotificationStore` captures the wire version.** The `InMemoryPushNotificationStore` (and any conforming implementation) reads `context.requestedVersion` on `save()` and persists it alongside the config as a `StoredPushNotificationConfig { config, wireVersion }`. The wire version is surfaced via the optional `loadWithMetadata` read method.
 
-2. **`DefaultPushNotificationSender` routes per wire version.** The sender resolves a `PushNotificationSerializer` per stored entry using the persisted wire version. It always registers `V1PushNotificationSerializer` under `'1.0'` and falls back to it (with a one-time warning per unknown version) when no serializer is registered for the entry's version.
+2. **`DefaultPushNotificationSender` routes per wire version.** The sender prefers `loadWithMetadata` when the store implements it, otherwise falls back to `load` and defaults every entry to wire version `'0.3'` per spec §3.6.2's absent-header rule. It always registers `V1PushNotificationSerializer` under `'1.0'` and falls back to it (with a one-time warning per unknown version) when no serializer is registered for the entry's version.
 
 ### Enabling v0.3 push delivery
 
@@ -71,3 +71,13 @@ const sender = createLegacyAwarePushNotificationSender(store);
 ```
 
 v1.0-registered webhooks continue to receive the canonical `StreamResponse` body with `application/a2a+json`; v0.3-registered webhooks receive the bare-event JSON with `application/json`. Custom serializers (or overrides for the built-in `'0.3'` / `'1.0'` entries) can be supplied via the `serializers` option; user-supplied entries take precedence.
+
+### Caveat for custom `PushNotificationStore` implementations
+
+The `PushNotificationStore.loadWithMetadata` method is optional. The SDK's `InMemoryPushNotificationStore` implements it; **custom store implementations that omit it cause the sender to default every stored config to wire version `'0.3'`** (per spec §3.6.2). The implications:
+
+- **v0.3-only deployments**: no concern — the default matches your transports.
+- **v1.0-only deployments**: also no concern unless you also opt into this compat layer (which you have no reason to do).
+- **Mixed v0.3 + v1.0 deployments backed by a custom store + the compat layer**: v1.0-registered webhooks will silently receive v0.3 bodies. Implement `loadWithMetadata` on your custom store (mirror `InMemoryPushNotificationStore`'s 3-line implementation) to preserve the originating wire version per config.
+
+This compat layer (and therefore the caveat above) is opt-in and will be retired once the ecosystem has migrated to v1.0.

src/compat/v0_3/server/push_notification/index.ts

@@ -14,21 +14,21 @@ import {
   type DefaultPushNotificationSenderOptions,
 } from '../../../../server/push_notification/default_push_notification_sender.js';
 import type { PushNotificationStore } from '../../../../server/push_notification/push_notification_store.js';
-import { A2A_LEGACY_PROTOCOL_VERSION } from '../../../../constants.js';
+import { ProtocolVersion } from '../../../../constants.js';
 import { V03PushNotificationSerializer } from './v03_push_notification_serializer.js';
 
 export { V03PushNotificationSerializer };
 
 /**
  * Constructs a {@link DefaultPushNotificationSender} with the v0.3
- * serializer pre-registered under the legacy version key
- * ({@link A2A_LEGACY_PROTOCOL_VERSION}, i.e. `'0.3'`).
+ * serializer pre-registered under {@link ProtocolVersion.V0_3} (`'0.3'`).
  *
  * Webhooks registered over v0.3 transports (e.g. legacy gRPC, legacy
  * JSON-RPC, legacy REST) carry their wire version through the
- * {@link PushNotificationStore} and are dispatched with the v0.3-shaped
- * body + `application/json` content type. Webhooks registered over the
- * canonical v1.0 transports continue to use the built-in v1.0 serializer.
+ * {@link PushNotificationStore} (when it implements `loadWithMetadata`)
+ * and are dispatched with the v0.3-shaped body + `application/json`
+ * content type. Webhooks registered over the canonical v1.0 transports
+ * continue to use the built-in v1.0 serializer.
  *
  * Callers can override the pre-registered v0.3 entry — or add serializers
  * for other versions — by supplying their own `serializers` map in
@@ -41,7 +41,7 @@ export function createLegacyAwarePushNotificationSender(
   return new DefaultPushNotificationSender(pushNotificationStore, {
     ...options,
     serializers: {
-      [A2A_LEGACY_PROTOCOL_VERSION]: new V03PushNotificationSerializer(),
+      [ProtocolVersion.V0_3]: new V03PushNotificationSerializer(),
       ...(options.serializers ?? {}),
     },
   });

src/compat/v0_3/server/push_notification/v03_push_notification_serializer.ts

@@ -4,36 +4,36 @@ import {
 } from '../../../../server/push_notification/push_notification_serializer.js';
 import { StreamResponse } from '../../../../index.js';
 import { LEGACY_JSON_CONTENT_TYPE } from '../../constants.js';
-import { toCompatTask } from '../../translate/tasks.js';
 import {
+  toCompatTask,
   toCompatTaskArtifactUpdateEvent,
   toCompatTaskStatusUpdateEvent,
 } from '../../translate/tasks.js';
+import { toCompatMessage } from '../../translate/messages.js';
 
 /**
  * The v0.3 push notification serializer.
  *
  * Per the v0.3 spec example (§9.5), the push-notification HTTP body is the
- * **bare event object** (a v0.3 JSON `Task`, `TaskStatusUpdateEvent`, or
- * `TaskArtifactUpdateEvent` discriminated by its `kind` field) with
- * `Content-Type: application/json`. Notably:
+ * **bare event object** (a v0.3 JSON `Task`, `Message`,
+ * `TaskStatusUpdateEvent`, or `TaskArtifactUpdateEvent` discriminated by
+ * its `kind` field) with `Content-Type: application/json`. Notably:
  *
  *  - It is **not** wrapped in a `StreamResponse` discriminator (no outer
- *    `task` / `statusUpdate` / `artifactUpdate` key) — that wrapper is a
- *    v1.0 addition (§4.3.3).
+ *    `task` / `message` / `statusUpdate` / `artifactUpdate` key) — that
+ *    wrapper is a v1.0 addition (§4.3.3).
  *  - It is **not** wrapped in a JSON-RPC envelope (no `jsonrpc`, `id`,
  *    `result`) — push notifications are unsolicited and have no in-flight
  *    request to correlate against; the JSON-RPC envelope only appears on
  *    the streaming (SSE) path in v0.3.
  *
  * The canonical {@link StreamResponse} payload is translated to the v0.3
  * JSON shape via the per-case `toCompat*` translators in
- * `compat/v0_3/translate/tasks.ts`, which also set the legacy `kind`
- * discriminator (`'task'`, `'status-update'`, or `'artifact-update'`) the
+ * `compat/v0_3/translate/`, which set the legacy `kind` discriminator
+ * (`'task'`, `'message'`, `'status-update'`, or `'artifact-update'`) the
  * v0.3 schema requires.
  *
- * `message` payloads are rejected (consistent with v1.0 behavior); push
- * notifications are defined for task / status / artifact events only.
+ * All four payload variants are handled per spec §4.3.3.
  */
 export class V03PushNotificationSerializer implements PushNotificationSerializer {
   serialize(streamResponse: StreamResponse): SerializedPushNotification {
@@ -47,14 +47,15 @@ export class V03PushNotificationSerializer implements PushNotificationSerializer
       case 'task':
         legacyEvent = toCompatTask(payload.value);
         break;
+      case 'message':
+        legacyEvent = toCompatMessage(payload.value);
+        break;
       case 'statusUpdate':
         legacyEvent = toCompatTaskStatusUpdateEvent(payload.value);
         break;
       case 'artifactUpdate':
         legacyEvent = toCompatTaskArtifactUpdateEvent(payload.value);
         break;
-      case 'message':
-        throw new Error('Push notification should not be sent for message payload.');
       default: {
         // Exhaustive check: keeps this switch in sync with the StreamResponse
         // payload union at compile time.

src/constants.ts

@@ -37,6 +37,26 @@ export const A2A_PROTOCOL_VERSION = '1.0';
  */
 export const A2A_LEGACY_PROTOCOL_VERSION = '0.3';
 
+/**
+ * Known A2A protocol wire versions.
+ *
+ * The string values match the canonical `Major.Minor` form transmitted in
+ * the `A2A-Version` HTTP header (§3.6.1) and stored on
+ * `ServerCallContext.requestedVersion`. Used as typed keys in version-keyed
+ * registries such as `DefaultPushNotificationSenderOptions.serializers`.
+ *
+ * Enum values are hard-coded string literals (TypeScript requires enum
+ * initializers to be constants); the matching exported string constants
+ * {@link A2A_PROTOCOL_VERSION} and {@link A2A_LEGACY_PROTOCOL_VERSION}
+ * remain the canonical sources of truth. The enum is interchangeable with
+ * those constants and with free-form `string` versions arriving over the
+ * wire.
+ */
+export enum ProtocolVersion {
+  V0_3 = '0.3',
+  V1_0 = '1.0',
+}
+
 /**
  * The JSON content type per §9.1.
  * JSON-RPC requests MUST use this content type.

src/server/push_notification/default_push_notification_sender.ts

@@ -1,5 +1,9 @@
 import { TaskPushNotificationConfig, StreamResponse } from '../../index.js';
-import { A2A_PROTOCOL_VERSION } from '../../constants.js';
+import {
+  A2A_LEGACY_PROTOCOL_VERSION,
+  A2A_PROTOCOL_VERSION,
+  ProtocolVersion,
+} from '../../constants.js';
 import { ServerCallContext } from '../context.js';
 import { PushNotificationSender } from './push_notification_sender.js';
 import { PushNotificationStore, StoredPushNotificationConfig } from './push_notification_store.js';
@@ -20,10 +24,12 @@ export interface DefaultPushNotificationSenderOptions {
    */
   tokenHeaderName?: string;
   /**
-   * Per-wire-version serializers. Keys are A2A wire versions (e.g. `'1.0'`,
-   * `'0.3'`); values are the {@link PushNotificationSerializer}
-   * implementations that produce the HTTP body and content type for
-   * notifications going out to webhooks registered over that wire version.
+   * Per-wire-version push-notification serializers. Keys are A2A wire
+   * versions ({@link ProtocolVersion.V1_0} = `'1.0'`,
+   * {@link ProtocolVersion.V0_3} = `'0.3'`); values are the
+   * {@link PushNotificationSerializer} implementations that produce the
+   * HTTP body and content type for notifications going out to webhooks
+   * registered over that wire version.
    *
    * The sender always registers a built-in `'1.0'` serializer
    * ({@link V1PushNotificationSerializer}) at construction time; entries
@@ -34,8 +40,12 @@ export interface DefaultPushNotificationSenderOptions {
    * When a stored config carries a wire version with no registered
    * serializer, the sender logs a warning and falls back to the `'1.0'`
    * serializer for that dispatch.
+   *
+   * The typed key set (`ProtocolVersion`) is a developer affordance; the
+   * underlying registry accepts any string at runtime to remain forward
+   * compatible with future or custom wire versions.
    */
-  serializers?: Record<string, PushNotificationSerializer>;
+  serializers?: Partial<Record<ProtocolVersion, PushNotificationSerializer>>;
 }
 
 export class DefaultPushNotificationSender implements PushNotificationSender {
@@ -65,22 +75,33 @@ export class DefaultPushNotificationSender implements PushNotificationSender {
     // serializer for testing or alternative encodings).
     const builtinV1 = new V1PushNotificationSerializer();
     this.serializers = new Map<string, PushNotificationSerializer>([
-      [A2A_PROTOCOL_VERSION, builtinV1],
+      [ProtocolVersion.V1_0, builtinV1],
     ]);
     if (options.serializers) {
       for (const [version, serializer] of Object.entries(options.serializers)) {
-        this.serializers.set(version, serializer);
+        if (serializer) {
+          this.serializers.set(version, serializer);
+        }
       }
     }
     // Cache the v1.0 serializer for unknown-version fallback. We resolve
     // this from the registry (not the local `builtinV1`) so a user who
     // overrode '1.0' has their custom serializer used for fallback too.
-    this.fallbackSerializer = this.serializers.get(A2A_PROTOCOL_VERSION) ?? builtinV1;
+    this.fallbackSerializer = this.serializers.get(ProtocolVersion.V1_0) ?? builtinV1;
   }
 
   async send(streamResponse: StreamResponse, context: ServerCallContext): Promise<void> {
     const taskId = this._getTaskId(streamResponse);
-    const storedConfigs = await this.pushNotificationStore.load(taskId, context);
+    // Stand-alone Messages (the message-only stream pattern in §3.1.2 with
+    // no task association) cannot have a registered push config — skip the
+    // store round-trip. This also keeps the dispatch silent when the
+    // request handler forwards a bare Message event for which no task
+    // exists.
+    if (!taskId) {
+      return;
+    }
+
+    const storedConfigs = await this._loadStoredConfigs(taskId, context);
     if (!storedConfigs || storedConfigs.length === 0) {
       return;
     }
@@ -114,6 +135,17 @@ export class DefaultPushNotificationSender implements PushNotificationSender {
     });
   }
 
+  /**
+   * Returns the task id associated with a {@link StreamResponse}.
+   *
+   * Per spec §4.3.3 all four payload variants (`task`, `message`,
+   * `statusUpdate`, `artifactUpdate`) are valid push-notification payloads.
+   * For task / status / artifact events the task id is always present.
+   * For message events the task id is present iff the message is bound to
+   * an existing task (§3.4.2); stand-alone messages from the message-only
+   * stream pattern carry an empty `taskId`, in which case there can be no
+   * registered push config and the sender simply skips dispatch.
+   */
   private _getTaskId(streamResponse: StreamResponse): string {
     const payload = streamResponse.payload;
     if (!payload) {
@@ -124,9 +156,8 @@ export class DefaultPushNotificationSender implements PushNotificationSender {
         return payload.value.id;
       case 'statusUpdate':
       case 'artifactUpdate':
-        return payload.value.taskId;
       case 'message':
-        throw new Error('Push notification should not be sent for message payload.');
+        return payload.value.taskId;
       default: {
         // Exhaustive check: if a new $case is added to the StreamResponse union
         // without updating this switch, TypeScript will report a compile error here.
@@ -136,6 +167,28 @@ export class DefaultPushNotificationSender implements PushNotificationSender {
     }
   }
 
+  /**
+   * Resolves stored configs from the {@link PushNotificationStore},
+   * preferring the wire-version-aware {@link PushNotificationStore.loadWithMetadata}
+   * when available.
+   *
+   * Stores that only implement the canonical {@link PushNotificationStore.load}
+   * method are silently lifted into the wrapped shape by tagging every
+   * entry with {@link A2A_LEGACY_PROTOCOL_VERSION} (`'0.3'`) per spec
+   * §3.6.2's absent-header default. See `src/compat/v0_3/README.md` for
+   * the implication on mixed-version deployments backed by custom stores.
+   */
+  private async _loadStoredConfigs(
+    taskId: string,
+    context: ServerCallContext
+  ): Promise<StoredPushNotificationConfig[]> {
+    if (this.pushNotificationStore.loadWithMetadata) {
+      return await this.pushNotificationStore.loadWithMetadata(taskId, context);
+    }
+    const plain = await this.pushNotificationStore.load(taskId, context);
+    return plain.map((config) => ({ config, wireVersion: A2A_LEGACY_PROTOCOL_VERSION }));
+  }
+
   /**
    * Resolves the serializer registered for the given wire version.
    *

src/server/push_notification/push_notification_serializer.ts

@@ -28,10 +28,10 @@ export interface PushNotificationSerializer {
    * Serializes a {@link StreamResponse} into the HTTP body + content type
    * for one push-notification dispatch.
    *
-   * Implementations MUST reject `message` payloads (push notifications are
-   * defined for `task` / `statusUpdate` / `artifactUpdate` only, per
-   * §4.3.3). Throwing from this method aborts the dispatch and is logged
-   * by the sender; it does NOT propagate to the event loop or the caller.
+   * Implementations MUST handle all four `StreamResponse` payload variants
+   * (`task`, `message`, `statusUpdate`, `artifactUpdate`) per spec §4.3.3.
+   * Any error thrown from this method aborts the dispatch and is logged by
+   * the sender; it does NOT propagate to the event loop or the caller.
    */
   serialize(streamResponse: StreamResponse): SerializedPushNotification;
 }