fix(mcp): rejection sniffer must capture res.write chunks, not only res.end

Live verification caught the miss: tool_call and search_zero_hit lines arrived,
tool_call_rejected did not. SDK 1.30 bridges the web-standard Response through
Hono's getRequestListener, which pipes ReadableStream bodies as Uint8Array
chunks via res.write followed by a BARE res.end() — the end(chunk) inspection
never saw the envelope. The sniffer (now attachToolCallRejectionSniffer in
telemetry.ts, unit-testable) wraps both write and end, reassembles up to 10 KB
and decodes string/Uint8Array chunks; tests pin the streaming path, the direct
end(string) path, the over-cap silence and the disabled no-wrap.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/McpServer/autopilot-monitor-mcp/src/__tests__/telemetry.test.ts

@@ -150,3 +150,70 @@ describe('logToolCallRejection', () => {
     expect((line.message as string).length).toBeLessThan(530);
   });
 });
+
+describe('attachToolCallRejectionSniffer', () => {
+  const ERROR_ENVELOPE = JSON.stringify({
+    jsonrpc: '2.0',
+    id: 1,
+    error: { code: -32602, message: 'Input validation error: Invalid arguments for tool get_session: sessionId must be a UUID' },
+  });
+
+  function fakeRes() {
+    const calls: string[] = [];
+    return {
+      calls,
+      write: vi.fn((..._args: never[]) => { calls.push('write'); return true; }),
+      end: vi.fn((..._args: never[]) => { calls.push('end'); return undefined as unknown; }),
+    };
+  }
+
+  it('catches the Hono streaming path: Uint8Array chunks via write, then a bare end()', async () => {
+    const { attachToolCallRejectionSniffer } = await loadTelemetry(true);
+    const spy = vi.spyOn(console, 'error').mockImplementation(() => {});
+    const res = fakeRes();
+    attachToolCallRejectionSniffer('get_session', res);
+
+    const bytes = new TextEncoder().encode(ERROR_ENVELOPE);
+    // Split mid-envelope to prove reassembly across chunks works.
+    res.write(bytes.slice(0, 40) as never);
+    res.write(bytes.slice(40) as never);
+    res.end();
+
+    const line = lastLoggedJson(spy);
+    expect(line.type).toBe('tool_call_rejected');
+    expect(line.tool).toBe('get_session');
+    expect(line.errorCode).toBe(-32602);
+    expect(line.message).toContain('sessionId must be a UUID');
+    // The response itself went through untouched.
+    expect(res.calls).toEqual(['write', 'write', 'end']);
+  });
+
+  it('catches the direct path: whole envelope as a single end(string)', async () => {
+    const { attachToolCallRejectionSniffer } = await loadTelemetry(true);
+    const spy = vi.spyOn(console, 'error').mockImplementation(() => {});
+    const res = fakeRes();
+    attachToolCallRejectionSniffer('query_table', res);
+    res.end(ERROR_ENVELOPE as never);
+    expect(lastLoggedJson(spy).tool).toBe('query_table');
+  });
+
+  it('stays silent for successful results and stops buffering past the cap', async () => {
+    const { attachToolCallRejectionSniffer } = await loadTelemetry(true);
+    const spy = vi.spyOn(console, 'error').mockImplementation(() => {});
+    const res = fakeRes();
+    attachToolCallRejectionSniffer('search_sessions', res);
+    // A large successful tool result (which even CONTAINS the word "error" in data).
+    const big = JSON.stringify({ jsonrpc: '2.0', id: 1, result: { content: [{ type: 'text', text: '"error" '.repeat(3000) }] } });
+    for (let i = 0; i < big.length; i += 1000) res.write(big.slice(i, i + 1000) as never);
+    res.end();
+    expect(spy).not.toHaveBeenCalled();
+  });
+
+  it('does not wrap the response at all when logging is disabled', async () => {
+    const { attachToolCallRejectionSniffer } = await loadTelemetry(false);
+    const res = fakeRes();
+    const originalEnd = res.end;
+    attachToolCallRejectionSniffer('get_session', res);
+    expect(res.end).toBe(originalEnd);
+  });
+});

src/McpServer/autopilot-monitor-mcp/src/index.ts

@@ -18,7 +18,7 @@ import { buildEventTypeSearchDocs } from './resource-catalog.js';
 import { createOAuthRouter } from './oauth.js';
 import { accessGuard } from './access-guard.js';
 import { hasGlobalScope, isGlobalAdmin, isDelegated, getDelegatedTenantIds, getHomeTenantId } from './client.js';
-import { toolLoggingEnabled, logToolCallRejection } from './telemetry.js';
+import { toolLoggingEnabled, attachToolCallRejectionSniffer } from './telemetry.js';
 import { API_BASE_URL } from './config.js';
 
 const __dirname = dirname(fileURLToPath(import.meta.url));
