transport: accept external abort signal in newTurn

Allow server-side route handlers to pass an external AbortSignal (e.g.
req.signal) into newTurn via the new `signal` option. When the signal
fires — due to client disconnect or serverless function timeout — the
turn's internal AbortController is aborted through the same path as
Ably cancel messages, stopping LLM generation and stream piping.

The listener is cleaned up on both end() and close() to avoid leaks.

Tests cover: signal abort, pre-aborted signal, in-flight stream
cancellation, listener cleanup on end() and close(), and
non-interference with Ably cancel routing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: JoaoDiasAbly

demo/vercel/react/use-chat/src/app/api/chat/route.ts

@@ -34,7 +34,7 @@ export async function POST(req: Request) {
   const channel = ably.channels.get(id);
 
   const transport = createServerTransport({ channel });
-  const turn = transport.newTurn({ turnId, clientId, parent, forkOf });
+  const turn = transport.newTurn({ turnId, clientId, parent, forkOf, signal: req.signal });
 
   await turn.start();
 

demo/vercel/react/use-client-transport/src/app/api/chat/route.ts

@@ -179,7 +179,7 @@ export async function POST(req: Request) {
     }
   };
 
-  const turn = transport.newTurn({ turnId, clientId, parent, forkOf, onMessage });
+  const turn = transport.newTurn({ turnId, clientId, parent, forkOf, onMessage, signal: req.signal });
 
   await turn.start();
 

src/core/transport/server-transport.ts

@@ -52,6 +52,8 @@ interface RegisteredTurn {
   controller: AbortController;
   onCancel?: (request: CancelRequest) => Promise<boolean>;
   onError?: (error: Ably.ErrorInfo) => void;
+  /** Removes the listener from the external signal, if one was provided. */
+  cleanupExternalSignal?: () => void;
 }
 
 // ---------------------------------------------------------------------------
@@ -116,6 +118,7 @@ class DefaultServerTransport<TEvent, TMessage> implements ServerTransport<TEvent
     this._logger?.trace('DefaultServerTransport.close();');
     this._channel.unsubscribe(EVENT_CANCEL, this._channelListener);
     for (const reg of this._registeredTurns.values()) {
+      reg.cleanupExternalSignal?.();
       reg.controller.abort();
     }
     this._registeredTurns.clear();
@@ -249,19 +252,44 @@ class DefaultServerTransport<TEvent, TMessage> implements ServerTransport<TEvent
       onError: turnOnError,
       parent: turnParent,
       forkOf: turnForkOf,
+      signal: externalSignal,
     } = turnOpts;
 
     const controller = new AbortController();
     let started = false;
     let ended = false;
 
+    // Link external signal (e.g. req.signal) to the internal controller so
+    // platform-level cancellation (client disconnect, function timeout) aborts
+    // the turn through the same path as Ably cancel messages.
+    let cleanupExternalSignal: (() => void) | undefined;
+    if (externalSignal) {
+      if (externalSignal.aborted) {
+        this._logger?.debug('DefaultServerTransport._createTurn(); external signal already aborted', { turnId });
+        controller.abort();
+      } else {
+        const listener = () => {
+          this._logger?.debug('DefaultServerTransport._createTurn(); external signal fired, aborting turn', {
+            turnId,
+          });
+          controller.abort();
+        };
+        externalSignal.addEventListener('abort', listener, { once: true });
+        cleanupExternalSignal = () => {
+          externalSignal.removeEventListener('abort', listener);
+        };
+        this._logger?.debug('DefaultServerTransport._createTurn(); linked external signal', { turnId });
+      }
+    }
+
     // Spec: AIT-ST3a — register immediately so early cancels can fire the abort signal.
     const registration: RegisteredTurn = {
       turnId,
       clientId: turnClientId ?? '',
       controller,
       onCancel,
       onError: turnOnError,
+      cleanupExternalSignal,
     };
     this._registeredTurns.set(turnId, registration);
 
@@ -468,6 +496,7 @@ class DefaultServerTransport<TEvent, TMessage> implements ServerTransport<TEvent
           throw errInfo;
         } finally {
           registeredTurns.delete(turnId);
+          cleanupExternalSignal?.();
         }
 
         logger?.debug('Turn.end(); turn ended', { turnId, reason });

