diff --git a/.changeset/onrpcbroken-abort-signal.md b/.changeset/onrpcbroken-abort-signal.md
new file mode 100644
index 00000000..8b17c3f8
--- /dev/null
+++ b/.changeset/onrpcbroken-abort-signal.md
@@ -0,0 +1,7 @@
+---
+"capnweb": minor
+---
+
+`onRpcBroken()` now takes an optional `AbortSignal`: `stub.onRpcBroken(cb, { signal })`. Aborting the signal drops the callback, and a signal that is already aborted registers nothing, even when the stub is already broken.
+
+Disposing a stub now drops the callbacks registered on it. This changes existing behavior. Previously, disposing a stub or promise left its `onRpcBroken()` callbacks registered, so they still fired when the connection was later lost. Disposal is per-stub, so callbacks registered on other stubs pointing at the same object still fire.
diff --git a/README.md b/README.md
index f73d0170..6767c62d 100644
--- a/README.md
+++ b/README.md
@@ -433,6 +433,23 @@ If anything happens to the stub that would cause all further method calls and pr
 * The stub's underlying connection is lost.
 * The stub is a promise, and the promise rejects.
 
+To stop listening, pass an `AbortSignal`:
+
+```ts
+let controller = new AbortController();
+
+stub.onRpcBroken((error: any) => {
+  console.error(error);
+}, { signal: controller.signal });
+
+// Later: stop listening.
+controller.abort();
+```
+
+Aborting the signal drops the callback. A signal that is already aborted registers nothing, even if the stub is already broken.
+
+Disposing a stub also drops the callbacks registered on it.
+
 ## Security Considerations
 
 * The WebSocket API in browsers always permits cross-site connections, and does not permit setting headers. Because of this, you generally cannot use cookies nor other headers for authentication. Instead, we highly recommend the pattern shown in the second example above, in which authentication happens in-band via an RPC method that returns the authenticated API.
diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts
index cb933387..c36d3eda 100644
--- a/__tests__/index.test.ts
+++ b/__tests__/index.test.ts
@@ -2175,6 +2175,173 @@ describe("onRpcBroken", () => {
       {which: "hangingCall", error: new Error("test disconnect")},
     ]);
   });
