diff --git a/core/packages/gax/src/index.ts b/core/packages/gax/src/index.ts
index b0d09dd3b4d..a02da512ae8 100644
--- a/core/packages/gax/src/index.ts
+++ b/core/packages/gax/src/index.ts
@@ -120,7 +120,7 @@ export {
   checkTelemetryEnabled,
 } from './util';
 
-export {StaticTraceContext} from './observability/TracerHelper';
+export {StaticTraceContext, traceAttempt} from './observability/TracerHelper';
 
 export {ServiceError, ChannelCredentials} from '@grpc/grpc-js';
 export {warn} from './warnings';
diff --git a/core/packages/gax/src/observability/TracerHelper.ts b/core/packages/gax/src/observability/TracerHelper.ts
index d3077e8754f..f52e4862b6e 100644
--- a/core/packages/gax/src/observability/TracerHelper.ts
+++ b/core/packages/gax/src/observability/TracerHelper.ts
@@ -14,7 +14,9 @@
  * limitations under the License.
  */
 
+import {EventEmitter} from 'events';
 import {Span, trace, Tracer} from '@opentelemetry/api';
+import {GaxCallResult} from '../apitypes';
 
 /**
  * Static metadata about the Google Cloud client library used to populate
@@ -67,6 +69,128 @@ export function getGaxTracer(): Tracer {
   return trace.getTracer('google-gax');
 }
 
+/**
+ * Checks if a value behaves like a Promise or Thenable.
+ *
+ * Note: It is not sufficient to check `result instanceof Promise` because:
+ * 1. Custom classes implementing `CancellablePromise` or Thenables may not
+ *    inherit directly from the native JavaScript `Promise` prototype.
+ * 2. GAX callers or custom callers may return objects such as `OngoingCallPromise`
+ *    that hold the actual promise on a `.promise` property.
+ * 3. Promises originating from different execution realms (such as Node.js vm
+ *    contexts or different package bundles) fail `instanceof Promise` checks.
+ */
+function isPromiseLike<T = unknown>(value: unknown): value is PromiseLike<T> {
+  return (
+    value instanceof Promise ||
+    (value !== null &&
+      (typeof value === 'object' || typeof value === 'function') &&
+      typeof (value as {then?: unknown}).then === 'function')
+  );
+}
+
+/**
+ * Extracts a PromiseLike target from a value, supporting native Promises,
+ * custom Thenables, classes implementing CancellablePromise, and OngoingCallPromise wrappers.
+ */
+function getPromiseTarget<T = unknown>(value: unknown): PromiseLike<T> | null {
+  if (isPromiseLike<T>(value)) {
+    return value;
+  }
+  if (
+    value !== null &&
+    (typeof value === 'object' || typeof value === 'function') &&
+    'promise' in (value as object) &&
+    isPromiseLike<T>((value as {promise?: unknown}).promise)
+  ) {
+    return (value as {promise: PromiseLike<T>}).promise;
+  }
+  return null;
+}
+
+/**
+ * Manages span lifecycle for Promise-based operations.
+ *
+ * @template T
+ * @param {T} promise - The promise returned from the traced operation.
+ * @param {function} recordError - Callback to record errors on the span.
+ * @param {function} endSpan - Callback to end the span idempotently.
+ */
+export function handlePromise<T>(
+  promise: T,
+  recordError: (err: unknown) => void,
+  endSpan: () => void,
+): void {
+  let spanEnded = false;
+  const endSpanOnce = () => {
+    if (!spanEnded) {
+      spanEnded = true;
+      endSpan();
+    }
+  };
+
+  const target = getPromiseTarget(promise) ?? promise;
+  Promise.resolve(target)
+    .then(() => {
+      endSpanOnce();
+      return null;
+    })
+    .catch(err => {
+      if (!spanEnded) {
+        recordError(err);
+        endSpanOnce();
+      }
+    });
+}
+
+/**
+ * Manages span lifecycle for Stream-based operations and cleans up event listeners.
+ *
+ * @param {EventEmitter} stream - The stream returned from the traced operation.
+ * @param {function} recordError - Callback to record errors on the span.
+ * @param {function} endSpan - Callback to end the span idempotently.
+ */
+export function handleStream(
+  stream: EventEmitter,
+  recordError: (err: unknown) => void,
+  endSpan: () => void,
+): void {
+  let spanEnded = false;
+
+  const cleanup = () => {
+    stream.removeListener('error', onError);
+    stream.removeListener('end', onEnd);
+    stream.removeListener('close', onClose);
+  };
+
+  const endSpanOnce = () => {
+    if (!spanEnded) {
+      spanEnded = true;
+      cleanup();
+      endSpan();
+    }
+  };
+
+  const onError = (err: unknown) => {
+    if (!spanEnded) {
+      recordError(err);
+      endSpanOnce();
+    }
+  };
+
+  const onEnd = () => {
+    endSpanOnce();
+  };
+
+  const onClose = () => {
+    endSpanOnce();
+  };
+
+  stream.on('error', onError);
+  stream.on('end', onEnd);
+  stream.on('close', onClose);
+}
+
 /**
  * Executes a function within an active OpenTelemetry span, populating standard
  * GCP telemetry attributes and recording errors/exceptions if thrown.
@@ -74,16 +198,36 @@ export function getGaxTracer(): Tracer {
  * @template T
  * @param {DynamicTraceContext} dynamicArgs - Dynamic trace context for the RPC call.
  * @param {StaticTraceContext} staticArgs - Static trace context for the client library.
- * @param {() => Promise<T>} fn - The asynchronous operation to trace.
- * @returns {Promise<T>} The result of the traced operation.
+ * @param {function} fn - The operation to trace.
+ * @param {boolean} [isStreamCall=false] - Whether the operation is a stream call (true) or promise call (false).
+ * @returns {T} The result of the traced operation.
  */
