Fix: release the start-up watchdog timer and listeners in BaseTestServer.startProcess()

tests/integration/test_case_utils.ts

@@ -158,6 +158,13 @@ const ADK_EVENT_ID_REGEX = /^[a-zA-Z0-9]{8}$/;
 const INVOCATION_ID_REGEX =
   /^e-[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$/;
 
+/**
+ * Matches the loopback URL a test server prints on start-up, e.g.
+ * `http://localhost:41234`, `http://127.0.0.1:41234` or `http://[::1]:41234`.
+ */
+const SERVER_URL_REGEX =
+  /http:\/\/(?:localhost|127\.0\.0\.1|\[::1\]):([0-9]+)/i;
+
 const IGNORE_FIELDS = [
   'id',
   'invocationId',
@@ -289,70 +296,85 @@ export abstract class BaseTestServer {
     serverName: string;
     timeoutMs: number;
   }): Promise<void> {
-    this.serverProcess = spawnProcess();
-
-    await new Promise<void>((resolve, reject) => {
-      let started = false;
-      const stdoutChunks: string[] = [];
-
-      this.serverProcess!.stdout.on('data', (data) => {
-        const message = data.toString();
-        stdoutChunks.push(message);
-
-        // Find URL like http://localhost:12345
-        const urlMatch = message.match(/http:\/\/localhost:([0-9]+)/i);
-        if (urlMatch && urlMatch[1]) {
-          const parsedPort = parseInt(urlMatch[1], 10);
-          if (parsedPort > 0) {
-            this.port = parsedPort;
-            this.url = `http://${this.host}:${this.port}`;
+    const serverProcess = spawnProcess();
+    this.serverProcess = serverProcess;
+
+    // Appended to only during the handshake, to explain a premature exit.
+    const stdoutChunks: string[] = [];
+    let releaseStartHandshake = () => {};
+
+    try {
+      await new Promise<void>((resolve, reject) => {
+        const onStdout = (data: Buffer) => {
+          const message = data.toString();
+          stdoutChunks.push(message);
+
+          const urlMatch = message.match(SERVER_URL_REGEX);
+          if (urlMatch) {
+            const parsedPort = parseInt(urlMatch[1], 10);
+            if (parsedPort > 0) {
+              this.port = parsedPort;
+              this.url = `http://${this.host}:${this.port}`;
+            }
           }
-        }
-
-        if (message.includes(startMessage)) {
-          started = true;
-          console.log(successLogMessage);
-          resolve();
-        }
-      });
-
-      this.serverProcess!.stderr.on('data', (data) => {
-        console.error(`${serverName} Stderr: ${data.toString()}`);
-      });
-
-      this.serverProcess!.on('error', (error) => {
-        console.error(`${serverName} Error: ${error.message}`);
-
-        reject(
-          new Error(
-            `Failed to start ${serverName.toLowerCase()}: ${error.message}`,
-          ),
-        );
-      });
 
-      this.serverProcess!.on('exit', (code) => {
-        console.error(`${serverName} exited with code ${code}`);
+          if (message.includes(startMessage)) {
+            console.log(successLogMessage);
+            resolve();
+          }
+        };
 
-        if (!started) {
+        // Only attached during the handshake, so any exit seen here is
+        // premature.
+        const onExit = (code: number | null) => {
           console.error(
             `${serverName} Captured stdout before premature exit:\n${stdoutChunks.join('')}`,
           );
           reject(
             new Error(`${serverName} exited prematurely with code ${code}`),
           );
-        }
-      });
+        };
 
-      setTimeout(() => {
-        if (!started) {
+        const startTimer = setTimeout(() => {
           reject(
             new Error(
               `Timeout waiting for ${serverName.toLowerCase()} to start.`,
             ),
           );
-        }
-      }, timeoutMs);
-    });
+        }, timeoutMs);
+
+        releaseStartHandshake = () => {
+          clearTimeout(startTimer);
+          // stdout stays in flowing mode after this, so it keeps draining and
+          // the child never blocks on a full pipe.
+          serverProcess.stdout.off('data', onStdout);
+          serverProcess.off('exit', onExit);
+        };
+
+        serverProcess.stdout.on('data', onStdout);
+        serverProcess.on('exit', onExit);
+
+        // stderr logging and the 'error' listener intentionally outlive the
+        // handshake: an 'error' event with no listener is thrown by
+        // EventEmitter and would crash the test worker, and a failed kill()
+        // emits one after the handshake.
+        serverProcess.stderr.on('data', (data: Buffer) => {
+          console.error(`${serverName} Stderr: ${data.toString()}`);
+        });
+
+        serverProcess.on('error', (error) => {
+          console.error(`${serverName} Error: ${error.message}`);
+
+          reject(
+            new Error(
+              `Failed to start ${serverName.toLowerCase()}: ${error.message}`,
+            ),
+          );
+        });
+      });
+    } finally {
+      releaseStartHandshake();
+    }
   }
 
   async stop(): Promise<void> {

tests/integration/test_case_utils_test.ts

@@ -0,0 +1,220 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import type {ChildProcessWithoutNullStreams} from 'node:child_process';
+import {spawn} from 'node:child_process';
+import {once} from 'node:events';
+import {afterEach, beforeEach, describe, expect, it, vi} from 'vitest';
+import {BaseTestServer} from './test_case_utils.js';
+
+const START_MESSAGE = 'READY';
+const SERVER_NAME = 'Fake';
+const DEFAULT_TIMEOUT_MS = 15000;
+/** Distinctive value so the watchdog is identifiable among all setTimeout calls. */
+const WATCHDOG_TIMEOUT_MS = 12345;
+/** Port handed to the constructor, so a read-back is visible as a change. */
+const CONSTRUCTOR_PORT = 19999;
+
+/** Overrides a {@link FakeServer} applies to its start handshake. */
+interface FakeServerOptions {
+  startMessage?: string;
+  timeoutMs?: number;
+}
+
+/** Keeps a `node -e` child alive until the test kills it. */
+const KEEP_ALIVE = 'setTimeout(() => {}, 60000);';
+
+/** A `node -e` statement that writes one line to stdout. */
+function writeLine(line: string): string {
+  return `process.stdout.write(${JSON.stringify(`${line}\n`)});`;
+}
+
+/** `node -e` source that writes the given lines to stdout, then idles. */
+function serverScript(...lines: string[]): string {
+  return `${lines.map(writeLine).join('')}${KEEP_ALIVE}`;
+}
+
+/**
+ * Minimal {@link BaseTestServer} driven by a hermetic `node -e` child, so the
+ * start handshake can be exercised without a real server.
+ */
+class FakeServer extends BaseTestServer {
+  private child?: ChildProcessWithoutNullStreams;
+
+  constructor(
+    private readonly script: string,
+    private readonly options: FakeServerOptions = {},
+  ) {
+    super('localhost', CONSTRUCTOR_PORT);
+  }
+
+  async start(): Promise<void> {
+    await this.startProcess({
+      spawnProcess: () => {
+        this.child = spawn(process.execPath, ['-e', this.script]);
+        return this.child;
+      },
+      startMessage: this.options.startMessage ?? START_MESSAGE,
+      successLogMessage: 'fake server started',
+      serverName: SERVER_NAME,
+      timeoutMs: this.options.timeoutMs ?? DEFAULT_TIMEOUT_MS,
+    });
+  }
+
+  get childProcess(): ChildProcessWithoutNullStreams {
+    if (!this.child) {
+      expect.fail('the fake server was never spawned');
+    }
+    return this.child;
+  }
+}
+
+/** Resolves once `needle` has been seen on the child's stdout. */
+function waitForStdout(
+  child: ChildProcessWithoutNullStreams,
+  needle: string,
+): Promise<void> {
+  return new Promise<void>((resolve) => {
+    let output = '';
+    const onData = (data: Buffer) => {
+      output += data.toString();
+      if (output.includes(needle)) {
+        child.stdout.off('data', onData);
+        resolve();
+      }
+    };
+    child.stdout.on('data', onData);
+  });
+}
+
+describe('BaseTestServer.startProcess', () => {
+  const servers: FakeServer[] = [];
+  let consoleErrorSpy: ReturnType<typeof vi.spyOn>;
+
+  function createServer(
+    script: string,
+    options?: FakeServerOptions,
+  ): FakeServer {
+    const server = new FakeServer(script, options);
+    servers.push(server);
+    return server;
+  }
+
+  beforeEach(() => {
+    vi.spyOn(console, 'log').mockImplementation(() => {});
+    consoleErrorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
+  });
+
+  afterEach(async () => {
+    vi.restoreAllMocks();
+    for (const server of servers.splice(0)) {
+      await server.stop();
+    }
+  });
+
+  it('clears the start-up watchdog once the server reports success', async () => {
+    const setTimeoutSpy = vi.spyOn(globalThis, 'setTimeout');
+    const clearTimeoutSpy = vi.spyOn(globalThis, 'clearTimeout');
+    const server = createServer(serverScript(START_MESSAGE), {
+      timeoutMs: WATCHDOG_TIMEOUT_MS,
+    });
+
+    await server.start();
+
+    const watchdogIndex = setTimeoutSpy.mock.calls.findIndex(
+      ([, delay]) => delay === WATCHDOG_TIMEOUT_MS,
+    );
+    expect(watchdogIndex).toBeGreaterThanOrEqual(0);
+    expect(clearTimeoutSpy).toHaveBeenCalledWith(
+      setTimeoutSpy.mock.results[watchdogIndex].value,
+    );
+  });
+
+  it('detaches the start-up listeners once the server reports success', async () => {
+    const server = createServer(serverScript(START_MESSAGE));
+
+    await server.start();
+
+    const child = server.childProcess;
+    expect(child.stdout.listenerCount('data')).toBe(0);
+    expect(child.listenerCount('exit')).toBe(0);
+    expect(child.stdout.isPaused()).toBe(false);
+    // Deliberately retained: an unhandled 'error' event would crash the worker
+    // and stderr needs a draining consumer.
+    expect(child.listenerCount('error')).toBe(1);
+    expect(child.stderr.listenerCount('data')).toBe(1);
+  });
+
+  it('does not log an exit message when the server is stopped cleanly', async () => {
+    const server = createServer(serverScript(START_MESSAGE));
+    await server.start();
+    const child = server.childProcess;
+    const exited = once(child, 'exit');
+
+    await server.stop();
+    await exited;
+
+    const logged = consoleErrorSpy.mock.calls.flat().join('\n');
+    expect(logged).not.toMatch(/exited with code/);
+    // A clean shutdown must not reach the premature-exit diagnostic either.
+    expect(logged).toBe('');
+  });
+
+  it('ignores URLs printed after start-up', async () => {
+    const server = createServer(
+      `${writeLine('http://localhost:41111')}${writeLine(START_MESSAGE)}` +
+        `setTimeout(() => {${writeLine('http://localhost:49999')}}, 50);` +
+        KEEP_ALIVE,
+    );
+
+    await server.start();
+    await waitForStdout(server.childProcess, '49999');
+
+    expect(server.port).toBe(41111);
+    expect(server.url).toBe('http://localhost:41111');
+  });
+
+  it.each([
+    'http://localhost:41234',
+    'http://127.0.0.1:41234',
+    'http://[::1]:41234',
+  ])('reads the port back from the start-up banner %s', async (banner) => {
+    const server = createServer(
+      serverScript(`A2A Server started on ${banner}`),
+      {startMessage: 'A2A Server started on'},
+    );
+
+    await server.start();
+
+    expect(server.port).toBe(41234);
+    expect(server.url).toBe('http://localhost:41234');
+  });
+
+  it('rejects and releases the handshake when the server never starts', async () => {
+    const server = createServer(KEEP_ALIVE, {timeoutMs: 200});
+
+    await expect(server.start()).rejects.toThrow(
+      'Timeout waiting for fake to start.',
+    );
+
+    const child = server.childProcess;
+    expect(child.stdout.listenerCount('data')).toBe(0);
+    expect(child.listenerCount('exit')).toBe(0);
+  });
+
+  it('rejects with the captured stdout when the server exits prematurely', async () => {
+    const server = createServer(
+      `process.stdout.write(${JSON.stringify('boot log line\n')}, () => process.exit(3));`,
+    );
+
+    await expect(server.start()).rejects.toThrow(
+      'Fake exited prematurely with code 3',
+    );
+
+    const logged = consoleErrorSpy.mock.calls.flat().join('\n');
+    expect(logged).toContain('boot log line');
+  });
+});