+
+  it("never registers a callback whose signal is already aborted", async () => {
+    class TestBroken extends RpcTarget {
+      makeCounter() { return new Counter(0); }
+      throwError(): Promise<Counter> { throw new Error("test error"); }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let preAborted = new AbortController();
+    preAborted.abort();
+    let aborted = preAborted.signal;
+
+    let errors: string[] = [];
+
+    stub.onRpcBroken(() => { errors.push("stub"); }, {signal: aborted});
+
+    let counterPromise = stub.makeCounter();
+    counterPromise.onRpcBroken(() => { errors.push("counterPromise"); }, {signal: aborted});
+
+    // An already-broken stub normally reports synchronously; the aborted signal suppresses it.
+    let throwingPromise = stub.throwError();
+    await throwingPromise.catch(err => {});
+    throwingPromise.onRpcBroken(() => { errors.push("throwError"); }, {signal: aborted});
+    expect(errors).toStrictEqual([]);
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual([]);
+  });
+
+  it("deregisters a callback when its signal is aborted", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+    let canceled = new AbortController();
+
+    stub.onRpcBroken(() => { errors.push("kept1"); });
+    stub.onRpcBroken(() => { errors.push("canceled"); }, {signal: canceled.signal});
+    stub.onRpcBroken(() => { errors.push("kept2"); });
+
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    // The canceled callback is gone; the others still fire in registration order.
+    expect(errors).toStrictEqual(["kept1", "kept2"]);
+  });
+
+  it("honors the signal after the promise it was registered on has resolved", async () => {
+    class TestBroken extends RpcTarget {
+      makeCounter() { return new Counter(0); }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let errors: string[] = [];
+    let canceled = new AbortController();
+
+    // Register while the promise is still unresolved, so that the registration is later migrated
+    // onto the resolution.
+    let counterPromise = stub.makeCounter();
+    counterPromise.onRpcBroken(() => { errors.push("canceled"); }, {signal: canceled.signal});
+    counterPromise.onRpcBroken(() => { errors.push("kept"); });
+
+    await counterPromise;
+
+    // Abort only after the migration has happened.
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["kept"]);
+  });
+
+  it("removes the callback when the stub it was registered on is disposed", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+
+    // Register on a dup, so that disposing it leaves the underlying import (and the registration
+    // made through `stub` below) alive.
+    let dup = stub.dup();
+    dup.onRpcBroken(() => { errors.push("dup"); });
+    stub.onRpcBroken(() => { errors.push("stub"); });
+
+    dup[Symbol.dispose]();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    // Disposing the dup drops its own callback even though no AbortSignal was involved. The
+    // callback registered on `stub` still fires.
+    expect(errors).toStrictEqual(["stub"]);
+  });
+
+  it("removes the callback when the promise it was registered on is disposed", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+      hangingCall(): Promise<Counter> {
+        return new Promise(() => {});  // never resolves
+      }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+
+    let errors: string[] = [];
+
+    let hangingPromise = stub.hangingCall();
+    hangingPromise.onRpcBroken(() => { errors.push("hangingCall"); });
+    stub.onRpcBroken(() => { errors.push("stub"); });
+
+    hangingPromise[Symbol.dispose]();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["stub"]);
+  });
+
+  it("honors independent signals on separate dups of the same import", async () => {
+    class TestBroken extends RpcTarget {
+      getValue() { return 42; }
+    }
+
+    let harness = new TestHarness(new TestBroken());
+    let stub = harness.stub;
+    expect(await stub.getValue()).toBe(42);
+
+    let errors: string[] = [];
+
+    // Two dups of the same underlying import, each registering with its own signal. Aborting one
+    // signal must drop only that dup's callback, since each hook composes its own signal with the
+    // shared entry.
+    let dup1 = stub.dup();
+    let dup2 = stub.dup();
+    let canceled = new AbortController();
+    let kept = new AbortController();
+
+    dup1.onRpcBroken(() => { errors.push("dup1"); }, {signal: canceled.signal});
+    dup2.onRpcBroken(() => { errors.push("dup2"); }, {signal: kept.signal});
+
+    canceled.abort();
+
+    harness.clientTransport.forceReceiveError(new Error("test disconnect"));
+    await pumpMicrotasks();
+
+    expect(errors).toStrictEqual(["dup2"]);
+  });
 });
 
 // =======================================================================================