-export async function traceAttempt<T = unknown>(
+export function traceAttempt(
+  dynamicArgs: DynamicTraceContext,
+  staticArgs: StaticTraceContext,
+  fn: () => GaxCallResult,
+  isStreamCall?: boolean,
+): GaxCallResult;
+export function traceAttempt<T extends EventEmitter>(
+  dynamicArgs: DynamicTraceContext,
+  staticArgs: StaticTraceContext,
+  fn: () => T,
+  isStreamCall: true,
+): T;
+export function traceAttempt<T>(
   dynamicArgs: DynamicTraceContext,
   staticArgs: StaticTraceContext,
-  fn: () => Promise<T>,
-): Promise<T> {
+  fn: () => T,
+  isStreamCall?: false,
+): T;
+export function traceAttempt(
+  dynamicArgs: DynamicTraceContext,
+  staticArgs: StaticTraceContext,
+  fn: () => GaxCallResult,
+  isStreamCall = false,
+): GaxCallResult {
   const spanName = `${dynamicArgs.clientName}.${dynamicArgs.methodName}`;
-  return getGaxTracer().startActiveSpan(spanName, {}, async (span: Span) => {
+  return getGaxTracer().startActiveSpan(spanName, {}, (span: Span) => {
     span.setAttributes({
       'gcp.client.service': staticArgs.gcpClientService,
       'gcp.client.version': staticArgs.gcpVersion,
@@ -93,22 +237,51 @@ export async function traceAttempt<T = unknown>(
       'gcp.method.type': dynamicArgs.rpcType,
     });
 
+    let spanEnded = false;
+    const endSpan = () => {
+      if (!spanEnded) {
+        spanEnded = true;
+        span.end();
+      }
+    };
+
+    const recordError = (e: unknown) => {
+      if (e instanceof Error) {
+        span.setAttributes({
+          'error.message': e.message,
+          'error.type': e.constructor?.name ?? e.name,
+        });
+        span.recordException(e);
+        if (e.name) {
+          span.setAttribute('exception.type', e.name);
+        }
+      } else {
+        const message = String(e);
+        span.setAttributes({
+          'error.message': message,
+        });
+        span.recordException(message);
+      }
+    };
+
     try {
-      const result = await fn();
+      const result = fn();
+      // Use getPromiseTarget instead of `result instanceof Promise` to ensure custom
+      // thenables, CancellablePromise implementations, and OngoingCallPromise wrappers
+      // are properly tracked rather than leaving spans unclosed or ending them prematurely.
+      const promiseTarget = !isStreamCall ? getPromiseTarget(result) : null;
+      if (isStreamCall && result instanceof EventEmitter) {
+        handleStream(result, recordError, endSpan);
+      } else if (promiseTarget) {
+        handlePromise(promiseTarget, recordError, endSpan);
+      } else {
+        endSpan();
+      }
       return result;
     } catch (e) {
-      const err = e as Error;
-      span.setAttributes({
-        'error.message': err.message,
-        'error.type': err.constructor?.name ?? err.name,
-      });
-      span.recordException(err);
-      if (err.name) {
-        span.setAttribute('exception.type', err.name);
-      }
+      recordError(e);
+      endSpan();
       throw e;
-    } finally {
-      span.end();
     }
   });
 }
diff --git a/core/packages/gax/test/unit/tracerHelper.ts b/core/packages/gax/test/unit/tracerHelper.ts
index 06515c2f0c9..f0bd6e8d058 100644
--- a/core/packages/gax/test/unit/tracerHelper.ts
+++ b/core/packages/gax/test/unit/tracerHelper.ts
@@ -15,13 +15,22 @@
  */
 
 import * as assert from 'assert';
+import {EventEmitter} from 'events';
 import {describe, it, beforeEach, afterEach} from 'mocha';
 import {
   getGaxTracer,
   traceAttempt,
+  handlePromise,
+  handleStream,
   DynamicTraceContext,
   StaticTraceContext,
 } from '../../src/observability/TracerHelper';
+import {
+  GaxCallResult,
+  CancellableStream,
+  ResultTuple,
+} from '../../src/apitypes';
+import {OngoingCallPromise} from '../../src/call';
 import {OtelHarness} from './otelHarness';
 
 describe('TracerHelper', () => {
@@ -163,5 +172,678 @@ describe('TracerHelper', () => {
       assert.strictEqual(spans.length, 1);
       assert.strictEqual(spans[0].attributes['gcp.method.type'], 'http');
     });
+
+    it('manages span lifetime for resolved promises', async () => {
+      const result = await traceAttempt(dynamicArgs, staticArgs, () =>
+        Promise.resolve('async-result'),
+      );
+      assert.strictEqual(result, 'async-result');
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(spans[0].events.length, 0);
+    });
+
+    it('does not end span prematurely for pending asynchronous promises', async () => {
+      let resolvePromise: (val: string) => void;
+      const asyncPromise = new Promise<string>(resolve => {
+        resolvePromise = resolve;
+      });
+
+      const resultPromise = traceAttempt(
+        dynamicArgs,
+        staticArgs,
+        () => asyncPromise,
+      );
+
+      // Verify the span is NOT closed while the promise is pending
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      resolvePromise!('success');
+      const result = await resultPromise;
+      assert.strictEqual(result, 'success');
+
+      // Span should only be ended after resolution
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports Promise subclasses', async () => {
+      class CustomPromise<T> extends Promise<T> {}
+      const customPromise = new CustomPromise<string>(resolve => {
+        setTimeout(() => {
+          resolve('custom-result');
+        }, 10);
+      });
+
+      const result = traceAttempt(dynamicArgs, staticArgs, () => customPromise);
+      assert.strictEqual(result, customPromise);
+
+      const initialSpans = harness.getSpans('google-gax');
+      assert.strictEqual(initialSpans.length, 0);
+
+      await new Promise(resolve => setTimeout(resolve, 25));
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports custom thenables implementing CancellablePromise without inheriting from Promise', async () => {
+      class CustomCancellablePromise {
+        private readonly promise: Promise<string>;
+        constructor(executor: (resolve: (val: string) => void) => void) {
+          this.promise = new Promise(executor);
+        }
+        cancel(): void {}
+        then<TResult1 = string, TResult2 = never>(
+          onfulfilled?:
+            ((value: string) => TResult1 | PromiseLike<TResult1>) | null,
+          onrejected?:
+            ((reason: unknown) => TResult2 | PromiseLike<TResult2>) | null,
+        ): Promise<TResult1 | TResult2> {
+          return this.promise.then(onfulfilled, onrejected);
+        }
+        catch<TResult = never>(
+          onrejected?:
+            ((reason: unknown) => TResult | PromiseLike<TResult>) | null,
+        ): Promise<string | TResult> {
+          return this.promise.catch(onrejected);
+        }
+      }
+
+      let resolvePromise: (val: string) => void;
+      const customPromise = new CustomCancellablePromise(resolve => {
+        resolvePromise = resolve;
+      });
+
+      // Verify it is NOT an instance of native Promise
+      assert.strictEqual(customPromise instanceof Promise, false);
+
+      const result = traceAttempt(dynamicArgs, staticArgs, () => customPromise);
+      assert.strictEqual(result, customPromise);
+
+      // Verify the span is NOT closed while the custom promise is pending
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      resolvePromise!('custom-success');
+      await new Promise(resolve => setTimeout(resolve, 20));
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports OngoingCallPromise objects whose promise property resolves', async () => {
+      const ongoingCall = new OngoingCallPromise();
+      assert.strictEqual(ongoingCall instanceof Promise, false);
+
+      const result = traceAttempt(dynamicArgs, staticArgs, () => ongoingCall);
+      assert.strictEqual(result, ongoingCall);
+
+      // Verify the span is NOT closed while ongoingCall is in flight
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Complete the call via callback
+      ongoingCall.callback!(null, {data: 'result'});
+      await new Promise(resolve => setTimeout(resolve, 20));
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports OngoingCallPromise objects whose promise property rejects', async () => {
+      const ongoingCall = new OngoingCallPromise();
+      assert.strictEqual(ongoingCall instanceof Promise, false);
+
+      const error = new Error('ongoing call failed');
+      const result = traceAttempt(dynamicArgs, staticArgs, () => ongoingCall);
+      assert.strictEqual(result, ongoingCall);
+
+      // Verify the span is NOT closed while ongoingCall is in flight
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Fail the call via callback
+      ongoingCall.callback!(error);
+      await new Promise(resolve => setTimeout(resolve, 20));
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(
+        spans[0].attributes['error.message'],
+        'ongoing call failed',
+      );
+    });
+
+    it('ends span synchronously if result is not a Promise', () => {
+      const syncResult = {data: 'sync-data'};
+
+      const result = traceAttempt(dynamicArgs, staticArgs, () => syncResult);
+      assert.strictEqual(result, syncResult);
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('does not end span prematurely until asynchronous promise rejects', async () => {
+      let rejectPromise: (err: Error) => void;
+      const asyncPromise = new Promise((_resolve, reject) => {
+        rejectPromise = reject;
+      });
+
+      const error = new Error('async promise failure');
+      void traceAttempt(dynamicArgs, staticArgs, () => asyncPromise);
+
+      // Verify the span is NOT closed while the promise is pending
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      rejectPromise!(error);
+      await new Promise(resolve => setTimeout(resolve, 15));
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(
+        spans[0].attributes['error.message'],
+        'async promise failure',
+      );
+      assert.strictEqual(spans[0].events.length, 1);
+    });
+
+    it('does not end span prematurely while stream is active and emitting data', () => {
+      const emitter = new EventEmitter();
+      const result = traceAttempt(dynamicArgs, staticArgs, () => emitter, true);
+      assert.strictEqual(result, emitter);
+
+      // Span must not be finished when stream is created
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Emitting data chunks should not close the span
+      emitter.emit('data', 'chunk 1');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      emitter.emit('data', 'chunk 2');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Only when the stream finishes does the span end
+      emitter.emit('end');
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(spans[0].events.length, 0);
+    });
+
+    it('does not end span prematurely until stream emits error event', () => {
+      const emitter = new EventEmitter();
+      traceAttempt(dynamicArgs, staticArgs, () => emitter, true);
+
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      emitter.emit('data', 'chunk');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      const error = new Error('stream failure');
+      emitter.emit('error', error);
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(
+        spans[0].attributes['error.message'],
+        'stream failure',
+      );
+      assert.strictEqual(spans[0].events.length, 1);
+      assert.strictEqual(spans[0].events[0].name, 'exception');
+    });
+
+    it('does not end span prematurely until stream emits close event', () => {
+      const emitter = new EventEmitter();
+      traceAttempt(dynamicArgs, staticArgs, () => emitter, true);
+
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      emitter.emit('data', 'chunk');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      emitter.emit('close');
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports isStreamCall explicitly set to false', async () => {
+      const result = await traceAttempt(
+        dynamicArgs,
+        staticArgs,
+        () => Promise.resolve('explicit-false'),
+        false,
+      );
+      assert.strictEqual(result, 'explicit-false');
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('ends span synchronously if isStreamCall is true but result is not an EventEmitter', () => {
+      const nonEmitter = {data: 'not-an-emitter'};
+      const result = traceAttempt(
+        dynamicArgs,
+        staticArgs,
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        () => nonEmitter as any,
+        true,
+      );
+      assert.strictEqual(result, nonEmitter);
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports GaxCallResult promise operations', async () => {
+      const cancellablePromise = Object.assign(
+        Promise.resolve([{}, undefined, undefined] as ResultTuple),
+        {
+          cancel: () => {},
+        },
+      ) as GaxCallResult;
+
+      const result = traceAttempt(
+        dynamicArgs,
+        staticArgs,
+        () => cancellablePromise,
+      );
+      assert.strictEqual(result, cancellablePromise);
+      await result;
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('supports GaxCallResult stream operations', () => {
+      const stream = Object.assign(new EventEmitter(), {
+        cancel: () => {},
+      }) as unknown as CancellableStream;
+
+      const result = traceAttempt(dynamicArgs, staticArgs, () => stream, true);
+      assert.strictEqual(result, stream);
+
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+      stream.emit('end');
+
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+    });
+
+    it('correctly creates separate spans and cleans up listeners across retried stream attempts', () => {
+      const attempt1Stream = new EventEmitter();
+      const attempt2Stream = new EventEmitter();
+      const retryableError = new Error('transient stream failure');
+
+      let attempt = 0;
+      const executeStreamingCall = () => {
+        attempt++;
+        const currentStream = attempt === 1 ? attempt1Stream : attempt2Stream;
+        return traceAttempt(dynamicArgs, staticArgs, () => currentStream, true);
+      };
+
+      // Attempt 1
+      const stream1 = executeStreamingCall();
+      assert.strictEqual(stream1, attempt1Stream);
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Attempt 1 fails with transient error
+      attempt1Stream.emit('error', retryableError);
+
+      const spansAfterAttempt1 = harness.getSpans('google-gax');
+      assert.strictEqual(spansAfterAttempt1.length, 1);
+      assert.strictEqual(spansAfterAttempt1[0].ended, true);
+      assert.strictEqual(
+        spansAfterAttempt1[0].attributes['error.message'],
+        'transient stream failure',
+      );
+      assert.strictEqual(spansAfterAttempt1[0].events.length, 1);
+      assert.strictEqual(attempt1Stream.listenerCount('error'), 0);
+
+      // Attempt 2 (retry)
+      const stream2 = executeStreamingCall();
+      assert.strictEqual(stream2, attempt2Stream);
+      assert.strictEqual(harness.getSpans('google-gax').length, 1);
+
+      // Data chunks received on attempt 2
+      attempt2Stream.emit('data', 'retry chunk 1');
+      assert.strictEqual(harness.getSpans('google-gax').length, 1);
+
+      // Attempt 2 completes successfully
+      attempt2Stream.emit('end');
+
+      const spansAfterAttempt2 = harness.getSpans('google-gax');
+      assert.strictEqual(spansAfterAttempt2.length, 2);
+      assert.strictEqual(spansAfterAttempt2[1].ended, true);
+      assert.strictEqual(spansAfterAttempt2[1].events.length, 0);
+      assert.strictEqual(attempt2Stream.listenerCount('end'), 0);
+    });
+
+    it('keeps span active when a stream handles retries internally before completing', () => {
+      const outerStream = new EventEmitter();
+      const result = traceAttempt(
+        dynamicArgs,
+        staticArgs,
+        () => outerStream,
+        true,
+      );
+      assert.strictEqual(result, outerStream);
+
+      // Initial chunk before internal retry
+      outerStream.emit('data', 'chunk-before-retry');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Internal retry transparently recovers and delivers more data
+      outerStream.emit('data', 'chunk-after-retry');
+      assert.strictEqual(harness.getSpans('google-gax').length, 0);
+
+      // Final completion
+      outerStream.emit('end');
+      const spans = harness.getSpans('google-gax');
+      assert.strictEqual(spans.length, 1);
+      assert.strictEqual(spans[0].ended, true);
+      assert.strictEqual(spans[0].events.length, 0);
+    });
+  });
+
+  describe('handlePromise', () => {
+    it('waits for promise resolution before ending span', async () => {
+      let ended = false;
+      let resolvePromise: () => void;
+      const promise = new Promise<void>(resolve => {
+        resolvePromise = resolve;
+      });
+
+      handlePromise(
+        promise,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+      assert.strictEqual(ended, false);
+
+      resolvePromise!();
+      await new Promise(resolve => setTimeout(resolve, 15));
+      assert.strictEqual(ended, true);
+    });
+
+    it('records error and ends span when promise rejects', async () => {
+      let ended = false;
+      let recordedError: unknown;
+      const error = new Error('promise error');
+
+      handlePromise(
+        Promise.reject(error),
+        err => {
+          recordedError = err;
+        },
+        () => {
+          ended = true;
+        },
+      );
+
+      await new Promise(resolve => setTimeout(resolve, 15));
+      assert.strictEqual(ended, true);
+      assert.strictEqual(recordedError, error);
+    });
+
+    it('supports custom thenables', async () => {
+      let ended = false;
+      const thenable = {
+        then(onfulfilled?: (val?: unknown) => unknown) {
+          setTimeout(() => {
+            onfulfilled?.();
+          }, 10);
+        },
+      };
+
+      handlePromise(
+        thenable,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+      assert.strictEqual(ended, false);
+
+      await new Promise(resolve => setTimeout(resolve, 20));
+      assert.strictEqual(ended, true);
+    });
+
+    it('supports OngoingCallPromise wrappers in handlePromise', async () => {
+      let ended = false;
+      const ongoingCall = new OngoingCallPromise();
+
+      handlePromise(
+        ongoingCall,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+      assert.strictEqual(ended, false);
+
+      ongoingCall.callback!(null, {data: 'hello'});
+      await new Promise(resolve => setTimeout(resolve, 20));
+      assert.strictEqual(ended, true);
+    });
+
+    it('ensures endSpan is called only once even with thenables that trigger both resolve and reject', async () => {
+      let endSpanCount = 0;
+      const buggyThenable = {
+        then(
+          onfulfilled?: (val?: unknown) => unknown,
+          onrejected?: (err: unknown) => unknown,
+        ) {
+          onfulfilled?.();
+          onrejected?.(new Error('buggy error'));
+        },
+      };
+
+      handlePromise(
+        buggyThenable,
+        () => {},
+        () => {
+          endSpanCount++;
+        },
+      );
+
+      await new Promise(resolve => setTimeout(resolve, 20));
+      assert.strictEqual(endSpanCount, 1);
+    });
+  });
+
+  describe('handleStream', () => {
+    it('manages stream events, ends span, and cleans up listeners on end', () => {
+      let ended = false;
+      const emitter = new EventEmitter();
+      handleStream(
+        emitter,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+
+      assert.strictEqual(ended, false);
+      assert.strictEqual(emitter.listenerCount('end'), 1);
+      assert.strictEqual(emitter.listenerCount('close'), 1);
+      assert.strictEqual(emitter.listenerCount('error'), 1);
+
+      emitter.emit('data', 'chunk');
+      assert.strictEqual(ended, false);
+
+      emitter.emit('end');
+      assert.strictEqual(ended, true);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('ends span and cleans up listeners on stream close', () => {
+      let ended = false;
+      const emitter = new EventEmitter();
+      handleStream(
+        emitter,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+
+      assert.strictEqual(ended, false);
+      assert.strictEqual(emitter.listenerCount('close'), 1);
+
+      emitter.emit('close');
+      assert.strictEqual(ended, true);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('records error, ends span, and cleans up listeners on stream error', () => {
+      let ended = false;
+      let recordedError: unknown;
+      const order: string[] = [];
+      const error = new Error('stream failure');
+      const emitter = new EventEmitter();
+
+      handleStream(
+        emitter,
+        err => {
+          order.push('recordError');
+          recordedError = err;
+        },
+        () => {
+          order.push('endSpan');
+          ended = true;
+        },
+      );
+
+      assert.strictEqual(ended, false);
+      assert.strictEqual(emitter.listenerCount('error'), 1);
+
+      emitter.emit('error', error);
+      assert.strictEqual(ended, true);
+      assert.strictEqual(recordedError, error);
+      assert.deepStrictEqual(order, ['recordError', 'endSpan']);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('ensures endSpan is called only once if stream emits error followed by close', () => {
+      let endSpanCount = 0;
+      let recordedError: unknown;
+      const error = new Error('stream error');
+      const emitter = new EventEmitter();
+
+      handleStream(
+        emitter,
+        err => {
+          recordedError = err;
+        },
+        () => {
+          endSpanCount++;
+        },
+      );
+
+      assert.strictEqual(endSpanCount, 0);
+      emitter.emit('error', error);
+      emitter.emit('close');
+
+      assert.strictEqual(endSpanCount, 1);
+      assert.strictEqual(recordedError, error);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('ensures endSpan is called only once if stream emits end followed by close', () => {
+      let endSpanCount = 0;
+      const emitter = new EventEmitter();
+
+      handleStream(
+        emitter,
+        () => {},
+        () => {
+          endSpanCount++;
+        },
+      );
+
+      assert.strictEqual(endSpanCount, 0);
+      emitter.emit('end');
+      emitter.emit('close');
+
+      assert.strictEqual(endSpanCount, 1);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('ensures endSpan is called only once if stream emits close followed by end', () => {
+      let endSpanCount = 0;
+      const emitter = new EventEmitter();
+
+      handleStream(
+        emitter,
+        () => {},
+        () => {
+          endSpanCount++;
+        },
+      );
+
+      assert.strictEqual(endSpanCount, 0);
+      emitter.emit('close');
+      emitter.emit('end');
+
+      assert.strictEqual(endSpanCount, 1);
+      assert.strictEqual(emitter.listenerCount('end'), 0);
+      assert.strictEqual(emitter.listenerCount('close'), 0);
+      assert.strictEqual(emitter.listenerCount('error'), 0);
+    });
+
+    it('does not remove other error or event listeners (such as retry handlers) on cleanup', () => {
+      let ended = false;
+      let otherErrorHandled = false;
+      const error = new Error('retryable error');
+      const emitter = new EventEmitter();
+
+      // Simulate an external retry handler or middleware attached to the stream
+      emitter.on('error', err => {
+        assert.strictEqual(err, error);
+        otherErrorHandled = true;
+      });
+
+      handleStream(
+        emitter,
+        () => {},
+        () => {
+          ended = true;
+        },
+      );
+
+      // Verify two error listeners are present (external retry listener and handleStream listener)
+      assert.strictEqual(emitter.listenerCount('error'), 2);
+
+      emitter.emit('error', error);
+
+      assert.strictEqual(ended, true);
+      assert.strictEqual(otherErrorHandled, true);
+      // handleStream removed its own listener, but the external retry listener is preserved
+      assert.strictEqual(emitter.listenerCount('error'), 1);
+    });
   });
 });