From 89afc57c443889eb09f805609ccfd6bf557629fc Mon Sep 17 00:00:00 2001 From: Khaliq Date: Sun, 9 Aug 2026 23:47:25 +0200 Subject: [PATCH 1/6] feat(a2a): federate Ratify metadata --- CHANGELOG.md | 1 + README.md | 5 + docs/a2a-ratify-federation.md | 99 +++++ openapi.yaml | 85 ++++ package-lock.json | 112 +++++ packages/a2a/src/__tests__/schemas.test.ts | 66 +++ packages/a2a/src/index.ts | 72 ++++ packages/engine/CHANGELOG.md | 6 +- packages/engine/package.json | 1 + .../conformance/a2aFederation.test.ts | 398 ++++++++++++++++++ .../engine/src/engine/__tests__/a2a.test.ts | 16 + packages/engine/src/engine/a2a.ts | 97 ++++- packages/engine/src/engine/delivery.ts | 2 + packages/engine/src/engine/dm.ts | 32 +- packages/engine/src/routes/a2a.ts | 138 ++++++ packages/engine/src/routes/dm.ts | 14 +- packages/sdk-typescript/CHANGELOG.md | 6 +- packages/sdk-typescript/src/agent.ts | 7 +- packages/types/CHANGELOG.md | 6 +- packages/types/src/dm.ts | 1 + packages/types/src/message.ts | 1 + 21 files changed, 1153 insertions(+), 12 deletions(-) create mode 100644 docs/a2a-ratify-federation.md create mode 100644 packages/engine/src/__tests__/conformance/a2aFederation.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 16b22625..13a487e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ Packages without a separate changelog are covered by the cross-package notes bel ### Added - Added a pinned, multi-architecture self-host container and Compose runbook with persistent SQLite/files state, health checks, and mandatory HTTPS deployment-authority validation. +- A2A federation now preserves structured DM metadata end to end, defines a versioned `com.agentrelay.ratify` carrier for proof bundles and signed revocation lists, and supports reciprocal authenticated delivery between registered peers. ### Fixed diff --git a/README.md b/README.md index 6c7eebe9..7a8892ef 100644 --- a/README.md +++ b/README.md @@ -647,6 +647,7 @@ A2A (Agent-to-Agent) gateway endpoints: ```text POST /v1/a2a/register Register an external A2A agent GET /v1/a2a/agents List registered A2A agents +PATCH /v1/a2a/agents/:name Complete or rotate an A2A connection DELETE /v1/a2a/agents/:name Remove an A2A agent GET /v1/a2a/agents/:name/card Get agent card for a registered agent GET /.well-known/agent-card.json A2A agent card (root-level; ?workspace= selects on multi-tenant) @@ -655,6 +656,10 @@ POST /a2a/rpc A2A JSON-RPC gateway (root-level) POST /a2a/webhook/:ws/:name Inbound webhook for relay agents ``` +Single-workspace deployments serve the standard bare agent-card URL without a +workspace query parameter. A2A messages can carry versioned Ratify proof and +revocation metadata; see [Ratify over A2A](docs/a2a-ratify-federation.md). + Programmability, directory & observability: ```text diff --git a/docs/a2a-ratify-federation.md b/docs/a2a-ratify-federation.md new file mode 100644 index 00000000..e19c9ed6 --- /dev/null +++ b/docs/a2a-ratify-federation.md @@ -0,0 +1,99 @@ +# Ratify over A2A + +Relaycast carries Ratify Protocol proofs and revocations in the A2A message +metadata key `com.agentrelay.ratify`. The key lives at +`params.message.metadata["com.agentrelay.ratify"]`, not in the gateway's +top-level `params.metadata` routing fields. + +The version 1 payload is a closed, discriminated shape. Wire field names are +snake_case. A receiver must reject an unknown `version` or `kind` rather than +guessing how to interpret it. + +## Proof bundle + +```json +{ + "version": 1, + "kind": "proof_bundle", + "correlation_id": "challenge-or-invocation-id", + "bundle": "{canonical Ratify ProofBundle JSON}", + "grant": "{optional canonical Ratify DelegationCert JSON}", + "operation": { + "invocation_id": "task-42" + }, + "task": { + "title": "Update the runbook", + "instructions": "Edit the deployment section", + "path": "docs/" + } +} +``` + +`bundle` is required in both presentation directions. `grant`, `operation`, +and `task` are optional so the same carrier supports a mutual presentation or +a delegated task handoff. When present, `grant` is the canonical Ratify +DelegationCert wire JSON. The UTF-8 byte length of `bundle` may not exceed +131,072 bytes (`MAX_PROOF_BUNDLE_BYTES`, 128 KiB). The surrounding A2A JSON-RPC +request is necessarily larger and must not be capped at 128 KiB by a proxy. + +## Revocation list + +```json +{ + "version": 1, + "kind": "revocation_list", + "issuer_id": "human:northwind.example:alice", + "updated_at": 1786310000, + "revoked_certs": ["cert_01"], + "issuer_pub_key": { + "ed25519": "base64...", + "ml_dsa_65": "base64..." + }, + "signature": { + "ed25519": "base64...", + "ml_dsa_65": "base64..." + } +} +``` + +The receiver must not apply any `revoked_certs` merely because this metadata +arrived over an authenticated A2A connection. It must: + +1. resolve `issuer_id` to an already trusted issuer public key; +2. require `issuer_pub_key` to match that trusted key; +3. reconstruct the Ratify `RevocationList` from `issuer_id`, `updated_at`, + `revoked_certs`, and `signature`; and +4. call Ratify `verifyRevocationList` with the trusted issuer key before + changing local revocation state. + +The public key beside a signature is not its own trust anchor. A self-signed +attacker payload that is not bound to the expected `issuer_id` must fail closed. + +## Two-Relaycast handshake and delivery + +For deployments A and B to exchange messages in both directions: + +1. B registers A's agent card with `POST /v1/a2a/register`, setting the skill + on A as `target_agent`. B returns a relay proxy name and bearer token for A. +2. A registers B's card, stores B's returned token as `auth_credential`, and + sets the skill on B as `target_agent`. A returns its proxy token for B. +3. B completes the reciprocal connection with + `PATCH /v1/a2a/agents/{a-proxy-name}`, setting A's returned token as + `auth_credential`. The patch does not rotate either already-exchanged token. + A card with exactly one skill infers `target_agent`; multi-skill cards should + set it explicitly. +4. A sends a DM to B's local A2A proxy with the Ratify envelope in the DM + request's `data`. Relaycast places it in A2A message metadata unchanged. +5. B authenticates the bearer as the registered proxy for A, delivers the + message and metadata to the selected local agent, and emits the normal + `dm.received` delivery. B-to-A delivery follows the same path with the other + stored credential. + +A workspace key can use `/a2a/rpc` as an outbound gateway, but it cannot inject +a message directly into a local agent. Only the agent token issued to a +registered A2A proxy can use the inbound path, and that token cannot relay to a +second external A2A agent. + +Workspace agent cards advertise this extension under +`capabilities.extensions["com.agentrelay.ratify"]`, including supported +versions, kinds, and the proof-bundle byte maximum. diff --git a/openapi.yaml b/openapi.yaml index 9ae560ce..eb653601 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -290,6 +290,7 @@ components: $ref: '#/components/schemas/FileAttachment' metadata: type: object + additionalProperties: true description: User message metadata. Internal `__relaycast_*` keys are stripped from public responses and caller metadata cannot override `agent_name`. has_attachments: type: boolean @@ -350,6 +351,10 @@ components: type: array items: $ref: '#/components/schemas/FileAttachment' + metadata: + type: object + additionalProperties: true + description: Public message metadata, including A2A extension payloads DmSendResponse: type: object @@ -378,6 +383,10 @@ components: type: array items: $ref: '#/components/schemas/FileAttachment' + metadata: + type: object + additionalProperties: true + description: Public message metadata, including A2A extension payloads FlatDmMessage: type: object @@ -401,6 +410,10 @@ components: type: array items: $ref: '#/components/schemas/FileAttachment' + metadata: + type: object + additionalProperties: true + description: Public message metadata, including A2A extension payloads created_at: type: string format: date-time @@ -2698,6 +2711,13 @@ paths: items: type: string description: Optional file ids to attach to this DM + data: + type: object + nullable: true + additionalProperties: true + description: >- + Public structured message metadata. For Ratify over A2A, + place the versioned envelope at `com.agentrelay.ratify`. mode: type: string enum: [wait, steer] @@ -4878,6 +4898,28 @@ paths: post: summary: Register an external A2A agent tags: [A2A] + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + agent_card_url: + type: string + format: uri + agent_card: + type: object + auth_scheme: + type: string + enum: [bearer, api_key, none] + auth_credential: + type: string + target_agent: + type: string + description: >- + Remote card skill that DMs to the local proxy address. It + is inferred when the card advertises exactly one skill. responses: '201': description: Agent registered @@ -4899,6 +4941,43 @@ paths: type: object /a2a/agents/{name}: + patch: + summary: Complete or rotate an A2A connection + description: >- + Updates the outbound credential or selected remote skill without + rotating the inbound relay token. Use this after reciprocal + registration to complete authenticated delivery in both directions. + tags: [A2A] + parameters: + - name: name + in: path + required: true + schema: + type: string + requestBody: + required: true + content: + application/json: + schema: + type: object + minProperties: 1 + properties: + auth_scheme: + type: string + enum: [bearer, api_key, none] + auth_credential: + type: string + nullable: true + target_agent: + type: string + nullable: true + responses: + '200': + description: Connection updated + content: + application/json: + schema: + type: object delete: summary: Remove an A2A agent tags: [A2A] @@ -4942,6 +5021,12 @@ paths: description: Local development server post: summary: JSON-RPC gateway for A2A messages + description: >- + Sends to a registered external A2A agent or, when authenticated with + the bearer token issued by A2A registration, delivers to a local agent. + Ratify envelopes use + `params.message.metadata["com.agentrelay.ratify"]`; see + `docs/a2a-ratify-federation.md` for the version 1 shapes. tags: [A2A] responses: '200': diff --git a/package-lock.json b/package-lock.json index 21995aa8..686ee14a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1438,6 +1438,21 @@ "url": "https://github.com/sponsors/nzakas" } }, + "node_modules/@identities-ai/ratify-protocol": { + "version": "1.0.0-alpha.16", + "resolved": "https://registry.npmjs.org/@identities-ai/ratify-protocol/-/ratify-protocol-1.0.0-alpha.16.tgz", + "integrity": "sha512-VVtxteTa6s9Z7aPUmZjMwkmz8qJxGgYAIi+R9fbDIjC0DtQLdvgCCLoZfTdScyBdX86r9HDw1BHlCUjTm8UwSA==", + "dev": true, + "license": "Apache-2.0", + "dependencies": { + "@noble/ed25519": "^2.1.0", + "@noble/hashes": "^1.5.0", + "@noble/post-quantum": "^0.6.1" + }, + "engines": { + "node": ">=20" + } + }, "node_modules/@jridgewell/gen-mapping": { "version": "0.3.13", "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.13.tgz", @@ -1717,6 +1732,102 @@ "node": ">= 10" } }, + "node_modules/@noble/ciphers": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/@noble/ciphers/-/ciphers-2.2.0.tgz", + "integrity": "sha512-Z6pjIZ/8IJcCGzb2S/0Px5J81yij85xASuk1teLNeg75bfT07MV3a/O2Mtn1I2se43k3lkVEcFaR10N4cgQcZA==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/curves": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/@noble/curves/-/curves-2.2.0.tgz", + "integrity": "sha512-T/BoHgFXirb0ENSPBquzX0rcjXeM6Lo892a2jlYJkqk83LqZx0l1Of7DzlKJ6jkpvMrkHSnAcgb5JegL8SeIkQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@noble/hashes": "2.2.0" + }, + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/curves/node_modules/@noble/hashes": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-2.2.0.tgz", + "integrity": "sha512-IYqDGiTXab6FniAgnSdZwgWbomxpy9FtYvLKs7wCUs2a8RkITG+DFGO1DM9cr+E3/RgADRpFjrKVaJ1z6sjtEg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/ed25519": { + "version": "2.3.0", + "resolved": "https://registry.npmjs.org/@noble/ed25519/-/ed25519-2.3.0.tgz", + "integrity": "sha512-M7dvXL2B92/M7dw9+gzuydL8qn/jiqNHaoR3Q+cb1q1GHV7uwE17WCyFMG+Y+TZb5izcaXk5TdJRrDUxHXL78A==", + "dev": true, + "license": "MIT", + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/hashes": { + "version": "1.8.0", + "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-1.8.0.tgz", + "integrity": "sha512-jCs9ldd7NwzpgXDIf6P3+NrHh9/sD6CQdxHyjQI+h/6rDNo88ypBxxz45UDuZHz9r3tNz7N/VInSVoVdtXEI4A==", + "dev": true, + "license": "MIT", + "engines": { + "node": "^14.21.3 || >=16" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/post-quantum": { + "version": "0.6.1", + "resolved": "https://registry.npmjs.org/@noble/post-quantum/-/post-quantum-0.6.1.tgz", + "integrity": "sha512-+pormrDZwjRw05U8ADK4JpHejo87+gBd+muRBB/ozztH5yhDLMDF4jHQWN3NQQAsu1zBNPWTG0ZwVI0CR29H0A==", + "dev": true, + "license": "MIT", + "dependencies": { + "@noble/ciphers": "~2.2.0", + "@noble/curves": "~2.2.0", + "@noble/hashes": "~2.2.0" + }, + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, + "node_modules/@noble/post-quantum/node_modules/@noble/hashes": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/@noble/hashes/-/hashes-2.2.0.tgz", + "integrity": "sha512-IYqDGiTXab6FniAgnSdZwgWbomxpy9FtYvLKs7wCUs2a8RkITG+DFGO1DM9cr+E3/RgADRpFjrKVaJ1z6sjtEg==", + "dev": true, + "license": "MIT", + "engines": { + "node": ">= 20.19.0" + }, + "funding": { + "url": "https://paulmillr.com/funding/" + } + }, "node_modules/@relaycast/a2a": { "resolved": "packages/a2a", "link": true @@ -10620,6 +10731,7 @@ "relaycast-engine": "dist/bin/serve.js" }, "devDependencies": { + "@identities-ai/ratify-protocol": "1.0.0-alpha.16", "@types/better-sqlite3": "^7.6.11", "@types/node": "^25.2.1", "@types/ws": "^8.5.13", diff --git a/packages/a2a/src/__tests__/schemas.test.ts b/packages/a2a/src/__tests__/schemas.test.ts index 6659ce1f..8a6a4e3f 100644 --- a/packages/a2a/src/__tests__/schemas.test.ts +++ b/packages/a2a/src/__tests__/schemas.test.ts @@ -14,6 +14,9 @@ import { JsonRpcErrorSchema, JsonRpcResponseSchema, A2aResponseSchema, + MAX_PROOF_BUNDLE_BYTES, + RATIFY_A2A_METADATA_KEY, + RatifyA2aMetadataSchema, } from '../index.js'; // Helper: does any validation issue point at the given (possibly nested) path? @@ -288,6 +291,69 @@ describe('A2aMessageSchema', () => { }); }); +describe('Ratify A2A metadata', () => { + const revocation = { + version: 1, + kind: 'revocation_list', + issuer_id: 'human:northwind:alice', + updated_at: 1_786_310_000, + revoked_certs: ['cert-1'], + issuer_pub_key: { ed25519: 'pub-ed', ml_dsa_65: 'pub-pq' }, + signature: { ed25519: 'sig-ed', ml_dsa_65: 'sig-pq' }, + }; + + it('accepts the versioned proof and revocation shapes under the exact metadata key', () => { + const proofMessage = A2aMessageSchema.safeParse({ + message_id: 'proof-1', + parts: [{ kind: 'text', text: 'proofed task' }], + metadata: { + [RATIFY_A2A_METADATA_KEY]: { + version: 1, + kind: 'proof_bundle', + correlation_id: 'challenge-1', + bundle: '{"agent_id":"northwind"}', + operation: { action: 'write' }, + task: { title: 'Edit', instructions: 'Update docs', path: 'docs/' }, + }, + }, + }); + + expect(proofMessage.success).toBe(true); + expect(RatifyA2aMetadataSchema.safeParse(revocation).success).toBe(true); + }); + + it('accepts a proof bundle at the 128 KiB Ratify boundary and rejects one byte more', () => { + const atLimit = { + version: 1, + kind: 'proof_bundle', + correlation_id: 'boundary', + bundle: 'x'.repeat(MAX_PROOF_BUNDLE_BYTES), + }; + + expect(RatifyA2aMetadataSchema.safeParse(atLimit).success).toBe(true); + + const overLimit = RatifyA2aMetadataSchema.safeParse({ + ...atLimit, + bundle: `${atLimit.bundle}x`, + }); + expect(overLimit.success).toBe(false); + expect(hasIssueAtPath(overLimit, 'bundle')).toBe(true); + }); + + it('rejects an unknown version and incomplete hybrid signature components', () => { + const wrongVersion = RatifyA2aMetadataSchema.safeParse({ ...revocation, version: 2 }); + expect(wrongVersion.success).toBe(false); + expect(hasIssueAtPath(wrongVersion, 'version')).toBe(true); + + const incompleteSignature = RatifyA2aMetadataSchema.safeParse({ + ...revocation, + signature: { ed25519: 'sig-ed' }, + }); + expect(incompleteSignature.success).toBe(false); + expect(hasIssueAtPath(incompleteSignature, 'signature', 'ml_dsa_65')).toBe(true); + }); +}); + describe('A2aArtifactSchema', () => { it('accepts a valid artifact', () => { expect( diff --git a/packages/a2a/src/index.ts b/packages/a2a/src/index.ts index 33fa465d..6b55f509 100644 --- a/packages/a2a/src/index.ts +++ b/packages/a2a/src/index.ts @@ -1,5 +1,62 @@ import { z } from 'zod'; +/** Cross-deployment A2A message metadata key for Ratify Protocol payloads. */ +export const RATIFY_A2A_METADATA_KEY = 'com.agentrelay.ratify'; +export const RATIFY_A2A_WIRE_VERSION = 1; +export const MAX_PROOF_BUNDLE_BYTES = 128 * 1024; + +const RatifyHybridComponentSchema = z.object({ + ed25519: z.string().min(1), + ml_dsa_65: z.string().min(1), +}).strict(); + +const RatifyProofBundleWireSchema = z.string().min(1).superRefine((bundle, ctx) => { + const byteLength = new TextEncoder().encode(bundle).byteLength; + if (byteLength > MAX_PROOF_BUNDLE_BYTES) { + ctx.addIssue({ + code: 'custom', + message: `proof bundle exceeds ${MAX_PROOF_BUNDLE_BYTES} bytes`, + }); + } +}); + +export const RatifyProofBundleMetadataSchema = z.object({ + version: z.literal(RATIFY_A2A_WIRE_VERSION), + kind: z.literal('proof_bundle'), + correlation_id: z.string().min(1), + /** Canonical Ratify ProofBundle wire JSON. */ + bundle: RatifyProofBundleWireSchema, + /** Canonical Ratify DelegationCert wire JSON for a delegated task handoff. */ + grant: z.string().min(1).optional(), + operation: z.record(z.string(), z.unknown()).optional(), + task: z.object({ + title: z.string(), + instructions: z.string(), + path: z.string(), + }).strict().optional(), +}).strict(); + +export const RatifyRevocationListMetadataSchema = z.object({ + version: z.literal(RATIFY_A2A_WIRE_VERSION), + kind: z.literal('revocation_list'), + issuer_id: z.string().min(1), + updated_at: z.number().int().nonnegative(), + revoked_certs: z.array(z.string().min(1)), + /** Base64-encoded hybrid public key; receivers must bind it to issuer_id. */ + issuer_pub_key: RatifyHybridComponentSchema, + /** Base64-encoded issuer signature over the Ratify RevocationList fields. */ + signature: RatifyHybridComponentSchema, +}).strict(); + +export const RatifyA2aMetadataSchema = z.discriminatedUnion('kind', [ + RatifyProofBundleMetadataSchema, + RatifyRevocationListMetadataSchema, +]); + +export type RatifyProofBundleMetadata = z.infer; +export type RatifyRevocationListMetadata = z.infer; +export type RatifyA2aMetadata = z.infer; + export const A2aSkillSchema = z.object({ id: z.string().min(1).optional(), name: z.string().min(1), @@ -51,6 +108,21 @@ export const A2aMessageSchema = z.object({ role: z.enum(['user', 'agent', 'system']).default('user'), context_id: z.string().optional(), parts: z.array(A2aPartSchema).min(1), + metadata: z.record(z.string(), z.unknown()).optional(), +}).superRefine((message, ctx) => { + const ratify = message.metadata?.[RATIFY_A2A_METADATA_KEY]; + if (ratify === undefined) return; + + const parsed = RatifyA2aMetadataSchema.safeParse(ratify); + if (parsed.success) return; + + for (const issue of parsed.error.issues) { + ctx.addIssue({ + code: 'custom', + path: ['metadata', RATIFY_A2A_METADATA_KEY, ...issue.path], + message: issue.message, + }); + } }); export const A2aArtifactSchema = z.object({ diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index d825ed38..549a9116 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased - Patch] +## [Unreleased - Minor] + +### Added + +- A2A registration accepts a remote `target_agent`; a connection update endpoint completes reciprocal credentials, and registered peer tokens can deliver versioned message metadata to local DMs with normal realtime delivery. ### Fixed diff --git a/packages/engine/package.json b/packages/engine/package.json index d45cc33f..15890f25 100644 --- a/packages/engine/package.json +++ b/packages/engine/package.json @@ -68,6 +68,7 @@ "zod": "^4.3.6" }, "devDependencies": { + "@identities-ai/ratify-protocol": "1.0.0-alpha.16", "@types/better-sqlite3": "^7.6.11", "@types/node": "^25.2.1", "@types/ws": "^8.5.13", diff --git a/packages/engine/src/__tests__/conformance/a2aFederation.test.ts b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts new file mode 100644 index 00000000..638648b0 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts @@ -0,0 +1,398 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + MAX_PROOF_BUNDLE_BYTES, + RATIFY_A2A_METADATA_KEY, +} from '@relaycast/a2a'; +import { + base64StandardDecode, + base64StandardEncode, + generateHybridKeypair, + issueRevocationList, + verifyRevocationList, + type HybridPublicKey, + type RevocationList, +} from '@identities-ai/ratify-protocol'; +import { + createWorkspace, + makeNodeStack, + registerAgent, + type TestStack, +} from './harness.js'; + +type JsonRecord = Record; + +async function jsonData(response: Response): Promise { + const body = await response.json() as { data?: JsonRecord }; + return body.data ?? {}; +} + +describe('A2A federation between Relaycast deployments', () => { + let northwind: TestStack; + let borealis: TestStack; + let originalFetch: typeof globalThis.fetch; + let forwardedRpcBytes: number[]; + let transportDelayMs: number; + + beforeEach(() => { + northwind = makeNodeStack(); + borealis = makeNodeStack(); + originalFetch = globalThis.fetch; + forwardedRpcBytes = []; + transportDelayMs = 0; + + globalThis.fetch = vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + const request = input instanceof Request ? input : new Request(input, init); + const url = new URL(request.url); + if (url.pathname === '/a2a/rpc' && request.method === 'POST' && transportDelayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, transportDelayMs)); + } + if (url.hostname === 'northwind.example') { + return northwind.app.request(request); + } + if (url.hostname === 'borealis.example') { + if (url.pathname === '/a2a/rpc' && request.method === 'POST') { + const body = await request.clone().text(); + if (body.includes(RATIFY_A2A_METADATA_KEY)) { + forwardedRpcBytes.push(new TextEncoder().encode(body).byteLength); + } + } + return borealis.app.request(request); + } + throw new Error(`Unexpected federated test URL: ${url.toString()}`); + }) as typeof globalThis.fetch; + }); + + afterEach(() => { + globalThis.fetch = originalFetch; + northwind.close(); + borealis.close(); + }); + + async function connectDeployments() { + const northwindWorkspace = await createWorkspace(northwind.app, 'northwind'); + const borealisWorkspace = await createWorkspace(borealis.app, 'borealis'); + const lead = await registerAgent(northwind.app, northwindWorkspace.workspaceKey, 'lead'); + const worker = await registerAgent(borealis.app, borealisWorkspace.workspaceKey, 'worker'); + + const registerNorthwind = await borealis.app.request('/v1/a2a/register', { + method: 'POST', + headers: { + authorization: `Bearer ${borealisWorkspace.workspaceKey}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + agent_card: { + name: 'northwind', + url: 'https://northwind.example/a2a/rpc', + version: '1.0.0', + skills: [{ id: 'lead', name: 'lead' }], + }, + target_agent: 'lead', + }), + }); + expect(registerNorthwind.status).toBe(201); + const northwindProxyOnBorealis = await jsonData(registerNorthwind); + + const registerBorealis = await northwind.app.request('/v1/a2a/register', { + method: 'POST', + headers: { + authorization: `Bearer ${northwindWorkspace.workspaceKey}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + agent_card: { + name: 'borealis', + url: 'https://borealis.example/a2a/rpc', + version: '1.0.0', + skills: [{ id: 'worker', name: 'worker' }], + }, + auth_scheme: 'bearer', + auth_credential: northwindProxyOnBorealis.relay_token, + target_agent: 'worker', + }), + }); + expect(registerBorealis.status).toBe(201); + const borealisProxyOnNorthwind = await jsonData(registerBorealis); + + // The second registration returns Borealis's bearer token on Northwind. + // Store it on Borealis's existing Northwind proxy to complete reciprocal, + // independently authenticated delivery without rotating either token. + const completeBorealisConnection = await borealis.app.request( + `/v1/a2a/agents/${encodeURIComponent(String(northwindProxyOnBorealis.relay_name))}`, + { + method: 'PATCH', + headers: { + authorization: `Bearer ${borealisWorkspace.workspaceKey}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + auth_scheme: 'bearer', + auth_credential: borealisProxyOnNorthwind.relay_token, + }), + }, + ); + expect(completeBorealisConnection.status).toBe(200); + + return { + northwindWorkspace, + borealisWorkspace, + lead, + worker, + borealisProxyName: String(borealisProxyOnNorthwind.relay_name), + northwindProxyName: String(northwindProxyOnBorealis.relay_name), + }; + } + + async function agentMessages(stack: TestStack, agentToken: string): Promise { + const conversationsResponse = await stack.app.request('/v1/dm/conversations', { + headers: { authorization: `Bearer ${agentToken}` }, + }); + expect(conversationsResponse.status).toBe(200); + const conversations = (await conversationsResponse.json() as { data: JsonRecord[] }).data; + if (conversations.length === 0) return []; + + const response = await stack.app.request( + `/v1/dm/${String(conversations[0]!.id)}/messages`, + { headers: { authorization: `Bearer ${agentToken}` } }, + ); + expect(response.status).toBe(200); + return (await response.json() as { data: JsonRecord[] }).data; + } + + it('carries a full 128 KiB proof bundle end to end and rejects one byte more before egress', async () => { + const federation = await connectDeployments(); + const metadata = { + [RATIFY_A2A_METADATA_KEY]: { + version: 1, + kind: 'proof_bundle', + correlation_id: 'full-size-proof', + bundle: 'x'.repeat(MAX_PROOF_BUNDLE_BYTES), + }, + }; + + const sent = await northwind.app.request('/v1/dm', { + method: 'POST', + headers: { + authorization: `Bearer ${federation.lead.token}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + to: federation.borealisProxyName, + text: 'proofed task', + data: metadata, + }), + }); + expect(sent.status).toBe(201); + + const received = await agentMessages(borealis, federation.worker.token); + expect(received).toHaveLength(1); + expect(received[0]!.metadata).toMatchObject(metadata); + expect(forwardedRpcBytes).toHaveLength(1); + expect(forwardedRpcBytes[0]).toBeGreaterThan(MAX_PROOF_BUNDLE_BYTES); + console.info( + `A2A full proof: bundle=${MAX_PROOF_BUNDLE_BYTES} bytes, JSON-RPC body=${forwardedRpcBytes[0]} bytes`, + ); + + const rpcCountBeforeOversize = forwardedRpcBytes.length; + const oversized = await northwind.app.request('/v1/dm', { + method: 'POST', + headers: { + authorization: `Bearer ${federation.lead.token}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + to: federation.borealisProxyName, + text: 'must not leave northwind', + data: { + [RATIFY_A2A_METADATA_KEY]: { + ...metadata[RATIFY_A2A_METADATA_KEY], + bundle: 'x'.repeat(MAX_PROOF_BUNDLE_BYTES + 1), + }, + }, + }), + }); + expect(oversized.status).toBe(400); + expect(forwardedRpcBytes).toHaveLength(rpcCountBeforeOversize); + expect(await agentMessages(borealis, federation.worker.token)).toHaveLength(1); + }); + + it('blocks unauthenticated and unregistered peers from injecting Ratify proof metadata', async () => { + const federation = await connectDeployments(); + const request = { + jsonrpc: '2.0', + id: 'unregistered-caller', + method: 'message/send', + params: { + target_agent: 'worker', + message: { + message_id: 'unregistered-caller', + role: 'user', + parts: [{ kind: 'text', text: 'must not arrive' }], + metadata: { + [RATIFY_A2A_METADATA_KEY]: { + version: 1, + kind: 'proof_bundle', + correlation_id: 'unauthorized-proof', + bundle: '{"attacker":true}', + }, + }, + }, + }, + }; + + const unauthenticated = await borealis.app.request('/a2a/rpc', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(request), + }); + expect(unauthenticated.status).toBe(401); + + const rejected = await borealis.app.request('/a2a/rpc', { + method: 'POST', + headers: { + authorization: `Bearer ${federation.borealisWorkspace.workspaceKey}`, + 'content-type': 'application/json', + }, + body: JSON.stringify(request), + }); + expect(rejected.status).toBe(400); + await expect(rejected.json()).resolves.toMatchObject({ + error: { code: -32004 }, + }); + expect(await agentMessages(borealis, federation.worker.token)).toHaveLength(0); + }); + + it('applies signed revocations in the authenticated reverse direction and reports latency tails', async () => { + const federation = await connectDeployments(); + const issuer = await generateHybridKeypair(); + const issuerId = 'human:borealis:bob'; + type WireRevocation = { + version: 1; + kind: 'revocation_list'; + issuer_id: string; + updated_at: number; + revoked_certs: string[]; + issuer_pub_key: { ed25519: string; ml_dsa_65: string }; + signature: { ed25519: string; ml_dsa_65: string }; + }; + + const revoked = new Set(); + const trustedIssuers = new Map([[issuerId, issuer.publicKey]]); + const applyIfValid = async (wire: WireRevocation): Promise => { + const trustedKey = trustedIssuers.get(wire.issuer_id); + if (!trustedKey) return false; + const carriedKey: HybridPublicKey = { + ed25519: base64StandardDecode(wire.issuer_pub_key.ed25519), + ml_dsa_65: base64StandardDecode(wire.issuer_pub_key.ml_dsa_65), + }; + if ( + base64StandardEncode(carriedKey.ed25519) !== base64StandardEncode(trustedKey.ed25519) + || base64StandardEncode(carriedKey.ml_dsa_65) !== base64StandardEncode(trustedKey.ml_dsa_65) + ) return false; + + const candidate: RevocationList = { + issuer_id: wire.issuer_id, + updated_at: wire.updated_at, + revoked_certs: wire.revoked_certs, + signature: { + ed25519: base64StandardDecode(wire.signature.ed25519), + ml_dsa_65: base64StandardDecode(wire.signature.ml_dsa_65), + }, + }; + if (!(await verifyRevocationList(candidate, trustedKey))) return false; + for (const certId of candidate.revoked_certs) revoked.add(certId); + return true; + }; + + const wouldAcceptGrant = (certId: string) => !revoked.has(certId); + + const sendAndApply = async ( + certId: string, + sequence: number, + ): Promise<{ latencyMs: number; wire: WireRevocation }> => { + const signed: RevocationList = { + issuer_id: issuerId, + updated_at: Math.floor(Date.now() / 1000) + sequence, + revoked_certs: [certId], + signature: { ed25519: new Uint8Array(), ml_dsa_65: new Uint8Array() }, + }; + await issueRevocationList(signed, issuer.privateKey); + + const wire: WireRevocation = { + version: 1, + kind: 'revocation_list', + issuer_id: signed.issuer_id, + updated_at: signed.updated_at, + revoked_certs: signed.revoked_certs, + issuer_pub_key: { + ed25519: base64StandardEncode(issuer.publicKey.ed25519), + ml_dsa_65: base64StandardEncode(issuer.publicKey.ml_dsa_65), + }, + signature: { + ed25519: base64StandardEncode(signed.signature.ed25519), + ml_dsa_65: base64StandardEncode(signed.signature.ml_dsa_65), + }, + }; + const issuedAt = performance.now(); + + const sent = await borealis.app.request('/v1/dm', { + method: 'POST', + headers: { + authorization: `Bearer ${federation.worker.token}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + to: federation.northwindProxyName, + text: `issuer revocation ${sequence}`, + data: { [RATIFY_A2A_METADATA_KEY]: wire }, + }), + }); + expect(sent.status).toBe(201); + + const received = await agentMessages(northwind, federation.lead.token); + const matching = received.find((message) => { + const metadata = message.metadata as JsonRecord | undefined; + const candidate = metadata?.[RATIFY_A2A_METADATA_KEY] as Partial | undefined; + return candidate?.kind === 'revocation_list' && candidate.revoked_certs?.includes(certId); + }); + expect(matching).toBeDefined(); + const receivedWire = (matching!.metadata as JsonRecord)[RATIFY_A2A_METADATA_KEY] as WireRevocation; + expect(await applyIfValid(receivedWire)).toBe(true); + expect(wouldAcceptGrant(certId)).toBe(false); + + return { latencyMs: performance.now() - issuedAt, wire: receivedWire }; + }; + + const samples: number[] = []; + let lastWire: WireRevocation | undefined; + for (let index = 0; index < 20; index += 1) { + const result = await sendAndApply(`cert-live-grant-${index}`, index); + samples.push(result.latencyMs); + lastWire = result.wire; + } + + const sorted = [...samples].sort((left, right) => left - right); + const middle = sorted.length / 2; + const medianMs = (sorted[middle - 1]! + sorted[middle]!) / 2; + const p95Ms = sorted[Math.ceil(sorted.length * 0.95) - 1]!; + const maxMs = sorted.at(-1)!; + expect(samples).toHaveLength(20); + + transportDelayMs = 75; + const delayed = await sendAndApply('cert-injected-delay', 20); + transportDelayMs = 0; + expect(delayed.latencyMs).toBeGreaterThanOrEqual(60); + console.info( + `A2A revocation latency: n=${samples.length}, median=${medianMs.toFixed(2)} ms, ` + + `p95=${p95Ms.toFixed(2)} ms, max=${maxMs.toFixed(2)} ms, ` + + `injected_transport=75 ms -> ${delayed.latencyMs.toFixed(2)} ms`, + ); + + const tampered = { + ...lastWire!, + revoked_certs: ['cert-attacker-chose'], + }; + expect(await applyIfValid(tampered)).toBe(false); + expect(revoked.has('cert-attacker-chose')).toBe(false); + }); +}); diff --git a/packages/engine/src/engine/__tests__/a2a.test.ts b/packages/engine/src/engine/__tests__/a2a.test.ts index badacc7f..f6552acc 100644 --- a/packages/engine/src/engine/__tests__/a2a.test.ts +++ b/packages/engine/src/engine/__tests__/a2a.test.ts @@ -325,6 +325,14 @@ describe('translateRelayToA2a <-> translateA2aToRelay round-trip', () => { text: 'round trip', thread_id: 'rt-thread', attachments: [{ file_id: 'f-1', filename: 'note.md', content_type: 'text/markdown', size_bytes: 11 }], + metadata: { + 'com.agentrelay.ratify': { + version: 1, + kind: 'proof_bundle', + correlation_id: 'rt-1', + bundle: '{"agent_id":"alice"}', + }, + }, }; const back = translateA2aToRelay(translateRelayToA2a(dm)); @@ -334,6 +342,7 @@ describe('translateRelayToA2a <-> translateA2aToRelay round-trip', () => { expect(back.attachments).toEqual([ { file_id: 'f-1', filename: 'note.md', content_type: 'text/markdown', size_bytes: 11 }, ]); + expect(back.metadata).toEqual(dm.metadata); // The text gains a [file] marker line for the recovered file part. expect(back.text).toBe('round trip\n[file] note.md'); }); @@ -437,6 +446,13 @@ describe('getWorkspaceAgentCard', () => { expect(card.documentation_url).toBe('https://github.com/AgentWorkforce/relaycast'); expect(card.capabilities).toEqual({ methods: ['message/send', 'message/stream', 'task/get', 'task/cancel'], + extensions: { + 'com.agentrelay.ratify': { + versions: [1], + kinds: ['proof_bundle', 'revocation_list'], + max_proof_bundle_bytes: 131072, + }, + }, }); expect(card.provider).toEqual({ organization: 'Relaycast', diff --git a/packages/engine/src/engine/a2a.ts b/packages/engine/src/engine/a2a.ts index a7865a58..4cbaf677 100644 --- a/packages/engine/src/engine/a2a.ts +++ b/packages/engine/src/engine/a2a.ts @@ -8,6 +8,10 @@ import { A2aTaskStateSchema, JsonRpcRequestSchema, JsonRpcResponseSchema, + MAX_PROOF_BUNDLE_BYTES, + RATIFY_A2A_METADATA_KEY, + RATIFY_A2A_WIRE_VERSION, + RatifyA2aMetadataSchema, type A2aAgentCard, type A2aJsonRpcRequest, type A2aJsonRpcResponse, @@ -48,6 +52,7 @@ const RelayDMSchema = z.object({ created_at: z.string().optional(), thread_id: z.string().nullable().optional(), attachments: z.array(RelayFileAttachmentSchema).optional(), + metadata: z.record(z.string(), z.unknown()).optional(), }); export { @@ -59,6 +64,10 @@ export { A2aTaskStateSchema, JsonRpcRequestSchema, JsonRpcResponseSchema, + RatifyA2aMetadataSchema, + MAX_PROOF_BUNDLE_BYTES, + RATIFY_A2A_METADATA_KEY, + RATIFY_A2A_WIRE_VERSION, }; export type { @@ -79,6 +88,8 @@ export interface RegisterA2aAgentInput { agentCard?: A2aAgentCard; authScheme?: 'bearer' | 'api_key' | 'none'; authCredential?: string; + /** Remote skill/agent that DMs to this proxy should address. */ + targetAgent?: string; } export interface RegisterA2aAgentResult { @@ -88,6 +99,13 @@ export interface RegisterA2aAgentResult { certification: 'level_0' | 'level_1'; } +export interface UpdateA2aAgentConnectionInput { + authScheme?: 'bearer' | 'api_key' | 'none'; + authCredential?: string | null; + /** Remote skill/agent that DMs to this proxy should address. */ + targetAgent?: string | null; +} + export interface A2aAgentRecord { id: string; workspace_id: string; @@ -272,6 +290,16 @@ export async function registerA2aAgent( const agentCard = A2aAgentCardSchema.parse(resolvedCard); const externalUrl = normalizeBaseUrl(agentCard.url); const relayName = await deriveRelayName(agentCard); + const remoteSkillNames = new Set(agentCard.skills.flatMap((skill) => [skill.id, skill.name].filter(Boolean) as string[])); + if (input.targetAgent && !remoteSkillNames.has(input.targetAgent)) { + throw codedError( + `A2A target agent "${input.targetAgent}" is not advertised by the agent card`, + 'a2a_target_agent_not_advertised', + 400, + ); + } + const targetAgent = input.targetAgent + ?? (agentCard.skills.length === 1 ? (agentCard.skills[0]!.id ?? agentCard.skills[0]!.name) : undefined); const [existing] = await db .select({ id: a2aAgents.id }) @@ -288,6 +316,7 @@ export async function registerA2aAgent( a2a_external_url: externalUrl, a2a_skills: agentCard.skills, a2a_active: true, + ...(targetAgent ? { a2a_target_agent: targetAgent } : {}), }; // A prior removeA2aAgent soft-removes the A2A agent: it deletes the a2a_agents @@ -422,6 +451,58 @@ export async function getA2aAgentByRelayName( return row ? formatA2aRecord(row) : null; } +/** + * Complete or rotate the outbound half of an A2A registration. + * + * Reciprocal registration necessarily yields the second deployment's bearer + * token after the first registration row already exists. Updating that row is + * what lets both deployments authenticate sends without deleting proxies and + * invalidating the token the other side just stored. + */ +export async function updateA2aAgentConnection( + db: Db, + workspaceId: string, + relayName: string, + input: UpdateA2aAgentConnectionInput, +): Promise { + const agentRecord = await getA2aAgentByRelayName(db, workspaceId, relayName); + if (!agentRecord) return null; + + if (input.targetAgent !== undefined && input.targetAgent !== null) { + const advertisedNames = new Set( + agentRecord.agent_card.skills.flatMap((skill) => [skill.id, skill.name].filter(Boolean) as string[]), + ); + if (!advertisedNames.has(input.targetAgent)) { + throw codedError( + `A2A target agent "${input.targetAgent}" is not advertised by the agent card`, + 'a2a_target_agent_not_advertised', + 400, + ); + } + } + + await db + .update(a2aAgents) + .set({ + ...(input.authScheme !== undefined ? { authScheme: input.authScheme } : {}), + ...(input.authCredential !== undefined ? { authCredential: input.authCredential } : {}), + updatedAt: new Date(), + }) + .where(eq(a2aAgents.id, agentRecord.id)); + + if (input.targetAgent !== undefined) { + const metadata = { ...(agentRecord.relay_metadata ?? {}) }; + if (input.targetAgent === null) { + delete metadata.a2a_target_agent; + } else { + metadata.a2a_target_agent = input.targetAgent; + } + await updateAgent(db, workspaceId, relayName, { metadata }); + } + + return getA2aAgentByRelayName(db, workspaceId, relayName); +} + export async function removeA2aAgent(db: Db, workspaceId: string, relayName: string): Promise { const agentRecord = await getA2aAgentByRelayName(db, workspaceId, relayName); if (!agentRecord) return false; @@ -468,6 +549,7 @@ export function translateRelayToA2a(message: RelayDM): A2aJsonRpcRequest { role: 'user', context_id: relay.thread_id ?? undefined, parts, + metadata: relay.metadata, }, }, }; @@ -485,6 +567,7 @@ export function translateA2aToRelay(jsonRpc: A2aJsonRpcRequest | A2aJsonRpcRespo text: extractTextFromParts(message.parts), thread_id: message.context_id ?? null, attachments: extractAttachmentsFromParts(message.parts), + metadata: message.metadata, }; } @@ -500,6 +583,7 @@ export function translateA2aToRelay(jsonRpc: A2aJsonRpcRequest | A2aJsonRpcRespo text: parts.length > 0 ? extractTextFromParts(parts) : task?.status.message ?? '', thread_id: responseMessage?.context_id ?? task?.context_id ?? null, attachments: extractAttachmentsFromParts(parts), + metadata: responseMessage?.metadata, }; } @@ -508,7 +592,11 @@ export async function sendToExternalAgent( jsonRpcPayload: A2aJsonRpcRequest, auth?: { scheme?: 'bearer' | 'api_key' | 'none' | string | null; credential?: string | null }, ): Promise { - const request = JsonRpcRequestSchema.parse(jsonRpcPayload); + const parsedRequest = JsonRpcRequestSchema.safeParse(jsonRpcPayload); + if (!parsedRequest.success) { + throw codedError('Invalid A2A JSON-RPC request or metadata', 'invalid_a2a_request', 400); + } + const request = parsedRequest.data; const targetUrl = normalizeBaseUrl(agentUrl); if (!isSafeExternalUrl(targetUrl)) { @@ -684,6 +772,13 @@ export async function getWorkspaceAgentCard( }, capabilities: { methods: ['message/send', 'message/stream', 'task/get', 'task/cancel'], + extensions: { + [RATIFY_A2A_METADATA_KEY]: { + versions: [RATIFY_A2A_WIRE_VERSION], + kinds: ['proof_bundle', 'revocation_list'], + max_proof_bundle_bytes: MAX_PROOF_BUNDLE_BYTES, + }, + }, }, default_input_modes: ['text/plain', 'application/json'], default_output_modes: ['text/plain', 'application/json'], diff --git a/packages/engine/src/engine/delivery.ts b/packages/engine/src/engine/delivery.ts index 19ec009e..671fd8f3 100644 --- a/packages/engine/src/engine/delivery.ts +++ b/packages/engine/src/engine/delivery.ts @@ -544,6 +544,7 @@ function buildRoutableDeliveryEvent( text: row.body, injection_mode: injectionMode, attachments, + metadata: publicMessageMetadata(row.metadata as Record | null), }, created_at: row.createdAt.toISOString(), id: row.delivery.messageId, @@ -552,6 +553,7 @@ function buildRoutableDeliveryEvent( text: row.body, injection_mode: injectionMode, attachments, + metadata: publicMessageMetadata(row.metadata as Record | null), }, { fromName: senderName }), }; } diff --git a/packages/engine/src/engine/dm.ts b/packages/engine/src/engine/dm.ts index 029c8312..dd82c3ac 100644 --- a/packages/engine/src/engine/dm.ts +++ b/packages/engine/src/engine/dm.ts @@ -23,6 +23,7 @@ import { import { DEFAULT_MAILBOX_DEPTH_CAP, DEFAULT_MAILBOX_TTL_MS, type MailboxConfig } from './mailboxConfig.js'; import { codedError } from '../lib/httpError.js'; import { fetchAttachmentsBatch, resolveSendAttachments, type AttachmentRow } from './attachments.js'; +import { publicMessageMetadata, sanitizeUserMessageMetadata } from './messageMetadata.js'; type Db = ReturnType; @@ -246,7 +247,12 @@ function buildDmMessageWrites( workspaceId: string, fromAgentId: string, channelId: string, - data: { text: string; attachments?: string[]; mode?: 'wait' | 'steer' }, + data: { + text: string; + attachments?: string[]; + mode?: 'wait' | 'steer'; + data?: Record | null; + }, attachments: AttachmentRow[], messageId: string, ): AtomicWrite[] { @@ -261,7 +267,12 @@ function buildDmMessageWrites( agentId: fromAgentId, body: data.text, hasAttachments, - metadata: { injection_mode: data.mode ?? 'wait' }, + metadata: { + // Keep the server-owned delivery mode after caller metadata so a + // federated peer cannot override how the local runtime is injected. + ...sanitizeUserMessageMetadata(data.data), + injection_mode: data.mode ?? 'wait', + }, }) .returning(), ]; @@ -282,7 +293,13 @@ export async function sendDm( db: Db, workspaceId: string, fromAgentId: string, - data: { to: string; text: string; attachments?: string[]; mode?: 'wait' | 'steer' }, + data: { + to: string; + text: string; + attachments?: string[]; + mode?: 'wait' | 'steer'; + data?: Record | null; + }, options: SendDmOptions = {}, ) { const startedAtMs = Date.now(); @@ -332,11 +349,15 @@ export async function sendDm( created_at: new Date().toISOString(), thread_id: conv.id, attachments, + metadata: sanitizeUserMessageMetadata(data.data), }); payload.params = { ...payload.params, - target_agent: toAgent.name, + target_agent: + typeof a2aTarget.relay_metadata?.a2a_target_agent === 'string' + ? a2aTarget.relay_metadata.a2a_target_agent + : toAgent.name, metadata: { target_agent: fromAgent.name, relay_conversation_id: conv.id, @@ -412,6 +433,7 @@ export async function sendDm( text: message.body, injection_mode: injectionMode, attachments, + metadata: publicMessageMetadata(message.metadata), }, created_at: message.createdAt.toISOString(), @@ -422,6 +444,7 @@ export async function sendDm( text: message.body, injection_mode: injectionMode, attachments, + metadata: publicMessageMetadata(message.metadata), // Internal: delivery record for the recipient — stripped by route before response _delivery: dmDelivery, @@ -602,6 +625,7 @@ export async function getDmMessages( agent_name: r.agentName, text: r.body, injection_mode: r.metadata?.injection_mode as 'wait' | 'steer' | undefined, + metadata: publicMessageMetadata(r.metadata), attachments: attachmentMap.get(r.id) || [], created_at: r.createdAt.toISOString(), })); diff --git a/packages/engine/src/routes/a2a.ts b/packages/engine/src/routes/a2a.ts index 6dc11b32..0f625420 100644 --- a/packages/engine/src/routes/a2a.ts +++ b/packages/engine/src/routes/a2a.ts @@ -9,6 +9,10 @@ import { asCodedError, errorResponse, type CodedError } from '../lib/httpError.j import { rateLimit } from '../middleware/rateLimit.js'; import * as a2aEngine from '../engine/a2a.js'; import * as dmEngine from '../engine/dm.js'; +import { buildDmReceivedEventData } from '../engine/deliveryWire.js'; +import { publishWorkspaceEvent } from './fanout.js'; +import { notifyDeliveryRejections, routeDeliveryOutcomes } from './deliveryRouting.js'; +import { sendWebhookEvent } from './webhookOutbox.js'; import { jsonCreated, jsonError, @@ -24,11 +28,21 @@ const registerA2aSchema = z.object({ agent_card: a2aEngine.A2aAgentCardSchema.optional(), auth_scheme: z.enum(['bearer', 'api_key', 'none']).optional(), auth_credential: z.string().optional(), + target_agent: z.string().min(1).optional(), }).refine((value) => value.agent_card_url || value.agent_card, { message: 'agent_card_url or agent_card is required', path: ['agent_card_url'], }); +const updateA2aConnectionSchema = z.object({ + auth_scheme: z.enum(['bearer', 'api_key', 'none']).optional(), + auth_credential: z.string().min(1).nullable().optional(), + target_agent: z.string().min(1).nullable().optional(), +}).refine( + (value) => Object.values(value).some((field) => field !== undefined), + { message: 'At least one connection field is required' }, +); + const rpcRequestSchema = a2aEngine.JsonRpcRequestSchema; const rpcWebhookSchema = z.union([a2aEngine.JsonRpcRequestSchema, a2aEngine.JsonRpcResponseSchema]); @@ -153,6 +167,7 @@ a2aRoutes.post('/v1/a2a/register', requireAuth, rateLimit, async (c) => { agentCard: parsed.data.agent_card, authScheme: parsed.data.auth_scheme, authCredential: parsed.data.auth_credential, + targetAgent: parsed.data.target_agent, }); return jsonCreated(c, { @@ -166,6 +181,39 @@ a2aRoutes.post('/v1/a2a/register', requireAuth, rateLimit, async (c) => { } }); +// PATCH /v1/a2a/agents/:name +a2aRoutes.patch('/v1/a2a/agents/:name', requireAuth, rateLimit, async (c) => { + try { + const parsed = await parseJsonBody(c, updateA2aConnectionSchema, 'At least one connection field is required'); + if (!parsed.ok) { + return parsed.response; + } + + const updated = await a2aEngine.updateA2aAgentConnection( + c.get('db'), + c.get('workspace').id, + c.req.param('name'), + { + authScheme: parsed.data.auth_scheme, + authCredential: parsed.data.auth_credential, + targetAgent: parsed.data.target_agent, + }, + ); + if (!updated) { + return a2aAgentNotFound(c); + } + + return jsonOk(c, { + relay_name: updated.relay_name, + auth_scheme: updated.auth_scheme, + target_agent: updated.relay_metadata?.a2a_target_agent ?? null, + updated: true, + }); + } catch (err: unknown) { + return codedJsonError(c, err); + } +}); + // DELETE /v1/a2a/agents/:name a2aRoutes.delete('/v1/a2a/agents/:name', requireAuth, rateLimit, async (c) => { try { @@ -304,6 +352,95 @@ a2aRoutes.post('/a2a/rpc', requireAuth, rateLimit, async (c) => { } const target = await a2aEngine.getA2aAgentByRelayName(db, workspace.id, targetAgentName); + + // A bearer token issued by A2A registration identifies the remote + // deployment's local proxy. Requests from that proxy land on a real local + // agent; they must never be used as an authenticated open relay to another + // external A2A target. + const authenticatedAgent = c.get('agent'); + const [registeredCaller] = authenticatedAgent + ? await db + .select({ id: a2aAgents.id }) + .from(a2aAgents) + .where(and( + eq(a2aAgents.workspaceId, workspace.id), + eq(a2aAgents.relayAgentId, authenticatedAgent.id), + )) + : []; + + if (registeredCaller) { + if (request.method !== 'message/send' && request.method !== 'message/stream') { + const response = a2aEngine.jsonRpcError(request.id, -32601, `Unsupported inbound method "${request.method}"`); + return jsonResponse(c, response, jsonRpcHttpStatus(response)); + } + if (!request.params?.message) { + const response = a2aEngine.jsonRpcError(request.id, -32602, 'message is required'); + return jsonResponse(c, response, jsonRpcHttpStatus(response)); + } + if (target) { + const response = a2aEngine.jsonRpcError(request.id, -32003, 'A registered A2A caller cannot relay to another external agent'); + return jsonResponse(c, response, 403); + } + + const [localTarget] = await db + .select({ id: agents.id }) + .from(agents) + .where(and(eq(agents.workspaceId, workspace.id), eq(agents.name, targetAgentName))); + if (!localTarget) { + const response = a2aEngine.jsonRpcError(request.id, -32004, `Unknown local agent "${targetAgentName}"`); + return jsonResponse(c, response, jsonRpcHttpStatus(response)); + } + + const relayMessage = a2aEngine.translateA2aToRelay(request); + const sent = await dmEngine.sendDm(db, workspace.id, authenticatedAgent!.id, { + to: targetAgentName, + text: relayMessage.text, + mode: 'wait', + data: relayMessage.metadata, + }, { + skipA2aIntercept: true, + }); + await a2aEngine.incrementA2aMessagesReceived(db, registeredCaller.id); + + const eventData = buildDmReceivedEventData(sent, { fromName: authenticatedAgent!.name }); + await sendWebhookEvent(c, { + type: 'dm.received', + workspaceId: workspace.id, + data: eventData, + }); + await publishWorkspaceEvent(c, 'dm.received', eventData); + if (sent._delivery) { + await routeDeliveryOutcomes(c, [sent._delivery], 'dm.received', eventData); + } + if (sent._delivery_rejections.length > 0) { + await notifyDeliveryRejections(c, authenticatedAgent!.id, sent._delivery_rejections); + } + + const response = a2aEngine.jsonRpcSuccess(request.id, { + task: { + id: sent.message.id, + context_id: relayMessage.thread_id ?? sent.conversation_id, + status: { + state: a2aEngine.mapRelayTaskState('queued'), + message: 'Message accepted by Relaycast', + }, + history: [{ + message_id: sent.message.id, + role: 'agent', + context_id: relayMessage.thread_id ?? sent.conversation_id, + parts: [{ kind: 'text', text: sent.message.text }], + metadata: relayMessage.metadata, + }], + metadata: { + conversation_id: sent.conversation_id, + relay_agent: authenticatedAgent!.name, + target_agent: targetAgentName, + }, + }, + }); + return jsonResponse(c, response); + } + if (!target) { const response = a2aEngine.jsonRpcError(request.id, -32004, `Unknown A2A agent "${targetAgentName}"`); return jsonResponse(c, response, jsonRpcHttpStatus(response)); @@ -397,6 +534,7 @@ a2aRoutes.post('/a2a/webhook/:workspace_id/:agent_name', async (c) => { to: targetAgentName, text: relayMessage.text, mode: 'wait', + data: relayMessage.metadata, }, { skipA2aIntercept: true, }); diff --git a/packages/engine/src/routes/dm.ts b/packages/engine/src/routes/dm.ts index 491e87a9..1462d540 100644 --- a/packages/engine/src/routes/dm.ts +++ b/packages/engine/src/routes/dm.ts @@ -22,6 +22,7 @@ const sendDmSchema = z.object({ to: z.string().min(1), text: z.string().min(1), attachments: z.array(z.string()).optional(), + data: z.record(z.string(), z.unknown()).nullable().optional(), mode: z.enum(['wait', 'steer']).default('wait'), }); @@ -47,8 +48,14 @@ dmRoutes.post( if (!parsed.ok) { return parsed.response; } - const { to, text, attachments, mode } = parsed.data; + const { to, text, attachments, data, mode } = parsed.data; const normalizedAttachments = attachments && attachments.length > 0 ? attachments : undefined; + const fingerprintBody = { + to, + text, + ...(normalizedAttachments ? { attachments: normalizedAttachments } : {}), + ...(data !== undefined ? { data } : {}), + }; const { key: idempotencyKey, error: idempotencyError } = parseIdempotencyKey(c.req.header('Idempotency-Key')); if (idempotencyError) { @@ -69,13 +76,14 @@ dmRoutes.post( // Backward compatibility: historical fingerprint excluded mode (equivalent to wait). // Only include mode when explicit steer is requested. fingerprint: mode === 'steer' - ? JSON.stringify({ to, text, ...(normalizedAttachments ? { attachments: normalizedAttachments } : {}), mode }) - : JSON.stringify({ to, text, ...(normalizedAttachments ? { attachments: normalizedAttachments } : {}) }), + ? JSON.stringify({ ...fingerprintBody, mode }) + : JSON.stringify(fingerprintBody), kv: c.get('engine').kv, operation: () => dmEngine.sendDm(db, workspace.id, agent!.id, { to, text, attachments: normalizedAttachments, + data, mode, }, { mailbox }), afterOperation: async (data) => { diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index 5522bb8c..e070f079 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Minor] + +### Added + +- `AgentClient.dm()` accepts structured `data` and returns public DM metadata, enabling versioned A2A extensions such as Ratify proofs and revocations. ## [7.0.0] - 2026-08-07 diff --git a/packages/sdk-typescript/src/agent.ts b/packages/sdk-typescript/src/agent.ts index c694511d..8f03cc97 100644 --- a/packages/sdk-typescript/src/agent.ts +++ b/packages/sdk-typescript/src/agent.ts @@ -561,12 +561,17 @@ export class AgentClient { async dm( agent: string, text: string, - opts?: (IdempotencyOption & { mode?: 'wait' | 'steer'; attachments?: string[] }), + opts?: (IdempotencyOption & { + mode?: 'wait' | 'steer'; + attachments?: string[]; + data?: Record | null; + }), ): Promise { const body: SendDmRequest = { to: agent, text, ...(opts?.attachments ? { attachments: opts.attachments } : {}), + ...(opts?.data !== undefined ? { data: opts.data } : {}), mode: opts?.mode ?? 'wait', }; return this.client.post('/v1/dm', body, idempotencyHeaders(opts)); diff --git a/packages/types/CHANGELOG.md b/packages/types/CHANGELOG.md index a31e95c6..3f27f0c2 100644 --- a/packages/types/CHANGELOG.md +++ b/packages/types/CHANGELOG.md @@ -7,7 +7,11 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [Unreleased - Minor] + +### Added + +- Direct-message requests accept structured `data`, and DM payloads expose the resulting public `metadata`. ## [7.0.0] - 2026-08-07 diff --git a/packages/types/src/dm.ts b/packages/types/src/dm.ts index 37d2f904..fc863b4c 100644 --- a/packages/types/src/dm.ts +++ b/packages/types/src/dm.ts @@ -29,6 +29,7 @@ export const SendDmRequestSchema = z.object({ to: z.string(), text: z.string(), attachments: z.array(z.string()).optional(), + data: z.record(z.string(), z.unknown()).nullable().optional(), mode: DmInjectionModeSchema.default('wait'), }); export type SendDmRequest = z.infer; diff --git a/packages/types/src/message.ts b/packages/types/src/message.ts index 9af9c4da..f1cfb36f 100644 --- a/packages/types/src/message.ts +++ b/packages/types/src/message.ts @@ -78,6 +78,7 @@ export const CoreMessagePayloadSchema = z.object({ text: z.string(), injection_mode: MessageInjectionModeSchema.optional(), attachments: z.array(FileAttachmentSchema).optional(), + metadata: z.record(z.string(), z.unknown()).optional(), }); export type CoreMessagePayload = z.infer; From d98332a06cc76d0558d8a13e84fdb65dd0415871 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 10 Aug 2026 13:29:20 +0200 Subject: [PATCH 2/6] address review: verbatim signed payloads, inbound idempotency, Swift metadata MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Devin 🔴 / cubic P1 — the SDK's automatic key transform rewrote the inner field names of message `data` and `metadata`. For a Ratify envelope that is not cosmetic: a RevocationList whose `revoked_certs` arrives as `revokedCerts` cannot be reconstructed byte-for-byte, so its signature no longer verifies and a cross-deployment revocation is rejected for the wrong reason. Both keys join VERBATIM_VALUE_KEYS alongside `headers`, `input` and `input_schema`, which are already exempt on exactly this reasoning: their keys are caller data, not wire protocol. BEHAVIOUR CHANGE, called out in the SDK changelog rather than buried: callers who relied on the SDK snake_casing their metadata keys on the wire now see the keys exactly as written, including for agent registration metadata. The setup.test.ts assertion that expected `favorite_color` encoded the corruption rather than the contract — a caller who deliberately wrote `favoriteColor` got a different key back — so that expectation is corrected rather than preserved. Reviewers should weigh this: it is the one part of this PR that changes behaviour for existing consumers. Codex P1 — inbound A2A `message/send` called sendDm directly, bypassing the idempotency the DM route already applies to every send. `sendToExternalAgent` retries on 5xx, and the counter, webhook, workspace event and delivery routing all run after the durable write, so a transient failure there delivered the counterparty's proof or task twice. Inbound now runs through runIdempotent, keyed on the caller's message_id (falling back to the JSON-RPC id) and scoped to the registered caller so two counterparties cannot collide. Replays skip the side effects and return the original result. cubic P1 — Swift's CoreMessagePayload silently dropped the newly preserved metadata during decoding, so Swift consumers could not see Ratify envelopes at all. Added as optional [String: JSONValue], mirroring CoreMessagePayloadSchema. Both new tests verified non-vacuous: reverting only the casing change fails the two round-trip assertions, and disabling only the idempotency key delivers the retried message twice. Also drops the agent-card discovery changes from this branch — they ship separately in #319, so the two PRs are now disjoint. engine 53 files / 556 tests, sdk-typescript 419 tests, full turbo build green. Co-Authored-By: Claude Opus 5 --- .../conformance/a2aFederation.test.ts | 50 +++++++++++ packages/engine/src/routes/a2a.ts | 83 ++++++++++++++----- .../sdk-swift/Sources/Relaycast/Models.swift | 8 ++ packages/sdk-typescript/CHANGELOG.md | 4 + .../src/__tests__/casing.test.ts | 37 +++++++++ .../src/__tests__/setup.test.ts | 9 +- packages/sdk-typescript/src/casing.ts | 11 +++ 7 files changed, 181 insertions(+), 21 deletions(-) diff --git a/packages/engine/src/__tests__/conformance/a2aFederation.test.ts b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts index 638648b0..cf91546f 100644 --- a/packages/engine/src/__tests__/conformance/a2aFederation.test.ts +++ b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts @@ -140,6 +140,8 @@ describe('A2A federation between Relaycast deployments', () => { worker, borealisProxyName: String(borealisProxyOnNorthwind.relay_name), northwindProxyName: String(northwindProxyOnBorealis.relay_name), + // The bearer Northwind presents when calling Borealis's /a2a/rpc. + northwindProxyToken: String(northwindProxyOnBorealis.relay_token), }; } @@ -262,6 +264,54 @@ describe('A2A federation between Relaycast deployments', () => { expect(await agentMessages(borealis, federation.worker.token)).toHaveLength(0); }); + it('delivers a retried inbound message once, not twice', async () => { + // The sending side retries on 5xx, and everything after the durable DM + // write on the receiving side — the counter, the webhook, the workspace + // event, delivery routing — can still fail. Without an idempotency key the + // retry writes a second DM, and the counterparty's proof or task is + // delivered twice. Same message_id must mean one delivery. + const federation = await connectDeployments(); + const request = { + jsonrpc: '2.0', + id: 'retried-rpc-id', + method: 'message/send', + params: { + target_agent: 'worker', + message: { + message_id: 'retried-message-id', + role: 'user', + parts: [{ kind: 'text', text: 'delivered once' }], + metadata: { + [RATIFY_A2A_METADATA_KEY]: { + version: 1, + kind: 'proof_bundle', + correlation_id: 'retried-proof', + bundle: '{"proof":true}', + }, + }, + }, + }, + }; + + const send = async () => + borealis.app.request('/a2a/rpc', { + method: 'POST', + headers: { + authorization: `Bearer ${federation.northwindProxyToken}`, + 'content-type': 'application/json', + }, + body: JSON.stringify(request), + }); + + const first = await send(); + expect(first.status).toBe(200); + const second = await send(); + expect(second.status).toBe(200); + + const received = await agentMessages(borealis, federation.worker.token); + expect(received).toHaveLength(1); + }); + it('applies signed revocations in the authenticated reverse direction and reports latency tails', async () => { const federation = await connectDeployments(); const issuer = await generateHybridKeypair(); diff --git a/packages/engine/src/routes/a2a.ts b/packages/engine/src/routes/a2a.ts index 0f625420..de68d424 100644 --- a/packages/engine/src/routes/a2a.ts +++ b/packages/engine/src/routes/a2a.ts @@ -7,6 +7,7 @@ import { a2aAgents, agents, messages, workspaces } from '../db/schema.js'; import { requireAuth, hashToken } from '../middleware/auth.js'; import { asCodedError, errorResponse, type CodedError } from '../lib/httpError.js'; import { rateLimit } from '../middleware/rateLimit.js'; +import { runIdempotent } from '../middleware/idempotency.js'; import * as a2aEngine from '../engine/a2a.js'; import * as dmEngine from '../engine/dm.js'; import { buildDmReceivedEventData } from '../engine/deliveryWire.js'; @@ -91,7 +92,7 @@ function extractWorkspaceHint(c: Context): string | null { return hostSegments[0]!; } - return null; + return c.req.param('workspace') || null; } function extractTargetAgentName(params: Record | undefined, fallbackContextId?: string): string | null { @@ -392,28 +393,70 @@ a2aRoutes.post('/a2a/rpc', requireAuth, rateLimit, async (c) => { } const relayMessage = a2aEngine.translateA2aToRelay(request); - const sent = await dmEngine.sendDm(db, workspace.id, authenticatedAgent!.id, { - to: targetAgentName, - text: relayMessage.text, - mode: 'wait', - data: relayMessage.metadata, - }, { - skipA2aIntercept: true, - }); - await a2aEngine.incrementA2aMessagesReceived(db, registeredCaller.id); - const eventData = buildDmReceivedEventData(sent, { fromName: authenticatedAgent!.name }); - await sendWebhookEvent(c, { - type: 'dm.received', + // Inbound deliveries must be idempotent, because the sending side retries. + // `sendToExternalAgent` re-sends on 5xx, and everything after the durable + // DM write here — the counter, the webhook, the workspace event, delivery + // routing — can still fail. Without a key, that retry writes a second DM + // and the counterparty's proof or task is delivered twice. The DM route + // already runs every send through `runIdempotent`; this path called + // `sendDm` directly and skipped it. + // + // The key is the caller's own message id where it supplies one, falling + // back to the JSON-RPC request id, scoped to the registered caller so two + // counterparties cannot collide on the same value. + const inboundMessageId = + typeof request.params?.message?.message_id === 'string' + ? request.params.message.message_id + : typeof request.id === 'string' || typeof request.id === 'number' + ? String(request.id) + : null; + + const idempotent = await runIdempotent({ workspaceId: workspace.id, - data: eventData, + actorId: authenticatedAgent!.id, + scope: 'a2a:inbound', + key: inboundMessageId ? `${registeredCaller.id}:${inboundMessageId}` : undefined, + status: 200, + fingerprint: JSON.stringify({ + to: targetAgentName, + text: relayMessage.text, + data: relayMessage.metadata ?? null, + }), + kv: c.get('engine').kv, + operation: () => dmEngine.sendDm(db, workspace.id, authenticatedAgent!.id, { + to: targetAgentName, + text: relayMessage.text, + mode: 'wait', + data: relayMessage.metadata, + }, { + skipA2aIntercept: true, + }), }); - await publishWorkspaceEvent(c, 'dm.received', eventData); - if (sent._delivery) { - await routeDeliveryOutcomes(c, [sent._delivery], 'dm.received', eventData); - } - if (sent._delivery_rejections.length > 0) { - await notifyDeliveryRejections(c, authenticatedAgent!.id, sent._delivery_rejections); + const sent = idempotent.data; + + // A replay returns the original result and must not repeat the side + // effects: re-counting the message, re-firing dm.received to webhooks and + // the workspace stream, or re-routing delivery would make a retried + // request observably different from a single one, which is the thing + // idempotency exists to prevent. The response below is identical either + // way, so the caller cannot tell — which is the point. + if (!idempotent.replayed) { + await a2aEngine.incrementA2aMessagesReceived(db, registeredCaller.id); + + const eventData = buildDmReceivedEventData(sent, { fromName: authenticatedAgent!.name }); + await sendWebhookEvent(c, { + type: 'dm.received', + workspaceId: workspace.id, + data: eventData, + }); + await publishWorkspaceEvent(c, 'dm.received', eventData); + if (sent._delivery) { + await routeDeliveryOutcomes(c, [sent._delivery], 'dm.received', eventData); + } + if (sent._delivery_rejections.length > 0) { + await notifyDeliveryRejections(c, authenticatedAgent!.id, sent._delivery_rejections); + } } const response = a2aEngine.jsonRpcSuccess(request.id, { diff --git a/packages/sdk-swift/Sources/Relaycast/Models.swift b/packages/sdk-swift/Sources/Relaycast/Models.swift index fb6b48e7..ea52b7ea 100644 --- a/packages/sdk-swift/Sources/Relaycast/Models.swift +++ b/packages/sdk-swift/Sources/Relaycast/Models.swift @@ -604,6 +604,14 @@ public struct CoreMessagePayload: Codable, Equatable, Sendable { public let text: String public let injectionMode: MessageInjectionMode? public let attachments: [FileAttachment]? + /// Caller-authored message metadata, passed through verbatim. + /// + /// Mirrors `metadata` on `CoreMessagePayloadSchema`. Without it a Swift + /// consumer decodes the payload successfully and silently loses the field, + /// which now carries delegated-authority proofs and signed revocation + /// lists on federated deliveries — data that cannot be reconstructed once + /// dropped. + public let metadata: [String: JSONValue]? } public struct MessageWithMeta: Codable, Equatable, Sendable { diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index e070f079..b0d59188 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -13,6 +13,10 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht - `AgentClient.dm()` accepts structured `data` and returns public DM metadata, enabling versioned A2A extensions such as Ratify proofs and revocations. +### Changed + +- **Breaking for callers who rely on key rewriting inside `data`/`metadata`.** Both are now passed through verbatim, like `headers`, `input` and `input_schema`, instead of being snake_cased on send and camelCased on read. Their keys are caller-authored data: rewriting them corrupted any document whose field names are meaningful, and made signed payloads unverifiable — a `RevocationList` whose `revoked_certs` arrived as `revokedCerts` cannot be reconstructed, so its signature fails. Callers who wrote camelCase keys and read them back unchanged are unaffected; callers who depended on the SDK converting their metadata keys to snake_case on the wire will now see the keys exactly as written, including for agent registration metadata. + ## [7.0.0] - 2026-08-07 ### Changed diff --git a/packages/sdk-typescript/src/__tests__/casing.test.ts b/packages/sdk-typescript/src/__tests__/casing.test.ts index 5097ab3d..53dc2910 100644 --- a/packages/sdk-typescript/src/__tests__/casing.test.ts +++ b/packages/sdk-typescript/src/__tests__/casing.test.ts @@ -13,7 +13,44 @@ const schema = { additionalProperties: false, }; +// A signed document is the strictest case: it must survive a round trip +// byte-for-byte or its signature stops verifying. Wire field names here are +// snake_case by protocol, not by convention, and camelizing them destroys the +// signable bytes. +const signedRevocationList = { + revoked_certs: ['cert-a', 'cert-b'], + issued_at: 1754870400, + issuer_pub_key: { ed25519: 'AAA', ml_dsa_65: 'BBB' }, + signature: { ed25519: 'CCC', ml_dsa_65: 'DDD' }, +}; + describe('casing transforms', () => { + it('passes message data and metadata through verbatim in both directions', () => { + const wire = decamelizeKeys({ + to: 'agent-b', + text: 'work', + data: signedRevocationList, + }) as Record; + expect(wire.data).toEqual(signedRevocationList); + + const read = camelizeKeys({ + thread_id: 't1', + metadata: signedRevocationList, + }) as { threadId: string; metadata: unknown }; + expect(read.threadId).toBe('t1'); + expect(read.metadata).toEqual(signedRevocationList); + }); + + it('a signed payload round-trips byte-identical through both transforms', () => { + // The regression this guards: revoked_certs arriving as revokedCerts means + // the list cannot be reconstructed, its signature fails, and a + // cross-deployment revocation is rejected for the wrong reason. + const sent = { data: signedRevocationList }; + const read = camelizeKeys(decamelizeKeys(sent)) as { data: unknown }; + expect(JSON.stringify(read.data)).toBe(JSON.stringify(signedRevocationList)); + }); + + it('decamelizeKeys renames wire fields but passes schema subtrees verbatim', () => { const wire = decamelizeKeys({ name: 'crm.get_person_batch', diff --git a/packages/sdk-typescript/src/__tests__/setup.test.ts b/packages/sdk-typescript/src/__tests__/setup.test.ts index b2b1b02d..1a69c0f3 100644 --- a/packages/sdk-typescript/src/__tests__/setup.test.ts +++ b/packages/sdk-typescript/src/__tests__/setup.test.ts @@ -437,8 +437,15 @@ describe('WorkspaceHandle', () => { name: 'Alice', type: 'human', persona: 'planner', + // Wire fields are snake_cased; the metadata VALUE is not. Its keys are + // caller-authored data, so the transform leaves them exactly as written — + // the same rule already applied to `headers`, `input` and `input_schema`. + // This assertion previously expected `favorite_color`, which encoded the + // corruption rather than the contract: a caller who deliberately wrote + // `favoriteColor` got a different key back, and anything signed could not + // survive the round trip at all. metadata: { - favorite_color: 'blue', + favoriteColor: 'blue', }, }); }); diff --git a/packages/sdk-typescript/src/casing.ts b/packages/sdk-typescript/src/casing.ts index 2c372cd9..909f78c9 100644 --- a/packages/sdk-typescript/src/casing.ts +++ b/packages/sdk-typescript/src/casing.ts @@ -33,6 +33,15 @@ function toSnakeKey(key: string): string { * transforming them corrupts the document (e.g. a schema's * `properties.batchSize` must not become `properties.batch_size`) — so both * transforms rename the field itself but pass the value through verbatim. + * + * `data` and `metadata` belong here for the same reason, and their absence was + * a live corruption rather than a latent one. Message `data`/`metadata` carry + * caller-authored documents, and once Ratify proof bundles and signed + * revocation lists travel in them the transform is not merely untidy: a + * `RevocationList` whose `revoked_certs` arrives as `revokedCerts` cannot be + * reconstructed byte-for-byte, so its signature no longer verifies and the + * cross-deployment kill switch fails closed for the wrong reason. Anything + * signed must survive a round trip unmodified. */ const VERBATIM_VALUE_KEYS = new Set([ 'headers', @@ -42,6 +51,8 @@ const VERBATIM_VALUE_KEYS = new Set([ 'inputSchema', 'output_schema', 'outputSchema', + 'data', + 'metadata', ]); export function camelizeKeys(value: T): Camelize { From 89bda63a60be0f973d46952740b5af28493c2ae9 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 10 Aug 2026 13:33:37 +0200 Subject: [PATCH 3/6] changelog: state the measured scope of the casing change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous entry led with 'Breaking', which is technically true and measurably alarmist. Checked against production before claiming a blast radius: 1025 agents, 723 with metadata, 8 multi-word keys in use (node_id, registered_at, cloud_workspace_id, cloud_agent_id, invocation_id, identity_key, room_membership_id, room_identity_id) — all stored snake_case. Single-word keys are identical in both casings and cannot be affected. Grepped relay, relaycast, cloud and the VS Code extension for the camelCase forms these would be read as today (metadata.nodeId, metadata.cloudWorkspaceId, …): zero hits. The one consumer that reads agent metadata at all, cloud/packages/web/lib/room/relaycast-access.ts, checks metadata.cloud_workspace_id in snake_case and reaches the API through globalThis.fetch rather than this SDK, so it reads the wire format and is unaffected either way. Entry now states what actually changes and for whom. Co-Authored-By: Claude Opus 5 --- packages/sdk-typescript/CHANGELOG.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index b0d59188..669ec062 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -15,7 +15,9 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht ### Changed -- **Breaking for callers who rely on key rewriting inside `data`/`metadata`.** Both are now passed through verbatim, like `headers`, `input` and `input_schema`, instead of being snake_cased on send and camelCased on read. Their keys are caller-authored data: rewriting them corrupted any document whose field names are meaningful, and made signed payloads unverifiable — a `RevocationList` whose `revoked_certs` arrived as `revokedCerts` cannot be reconstructed, so its signature fails. Callers who wrote camelCase keys and read them back unchanged are unaffected; callers who depended on the SDK converting their metadata keys to snake_case on the wire will now see the keys exactly as written, including for agent registration metadata. +- `data` and `metadata` are now passed through verbatim, like `headers`, `input` and `input_schema`, instead of being snake_cased on send and camelCased on read. Their keys are caller-authored data: rewriting them corrupted any document whose field names are meaningful, and made signed payloads unverifiable — a `RevocationList` whose `revoked_certs` arrived as `revokedCerts` cannot be reconstructed, so its signature fails. + + Scope of the change: only **multi-word** keys differ, and only for callers reading or writing these fields through this SDK. A reader who previously saw `nodeId` for a stored `node_id` now sees `node_id`. Callers who wrote camelCase keys and read them back unchanged are unaffected, and callers who deliberately wrote snake_case keys were already getting a different key back — for them this is a fix. ## [7.0.0] - 2026-08-07 From ca3a7a172779ae06fac9c591c345663f18507273 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 10 Aug 2026 14:02:55 +0200 Subject: [PATCH 4/6] raise the pending SDK release to Major for the casing change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cubic P2, accepted — with a note, because there is precedent pointing the other way and it deserves stating rather than silently overriding. 6.2.0 shipped the identical class of change (input_schema/output_schema, input/ output and headers stopped being key-rewritten) as a Minor, framed as a Fix, on the reasoning that rewriting user-authored keys was corruption rather than contract. By that precedent this entry is Minor too: data/metadata are the same kind of value and were simply missed from the same set. Taking Major anyway, because the two differ in one material way. input/output and headers are action payloads whose reader is essentially the code that produced them, so a rename is self-consistent. Agent and message metadata is written by one system and read by another — fleet and cloud write node_id, registered_at, cloud_workspace_id; other code reads them — and a silent rename across a system boundary is exactly where this bites. A behavioural change on a cross-system field-name contract should not ride a minor bump. Root and sdk-typescript headings raised; the engine and types headings stay Minor, since neither changes behaviour for existing callers. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 2 +- packages/sdk-typescript/CHANGELOG.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 13a487e4..dc2fee06 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ This project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). Packages without a separate changelog are covered by the cross-package notes below. -## [Unreleased - Minor] +## [Unreleased - Major] ### Added diff --git a/packages/sdk-typescript/CHANGELOG.md b/packages/sdk-typescript/CHANGELOG.md index 669ec062..926117f9 100644 --- a/packages/sdk-typescript/CHANGELOG.md +++ b/packages/sdk-typescript/CHANGELOG.md @@ -7,7 +7,7 @@ See the [root changelog](../../CHANGELOG.md) for cross-package release highlight The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased - Minor] +## [Unreleased - Major] ### Added From 408961bf29aca4483a783d6b6077129b7758806b Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 10 Aug 2026 14:33:08 +0200 Subject: [PATCH 5/6] address review: the remaining P2/P3 items MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mailbox config (two comments, same defect): inbound A2A delivery passed only skipA2aIntercept, so sendDm fell back to its fixed one-hour / 1000-message defaults and registered-peer deliveries were the one path on the deployment exempt from its own configured TTL and depth cap. Now resolves and passes the mailbox config as /v1/dm does. Workspace/observer DM history dropped metadata: getDmMessagesForWorkspace never selected or projected messages.metadata, so a federated delivery's Ratify envelope was visible to the recipient and invisible in history — the same message reading differently depending on which door you came through. Projected through publicMessageMetadata, which strips the internal __relaycast_* keys. Idempotency fingerprint size: `data` was embedded whole. A proof bundle runs to 128 KiB, and the fingerprint is serialized into the stored record, kept for the TTL, and string-compared on every replay — roughly 256 KiB per DM in KV and a full-payload comparison each time. Now digested with sha256Hex, which answers the only question the fingerprint asks in constant size. Inbound acknowledgement latency: fanout, delivery routing and rejection notification were awaited, so the counterparty's "message accepted" waited on our recipient's delivery, including a slow HTTP-push receiver. Moved to runInBackground, matching /v1/dm. The durable write is what the response attests to. message/stream: accepted and then served as a one-shot, returning a task already terminal with no stream channel. A client calling message/stream expects a working task plus somewhere to subscribe; a completed one-shot is a wrong answer dressed as a right one and would silently truncate a conversation the caller believes is open. Now refused with -32601 and a message naming message/send, so the caller falls back immediately. NOTE: the agent card still advertises message/stream — that is a wider decision than this PR and is flagged rather than changed here. Base64 validation: issuer_pub_key and signature are documented as base64 but only checked non-empty, so a value that could never decode passed validation at the edge and failed later inside the verifier as an opaque error. Validated as standard base64. Existing test fixtures used placeholders ('pub-ed', 'sig-ed') that are not base64; replaced with real values and added cases for the rule. Proof-bundle size guard: the whole string was encoded before deciding it was oversized, so an attacker could force a multi-megabyte encode and a second same-size allocation before rejection — on unauthenticated inbound federation traffic. String length is checked first; UTF-8 cannot shrink, so length over the cap already proves byte length over it. Revocation test: asserts the received wire deep-equals the sent one, not just that its signature verifies. The engine is not the enforcement point — a Ratify verifier is — so the property the engine owns is that a signed document crosses unmodified. A key rename now fails with a readable diff rather than an opaque signature failure. turbo test 18/18 tasks. Co-Authored-By: Claude Opus 5 --- packages/a2a/src/__tests__/schemas.test.ts | 27 ++++++++- packages/a2a/src/index.ts | 34 ++++++++++- .../conformance/a2aFederation.test.ts | 8 +++ packages/engine/src/engine/dmAll.ts | 8 +++ packages/engine/src/routes/a2a.ts | 57 +++++++++++++++---- packages/engine/src/routes/dm.ts | 10 +++- 6 files changed, 128 insertions(+), 16 deletions(-) diff --git a/packages/a2a/src/__tests__/schemas.test.ts b/packages/a2a/src/__tests__/schemas.test.ts index 8a6a4e3f..ce6e8480 100644 --- a/packages/a2a/src/__tests__/schemas.test.ts +++ b/packages/a2a/src/__tests__/schemas.test.ts @@ -298,8 +298,11 @@ describe('Ratify A2A metadata', () => { issuer_id: 'human:northwind:alice', updated_at: 1_786_310_000, revoked_certs: ['cert-1'], - issuer_pub_key: { ed25519: 'pub-ed', ml_dsa_65: 'pub-pq' }, - signature: { ed25519: 'sig-ed', ml_dsa_65: 'sig-pq' }, + // Real base64. These fields are documented as base64-encoded keys and + // signatures and are now validated as such, so placeholders like 'pub-ed' + // (not base64 — the hyphen is outside the alphabet) no longer stand in. + issuer_pub_key: { ed25519: 'cHViLWVk', ml_dsa_65: 'cHViLXBx' }, + signature: { ed25519: 'c2lnLWVk', ml_dsa_65: 'c2lnLXBx' }, }; it('accepts the versioned proof and revocation shapes under the exact metadata key', () => { @@ -347,10 +350,28 @@ describe('Ratify A2A metadata', () => { const incompleteSignature = RatifyA2aMetadataSchema.safeParse({ ...revocation, - signature: { ed25519: 'sig-ed' }, + signature: { ed25519: 'c2lnLWVk' }, }); expect(incompleteSignature.success).toBe(false); expect(hasIssueAtPath(incompleteSignature, 'signature', 'ml_dsa_65')).toBe(true); + + // These fields are documented as base64. A non-empty check accepted values + // that could never decode, so a malformed key or signature passed A2A + // validation at the edge and failed later inside the verifier as an opaque + // error, far from the input that caused it. + for (const bad of ['sig-ed', 'not base64', 'AAAA=AAA', 'QUJD']) { + const result = RatifyA2aMetadataSchema.safeParse({ + ...revocation, + signature: { ...revocation.signature, ed25519: bad }, + }); + if (bad === 'QUJD') { + // Valid base64, length a multiple of 4 — must still be accepted. + expect(result.success).toBe(true); + continue; + } + expect(result.success).toBe(false); + expect(hasIssueAtPath(result, 'signature', 'ed25519')).toBe(true); + } }); }); diff --git a/packages/a2a/src/index.ts b/packages/a2a/src/index.ts index 6b55f509..6d396c80 100644 --- a/packages/a2a/src/index.ts +++ b/packages/a2a/src/index.ts @@ -5,12 +5,42 @@ export const RATIFY_A2A_METADATA_KEY = 'com.agentrelay.ratify'; export const RATIFY_A2A_WIRE_VERSION = 1; export const MAX_PROOF_BUNDLE_BYTES = 128 * 1024; +/** + * Standard base64, optionally padded. These fields are documented as + * base64-encoded keys and signatures, and a non-empty check accepted anything — + * so a value with `-`, `_`, spaces or any other stray character passed A2A + * validation and only failed later at decode, inside the verifier, as an opaque + * error. Rejecting at the edge keeps the schema's promise honest and puts the + * error where the malformed input entered. + */ +const BASE64 = /^[A-Za-z0-9+/]+={0,2}$/; +const base64Field = z + .string() + .min(1) + .refine((v) => BASE64.test(v) && v.length % 4 === 0, { + message: 'expected standard base64', + }); + const RatifyHybridComponentSchema = z.object({ - ed25519: z.string().min(1), - ml_dsa_65: z.string().min(1), + ed25519: base64Field, + ml_dsa_65: base64Field, }).strict(); const RatifyProofBundleWireSchema = z.string().min(1).superRefine((bundle, ctx) => { + // Reject on string length before encoding. UTF-8 is at most 3 bytes per UTF-16 + // code unit for anything expressible in a JS string, so `length` over the cap + // already guarantees the byte length is over it. Encoding first meant an + // attacker could force a full multi-megabyte encode — allocating a second + // buffer of the same size — before the payload was rejected for being too + // large. The cheap check has to come first, precisely because this schema runs + // on unauthenticated inbound federation traffic. + if (bundle.length > MAX_PROOF_BUNDLE_BYTES) { + ctx.addIssue({ + code: 'custom', + message: `proof bundle exceeds ${MAX_PROOF_BUNDLE_BYTES} bytes`, + }); + return; + } const byteLength = new TextEncoder().encode(bundle).byteLength; if (byteLength > MAX_PROOF_BUNDLE_BYTES) { ctx.addIssue({ diff --git a/packages/engine/src/__tests__/conformance/a2aFederation.test.ts b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts index cf91546f..de14aad9 100644 --- a/packages/engine/src/__tests__/conformance/a2aFederation.test.ts +++ b/packages/engine/src/__tests__/conformance/a2aFederation.test.ts @@ -407,6 +407,14 @@ describe('A2A federation between Relaycast deployments', () => { }); expect(matching).toBeDefined(); const receivedWire = (matching!.metadata as JsonRecord)[RATIFY_A2A_METADATA_KEY] as WireRevocation; + + // The engine is not the enforcement point — a Ratify verifier is, and + // `applyIfValid` models one here. What the engine is responsible for is + // that a signed document crosses the boundary unmodified, so assert that + // directly rather than leaving it implied by the signature check. A key + // rename or a re-encode anywhere in transport fails this line first, with + // a readable diff, instead of surfacing as an opaque signature failure. + expect(receivedWire).toEqual(wire); expect(await applyIfValid(receivedWire)).toBe(true); expect(wouldAcceptGrant(certId)).toBe(false); diff --git a/packages/engine/src/engine/dmAll.ts b/packages/engine/src/engine/dmAll.ts index c8f79838..a6ca03d7 100644 --- a/packages/engine/src/engine/dmAll.ts +++ b/packages/engine/src/engine/dmAll.ts @@ -8,6 +8,7 @@ import { agents, } from '../db/schema.js'; import { codedError } from '../lib/httpError.js'; +import { publicMessageMetadata } from './messageMetadata.js'; type Db = ReturnType; @@ -136,6 +137,7 @@ export async function getDmMessagesForWorkspace( agentId: messages.agentId, agentName: agents.name, body: messages.body, + metadata: messages.metadata, createdAt: messages.createdAt, }) .from(messages) @@ -149,6 +151,12 @@ export async function getDmMessagesForWorkspace( agent_id: r.agentId, agent_name: r.agentName, text: r.body, + // Workspace/observer history is a public DM read path and must project + // metadata like the agent-facing one, or a federated delivery's Ratify + // envelope is visible to the recipient and invisible in history — the same + // message reading differently depending on which door you came through. + // `publicMessageMetadata` strips the internal `__relaycast_*` keys. + metadata: publicMessageMetadata(r.metadata as Record | null), created_at: r.createdAt.toISOString(), })); } diff --git a/packages/engine/src/routes/a2a.ts b/packages/engine/src/routes/a2a.ts index de68d424..fc48ff26 100644 --- a/packages/engine/src/routes/a2a.ts +++ b/packages/engine/src/routes/a2a.ts @@ -8,6 +8,8 @@ import { requireAuth, hashToken } from '../middleware/auth.js'; import { asCodedError, errorResponse, type CodedError } from '../lib/httpError.js'; import { rateLimit } from '../middleware/rateLimit.js'; import { runIdempotent } from '../middleware/idempotency.js'; +import { runInBackground } from './background.js'; +import { resolveMailboxConfig } from '../engine/mailboxConfig.js'; import * as a2aEngine from '../engine/a2a.js'; import * as dmEngine from '../engine/dm.js'; import { buildDmReceivedEventData } from '../engine/deliveryWire.js'; @@ -370,8 +372,20 @@ a2aRoutes.post('/a2a/rpc', requireAuth, rateLimit, async (c) => { : []; if (registeredCaller) { - if (request.method !== 'message/send' && request.method !== 'message/stream') { - const response = a2aEngine.jsonRpcError(request.id, -32601, `Unsupported inbound method "${request.method}"`); + if (request.method !== 'message/send') { + // `message/stream` is deliberately refused here rather than served. + // + // This branch performs a single sendDm and returns a task already in a + // terminal `queued` state with no stream channel. A client that calls + // message/stream expects a `working` task plus somewhere to subscribe + // for incremental updates; handing it a completed one-shot is a wrong + // answer dressed as a right one, and it would silently truncate a + // conversation the caller believes is still open. Refusing lets the + // caller fall back to message/send immediately and correctly. + const detail = request.method === 'message/stream' + ? 'message/stream is not supported on inbound federated delivery; use message/send' + : `Unsupported inbound method "${request.method}"`; + const response = a2aEngine.jsonRpcError(request.id, -32601, detail); return jsonResponse(c, response, jsonRpcHttpStatus(response)); } if (!request.params?.message) { @@ -431,6 +445,11 @@ a2aRoutes.post('/a2a/rpc', requireAuth, rateLimit, async (c) => { data: relayMessage.metadata, }, { skipA2aIntercept: true, + // Without this, sendDm falls back to its fixed one-hour / 1000-message + // defaults and a registered peer's deliveries quietly ignore whatever + // TTL and depth cap the operator configured — the one delivery path on + // the deployment that is exempt from its own backpressure settings. + mailbox: resolveMailboxConfig(c.get('engine').config, workspace.id), }), }); const sent = idempotent.data; @@ -444,18 +463,36 @@ a2aRoutes.post('/a2a/rpc', requireAuth, rateLimit, async (c) => { if (!idempotent.replayed) { await a2aEngine.incrementA2aMessagesReceived(db, registeredCaller.id); + // Fanout and delivery routing run in the background, as `/v1/dm` does. + // Awaiting them made the counterparty's "message accepted" wait on our + // recipient's delivery — including a slow HTTP-push receiver — so a + // sluggish or failing local subscriber could delay or fail an A2A call + // that had already been durably accepted. The write above is what the + // response attests to; everything here is downstream of it. const eventData = buildDmReceivedEventData(sent, { fromName: authenticatedAgent!.name }); - await sendWebhookEvent(c, { - type: 'dm.received', - workspaceId: workspace.id, - data: eventData, - }); - await publishWorkspaceEvent(c, 'dm.received', eventData); + runInBackground( + c, + sendWebhookEvent(c, { type: 'dm.received', workspaceId: workspace.id, data: eventData }), + 'a2a inbound webhook dm.received', + ); + runInBackground( + c, + publishWorkspaceEvent(c, 'dm.received', eventData), + 'a2a inbound publish dm.received', + ); if (sent._delivery) { - await routeDeliveryOutcomes(c, [sent._delivery], 'dm.received', eventData); + runInBackground( + c, + routeDeliveryOutcomes(c, [sent._delivery], 'dm.received', eventData), + 'a2a inbound route dm delivery', + ); } if (sent._delivery_rejections.length > 0) { - await notifyDeliveryRejections(c, authenticatedAgent!.id, sent._delivery_rejections); + runInBackground( + c, + notifyDeliveryRejections(c, authenticatedAgent!.id, sent._delivery_rejections), + 'a2a inbound fanout delivery rejected', + ); } } diff --git a/packages/engine/src/routes/dm.ts b/packages/engine/src/routes/dm.ts index 1462d540..4520e83d 100644 --- a/packages/engine/src/routes/dm.ts +++ b/packages/engine/src/routes/dm.ts @@ -4,6 +4,7 @@ import type { AppEnv } from '../env.js'; import { requireAgentToken } from '../middleware/auth.js'; import { rateLimit } from '../middleware/rateLimit.js'; import { jsonIdempotentOk, parseIdempotencyKey, runIdempotent } from '../middleware/idempotency.js'; +import { sha256Hex } from '../lib/crypto.js'; import * as dmEngine from '../engine/dm.js'; import { resolveMailboxConfig } from '../engine/mailboxConfig.js'; import { publishWorkspaceEvent } from './fanout.js'; @@ -50,11 +51,18 @@ dmRoutes.post( } const { to, text, attachments, data, mode } = parsed.data; const normalizedAttachments = attachments && attachments.length > 0 ? attachments : undefined; + // `data` is digested rather than embedded. It is caller-supplied and can + // be large — a Ratify proof bundle runs to MAX_PROOF_BUNDLE_BYTES (128 + // KiB) — and the fingerprint is serialized into the stored idempotency + // record, kept for the TTL, and string-compared on every replay. Inlining + // it put ~256 KiB per DM into the KV record and made each replay compare + // the whole payload. A digest answers the only question the fingerprint + // asks — "is this the same request?" — in constant size. const fingerprintBody = { to, text, ...(normalizedAttachments ? { attachments: normalizedAttachments } : {}), - ...(data !== undefined ? { data } : {}), + ...(data !== undefined ? { data_sha256: await sha256Hex(JSON.stringify(data)) } : {}), }; const { key: idempotencyKey, error: idempotencyError } = parseIdempotencyKey(c.req.header('Idempotency-Key')); From c14e841d89982c4fa61737e81fb980c24008b5e8 Mon Sep 17 00:00:00 2001 From: Khaliq Date: Mon, 10 Aug 2026 15:22:36 +0200 Subject: [PATCH 6/6] fix: accept unpadded base64 in Ratify hybrid key and signature fields MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cubic P1, and a regression I introduced while hardening validation. Requiring a multiple-of-four length rejected unpadded base64 — legal and commonly emitted — and the value it would have rejected is a signed revocation list. Over-strict validation there fails the kill switch closed over a formatting preference, which is worse than the loose check it replaced. Padding is now optional: a padded value must still be a multiple of four, and an unpadded one is rejected only at length ≡ 1 (mod 4), which no base64 quantum can produce. Tests cover 'QUJDRA' and 'QUJDRA==' (same bytes, both accepted) alongside the malformed cases. turbo test 18/18. Co-Authored-By: Claude Opus 5 --- packages/a2a/src/__tests__/schemas.test.ts | 26 +++++++++++++++------- packages/a2a/src/index.ts | 21 +++++++++++++---- 2 files changed, 35 insertions(+), 12 deletions(-) diff --git a/packages/a2a/src/__tests__/schemas.test.ts b/packages/a2a/src/__tests__/schemas.test.ts index ce6e8480..3d133212 100644 --- a/packages/a2a/src/__tests__/schemas.test.ts +++ b/packages/a2a/src/__tests__/schemas.test.ts @@ -359,19 +359,29 @@ describe('Ratify A2A metadata', () => { // that could never decode, so a malformed key or signature passed A2A // validation at the edge and failed later inside the verifier as an opaque // error, far from the input that caused it. - for (const bad of ['sig-ed', 'not base64', 'AAAA=AAA', 'QUJD']) { - const result = RatifyA2aMetadataSchema.safeParse({ + const parseWithSignature = (ed25519: string) => + RatifyA2aMetadataSchema.safeParse({ ...revocation, - signature: { ...revocation.signature, ed25519: bad }, + signature: { ...revocation.signature, ed25519 }, }); - if (bad === 'QUJD') { - // Valid base64, length a multiple of 4 — must still be accepted. - expect(result.success).toBe(true); - continue; - } + + for (const bad of ['sig-ed', 'not base64', 'AAAA=AAA', 'QUJDR']) { + const result = parseWithSignature(bad); expect(result.success).toBe(false); expect(hasIssueAtPath(result, 'signature', 'ed25519')).toBe(true); } + + // Padding is optional, and this matters more than it looks: a peer that + // emits unpadded base64 sends a perfectly valid signed revocation list, and + // rejecting it would fail the kill switch closed over a formatting + // preference. 'QUJDRA' (6 chars, no padding) and 'QUJDRA==' are the same + // bytes; both must parse. + for (const good of ['QUJD', 'QUJDRA', 'QUJDRA==', 'QUJDRQ=']) { + const result = parseWithSignature(good); + // 'QUJDRQ=' is 7 chars with padding — not a multiple of four — so it is + // the one malformed case in this list. + expect(result.success).toBe(good !== 'QUJDRQ='); + } }); }); diff --git a/packages/a2a/src/index.ts b/packages/a2a/src/index.ts index 6d396c80..bd12344a 100644 --- a/packages/a2a/src/index.ts +++ b/packages/a2a/src/index.ts @@ -13,13 +13,26 @@ export const MAX_PROOF_BUNDLE_BYTES = 128 * 1024; * error. Rejecting at the edge keeps the schema's promise honest and puts the * error where the malformed input entered. */ -const BASE64 = /^[A-Za-z0-9+/]+={0,2}$/; +const BASE64_CHARS = /^[A-Za-z0-9+/]+={0,2}$/; + +/** + * Padding is optional. Requiring a multiple-of-four length would reject a peer + * that emits unpadded base64 — a legal and common encoding — and the value it + * would reject is a signed revocation list, so over-strict validation here + * fails the kill switch closed for a formatting preference. What is genuinely + * invalid is a length ≡ 1 (mod 4), which no base64 quantum can produce, and a + * padded value whose total length is not a multiple of four. + */ +function isBase64(value: string): boolean { + if (!BASE64_CHARS.test(value)) return false; + const padding = value.endsWith('==') ? 2 : value.endsWith('=') ? 1 : 0; + return padding > 0 ? value.length % 4 === 0 : value.length % 4 !== 1; +} + const base64Field = z .string() .min(1) - .refine((v) => BASE64.test(v) && v.length % 4 === 0, { - message: 'expected standard base64', - }); + .refine(isBase64, { message: 'expected base64 (padding optional)' }); const RatifyHybridComponentSchema = z.object({ ed25519: base64Field,