diff --git a/__tests__/signal.test.ts b/__tests__/signal.test.ts
new file mode 100644
index 00000000..e21344ad
--- /dev/null
+++ b/__tests__/signal.test.ts
@@ -0,0 +1,110 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the MIT license found in the LICENSE.txt file or at:
+//     https://opensource.org/license/mit
+
+// Tests for `anySignal()`, the `AbortSignal.any` shim in src/signal.ts. Covers both the native
+// path (when `AbortSignal.any` exists) and the hand-rolled fallback used on runtimes that lack it.
+
+import { expect, it, describe, afterEach } from "vitest"
+import { anySignal } from "../src/signal.js"
+
+// Run the given callback with `AbortSignal.any` removed, forcing `anySignal()` down its fallback
+// path. The original is always restored, even if the callback throws.
+function withoutNativeAny(fn: () => void) {
+  let original = Object.getOwnPropertyDescriptor(AbortSignal, "any");
+  // @ts-expect-error - deleting an optional static for the duration of the test.
+  delete AbortSignal.any;
+  try {
+    expect(AbortSignal.any).toBeUndefined();
+    fn();
+  } finally {
+    if (original) Object.defineProperty(AbortSignal, "any", original);
+  }
+}
+
+describe("anySignal", () => {
+  afterEach(() => {
+    // Guard against a test leaving `AbortSignal.any` deleted if it somehow escaped the finally.
+    expect(typeof AbortSignal.any).toBe("function");
+  });
+
+  describe("native path", () => {
+    it("aborts when any source aborts", () => {
+      let a = new AbortController();
+      let b = new AbortController();
+      let composite = anySignal([a.signal, b.signal]);
+
+      expect(composite.aborted).toBe(false);
+      b.abort(new Error("boom"));
+      expect(composite.aborted).toBe(true);
+      expect((composite.reason as Error).message).toBe("boom");
+    });
+
+    it("is already aborted when a source is already aborted", () => {
+      let a = new AbortController();
+      a.abort(new Error("pre"));
+      let composite = anySignal([a.signal, new AbortController().signal]);
+
+      expect(composite.aborted).toBe(true);
+      expect((composite.reason as Error).message).toBe("pre");
+    });
+  });
+
+  describe("fallback path (AbortSignal.any unavailable)", () => {
+    it("aborts when any source aborts later, propagating the reason", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        expect(composite.aborted).toBe(false);
+        b.abort(new Error("boom"));
+        expect(composite.aborted).toBe(true);
+        expect((composite.reason as Error).message).toBe("boom");
+      });
+    });
+
+    it("is already aborted when a source is already aborted, propagating the reason", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        a.abort(new Error("pre"));
+        let composite = anySignal([a.signal, new AbortController().signal]);
+
+        expect(composite.aborted).toBe(true);
+        expect((composite.reason as Error).message).toBe("pre");
+      });
+    });
+
+    it("only fires once even if multiple sources abort", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        let reasons: unknown[] = [];
+        composite.addEventListener("abort", () => { reasons.push(composite.reason); });
+
+        a.abort(new Error("first"));
+        b.abort(new Error("second"));
+
+        expect(reasons.length).toBe(1);
+        expect((reasons[0] as Error).message).toBe("first");
+      });
+    });
+
+    it("stops listening to a source once the composite has aborted", () => {
+      withoutNativeAny(() => {
+        let a = new AbortController();
+        let b = new AbortController();
+        let composite = anySignal([a.signal, b.signal]);
+
+        a.abort(new Error("first"));
+
+        // `b` outlives the composite; aborting it must not touch the already-settled reason, and
+        // its listener should have been dropped when the composite aborted.
+        b.abort(new Error("second"));
+        expect((composite.reason as Error).message).toBe("first");
+      });
+    });
+  });
+});
diff --git a/__type-tests__/rpc-promise-semantics.test.ts b/__type-tests__/rpc-promise-semantics.test.ts
index 92f80eea..947846c0 100644
--- a/__type-tests__/rpc-promise-semantics.test.ts
+++ b/__type-tests__/rpc-promise-semantics.test.ts
@@ -67,6 +67,11 @@ userPromise.onRpcBroken((_error) => {})
 counterPromise.onRpcBroken((_error) => {})
 idPromise.onRpcBroken((_error) => {})
 
+// The options bag is optional and takes an AbortSignal.
+userPromise.onRpcBroken((_error) => {}, {})
+counterPromise.onRpcBroken((_error) => {}, { signal: new AbortController().signal })
+idPromise.onRpcBroken((_error) => {}, { signal: undefined })
+
 expectAssignable<Promise<number>>(counterPromise.increment(3))
 expectAssignable<Promise<number>>(counterPromise.value)
 expectAssignable<Promise<string>>(userPromise.getName())
diff --git a/packages/capnweb-validate/src/internal/core.ts b/packages/capnweb-validate/src/internal/core.ts
index b88cf4d6..df164951 100644
--- a/packages/capnweb-validate/src/internal/core.ts
+++ b/packages/capnweb-validate/src/internal/core.ts
@@ -56,7 +56,10 @@ type WrapSide = "server" | "client";
 
 interface StubBase<T = unknown> extends Disposable {
   dup(): this;
-  onRpcBroken(callback: (error: unknown) => void): void;
+  onRpcBroken(
+    callback: (error: unknown) => void,
+    options?: { signal?: AbortSignal }
+  ): void;
   readonly __RPC_STUB_BRAND: T;
 }
 
diff --git a/src/core.ts b/src/core.ts
index 86d8c032..f2cf5c3b 100644
--- a/src/core.ts
+++ b/src/core.ts
@@ -2,7 +2,7 @@
 // Licensed under the MIT license found in the LICENSE.txt file or at:
 //     https://opensource.org/license/mit
 
