From 2551175aba07ef365f116af271dda73be3d626f3 Mon Sep 17 00:00:00 2001 From: Amaad Martin Date: Thu, 6 Aug 2026 05:31:56 -0700 Subject: [PATCH 1/3] Fix: forward async route handler rejections in the dev API server Express 4 discards the promise an async route handler returns, so a rejection outside the handler's own try/catch left the request hanging and raised a process-level unhandledRejection, which kills the dev server under node's default mode. Wrap all 15 async handlers in asyncHandler, replace the floating async 'end' listener on /api/reasoning_engine with an awaited readRawBody, move the app.listen callback body onto a private method, and register a terminal error middleware that answers with the {error} JSON shape the routes already use. --- dev/src/server/adk_api_server.ts | 408 ++++++++++++++++--------------- dev/src/utils/express_utils.ts | 98 ++++++++ 2 files changed, 312 insertions(+), 194 deletions(-) create mode 100644 dev/src/utils/express_utils.ts diff --git a/dev/src/server/adk_api_server.ts b/dev/src/server/adk_api_server.ts index a52ad63ab..3c71d8d6e 100644 --- a/dev/src/server/adk_api_server.ts +++ b/dev/src/server/adk_api_server.ts @@ -34,6 +34,11 @@ import * as http from 'node:http'; import * as path from 'node:path'; import {AgentFileOptions, AgentLoader} from '../utils/agent_loader.js'; +import { + asyncHandler, + errorHandler, + readRawBody, +} from '../utils/express_utils.js'; import {AdkLogger} from '../utils/logger.js'; import { ApiServerSpanExporter, @@ -241,19 +246,22 @@ export class AdkApiServer { next(); }); - app.get('/list-apps', async (req: Request, res: Response) => { - try { - const apps = await this.agentLoader.listAgents(); - res.json(apps); - } catch (e: unknown) { - const error = `Failed to list apps: ${e}`; + app.get( + '/list-apps', + asyncHandler(async (req: Request, res: Response) => { + try { + const apps = await this.agentLoader.listAgents(); + res.json(apps); + } catch (e: unknown) { + const error = `Failed to list apps: ${e}`; - res.status(500).json({error}); - this.logger.error(error); + res.status(500).json({error}); + this.logger.error(error); - return; - } - }); + return; + } + }), + ); app.get('/debug/trace/:eventId', (req: Request, res: Response) => { try { @@ -308,7 +316,7 @@ export class AdkApiServer { app.get( '/apps/:appName/users/:userId/sessions/:sessionId/events/:eventId/graph', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -382,13 +390,13 @@ export class AdkApiServer { this.logger.error(error); return; } - }, + }), ); // ------------------------- Session related endpoints --------------------- app.get( '/apps/:appName/users/:userId/sessions/:sessionId', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -412,12 +420,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.get( '/apps/:appName/users/:userId/sessions', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -434,12 +442,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.post( '/apps/:appName/users/:userId/sessions/:sessionId', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -473,12 +481,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.post( '/apps/:appName/users/:userId/sessions', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -497,12 +505,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.delete( '/apps/:appName/users/:userId/sessions/:sessionId', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -532,13 +540,13 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); // ----------------------- Artifact related endpoints ---------------------- app.get( '/apps/:appName/users/:userId/sessions/:sessionId/artifacts/:artifactName', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -566,12 +574,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.get( '/apps/:appName/users/:userId/sessions/:sessionId/artifacts/:artifactName/versions/:versionId', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -601,12 +609,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.get( '/apps/:appName/users/:userId/sessions/:sessionId/artifacts', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -625,12 +633,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.get( '/apps/:appName/users/:userId/sessions/:sessionId/artifacts/:artifactName/versions', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -651,12 +659,12 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); app.delete( '/apps/:appName/users/:userId/sessions/:sessionId/artifacts/:artifactName', - async (req: Request, res: Response) => { + asyncHandler(async (req: Request, res: Response) => { try { const appName = req.params['appName']; const userId = req.params['userId']; @@ -677,7 +685,7 @@ export class AdkApiServer { res.status(500).json({error}); this.logger.error(error); } - }, + }), ); // --------------------- Eval Sets related endpoints ----------------------- @@ -753,84 +761,35 @@ export class AdkApiServer { }); // -------------------------- Run related endpoints ------------------------ - app.post('/run', async (req: Request, res: Response) => { - const {appName, userId, sessionId, newMessage, stateDelta} = req.body; - const session = await this.sessionService.getSession({ - appName, - userId, - sessionId, - }); - - if (!session) { - res.status(404).json({error: `Session not found: ${sessionId}`}); - return; - } - - const abortController = new AbortController(); - let responseCompleted = false; - - req.on('close', () => { - if (!responseCompleted) { - this.logger.info( - `HTTP connection closed. Aborting agent execution for session ${sessionId}`, - ); - abortController.abort(); - } - }); - - try { - const events: Event[] = []; - for await (const e of this.executeAgentRun({ + app.post( + '/run', + asyncHandler(async (req: Request, res: Response) => { + const {appName, userId, sessionId, newMessage, stateDelta} = req.body; + const session = await this.sessionService.getSession({ appName, userId, sessionId, - newMessage, - stateDelta, - abortSignal: abortController.signal, - })) { - events.push(e); - } + }); - responseCompleted = true; - res.json(events); - } catch (e: unknown) { - const error = `Failed to run agent: ${e}`; + if (!session) { + res.status(404).json({error: `Session not found: ${sessionId}`}); + return; + } - res.status(500).json({error}); - this.logger.error(error); - } - }); + const abortController = new AbortController(); + let responseCompleted = false; - app.post('/api/reasoning_engine', async (req: Request, res: Response) => { - this.logger.info( - `Received Reasoning Engine query headers: ${JSON.stringify(req.headers)}`, - ); + req.on('close', () => { + if (!responseCompleted) { + this.logger.info( + `HTTP connection closed. Aborting agent execution for session ${sessionId}`, + ); + abortController.abort(); + } + }); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - const executeQuery = async (body: any) => { - const input = body.input || {}; - const appName = input.appName || body.appName; - const userId = input.userId || body.userId || 'default-user'; - const sessionId = - input.sessionId || body.sessionId || 'default-session'; - const newMessage = input.newMessage || body.newMessage; - const stateDelta = input.stateDelta || body.stateDelta; - if (!appName) { - res.status(400).json({error: 'appName is required in input'}); - return; - } try { - await this.sessionService.getOrCreateSession({ - appName, - userId, - sessionId, - state: {}, - }); const events: Event[] = []; - const abortController = new AbortController(); - req.on('close', () => { - abortController.abort(); - }); for await (const e of this.executeAgentRun({ appName, userId, @@ -841,30 +800,79 @@ export class AdkApiServer { })) { events.push(e); } - res.json({output: events}); + + responseCompleted = true; + res.json(events); } catch (e: unknown) { - const error = `Failed to run agent via Reasoning Engine API: ${e}`; + const error = `Failed to run agent: ${e}`; + res.status(500).json({error}); this.logger.error(error); } - }; + }), + ); - const isParsed = - req.body && (Object.keys(req.body).length > 0 || !req.readable); - if (isParsed) { + app.post( + '/api/reasoning_engine', + asyncHandler(async (req: Request, res: Response) => { this.logger.info( - `Using already parsed body: ${JSON.stringify(req.body)}`, + `Received Reasoning Engine query headers: ${JSON.stringify(req.headers)}`, ); - await executeQuery(req.body); - } else { - let rawBody = ''; - req.on('data', (chunk) => { - rawBody += chunk; - }); - req.on('end', async () => { + + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const executeQuery = async (body: any) => { + const input = body.input || {}; + const appName = input.appName || body.appName; + const userId = input.userId || body.userId || 'default-user'; + const sessionId = + input.sessionId || body.sessionId || 'default-session'; + const newMessage = input.newMessage || body.newMessage; + const stateDelta = input.stateDelta || body.stateDelta; + if (!appName) { + res.status(400).json({error: 'appName is required in input'}); + return; + } + try { + await this.sessionService.getOrCreateSession({ + appName, + userId, + sessionId, + state: {}, + }); + const events: Event[] = []; + const abortController = new AbortController(); + req.on('close', () => { + abortController.abort(); + }); + for await (const e of this.executeAgentRun({ + appName, + userId, + sessionId, + newMessage, + stateDelta, + abortSignal: abortController.signal, + })) { + events.push(e); + } + res.json({output: events}); + } catch (e: unknown) { + const error = `Failed to run agent via Reasoning Engine API: ${e}`; + res.status(500).json({error}); + this.logger.error(error); + } + }; + + const isParsed = + req.body && (Object.keys(req.body).length > 0 || !req.readable); + if (isParsed) { + this.logger.info( + `Using already parsed body: ${JSON.stringify(req.body)}`, + ); + await executeQuery(req.body); + } else { + const rawBody = await readRawBody(req); this.logger.info(`Received Reasoning Engine raw body: ${rawBody}`); - // eslint-disable-next-line @typescript-eslint/no-explicit-any - let body: any = {}; + let body: unknown = {}; if (rawBody) { try { body = JSON.parse(rawBody); @@ -873,104 +881,116 @@ export class AdkApiServer { } } await executeQuery(body); - }); - } - }); - - app.post('/run_sse', async (req: Request, res: Response) => { - const {appName, userId, sessionId, newMessage, streaming, stateDelta} = - req.body; - - const session = await this.sessionService.getSession({ - appName, - userId, - sessionId, - }); - - if (!session) { - const error = `Session not found: ${sessionId}`; - - res.status(404).json({error}); - this.logger.error(error); - return; - } - - const abortController = new AbortController(); - let responseCompleted = false; - - req.on('close', () => { - if (!responseCompleted) { - this.logger.info( - `HTTP connection closed. Aborting agent SSE execution for session ${sessionId}`, - ); - abortController.abort(); } - }); + }), + ); - try { - res.setHeader('Cache-Control', 'no-cache'); - res.setHeader('Content-Type', 'text/event-stream'); - res.setHeader('Connection', 'keep-alive'); - res.flushHeaders(); + app.post( + '/run_sse', + asyncHandler(async (req: Request, res: Response) => { + const {appName, userId, sessionId, newMessage, streaming, stateDelta} = + req.body; - for await (const event of this.executeAgentRun({ + const session = await this.sessionService.getSession({ appName, userId, sessionId, - newMessage, - stateDelta, - runConfig: { - streamingMode: streaming ? StreamingMode.SSE : StreamingMode.NONE, - }, - abortSignal: abortController.signal, - })) { - res.write(`data: ${JSON.stringify(event)}\n\n`); - } + }); - responseCompleted = true; - res.end(); - } catch (e: unknown) { - if (res.headersSent) { - if (!responseCompleted) { - const error = (e as Error).message; - this.logger.error(error); - try { - res.end(`data: ${JSON.stringify({error})}\n\n`); - } catch { - // Ignore errors from res.end when the response has already been sent. - } - } - } else { - const error = `Failed to run agent: ${e}`; + if (!session) { + const error = `Session not found: ${sessionId}`; - res.status(500).json({error}); + res.status(404).json({error}); this.logger.error(error); + return; } - } - }); - } - async start(): Promise { - await this.init(); + const abortController = new AbortController(); + let responseCompleted = false; + + req.on('close', () => { + if (!responseCompleted) { + this.logger.info( + `HTTP connection closed. Aborting agent SSE execution for session ${sessionId}`, + ); + abortController.abort(); + } + }); - return new Promise((resolve, reject) => { - this.server = this.app.listen(this.port, this.host, async () => { try { - if (this.a2a) { - await this.initA2A(); + res.setHeader('Cache-Control', 'no-cache'); + res.setHeader('Content-Type', 'text/event-stream'); + res.setHeader('Connection', 'keep-alive'); + res.flushHeaders(); + + for await (const event of this.executeAgentRun({ + appName, + userId, + sessionId, + newMessage, + stateDelta, + runConfig: { + streamingMode: streaming ? StreamingMode.SSE : StreamingMode.NONE, + }, + abortSignal: abortController.signal, + })) { + res.write(`data: ${JSON.stringify(event)}\n\n`); } - console.log(` + responseCompleted = true; + res.end(); + } catch (e: unknown) { + if (res.headersSent) { + if (!responseCompleted) { + const error = (e as Error).message; + this.logger.error(error); + try { + res.end(`data: ${JSON.stringify({error})}\n\n`); + } catch { + // Ignore errors from res.end when the response has already been sent. + } + } + } else { + const error = `Failed to run agent: ${e}`; + + res.status(500).json({error}); + this.logger.error(error); + } + } + }), + ); + + // Registered last: express only looks for an error handler in the layers + // that follow the one that failed. + app.use(errorHandler(this.logger)); + } + + /** + * Finishes startup once the socket is listening. Kept off the `listen` + * callback itself, which express expects to return `void`. + */ + private async onListening(): Promise { + if (this.a2a) { + await this.initA2A(); + } + + console.log(` +-----------------------------------------------------------------------------+ | ADK API Server started | | | | For local testing, access at ${this.url}.${''.padStart(39 - this.url.length)} | +-----------------------------------------------------------------------------+`); - resolve(); - } catch (error) { + } + + async start(): Promise { + await this.init(); + + return new Promise((resolve, reject) => { + this.server = this.app.listen(this.port, this.host, () => { + void this.onListening().then(resolve, (error: unknown) => { this.logger.error('Error during AdkApiServer startup:', error); reject(error); - } + }); }); this.server.on('error', (err: unknown) => { diff --git a/dev/src/utils/express_utils.ts b/dev/src/utils/express_utils.ts new file mode 100644 index 000000000..eff002df2 --- /dev/null +++ b/dev/src/utils/express_utils.ts @@ -0,0 +1,98 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {Logger} from '@google/adk'; +import { + ErrorRequestHandler, + NextFunction, + Request, + RequestHandler, + Response, +} from 'express'; +import {Readable} from 'node:stream'; + +/** Status used when an error carries no usable HTTP status of its own. */ +const INTERNAL_SERVER_ERROR = 500; + +/** + * An express handler that returns a promise. The resolved value is discarded, + * so a handler that ends in `return res.json(...)` fits this type without a + * cast. + */ +export type AsyncRequestHandler = ( + req: Request, + res: Response, + next: NextFunction, +) => Promise; + +/** + * Adapts a promise-returning handler to express 4's void-returning + * `RequestHandler`. + * + * Express 4 ignores the promise a handler returns, so a rejection reaches + * nobody: the client waits for a response that never arrives and node raises + * `unhandledRejection`, which terminates the process under node's default + * mode. Forwarding the rejection to `next` sends it to the error middleware + * instead. + */ +export function asyncHandler(handler: AsyncRequestHandler): RequestHandler { + return (req, res, next) => { + void handler(req, res, next).catch(next); + }; +} + +/** + * Buffers a request stream into a string. Unlike an `end` listener, this + * observes a stream `error` and rejects. + */ +export function readRawBody(stream: Readable): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + stream.on('data', (chunk: Buffer | string) => { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + }); + stream.on('end', () => { + resolve(Buffer.concat(chunks).toString('utf8')); + }); + stream.on('error', reject); + }); +} + +/** + * Reads the HTTP status an error carries. Express middleware such as the JSON + * body parser sets `status` on the errors it raises; dropping it would report + * a malformed or oversized request body as a server fault. + */ +export function errorStatus(err: unknown): number { + const status = + typeof err === 'object' && err !== null && 'status' in err + ? err.status + : undefined; + + return typeof status === 'number' && status >= 400 && status < 600 + ? status + : INTERNAL_SERVER_ERROR; +} + +/** + * Builds the terminal express error middleware. It answers with the `{error}` + * JSON shape that every route in the dev API server already uses, unless the + * response has started, in which case only express can tear the connection + * down cleanly. + */ +export function errorHandler(logger: Logger): ErrorRequestHandler { + return (err: unknown, req: Request, res: Response, next: NextFunction) => { + const error = `Failed to handle ${req.method} ${req.originalUrl}: ${err}`; + logger.error(error); + + if (res.headersSent) { + next(err); + return; + } + + res.status(errorStatus(err)).json({error}); + }; +} From ef7e7205a3a9d5926f20219f5e25ac32a1adc552 Mon Sep 17 00:00:00 2001 From: Amaad Martin Date: Thu, 6 Aug 2026 05:42:52 -0700 Subject: [PATCH 2/3] Test: pin the dev API server's async handler error paths Cover asyncHandler, readRawBody, errorStatus and errorHandler in a new express_utils test that drives a real express app, and add a server describe proving /run, /run_sse and the Reasoning Engine raw-body path answer 500 without leaking an unhandled rejection. --- dev/test/server/adk_api_server_test.ts | 182 +++++++++++++++++ dev/test/utils/express_utils_test.ts | 265 +++++++++++++++++++++++++ 2 files changed, 447 insertions(+) create mode 100644 dev/test/utils/express_utils_test.ts diff --git a/dev/test/server/adk_api_server_test.ts b/dev/test/server/adk_api_server_test.ts index d6db712d2..97e55ace3 100644 --- a/dev/test/server/adk_api_server_test.ts +++ b/dev/test/server/adk_api_server_test.ts @@ -18,6 +18,8 @@ import { InMemorySessionService, InvocationContext, LlmAgent, + Logger, + LogLevel, Runner, Session, } from '@google/adk'; @@ -132,6 +134,56 @@ class HttpClient { } } +/** Time given to node to classify a dropped promise as unhandled. */ +const REJECTION_SETTLE_MS = 50; + +/** + * Captures process-level unhandled rejections. A promise express drops is + * observable nowhere else, so this tells "the request was answered" apart from + * "the request was answered and nothing leaked". + */ +function trackUnhandledRejections(): { + settled: () => Promise; + stop: () => void; +} { + const seen: unknown[] = []; + const listener = (reason: unknown) => { + seen.push(reason); + }; + process.on('unhandledRejection', listener); + + return { + async settled(): Promise { + await new Promise((resolve) => setTimeout(resolve, REJECTION_SETTLE_MS)); + return seen; + }, + stop(): void { + process.off('unhandledRejection', listener); + }, + }; +} + +/** + * A logger that throws on one message. It injects a failure into the Reasoning + * Engine raw-body branch, whose other failure modes the inner try/catch + * already answers. + */ +class ExplodingLogger implements Logger { + constructor(private readonly failOn: string) {} + + setLogLevel(_level: LogLevel): void {} + log(_level: LogLevel, ..._args: unknown[]): void {} + debug(..._args: unknown[]): void {} + warn(..._args: unknown[]): void {} + error(..._args: unknown[]): void {} + + info(...args: unknown[]): void { + if (args.join(' ').includes(this.failOn)) { + throw new Error('logger exploded'); + } + } +} + class TestAgent extends LlmAgent { async *runAsyncImpl( context: InvocationContext, @@ -1188,6 +1240,19 @@ describe('AdkWebServer', () => { agentLoader.getAgentFile = originalGetAgentFile; } }); + + it('should fall back to an empty raw body that will not parse', async () => { + const response = await fetch(`${server.url}/api/reasoning_engine`, { + method: 'POST', + headers: {'Content-Type': 'application/json,application/json'}, + body: '{not json', + }); + + expect(response.status).toBe(400); + expect(((await response.json()) as {error: string}).error).toContain( + 'appName is required', + ); + }); }); describe('Startup', () => { @@ -1214,6 +1279,27 @@ describe('AdkWebServer', () => { } }); + it('should reject when mounting the A2A surface fails', async () => { + const originalListAgents = agentLoader.listAgents; + agentLoader.listAgents = () => + Promise.reject(new Error('agent scan failed')); + + const a2aServer = new AdkApiServer({ + agentLoader, + sessionService, + memoryService, + artifactService, + a2a: true, + }); + + try { + await expect(a2aServer.start()).rejects.toThrow('agent scan failed'); + } finally { + agentLoader.listAgents = originalListAgents; + await a2aServer.stop(); + } + }); + it('should default to listening on localhost', async () => { const defaultServer = new AdkApiServer({ agentLoader, @@ -1255,4 +1341,100 @@ describe('AdkWebServer', () => { } }); }); + + describe('Async handler failures', () => { + const newMessage = {parts: [{text: 'Hello'}], role: 'user'}; + + it('should answer 500 when the run_sse session lookup rejects', async () => { + const getSession = vi + .spyOn(sessionService, 'getSession') + .mockRejectedValue(new Error('session store offline')); + const rejections = trackUnhandledRejections(); + + try { + const response = await fetch(`${server.url}/run_sse`, { + method: 'POST', + headers: {'Content-Type': 'application/json'}, + body: JSON.stringify({ + appName: 'testApp', + userId: 'testUser', + sessionId: 'sessionId', + newMessage, + }), + }); + + expect(response.status).toBe(500); + expect(((await response.json()) as {error: string}).error).toContain( + 'session store offline', + ); + expect(await rejections.settled()).toEqual([]); + } finally { + rejections.stop(); + getSession.mockRestore(); + } + }); + + it('should answer 500 when the run session lookup rejects', async () => { + const getSession = vi + .spyOn(sessionService, 'getSession') + .mockRejectedValue(new Error('session store offline')); + const rejections = trackUnhandledRejections(); + + try { + const response = await fetch(`${server.url}/run`, { + method: 'POST', + headers: {'Content-Type': 'application/json'}, + body: JSON.stringify({ + appName: 'testApp', + userId: 'testUser', + sessionId: 'sessionId', + newMessage, + }), + }); + + expect(response.status).toBe(500); + expect(((await response.json()) as {error: string}).error).toContain( + 'session store offline', + ); + expect(await rejections.settled()).toEqual([]); + } finally { + rejections.stop(); + getSession.mockRestore(); + } + }); + + it('should answer 500 when the reasoning_engine raw body path throws', async () => { + const failingServer = new AdkApiServer({ + agentLoader, + sessionService, + memoryService, + artifactService, + logger: new ExplodingLogger('Received Reasoning Engine raw body'), + }); + await failingServer.start(); + const rejections = trackUnhandledRejections(); + + try { + const response = await fetch( + `${failingServer.url}/api/reasoning_engine`, + { + method: 'POST', + // A content type express's JSON parser does not recognise, which + // is what selects the raw-body branch. + headers: {'Content-Type': 'application/json,application/json'}, + body: JSON.stringify({input: {appName: 'testApp'}}), + }, + ); + + expect(response.status).toBe(500); + expect(((await response.json()) as {error: string}).error).toContain( + 'logger exploded', + ); + expect(await rejections.settled()).toEqual([]); + } finally { + rejections.stop(); + await failingServer.stop(); + } + }); + }); }); diff --git a/dev/test/utils/express_utils_test.ts b/dev/test/utils/express_utils_test.ts new file mode 100644 index 000000000..b142129fb --- /dev/null +++ b/dev/test/utils/express_utils_test.ts @@ -0,0 +1,265 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import {Logger, LogLevel} from '@google/adk'; +import express, {NextFunction, Request, Response} from 'express'; +import {once} from 'node:events'; +import {Readable} from 'node:stream'; +import {describe, expect, it} from 'vitest'; + +import { + asyncHandler, + errorHandler, + errorStatus, + readRawBody, +} from '../../src/utils/express_utils.js'; + +/** Records what the error middleware logs. */ +class RecordingLogger implements Logger { + readonly errors: string[] = []; + + setLogLevel(_level: LogLevel): void {} + log(_level: LogLevel, ..._args: unknown[]): void {} + debug(..._args: unknown[]): void {} + info(..._args: unknown[]): void {} + warn(..._args: unknown[]): void {} + error(...args: unknown[]): void { + this.errors.push(args.join(' ')); + } +} + +/** + * Runs `use` against a throwaway express app on an ephemeral port. The + * utilities under test only mean anything inside express's own dispatch, so + * the tests drive real requests rather than hand-built request objects. + */ +async function withApp( + configure: (app: express.Express) => void, + use: (baseUrl: string) => Promise, +): Promise { + const app = express(); + configure(app); + + const server = app.listen(0, '127.0.0.1'); + await once(server, 'listening'); + const address = server.address(); + if (address === null || typeof address === 'string') { + expect.fail('expected the test server to be listening on a TCP port'); + } + + try { + await use(`http://127.0.0.1:${address.port}`); + } finally { + server.closeAllConnections(); + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + } +} + +describe('asyncHandler', () => { + it('sends a rejection to the error middleware', async () => { + const logger = new RecordingLogger(); + + await withApp( + (app) => { + app.get( + '/boom', + asyncHandler(() => Promise.reject(new Error('handler exploded'))), + ); + app.use(errorHandler(logger)); + }, + async (baseUrl) => { + const response = await fetch(`${baseUrl}/boom`); + + expect(response.status).toBe(500); + expect(((await response.json()) as {error: string}).error).toContain( + 'handler exploded', + ); + }, + ); + + expect(logger.errors).toHaveLength(1); + }); + + it('leaves a resolving handler alone', async () => { + const logger = new RecordingLogger(); + + await withApp( + (app) => { + app.get( + '/ok', + // The `return res.json(...)` shape twelve dev-server routes use: it + // resolves to a `Response`, not to `void`. + asyncHandler((req, res) => Promise.resolve(res.json({ok: true}))), + ); + app.use(errorHandler(logger)); + }, + async (baseUrl) => { + const response = await fetch(`${baseUrl}/ok`); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ok: true}); + }, + ); + + expect(logger.errors).toHaveLength(0); + }); + + it('returns undefined so express sees a void-returning handler', async () => { + let returnValue: unknown = 'not called'; + + await withApp( + (app) => { + app.get('/void-contract', (req, res, next) => { + returnValue = asyncHandler(() => Promise.resolve('ignored'))( + req, + res, + next, + ); + res.json({ok: true}); + }); + }, + async (baseUrl) => { + const response = await fetch(`${baseUrl}/void-contract`); + expect(response.status).toBe(200); + }, + ); + + expect(returnValue).toBeUndefined(); + }); +}); + +describe('readRawBody', () => { + it('joins string chunks', async () => { + await expect(readRawBody(Readable.from(['{"a":', '1}']))).resolves.toBe( + '{"a":1}', + ); + }); + + it('decodes a multi-byte character split across buffer chunks', async () => { + const encoded = Buffer.from('é', 'utf8'); + + await expect( + readRawBody(Readable.from([encoded.subarray(0, 1), encoded.subarray(1)])), + ).resolves.toBe('é'); + }); + + it('resolves to an empty string for an empty stream', async () => { + await expect(readRawBody(Readable.from([]))).resolves.toBe(''); + }); + + it('rejects when the stream errors', async () => { + const stream = new Readable({ + read() { + this.destroy(new Error('socket died')); + }, + }); + + await expect(readRawBody(stream)).rejects.toThrow('socket died'); + }); +}); + +describe('errorStatus', () => { + it('defaults to 500', () => { + expect(errorStatus(new Error('boom'))).toBe(500); + }); + + it('defaults to 500 for a non-object error', () => { + expect(errorStatus('boom')).toBe(500); + }); + + it('defaults to 500 for null', () => { + expect(errorStatus(null)).toBe(500); + }); + + it('defaults to 500 when status is not a number', () => { + expect(errorStatus(Object.assign(new Error('boom'), {status: '400'}))).toBe( + 500, + ); + }); + + it('defaults to 500 for a status outside the error range', () => { + expect(errorStatus(Object.assign(new Error('boom'), {status: 200}))).toBe( + 500, + ); + expect(errorStatus(Object.assign(new Error('boom'), {status: 600}))).toBe( + 500, + ); + }); + + it('keeps a client error status set by middleware', () => { + expect(errorStatus(Object.assign(new Error('boom'), {status: 400}))).toBe( + 400, + ); + }); +}); + +describe('errorHandler', () => { + it('declares four parameters so express treats it as error middleware', () => { + expect(errorHandler(new RecordingLogger()).length).toBe(4); + }); + + it('keeps the status an express middleware error carries', async () => { + await withApp( + (app) => { + app.use(express.json()); + app.post('/echo', (req, res) => { + res.json(req.body); + }); + app.use(errorHandler(new RecordingLogger())); + }, + async (baseUrl) => { + const response = await fetch(`${baseUrl}/echo`, { + method: 'POST', + headers: {'Content-Type': 'application/json'}, + body: '{not json', + }); + + expect(response.status).toBe(400); + expect(((await response.json()) as {error: string}).error).toContain( + 'POST /echo', + ); + }, + ); + }); + + it('delegates the original error once the response has started', async () => { + const logger = new RecordingLogger(); + const failure = new Error('too late'); + let delegated: unknown = 'not called'; + + await withApp( + (app) => { + app.get( + '/late-failure', + asyncHandler((req, res) => { + res.write('partial'); + return Promise.reject(failure); + }), + ); + app.use(errorHandler(logger)); + app.use( + (err: unknown, req: Request, res: Response, next: NextFunction) => { + delegated = err; + next(err); + }, + ); + }, + async (baseUrl) => { + const response = await fetch(`${baseUrl}/late-failure`); + + // The status line was already flushed, so the error cannot change it; + // express destroys the socket instead, truncating the body. + expect(response.status).toBe(200); + await expect(response.text()).rejects.toThrow(); + }, + ); + + expect(delegated).toBe(failure); + expect(logger.errors[0]).toContain('too late'); + }); +}); From 9b5ad613c9f12a377338cb140c1c1cbb30082bab Mon Sep 17 00:00:00 2001 From: Amaad Martin Date: Thu, 6 Aug 2026 06:21:59 -0700 Subject: [PATCH 3/3] Refactor: use node's stream consumer and flatten server startup Replace the hand-rolled readRawBody with text() from node:stream/consumers, which node has shipped since 16.7 and which handles the same three edge cases the helper did. Move the A2A mount and the startup banner out of the app.listen callback into start()'s own control flow, so the callback stays void-returning without needing a separate private method. --- dev/src/server/adk_api_server.ts | 49 ++++++++++++---------------- dev/src/utils/express_utils.ts | 18 ---------- dev/test/utils/express_utils_test.ts | 32 ------------------ 3 files changed, 21 insertions(+), 78 deletions(-) diff --git a/dev/src/server/adk_api_server.ts b/dev/src/server/adk_api_server.ts index 3c71d8d6e..0278d2e00 100644 --- a/dev/src/server/adk_api_server.ts +++ b/dev/src/server/adk_api_server.ts @@ -32,13 +32,10 @@ import cors from 'cors'; import express, {Request, Response} from 'express'; import * as http from 'node:http'; import * as path from 'node:path'; +import {text} from 'node:stream/consumers'; import {AgentFileOptions, AgentLoader} from '../utils/agent_loader.js'; -import { - asyncHandler, - errorHandler, - readRawBody, -} from '../utils/express_utils.js'; +import {asyncHandler, errorHandler} from '../utils/express_utils.js'; import {AdkLogger} from '../utils/logger.js'; import { ApiServerSpanExporter, @@ -870,7 +867,7 @@ export class AdkApiServer { ); await executeQuery(req.body); } else { - const rawBody = await readRawBody(req); + const rawBody = await text(req); this.logger.info(`Received Reasoning Engine raw body: ${rawBody}`); let body: unknown = {}; if (rawBody) { @@ -965,32 +962,12 @@ export class AdkApiServer { app.use(errorHandler(this.logger)); } - /** - * Finishes startup once the socket is listening. Kept off the `listen` - * callback itself, which express expects to return `void`. - */ - private async onListening(): Promise { - if (this.a2a) { - await this.initA2A(); - } - - console.log(` -+-----------------------------------------------------------------------------+ -| ADK API Server started | -| | -| For local testing, access at ${this.url}.${''.padStart(39 - this.url.length)} | -+-----------------------------------------------------------------------------+`); - } - async start(): Promise { await this.init(); - return new Promise((resolve, reject) => { + await new Promise((resolve, reject) => { this.server = this.app.listen(this.port, this.host, () => { - void this.onListening().then(resolve, (error: unknown) => { - this.logger.error('Error during AdkApiServer startup:', error); - reject(error); - }); + resolve(); }); this.server.on('error', (err: unknown) => { @@ -1004,6 +981,22 @@ export class AdkApiServer { } }); }); + + try { + if (this.a2a) { + await this.initA2A(); + } + } catch (error) { + this.logger.error('Error during AdkApiServer startup:', error); + throw error; + } + + console.log(` ++-----------------------------------------------------------------------------+ +| ADK API Server started | +| | +| For local testing, access at ${this.url}.${''.padStart(39 - this.url.length)} | ++-----------------------------------------------------------------------------+`); } stop(): Promise { diff --git a/dev/src/utils/express_utils.ts b/dev/src/utils/express_utils.ts index eff002df2..1704aedd7 100644 --- a/dev/src/utils/express_utils.ts +++ b/dev/src/utils/express_utils.ts @@ -12,7 +12,6 @@ import { RequestHandler, Response, } from 'express'; -import {Readable} from 'node:stream'; /** Status used when an error carries no usable HTTP status of its own. */ const INTERNAL_SERVER_ERROR = 500; @@ -44,23 +43,6 @@ export function asyncHandler(handler: AsyncRequestHandler): RequestHandler { }; } -/** - * Buffers a request stream into a string. Unlike an `end` listener, this - * observes a stream `error` and rejects. - */ -export function readRawBody(stream: Readable): Promise { - return new Promise((resolve, reject) => { - const chunks: Buffer[] = []; - stream.on('data', (chunk: Buffer | string) => { - chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); - }); - stream.on('end', () => { - resolve(Buffer.concat(chunks).toString('utf8')); - }); - stream.on('error', reject); - }); -} - /** * Reads the HTTP status an error carries. Express middleware such as the JSON * body parser sets `status` on the errors it raises; dropping it would report diff --git a/dev/test/utils/express_utils_test.ts b/dev/test/utils/express_utils_test.ts index b142129fb..34e1af2ea 100644 --- a/dev/test/utils/express_utils_test.ts +++ b/dev/test/utils/express_utils_test.ts @@ -7,14 +7,12 @@ import {Logger, LogLevel} from '@google/adk'; import express, {NextFunction, Request, Response} from 'express'; import {once} from 'node:events'; -import {Readable} from 'node:stream'; import {describe, expect, it} from 'vitest'; import { asyncHandler, errorHandler, errorStatus, - readRawBody, } from '../../src/utils/express_utils.js'; /** Records what the error middleware logs. */ @@ -133,36 +131,6 @@ describe('asyncHandler', () => { }); }); -describe('readRawBody', () => { - it('joins string chunks', async () => { - await expect(readRawBody(Readable.from(['{"a":', '1}']))).resolves.toBe( - '{"a":1}', - ); - }); - - it('decodes a multi-byte character split across buffer chunks', async () => { - const encoded = Buffer.from('é', 'utf8'); - - await expect( - readRawBody(Readable.from([encoded.subarray(0, 1), encoded.subarray(1)])), - ).resolves.toBe('é'); - }); - - it('resolves to an empty string for an empty stream', async () => { - await expect(readRawBody(Readable.from([]))).resolves.toBe(''); - }); - - it('rejects when the stream errors', async () => { - const stream = new Readable({ - read() { - this.destroy(new Error('socket died')); - }, - }); - - await expect(readRawBody(stream)).rejects.toThrow('socket died'); - }); -}); - describe('errorStatus', () => { it('defaults to 500', () => { expect(errorStatus(new Error('boom'))).toBe(500);