fix: unify JSON-RPC message validation policy across transports (#212)

The WebSocket and SSE receive paths validated messages with the strict
isJsonRpcMessage guard and silently dropped anything that failed it, so
a lenient peer's response (e.g. error:null alongside result) or request
(e.g. missing the jsonrpc field) never settled the caller's promise —
the same shapes worked over stdio, which only skips non-object lines.

Relax the per-transport guards (ws-stream, ws-server, sse, HTTP server
POST) to an object check and let the connection layer own the
accept/reject policy: it dispatches lenient requests, fast-rejects
structurally invalid responses with RequestError.invalidRequest, and
logs anything else, as of #210. Non-object payloads are still rejected
at the transport with a warning (or HTTP 400) since they carry no id to
answer. Also consolidates ws-stream's private isRecord duplicate.

Fixes #211

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: benbrandt

src/server.ts

@@ -9,7 +9,7 @@ import {
   methodRequiresSessionHeader,
   sessionIdFromParams,
 } from "./protocol.js";
-import { isJsonRpcMessage, isResponseMessage } from "./jsonrpc.js";
+import { isRecord, isResponseMessage } from "./jsonrpc.js";
 import { AGENT_METHODS } from "./schema/index.js";
 import { serializeSseEvent, serializeSseKeepAlive } from "./sse.js";
 import { handleWebSocketConnection } from "./ws-server.js";
@@ -201,15 +201,18 @@ export class AcpServer {
       return textResponse("Batch JSON-RPC requests are not implemented", 501);
     }
 
-    if (!isJsonRpcMessage(body.value)) {
+    // Reject non-object bodies; anything object-shaped is left for the
+    // connection layer to validate.
+    if (!isRecord(body.value)) {
       return textResponse("Invalid JSON-RPC message", 400);
     }
 
+    const message = body.value as AnyMessage;
     const connectionId = req.headers.get(HEADER_CONNECTION_ID);
 
-    if (isInitializeRequest(body.value)) {
+    if (isInitializeRequest(message)) {
       if (!connectionId) {
-        return await this.handleInitialize(body.value, req.signal, options);
+        return await this.handleInitialize(message, req.signal, options);
       }
 
       return textResponse("Initialize not allowed on existing connection", 400);
@@ -227,7 +230,7 @@ export class AcpServer {
 
     const forwarded = await this.forwardConnectedMessage(
       connection,
-      body.value,
+      message,
       req.headers,
     );
     if (!forwarded.ok) {

src/sse.test.ts

@@ -149,6 +149,30 @@ describe("SSE transport helpers", () => {
     ).resolves.toEqual([message]);
   });
 
+  it("passes through object payloads without validating their shape", async () => {
+    // Lenient shapes are left for the connection layer to validate.
+    const lenient = { id: 1, result: { ok: true } }; // missing jsonrpc
+    await expect(
+      collectMessages(
+        streamFromChunks([`data: ${JSON.stringify(lenient)}\n\n`]),
+      ),
+    ).resolves.toEqual([lenient]);
+  });
+
+  it("skips non-object payloads", async () => {
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };
+
+    await expect(
+      collectMessages(
+        streamFromChunks(["data: 42\n\n", serializeSseEvent(message)]),
+      ),
+    ).resolves.toEqual([message]);
+    expect(warn).toHaveBeenCalledOnce();
+
+    warn.mockRestore();
+  });
+
   it("skips malformed JSON without throwing", async () => {
     const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
     const message: AnyMessage = { jsonrpc: "2.0", id: 1, result: { ok: true } };

src/sse.ts

@@ -1,5 +1,5 @@
 import type { AnyMessage } from "./jsonrpc.js";
-import { isJsonRpcMessage } from "./jsonrpc.js";
+import { isRecord } from "./jsonrpc.js";
 import { LineBuffer } from "./line-buffer.js";
 
 export function serializeSseEvent(msg: AnyMessage): string {
@@ -89,11 +89,13 @@ function parseSseEvent(eventLines: string[]): AnyMessage | undefined {
 
   try {
     const parsed: unknown = JSON.parse(data);
-    if (isJsonRpcMessage(parsed)) {
-      return parsed;
+    // Skip non-object payloads with a useful warning; anything object-shaped
+    // is left for the connection layer to validate.
+    if (isRecord(parsed)) {
+      return parsed as AnyMessage;
     }
 
-    console.warn("Skipping SSE payload that is not a JSON-RPC message");
+    console.warn("Skipping SSE payload that is not an object");
     return undefined;
   } catch (error) {
     console.warn("Failed to parse SSE JSON payload:", error);

src/ws-server.ts

@@ -1,8 +1,4 @@
-import {
-  isJsonRpcMessage,
-  isRequestMessage,
-  isResponseMessage,
-} from "./jsonrpc.js";
+import { isRecord, isRequestMessage, isResponseMessage } from "./jsonrpc.js";
 import {
   isInitializeRequest,
   messageIdKey,
@@ -141,18 +137,22 @@ class WebSocketServerSession implements WebSocketServerSessionHandle {
       return;
     }
 
-    if (!isJsonRpcMessage(value)) {
-      console.warn("Ignoring non-JSON-RPC ACP WebSocket message:", value);
+    // Skip non-object messages with a useful warning; anything object-shaped
+    // is left for the connection layer to validate.
+    if (!isRecord(value)) {
+      console.warn("Ignoring non-object ACP WebSocket message:", value);
       await this.shutdownIfUninitialized(1002, "Invalid JSON-RPC message");
       return;
     }
 
+    const message = value as AnyMessage;
+
     if (!this.connection) {
-      await this.handleInitialize(value);
+      await this.handleInitialize(message);
       return;
     }
 
-    const forwarded = await this.forwardMessage(value);
+    const forwarded = await this.forwardMessage(message);
     if (!forwarded.ok) {
       console.warn("Ignoring ACP WebSocket message:", forwarded.message);
     }

src/ws-stream.test.ts

@@ -321,7 +321,7 @@ describe("createWebSocketStream", () => {
     }
   });
 
-  it("ignores binary, malformed JSON, and non-JSON-RPC messages", async () => {
+  it("ignores binary, malformed JSON, and non-object messages, passing other objects through", async () => {
     const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
     const instances: FakeWebSocket[] = [];
     const stream = createWebSocketStream("ws://agent.example/acp", {
@@ -340,9 +340,12 @@ describe("createWebSocketStream", () => {
         new TextEncoder().encode(JSON.stringify(initializeResponse)).buffer,
       );
       socket.receive("not json");
+      socket.receive("42");
+      // Lenient shapes are left for the connection layer to validate.
       socket.receive(JSON.stringify({ hello: "world" }));
       socket.receive(JSON.stringify(initializeResponse));
 
+      expect(await readMessage(reader)).toEqual({ hello: "world" });
       expect(await readMessage(reader)).toEqual(initializeResponse);
       expect(warn).toHaveBeenCalledTimes(2);
     } finally {

src/ws-stream.ts

@@ -1,5 +1,5 @@
 import { MemoryAcpCookieStore } from "./cookie-store.js";
-import { isJsonRpcMessage } from "./jsonrpc.js";
+import { isRecord } from "./jsonrpc.js";
 import { onWebSocket, webSocketMessageToString } from "./ws-utils.js";
 import type { AcpCookieStore } from "./cookie-store.js";
 import type { WebSocketLike } from "./ws-utils.js";
@@ -202,12 +202,14 @@ class WebSocketStreamTransport {
       return;
     }
 
-    if (!isJsonRpcMessage(value)) {
-      console.warn("Ignoring non-JSON-RPC ACP WebSocket message:", value);
+    // Skip non-object messages with a useful warning; anything object-shaped
+    // is left for the connection layer to validate.
+    if (!isRecord(value)) {
+      console.warn("Ignoring non-object ACP WebSocket message:", value);
       return;
     }
 
-    this.readableController?.enqueue(value);
+    this.readableController?.enqueue(value as AnyMessage);
   }
 
   private close(): void {
@@ -342,10 +344,6 @@ function headersFromRecord(value: unknown): Headers | undefined {
   return headers;
 }
 
-function isRecord(value: unknown): value is Record<string, unknown> {
-  return typeof value === "object" && value !== null;
-}
-
 function resolveWebSocket(
   WebSocketCtor: WebSocketConstructor | undefined,
 ): WebSocketConstructor {