-import type { RpcTargetBranded, __RPC_TARGET_BRAND } from "./types.js";
+import type { OnRpcBrokenOptions, RpcTargetBranded, __RPC_TARGET_BRAND } from "./types.js";
 import { WORKERS_MODULE_SYMBOL } from "./symbols.js"
 
 // Polyfill Symbol.dispose for browsers that don't support it yet
@@ -312,7 +312,13 @@ export abstract class StubHook {
   // a disposed payload) or it may reject. It's safe to call dispose() multiple times.
   abstract dispose(): void;
 
-  abstract onBroken(callback: (error: any) => void): void;
+  // Registers a callback to be invoked if this hook becomes permanently broken, e.g. because the
+  // connection was lost. If the hook is already broken, the callback may be invoked synchronously.
+  //
+  // An implementation that invokes or stores `callback` must honor `options.signal`: skip the
+  // registration entirely if the signal is already aborted, and drop the callback when it aborts.
+  // An implementation that delegates passes `options` through unchanged.
+  abstract onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void;
 }
 
 export class ErrorStubHook extends StubHook {
@@ -325,7 +331,10 @@ export class ErrorStubHook extends StubHook {
   pull(): RpcPayload | Promise<RpcPayload> { return Promise.reject(this.error); }
   ignoreUnhandledRejections(): void {}
   dispose(): void {}
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
+    // The caller already canceled, so stay quiet even though we could report right now.
+    if (options?.signal?.aborted) return;
+
     try {
       callback(this.error);
     } catch (err) {
@@ -519,8 +528,8 @@ export class RpcStub extends RpcTarget {
     }
   }
 
-  onRpcBroken(callback: (error: any) => void) {
-    this[RAW_STUB].hook.onBroken(callback);
+  onRpcBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions) {
+    this[RAW_STUB].hook.onBroken(callback, options);
   }
 
   map(func: (value: RpcPromise) => unknown): RpcPromise {
@@ -1843,13 +1852,13 @@ export class PayloadStubHook extends ValueStubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
     if (this.payload) {
       if (this.payload.value instanceof RpcStub) {
         // Payload is a single stub, we should forward onRpcBroken to it.
         // TODO: Consider prohibiting PayloadStubHook created around a single stub; should always
         //   use the underlying stub's hook instead?
-        this.payload.value.onRpcBroken(callback);
+        this.payload.value.onRpcBroken(callback, options);
       }
 
       // TODO: Should native stubs be able to implement onRpcBroken?
@@ -1964,7 +1973,7 @@ class TargetStubHook extends ValueStubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
     // TODO: Should RpcTargets be able to implement onRpcBroken?
   }
 }
@@ -2067,13 +2076,19 @@ export class PromiseStubHook extends StubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
+    if (options?.signal?.aborted) return;
+
     if (this.resolution) {
-      this.resolution.onBroken(callback);
+      this.resolution.onBroken(callback, options);
     } else {
       this.promise.then(hook => {
-        hook.onBroken(callback);
-      }, callback);
+        hook.onBroken(callback, options);
+      }, error => {
+        // The signal may have aborted while we were waiting for the promise.
+        if (options?.signal?.aborted) return;
+        callback(error);
+      });
     }
   }
 }