src/core/transport/types.ts

@@ -158,6 +158,14 @@ export interface NewTurnOptions<TEvent> {
    * publish failure, encoder recovery failure, stream encoding errors.
    */
   onError?: (error: Ably.ErrorInfo) => void;
+
+  /**
+   * An external abort signal (typically the HTTP request's `req.signal`) that,
+   * when fired, aborts this turn. This allows platform-level cancellation —
+   * client disconnect, serverless function timeout — to stop LLM generation
+   * and stream piping gracefully.
+   */
+  signal?: AbortSignal;
 }
 
 // ---------------------------------------------------------------------------

test/core/transport/server-transport.test.ts

@@ -574,6 +574,75 @@ describe('ServerTransport', () => {
     });
   });
 
+  describe('external signal', () => {
+    it('aborts the turn when the external signal fires', async () => {
+      const externalController = new AbortController();
+      const turn = transport.newTurn({ turnId: 'turn-1', signal: externalController.signal });
+      await turn.start();
+
+      externalController.abort();
+
+      expect(turn.abortSignal.aborted).toBe(true);
+    });
+
+    it('aborts the turn immediately when the external signal is already aborted', () => {
+      const turn = transport.newTurn({ turnId: 'turn-1', signal: AbortSignal.abort() });
+
+      expect(turn.abortSignal.aborted).toBe(true);
+    });
+
+    it('start() throws when the external signal was already aborted', async () => {
+      const turn = transport.newTurn({ turnId: 'turn-1', signal: AbortSignal.abort() });
+
+      await expect(turn.start()).rejects.toBeErrorInfoWithCode(ErrorCode.InvalidArgument);
+    });
+
+    it('cancels an in-flight streamResponse when the external signal fires', async () => {
+      const externalController = new AbortController();
+      const turn = transport.newTurn({ turnId: 'turn-1', signal: externalController.signal });
+      await turn.start();
+
+      // A stream that never closes on its own — waits for cancellation.
+      const stream = new ReadableStream<TestEvent>({
+        start: (controller) => {
+          controller.enqueue({ type: 'text', text: 'partial' });
+        },
+      });
+
+      const resultPromise = turn.streamResponse(stream);
+      externalController.abort();
+
+      const result = await resultPromise;
+      expect(result.reason).toBe('cancelled');
+    });
+
+    it('removes the external signal listener when the turn ends', async () => {
+      const externalController = new AbortController();
+      const signal = externalController.signal;
+      const removeSpy = vi.spyOn(signal, 'removeEventListener');
+
+      const turn = transport.newTurn({ turnId: 'turn-1', signal });
+      await turn.start();
+      await turn.end('complete');
+
+      expect(removeSpy).toHaveBeenCalledWith('abort', expect.any(Function));
+    });
+
+    it('does not interfere with Ably cancel routing', async () => {
+      const externalController = new AbortController();
+      const turn = transport.newTurn({ turnId: 'turn-1', clientId: 'user-a', signal: externalController.signal });
+      await turn.start();
+
+      // Cancel via Ably channel message (not external signal)
+      simulateCancel(channel, { [HEADER_CANCEL_TURN_ID]: 'turn-1' });
+      await new Promise((r) => setTimeout(r, 10));
+
+      expect(turn.abortSignal.aborted).toBe(true);
+      // External signal was NOT fired
+      expect(externalController.signal.aborted).toBe(false);
+    });
+  });
+
   describe('close', () => {
     it('aborts all registered turns', async () => {
       const turn1 = transport.newTurn({ turnId: 'turn-1' });
@@ -591,5 +660,16 @@ describe('ServerTransport', () => {
       transport.close();
       expect(channel.unsubscribe).toHaveBeenCalledWith(EVENT_CANCEL, expect.any(Function));
     });
+
+    it('removes external signal listeners on close', () => {
+      const externalController = new AbortController();
+      const removeSpy = vi.spyOn(externalController.signal, 'removeEventListener');
+
+      transport.newTurn({ turnId: 'turn-1', signal: externalController.signal });
+
+      transport.close();
+
+      expect(removeSpy).toHaveBeenCalledWith('abort', expect.any(Function));
+    });
   });
 });