@@ -373,31 +373,15 @@ app.all('/mcp', async (req, res) => {
     return;
   }
 
-  // Sniff JSON-RPC error envelopes on tool calls (Zod argument rejection,
+  // Sniff the JSON-RPC error envelope on tool calls (Zod argument rejection,
   // unknown tool name): the SDK answers these BEFORE any tool handler runs,
-  // so wrapping the response write is the only place they are observable —
-  // and they are the strongest "the schema/description confuses the model"
-  // signal. Stateless mode guarantees one JSON-RPC request per POST and
-  // enableJsonResponse yields a single buffered body, so inspecting the
-  // res.end chunk sees the whole envelope. Error envelopes are tiny; the
-  // length guard skips parsing large (i.e. successful) tool results.
+  // so observing the response write is the only place they are visible — and
+  // they are the strongest "the schema/description confuses the model" signal.
+  // Stateless mode guarantees one JSON-RPC request per POST, so the tool name
+  // from the request body is unambiguous.
   if (toolLoggingEnabled && req.body?.method === 'tools/call') {
     const toolName = typeof req.body?.params?.name === 'string' ? req.body.params.name : 'unknown';
-    const originalEnd = res.end.bind(res);
-    res.end = ((chunk?: unknown, ...rest: unknown[]) => {
-      try {
-        const text = typeof chunk === 'string' ? chunk : Buffer.isBuffer(chunk) ? chunk.toString('utf8') : undefined;
-        if (text && text.length < 10_000 && text.includes('"error"')) {
-          const parsed = JSON.parse(text) as { error?: { code?: number; message?: unknown } };
-          if (parsed?.error?.code !== undefined) {
-            logToolCallRejection(toolName, parsed.error.code, String(parsed.error.message ?? ''));
-          }
-        }
-      } catch {
-        // Sniffing must never break the response.
-      }
-      return originalEnd(chunk as never, ...(rest as never[]));
-    }) as typeof res.end;
+    attachToolCallRejectionSniffer(toolName, res);
   }
 
   const transport = new StreamableHTTPServerTransport({

src/McpServer/autopilot-monitor-mcp/src/telemetry.ts

@@ -142,3 +142,66 @@ export function logToolCallRejection(toolName: string, errorCode: number, messag
     // Telemetry must never break a tool response.
   }
 }
+
+// Error envelopes are tiny; anything larger is a successful tool result and is
+// skipped instead of buffered.
+const SNIFF_BUFFER_CAP = 10_000;
+
+interface SniffableResponse {
+  write: (...args: never[]) => boolean;
+  end: (...args: never[]) => unknown;
+}
+
+/**
+ * Observe the JSON-RPC response of a tools/call POST for an error envelope and
+ * log it via logToolCallRejection. The SDK (1.30+) bridges the web-standard
+ * Response to the Node res through Hono's getRequestListener, which depending
+ * on the body type either calls res.end(string | Uint8Array) directly or pipes
+ * a ReadableStream through res.write chunks followed by a bare res.end() — so
+ * BOTH must be captured. Buffering stops at SNIFF_BUFFER_CAP (real rejections
+ * are a few hundred bytes). Never throws into the response path.
+ */
+export function attachToolCallRejectionSniffer(toolName: string, res: SniffableResponse): void {
+  if (!toolLoggingEnabled) return;
+
+  let captured = '';
+  let overflowed = false;
+  const capture = (chunk: unknown): void => {
+    if (overflowed || chunk == null) return;
+    let text: string | undefined;
+    if (typeof chunk === 'string') text = chunk;
+    // Covers Node Buffers too (Buffer extends Uint8Array).
+    else if (chunk instanceof Uint8Array) text = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString('utf8');
+    if (text === undefined) return;
+    captured += text;
+    if (captured.length > SNIFF_BUFFER_CAP) {
+      overflowed = true;
+      captured = '';
+    }
+  };
+
+  const originalWrite = res.write.bind(res);
+  const originalEnd = res.end.bind(res);
+  res.write = ((chunk: unknown, ...rest: unknown[]) => {
+    try {
+      capture(chunk);
+    } catch {
+      // Sniffing must never break the response.
+    }
+    return originalWrite(chunk as never, ...(rest as never[]));
+  }) as typeof res.write;
+  res.end = ((chunk?: unknown, ...rest: unknown[]) => {
+    try {
+      capture(chunk);
+      if (!overflowed && captured.includes('"error"')) {
+        const parsed = JSON.parse(captured) as { error?: { code?: number; message?: unknown } };
+        if (parsed?.error?.code !== undefined) {
+          logToolCallRejection(toolName, parsed.error.code, String(parsed.error.message ?? ''));
+        }
+      }
+    } catch {
+      // Sniffing must never break the response.
+    }
+    return originalEnd(chunk as never, ...(rest as never[]));
+  }) as typeof res.end;
+}