diff --git a/src/index.ts b/src/index.ts
index 3d80501a..a57566ef 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -7,7 +7,7 @@ import { serialize, deserialize, EncodingLevel } from "./serialize.js";
 import { RpcTransport, RpcTransportWithCustomEncoding, AnyRpcTransport, RpcSession as RpcSessionImpl, RpcSessionOptions } from "./rpc.js";
 import { RpcLimits, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH } from "./serialize.js";
 import { RpcTargetBranded, RpcCompatible, Stub, type RpcPromise as RpcPromiseType,
-         __RPC_TARGET_BRAND } from "./types.js";
+         type OnRpcBrokenOptions, __RPC_TARGET_BRAND } from "./types.js";
 import { newWebSocketRpcSession as newWebSocketRpcSessionImpl,
          newWorkersWebSocketRpcResponse, WebSocketTransport } from "./websocket.js";
 import { newHttpBatchRpcSession as newHttpBatchRpcSessionImpl,
@@ -23,7 +23,7 @@ forceInitStreams();
 export { serialize, deserialize, newWorkersWebSocketRpcResponse, newHttpBatchRpcResponse,
          nodeHttpBatchRpcResponse, WebSocketTransport, DEFAULT_LIMITS, DEFAULT_MAX_DEPTH };
 export type { RpcTransport, RpcTransportWithCustomEncoding, AnyRpcTransport,
-         RpcSessionOptions, RpcCompatible, EncodingLevel, RpcLimits };
+         RpcSessionOptions, RpcCompatible, EncodingLevel, RpcLimits, OnRpcBrokenOptions };
 
 // Hack the type system to make RpcStub's types work nicely!
 /**
diff --git a/src/map.ts b/src/map.ts
index b85cd7ca..ec660b43 100644
--- a/src/map.ts
+++ b/src/map.ts
@@ -4,6 +4,7 @@
 
 import { StubHook, PropertyPath, RpcPayload, RpcStub, RpcPromise, withCallInterceptor, ErrorStubHook, mapImpl, PayloadStubHook, unwrapStubAndPath, unwrapStubNoProperties } from "./core.js";
 import { Devaluator, Exporter, Importer, ExportId, ImportId, Evaluator, RpcLimits, DEFAULT_LIMITS } from "./serialize.js";
+import type { OnRpcBrokenOptions } from "./types.js";
 
 let currentMapBuilder: MapBuilder | undefined;
 
@@ -227,7 +228,7 @@ class MapVariableHook extends StubHook {
     // Probably never called but whatever.
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
     throwMapperBuilderUseError();
   }
 }
diff --git a/src/rpc.ts b/src/rpc.ts
index 1ea76c12..a55bb501 100644
--- a/src/rpc.ts
+++ b/src/rpc.ts
@@ -4,6 +4,8 @@
 
 import { StubHook, RpcPayload, RpcStub, PropertyPath, PayloadStubHook, ErrorStubHook, RpcTarget, unwrapStubAndPath, streamImpl } from "./core.js";
 import { Devaluator, Evaluator, ExportId, ImportId, Exporter, Importer, serialize, EncodingLevel, RpcLimits, DEFAULT_LIMITS } from "./serialize.js";
+import type { OnRpcBrokenOptions } from "./types.js";
+import { anySignal } from "./signal.js";
 
 /**
  * Interface for a string-based RPC transport. This is the default transport type — no
@@ -192,9 +194,13 @@ class ImportTableEntry {
   private activePull?: PromiseWithResolvers<void>;
   public resolution?: StubHook;
 
-  // List of integer indexes into session.onBrokenCallbacks which are callbacks registered on
-  // this import. Initialized on first use (so `undefined` is the same as an empty list).
-  private onBrokenRegistrations?: number[];
+  // Lazily initialized abort controller for local deregistration.
+  private abortController?: AbortController;
+
+  // Callbacks registered on this import, as indexes into session.onBrokenCallbacks, paired with the
+  // options they were registered with, so that resolve() can re-register them on the resolution.
+  // Initialized on first use (so `undefined` is the same as an empty list).
+  private onBrokenRegistrations?: {index: number, options?: OnRpcBrokenOptions}[];
 
   resolve(resolution: StubHook) {
     // TODO: Need embargo handling here? PayloadStubHook needs to be wrapped in a
@@ -217,10 +223,14 @@ class ImportTableEntry {
     if (this.onBrokenRegistrations) {
       // Delete all our callback registrations from this session and re-register them on the
       // target stub.
-      for (let i of this.onBrokenRegistrations) {
-        let callback = this.session.onBrokenCallbacks[i];
+      for (let {index, options} of this.onBrokenRegistrations) {
+        let callback = this.session.onBrokenCallbacks[index];
+        if (callback === undefined) {
+          // Already canceled via `options.signal`, so there's nothing to migrate.
+          continue;
+        }
         let endIndex = this.session.onBrokenCallbacks.length;
-        resolution.onBroken(callback);
+        resolution.onBroken(callback, options);
         if (this.session.onBrokenCallbacks[endIndex] === callback) {
           // Oh, calling onBroken() just registered the callback back on this connection again.
           // But when the connection dies, we want all the callbacks to be called in the order in
@@ -230,7 +240,7 @@ class ImportTableEntry {
           delete this.session.onBrokenCallbacks[endIndex];
         } else {
           // The callback is now registered elsewhere, so delete it from our session.
-          delete this.session.onBrokenCallbacks[i];
+          delete this.session.onBrokenCallbacks[index];
         }
       }
       this.onBrokenRegistrations = undefined;
@@ -254,6 +264,7 @@ class ImportTableEntry {
   dispose() {
     if (this.resolution) {
       this.resolution.dispose();
+      this.abortController?.abort();
     } else {
       this.abort(new Error("RPC was canceled because the RpcPromise was disposed."));
       this.sendRelease();
@@ -271,19 +282,32 @@ class ImportTableEntry {
 
       // The RpcSession itself will have called all our callbacks so we don't need to track the
       // registrations anymore.
+      this.abortController?.abort();
+      this.abortController = undefined;
       this.onBrokenRegistrations = undefined;
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
+    // The caller already canceled, so don't register.
+    if (options?.signal?.aborted) return;
+
     if (this.resolution) {
-      this.resolution.onBroken(callback);
+      this.resolution.onBroken(callback, options);
     } else {
       let index = this.session.onBrokenCallbacks.length;
       this.session.onBrokenCallbacks.push(callback);
 
       if (!this.onBrokenRegistrations) this.onBrokenRegistrations = [];
-      this.onBrokenRegistrations.push(index);
+      this.onBrokenRegistrations.push({ index, options });
+
+      let deregister = () => { delete this.session.onBrokenCallbacks[index]; };
+      (this.abortController ??= new AbortController()).signal.addEventListener("abort", deregister);
+      options?.signal?.addEventListener("abort",
+        deregister,
+        // Stop listening once this entry is disposed.
+        { once: true, signal: this.abortController!.signal },
+      );
     }
   }
 
@@ -298,6 +322,11 @@ class ImportTableEntry {
 class RpcImportHook extends StubHook {
   public entry?: ImportTableEntry;  // undefined when we're disposed
 
+  // Lazily initialized abort controller for deregistering the onBroken callbacks registered through
+  // this hook. It belongs to the hook rather than the import entry because the entry is shared by
+  // every hook pointing at it (via dup()) and is only disposed once the last of them is.
+  private abortController?: AbortController;
+
   // `pulling` is true if we already expect that this import is going to be resolved later, and
   // null if this import is not allowed to be pulled (i.e. it's a stub not a promise).
   constructor(public isPromise: boolean, entry: ImportTableEntry) {
@@ -395,6 +424,9 @@ class RpcImportHook extends StubHook {
   dispose(): void {
     let entry = this.entry;
     this.entry = undefined;
+    let controller = this.abortController;
+    this.abortController = undefined;
+    controller?.abort();
     if (entry) {
       if (--entry.localRefcount === 0) {
         entry.dispose();
@@ -402,9 +434,13 @@ class RpcImportHook extends StubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
     if (this.entry) {
-      this.entry.onBroken(callback);
+      this.abortController ??= new AbortController();
+      let signal = options?.signal
+          ? anySignal([this.abortController.signal, options.signal])
+          : this.abortController.signal;
+      this.entry.onBroken(callback, { signal });
     }
   }
 }
diff --git a/src/signal.ts b/src/signal.ts
new file mode 100644
index 00000000..c049fab9
--- /dev/null
+++ b/src/signal.ts
@@ -0,0 +1,34 @@
+// Copyright (c) 2026 Cloudflare, Inc.
+// Licensed under the MIT license found in the LICENSE.txt file or at:
+//     https://opensource.org/license/mit
+
+// Alternative for AbortSignal.any.
+export function anySignal(signals: AbortSignal[]): AbortSignal {
+  if (AbortSignal.any) {
+    // Prefer the native version if available.
+    return AbortSignal.any(signals);
+  }
+
+  let controller = new AbortController();
+
+  for (let signal of signals) {
+    if (signal.aborted) {
+      controller.abort(signal.reason);
+      return controller.signal;
+    }
+  }
+
+  for (let signal of signals) {
+    signal.addEventListener("abort", () => {
+      controller.abort(signal.reason);
+    }, {
+      once: true,
+
+      // Drop this listener as soon as the composite aborts, whichever source caused it. Otherwise
+      // a source signal that outlives the composite retains a listener.
+      signal: controller.signal,
+    });
+  }
+
+  return controller.signal;
+}
diff --git a/src/streams.ts b/src/streams.ts
index e3ff1f1c..72aa2ec9 100644
--- a/src/streams.ts
+++ b/src/streams.ts
@@ -5,6 +5,7 @@
 import {
   StubHook, RpcPayload, PropertyPath, ErrorStubHook, PayloadStubHook, PromiseStubHook, streamImpl
 } from "./core.js";
+import type { OnRpcBrokenOptions } from "./types.js";
 
 // =======================================================================================
 // WritableStreamStubHook - wraps a local WritableStream for export
@@ -116,9 +117,10 @@ class WritableStreamStubHook extends StubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
     // WritableStream stubs don't really have a "broken" state in the same way.
-    // The caller would notice when write/close/abort fails.
+    // The caller would notice when write/close/abort fails. We never store `callback`, so
+    // `options.signal` has nothing to cancel.
   }
 }
 
@@ -518,8 +520,8 @@ class ReadableStreamStubHook extends StubHook {
     }
   }
 
-  onBroken(callback: (error: any) => void): void {
-    // ReadableStream stubs don't have a "broken" state.
+  onBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void {
+    // ReadableStream stubs don't have a "broken" state, so `options.signal` has nothing to cancel.
   }
 }
 
diff --git a/src/types.d.ts b/src/types.d.ts
index 5a595ff0..232f17ae 100644
--- a/src/types.d.ts
+++ b/src/types.d.ts
@@ -49,12 +49,19 @@ export type RpcCompatible<T> =
   // Serialized as stubs, see `Stubify`
   | Stubable;
 
+// Options for `onRpcBroken()`.
+export interface OnRpcBrokenOptions {
+  // If given, aborting the signal drops the callback. A signal that is already aborted registers
+  // nothing, even if the stub is already broken.
+  signal?: AbortSignal;
+}
+
 // Base type for all RPC stubs, including common memory management methods.
 // `T` is used as a marker type for unwrapping `Stub`s later.
 interface StubBase<T = unknown> extends Disposable {
   [__RPC_STUB_BRAND]: T;
   dup(): this;
-  onRpcBroken(callback: (error: any) => void): void;
+  onRpcBroken(callback: (error: any) => void, options?: OnRpcBrokenOptions): void;
 }
 export type Stub<T extends RpcCompatible<T>> =
     T extends object ? Provider<T> & StubBase<T> : StubBase<T>;
diff --git a/vitest.config.ts b/vitest.config.ts
index 4752cede..d327ad3e 100644
--- a/vitest.config.ts
+++ b/vitest.config.ts
@@ -16,10 +16,10 @@ export default defineConfig({
         test: {
           name: 'node',
           // We throw flow-control test under Node only because it's testing straightforward
-          // JavaScript -- no need to run it on every runtime. The limits tests are likewise
-          // plain JavaScript receive-side guards, so Node coverage is sufficient.
+          // JavaScript -- no need to run it on every runtime. The limits and signal tests are
+          // likewise plain JavaScript, so Node coverage is sufficient.
           include: ['__tests__/index.test.ts', '__tests__/flow-control.test.ts',
-                    '__tests__/limits.test.ts',
+                    '__tests__/limits.test.ts', '__tests__/signal.test.ts',
                     'packages/capnweb-validate/__tests__/**/*.test.ts'],
           environment: 'node',
         },