diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/index.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/index.ts
index f1c7cb73d..fb4a52305 100644
--- a/packages/@aws-cdk/toolkit-lib/lib/api/io/index.ts
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/index.ts
@@ -1,3 +1,4 @@
 export * from './io-host';
 export * from './io-message';
 export * from './toolkit-action';
+export * from './listeners';
diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/io-message.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/io-message.ts
index 692f0461b..22700eff7 100644
--- a/packages/@aws-cdk/toolkit-lib/lib/api/io/io-message.ts
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/io-message.ts
@@ -82,3 +82,21 @@ export interface IoRequest<T, U> extends IoMessage<T> {
 
   readonly code: IoMessageCode;
 }
+
+/**
+ * A message matcher
+ *
+ * Decides whether a message matches a listener, narrowing its payload type `T`.
+ */
+export interface IMessageMatcher<T> {
+  is(msg: IoMessage<unknown>): msg is IoMessage<T>;
+}
+
+/**
+ * A request matcher.
+ *
+ * Carries the response type `U` so an answer can be typed.
+ */
+export interface IRequestMatcher<T, U> extends IMessageMatcher<T> {
+  is(msg: IoMessage<unknown>): msg is IoRequest<T, U>;
+}
diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/listeners.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/listeners.ts
new file mode 100644
index 000000000..b1250ff27
--- /dev/null
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/listeners.ts
@@ -0,0 +1,317 @@
+import type { IIoHost } from './io-host';
+import type { IoMessage, IoRequest, IMessageMatcher, IRequestMatcher, IoMessageCode, IoMessageLevel } from './io-message';
+import { ListenerRegistry } from './private/listener-registry';
+import type { MessageListenerResultOrPromise } from './private/listener-registry';
+
+// Re-export the listener result vocabulary from the shared registry so it is
+// part of the public API alongside `withListeners`.
+export type { MessageListenerResult, MessageListenerResultOrPromise } from './private/listener-registry';
+
+/**
+ * A predicate that decides whether a listener applies to a message.
+ *
+ * Use one to match a family of messages instead of a single code — e.g. every
+ * message of a given level, or any of several codes. It touches only the public
+ * `IoMessage` shape.
+ *
+ * @example
+ * ```ts
+ * host.on((msg) => msg.level === 'warn', listener);
+ * ```
+ */
+export type MessagePredicate = (msg: IoMessage<unknown>) => boolean;
+
+/**
+ * Options for `respond`/ `respondOnce`.
+ */
+export interface RespondOptions {
+  /**
+   * Whether also to suppress surfacing the question text.
+   * @default true - answer silently
+   */
+  readonly suppressQuestion?: boolean;
+}
+
+/**
+ * An `IIoHost` that additionally lets listeners be attached to it.
+ *
+ * The result of `withListeners`. It is still an `IIoHost`, so it drops straight
+ * into the `ioHost` slot of a new `Toolkit`, and it exposes methods to observe,
+ * reshape, or answer individual messages without subclassing a host.
+ *
+ * Listeners are keyed on a message `code` (e.g. `'CDK_TOOLKIT_I2901'`). The
+ * codes are listed in the message registry:
+ * https://docs.aws.amazon.com/cdk/api/toolkit-lib/message-registry/
+ */
+export interface IoHostWithListeners extends IIoHost {
+  /**
+   * Register a listener that is invoked for every message that matches — either
+   * a single message `code` (e.g. `'CDK_TOOLKIT_I2901'`), or a
+   * `MessagePredicate` that matches a family of messages.
+   *
+   * The listener may return a `MessageListenerResult` to update the message
+   * text and/or level or prevent the default handling (asking the wrapped host
+   * to write it); returning nothing leaves the message untouched. The listener
+   * may be async (return a `Promise`); it is awaited before the message is
+   * handled further. Returns a function that removes the listener again.
+   *
+   * The message payload is delivered as `unknown`; see the message registry for
+   * the shape carried by each code.
+   *
+   * @example
+   * ```ts
+   * const dispose = host.on('CDK_TOOLKIT_I2901', async (msg) => {
+   *   myCount += (msg.data as StackDetailsPayload).stacks.length;
+   *   await persist(myCount);
+   * });
+   * ```
+   *
+   * @example
+   * ```ts
+   * // A predicate matches a family of messages, e.g. every warning:
+   * const dispose = host.on((msg) => msg.level === 'warn', (msg) => {
+   *   warnings.push(msg.message);
+   * });
+   * ```
+   */
+  on<T>(
+    matcher: IMessageMatcher<T>,
+    listener: (msg: IoMessage<T>) => MessageListenerResultOrPromise,
+  ): () => void;
+  on(
+    selector: IoMessageCode | MessagePredicate,
+    listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
+  ): () => void;
+
+  /**
+   * Like `on`, but the listener is automatically removed after it has been
+   * invoked once.
+   */
+  once<T>(
+    matcher: IMessageMatcher<T>,
+    listener: (msg: IoMessage<T>) => MessageListenerResultOrPromise,
+  ): () => void;
+  once(
+    selector: IoMessageCode | MessagePredicate,
+    listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
+  ): () => void;
+
+  /**
+   * Register a formatter that replaces the printed text of matching messages —
+   * selected by a message `code`, a `MessagePredicate`, or a matcher. This lets
+   * a caller define _how_ a toolkit message is presented without the host
+   * needing to know about it.
+   *
+   * Optionally pass a `level` to also override the message's level. Syntactic
+   * sugar for an `on` listener that returns the new `message` and `level`.
+   * Returns a function that removes the formatter again.
+   *
+   * @example
+   * ```ts
+   * const dispose = host.rewrite('CDK_TOOLKIT_I2901', (msg) =>
+   *   `${(msg.data as StackDetailsPayload).stacks.length} stacks`);
+   * ```
+   */
+  rewrite<T>(
+    matcher: IMessageMatcher<T>,
+    formatter: (msg: IoMessage<T>) => string,
+    level?: IoMessageLevel,
+  ): () => void;
+  rewrite(
+    selector: IoMessageCode | MessagePredicate,
+    formatter: (msg: IoMessage<unknown>) => string,
+    level?: IoMessageLevel,
+  ): () => void;
+
+  /**
+   * Like `rewrite`, but the formatter is automatically removed after it has
+   * been applied once.
+   */
+  rewriteOnce<T>(
+    matcher: IMessageMatcher<T>,
+    formatter: (msg: IoMessage<T>) => string,
+    level?: IoMessageLevel,
+  ): () => void;
+  rewriteOnce(
+    selector: IoMessageCode | MessagePredicate,
+    formatter: (msg: IoMessage<unknown>) => string,
+    level?: IoMessageLevel,
+  ): () => void;
+
+  /**
+   * Answer a request (by its code) on the caller's behalf with a fixed value,
+   * so the wrapped host is not asked to prompt. Syntactic sugar for an `on`
+   * listener that responds with the value and prevents the default; for
+   * conditional answers or to also reword the question, use `on`/`once`
+   * directly. Returns a function that removes the responder again.
+   *
+   * By default the question is answered silently; pass
+   * `{ suppressQuestion: false }` to still surface the question while answering
+   * it.
+   *
+   * @example
+   * ```ts
+   * const dispose = host.respond('CDK_TOOLKIT_I7010', true);
+   * ```
+   */
+  respond<T, U>(
+    matcher: IRequestMatcher<T, U>,
+    value: U,
+    options?: RespondOptions,
+  ): () => void;
+  respond(
+    code: IoMessageCode,
+    value: unknown,
+    options?: RespondOptions,
+  ): () => void;
+
+  /**
+   * Like `respond`, but the answer is given only once and then removed.
+   */
+  respondOnce<T, U>(
+    matcher: IRequestMatcher<T, U>,
+    value: U,
+    options?: RespondOptions,
+  ): () => void;
+  respondOnce(
+    code: IoMessageCode,
+    value: unknown,
+    options?: RespondOptions,
+  ): () => void;
+}
+
+/**
+ * Wrap any `IIoHost` so listeners can be attached to it.
+ *
+ * The returned host forwards `notify` and `requestResponse` to the host you
+ * pass in, running any matching listeners in registration order in between. On
+ * `notify` it runs the listeners, applies any rewrite, and skips the wrapped
+ * host's write if a listener prevented the default. On `requestResponse` it runs
+ * them too, so a listener can rewrite the prompt text or answer it with
+ * `respond`, in which case the request resolves without asking the wrapped host.
+ *
+ * The result is still an `IIoHost`, so it drops straight into a new `Toolkit`.
+ * Its lifecycle stays yours: you wrap a host, register listeners, and pass it to
+ * the toolkit, all explicit.
+ *
+ * @example
+ * ```ts
+ * const base = new NonInteractiveIoHost();     // or your own host
+ * const host = withListeners(base);
+ * host.on('CDK_TOOLKIT_I2901', (m) => { count += (m.data as StackDetailsPayload).stacks.length; });
+ * const toolkit = new Toolkit({ ioHost: host });
+ * ```
+ */
+export function withListeners(host: IIoHost): IoHostWithListeners {
+  return new ListeningIoHost(host);
+}
+
+/**
+ * An `IIoHost` that runs a `ListenerRegistry` around a wrapped host.
+ *
+ * The registry is the shared listener engine (also used by the CLI's terminal
+ * host); this class adds the wrapped-host plumbing that turns it into an
+ * `IIoHost`, and adapts the public code-keyed API onto the registry's matchers.
+ */
+class ListeningIoHost implements IoHostWithListeners {
+  private readonly registry = new ListenerRegistry();
+
+  public constructor(private readonly inner: IIoHost) {
+  }
+
+  public async notify(msg: IoMessage<unknown>): Promise<void> {
+    const { message, preventDefault } = await this.registry.apply(msg);
+    if (preventDefault) {
+      return;
+    }
+    return this.inner.notify(message);
+  }
+
+  public async requestResponse<T>(msg: IoRequest<unknown, T>): Promise<T> {
+    const { message, preventDefault, responded } = await this.registry.apply(msg);
+
+    // A listener suppressed the default handling: resolve with the (possibly
+    // overridden) default response without asking the wrapped host.
+    if (preventDefault) {
+      return message.defaultResponse;
+    }
+
+    // A listener answered the request but wants the question surfaced: show it
+    // via the wrapped host, then resolve with the answer instead of prompting.
+    if (responded) {
+      await this.inner.notify(message);
+      return message.defaultResponse;
+    }
+
+    // No listener answered: let the wrapped host resolve the (possibly reworded)
+    // request as it sees fit (it may prompt, or use its own default).
+    return this.inner.requestResponse(message);
+  }
+
+  public on(
+    selector: IMessageMatcher<any> | IoMessageCode | MessagePredicate,
+    listener: (msg: IoMessage<any>) => MessageListenerResultOrPromise,
+  ): () => void {
+    return this.registry.on(toMatcher(selector), listener);
+  }
+
+  public once(
+    selector: IMessageMatcher<any> | IoMessageCode | MessagePredicate,
+    listener: (msg: IoMessage<any>) => MessageListenerResultOrPromise,
+  ): () => void {
+    return this.registry.once(toMatcher(selector), listener);
+  }
+
+  public rewrite(
+    selector: IMessageMatcher<any> | IoMessageCode | MessagePredicate,
+    formatter: (msg: IoMessage<any>) => string,
+    level?: IoMessageLevel,
+  ): () => void {
+    return this.registry.rewrite(toMatcher(selector), formatter, level);
+  }
+
+  public rewriteOnce(
+    selector: IMessageMatcher<any> | IoMessageCode | MessagePredicate,
+    formatter: (msg: IoMessage<any>) => string,
+    level?: IoMessageLevel,
+  ): () => void {
+    return this.registry.rewriteOnce(toMatcher(selector), formatter, level);
+  }
+
+  public respond(
+    selector: IRequestMatcher<any, any> | IoMessageCode,
+    value: unknown,
+    options: RespondOptions = {},
+  ): () => void {
+    return this.registry.respond(toMatcher(selector), value, options.suppressQuestion);
+  }
+
+  public respondOnce(
+    selector: IRequestMatcher<any, any> | IoMessageCode,
+    value: unknown,
+    options: RespondOptions = {},
+  ): () => void {
+    return this.registry.respondOnce(toMatcher(selector), value, options.suppressQuestion);
+  }
+}
+
+/**
+ * Resolve a code-or-predicate selector into the predicate the registry matches
+ * on: a code becomes a code-equality check, a predicate is used as-is.
+ */
+function toMatcher(selector: IMessageMatcher<any> | IoMessageCode | MessagePredicate): MessagePredicate {
+  if (typeof selector === 'string') {
+    return byCode(selector);
+  }
+  if (typeof selector === 'function') {
+    return selector;
+  }
+  return (msg) => selector.is(msg);
+}
+
+/**
+ * Build a matcher that fires for messages carrying the given code.
+ */
+function byCode(code: IoMessageCode): MessagePredicate {
+  return (msg) => msg.code === code;
+}
diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/private/index.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/index.ts
index e795860ff..77af47151 100644
--- a/packages/@aws-cdk/toolkit-lib/lib/api/io/private/index.ts
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/index.ts
@@ -5,4 +5,5 @@ export * from './level-priority';
 export * from './span';
 export * from './message-maker';
 export * from './messages';
+export * from './listener-registry';
 export * from './types';
diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/private/listener-registry.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/listener-registry.ts
new file mode 100644
index 000000000..9796a10b6
--- /dev/null
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/listener-registry.ts
@@ -0,0 +1,342 @@
+import type { IoMessage, IoMessageLevel, IMessageMatcher } from '../io-message';
+
+/**
+ * The result a message listener may return to influence how a message is handled.
+ *
+ * A listener may update the message _text_ and/or its _level_; it cannot change
+ * any other field of the message (such as its `code`), which keeps the
+ * code-keyed listener registry valid.
+ */
+export interface MessageListenerResult {
+  /**
+   * Replace the text that is printed for this message.
+   *
+   * @default - the message text is left unchanged
+   */
+  readonly message?: string;
+
+  /**
+   * Override the level of this message.
+   *
+   * A host may use the level for verbosity filtering and for deciding where to
+   * route the message, so overriding it can change whether and where the message
+   * is shown. The `code` is intentionally left unchanged.
+   *
+   * @default - the message level is left unchanged
+   */
+  readonly level?: IoMessageLevel;
+
+  /**
+   * Skip the default handling of the message.
+   *
+   * For a notification this means the host is not asked to handle it. For a
+   * request it stops processing entirely: the host is not asked to prompt, and
+   * the request resolves with its (possibly `respond`-overridden) default
+   * response.
+   *
+   * @default false
+   */
+  readonly preventDefault?: boolean;
+
+  /**
+   * For requests only: the value to resolve the request with. It is folded into
+   * the request's default response and skips the prompt (the host is not asked
+   * to answer). The question is still surfaced unless `preventDefault` is also
+   * set. Ignored for plain notifications.
+   *
+   * The presence of the key is what matters, so `false`/`0`/`''` are valid
+   * answers. Use the `respond`/`respondOnce` helpers for the common case.
+   *
+   * @default - this listener does not supply a response
+   */
+  readonly respond?: unknown;
+}
+
+/**
+ * What a message listener may return: nothing, a `MessageListenerResult`, or a
+ * `Promise` of either.
+ *
+ * Listeners may be async. The registry awaits each listener before running the
+ * next, so registration order — and the cumulative effect on the message — is
+ * preserved regardless of whether listeners are sync or async.
+ */
+export type MessageListenerResultOrPromise = void | MessageListenerResult | Promise<void | MessageListenerResult>;
+
+/**
+ * Selects which messages a listener applies to.
+ *
+ * Either an `IMessageMatcher` — the makers implement this, so a maker fires for
+ * its own messages — or a custom *predicate* over the message (e.g. to match a
+ * family of codes, or on the message level). Use `matchAny` to combine several.
+ */
+export type MessageSelector<T> =
+  | IMessageMatcher<T>
+  | ((msg: IoMessage<any>) => boolean);
+
+/**
+ * A function a listener runs when a matching message appears.
+ */
+export type MessageListenerFn = (msg: IoMessage<any>) => MessageListenerResultOrPromise;
+
+/**
+ * A registered message listener.
+ */
+interface MessageListener {
+  readonly once: boolean;
+  readonly fn: MessageListenerFn;
+  /**
+   * Decides which messages this listener applies to. For a listener registered
+   * with a maker this matches by `code`; for one registered with a predicate it
+   * is the predicate itself.
+   */
+  readonly matches: (msg: IoMessage<unknown>) => boolean;
+  /**
+   * Whether this is one of a host's own internal listeners (e.g. stack-activity
+   * routing). Internal listeners are not removed by `removeUserListeners`.
+   *
+   * @default false - a user listener registered via `on`/`once`/`rewrite`/`respond`
+   */
+  readonly internal?: boolean;
+}
+
+/**
+ * The outcome of running the registry's listeners over a single message.
+ */
+export interface AppliedListeners<T> {
+  /**
+   * The (possibly rewritten) message to hand to the host's default handling.
+   */
+  readonly message: T;
+
+  /**
+   * Whether a listener asked to skip the default handling.
+   */
+  readonly preventDefault: boolean;
+
+  /**
+   * Whether a listener answered a request (its answer is folded into
+   * `message.defaultResponse`).
+   */
+  readonly responded: boolean;
+}
+
+/**
+ * A registry of message listeners, keyed by code or predicate, run in
+ * registration order.
+ *
+ * This is the shared listener engine: both the CLI's terminal host and the
+ * public `withListeners` wrapper own one and run their messages through it, so
+ * there is a single implementation of matching, ordering, rewriting, and
+ * request answering. A host composes a registry and does its own I/O (writing,
+ * prompting, telemetry) around `apply`.
+ */
+export class ListenerRegistry {
+  // Listeners in registration order. Each carries a matcher (by code, or a
+  // custom predicate). See `on`/`once`/`rewrite`/`respond`.
+  private readonly listeners: MessageListener[] = [];
+
+  /**
+   * Register a listener that is invoked for every message that matches the
+   * selector. Returns a function that removes the listener again.
+   */
+  public on<T>(
+    selector: IMessageMatcher<T> | ((msg: IoMessage<any>) => msg is IoMessage<T>),
+    listener: (msg: IoMessage<T>) => MessageListenerResultOrPromise,
+  ): () => void;
+  public on(
+    predicate: (msg: IoMessage<any>) => boolean,
+    listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
+  ): () => void;
+  public on(selector: MessageSelector<any>, listener: MessageListenerFn): () => void {
+    return this.add({ once: false, fn: listener, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Like `on`, but the listener is automatically removed after it has been
+   * invoked once.
+   */
+  public once<T>(
+    selector: IMessageMatcher<T> | ((msg: IoMessage<any>) => msg is IoMessage<T>),
+    listener: (msg: IoMessage<T>) => MessageListenerResultOrPromise,
+  ): () => void;
+  public once(
+    predicate: (msg: IoMessage<any>) => boolean,
+    listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
+  ): () => void;
+  public once(selector: MessageSelector<any>, listener: MessageListenerFn): () => void {
+    return this.add({ once: true, fn: listener, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Register a formatter that replaces the printed text of matching messages,
+   * optionally also overriding the level. Syntactic sugar for an `on` listener
+   * that returns `{ message, level? }`.
+   */
+  public rewrite(
+    selector: MessageSelector<any>,
+    formatter: (msg: IoMessage<any>) => string,
+    level?: IoMessageLevel,
+  ): () => void {
+    const fn = (msg: IoMessage<any>) => ({ message: formatter(msg), ...(level !== undefined ? { level } : {}) });
+    return this.add({ once: false, fn, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Like `rewrite`, but the formatter is removed after it has been applied once.
+   */
+  public rewriteOnce(
+    selector: MessageSelector<any>,
+    formatter: (msg: IoMessage<any>) => string,
+    level?: IoMessageLevel,
+  ): () => void {
+    const fn = (msg: IoMessage<any>) => ({ message: formatter(msg), ...(level !== undefined ? { level } : {}) });
+    return this.add({ once: true, fn, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Answer a request (by its code) with a fixed value so the host does not
+   * prompt. Syntactic sugar for an `on` listener returning
+   * `{ respond: value, preventDefault: suppressQuestion }`.
+   *
+   * @param suppressQuestion - whether to also suppress surfacing the question
+   *   text. Defaults to `true` (answer silently).
+   */
+  public respond(selector: MessageSelector<any>, value: unknown, suppressQuestion = true): () => void {
+    const fn = (msg: IoMessage<unknown>) => ('defaultResponse' in msg ? { respond: value, preventDefault: suppressQuestion } : undefined);
+    return this.add({ once: false, fn, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Like `respond`, but the answer is given only once and then removed.
+   */
+  public respondOnce(selector: MessageSelector<any>, value: unknown, suppressQuestion = true): () => void {
+    const fn = (msg: IoMessage<unknown>) => ('defaultResponse' in msg ? { respond: value, preventDefault: suppressQuestion } : undefined);
+    return this.add({ once: true, fn, matches: messageMatcher(selector) });
+  }
+
+  /**
+   * Register one of the host's own internal listeners (e.g. stack-activity
+   * routing). Unlike listeners added via `on`/`once`/etc., an internal listener
+   * survives `removeUserListeners`. Returns a function that removes it.
+   */
+  public addInternal(matches: (msg: IoMessage<unknown>) => boolean, fn: MessageListenerFn): () => void {
+    return this.add({ once: false, internal: true, fn, matches });
+  }
+
+  /**
+   * Remove every listener registered via `on`/`once`/`rewrite`/`respond`,
+   * keeping the host's internal listeners so the host keeps working afterwards.
+   */
+  public removeUserListeners(): void {
+    // Drop user listeners in place (preserving array identity for any
+    // outstanding dispose closures); keep the host's internal listeners.
+    for (let i = this.listeners.length - 1; i >= 0; i--) {
+      if (!this.listeners[i].internal) {
+        this.listeners.splice(i, 1);
+      }
+    }
+  }
+
+  /**
+   * Run every registered listener that matches the message, in registration
+   * order. A listener matches by its code (maker) or its custom predicate.
+   *
+   * A listener may update the message text/level (passed on to subsequent
+   * listeners and the host), prevent the default handling, or (for requests)
+   * answer it. `once` listeners are removed after they have run. Matching is
+   * decided against the message as emitted, so a rewrite by an earlier listener
+   * does not change which later listeners apply.
+   *
+   * Returns the (possibly updated) message, whether the default handling was
+   * prevented, and whether a listener answered the request (folded into the
+   * message's `defaultResponse`).
+   */
+  public async apply<T extends IoMessage<unknown>>(msg: T): Promise<AppliedListeners<T>> {
+    let current = msg;
+    let preventDefault = false;
+    let responded = false;
+    // Iterate over a copy so that `once` listeners can remove themselves safely.
+    for (const listener of [...this.listeners]) {
+      // Match against the emitted message; a listener receives the cumulatively
+      // transformed `current` message.
+      if (!listener.matches(msg)) {
+        continue;
+      }
+
+      // Claim a `once` listener before the await; a concurrent `apply` that
+      // already removed it (index < 0) skips it, so it fires exactly once.
+      if (listener.once) {
+        const index = this.listeners.indexOf(listener);
+        if (index < 0) {
+          continue;
+        }
+        this.listeners.splice(index, 1);
+      }
+
+      // Listeners may be async; await each one before running the next so the
+      // cumulative effect on the message stays order-deterministic.
+      const result = await listener.fn(current);
+
+      if (result) {
+        if (result.message !== undefined) {
+          current = { ...current, message: result.message };
+        }
+        if (result.level !== undefined) {
+          current = { ...current, level: result.level };
+        }
+        if (result.preventDefault) {
+          preventDefault = true;
+        }
+        // The presence of the key is what matters (so `false`/`0`/`''` are valid
+        // answers); `'defaultResponse' in msg` tells a request from a notification.
+        if ('respond' in result && 'defaultResponse' in msg) {
+          current = { ...current, defaultResponse: result.respond };
+          responded = true;
+        }
+      }
+    }
+
+    return { message: current, preventDefault, responded };
+  }
+
+  /**
+   * Add a listener to the registry and return a function that removes it.
+   */
+  private add(listener: MessageListener): () => void {
+    this.listeners.push(listener);
+
+    return () => {
+      const index = this.listeners.indexOf(listener);
+      if (index >= 0) {
+        this.listeners.splice(index, 1);
+      }
+    };
+  }
+}
+
+/**
+ * Convert a `MessageSelector` into a predicate that decides whether a listener
+ * applies to a message. A matcher uses its `is` type guard; a predicate (any
+ * `(msg) => boolean`) is used as-is.
+ */
+export function messageMatcher(selector: MessageSelector<any>): (msg: IoMessage<unknown>) => boolean {
+  if (typeof selector === 'function') {
+    return selector;
+  }
+  return (msg) => selector.is(msg);
+}
+
+/**
+ * Combine several selectors into a single predicate that matches a message when
+ * *any* of them matches. Each selector may be a maker (matched by its `code`)
+ * or a predicate.
+ *
+ * @example
+ * ```ts
+ * host.on(matchAny(IO.CDK_TOOLKIT_I5501, IO.CDK_TOOLKIT_I5502), listener);
+ * ```
+ */
+export function matchAny(...selectors: MessageSelector<any>[]): (msg: IoMessage<unknown>) => boolean {
+  const matchers = selectors.map(messageMatcher);
+  return (msg) => matchers.some((matches) => matches(msg));
+}
diff --git a/packages/@aws-cdk/toolkit-lib/lib/api/io/private/message-maker.ts b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/message-maker.ts
index 4eae4b95a..307cb80ec 100644
--- a/packages/@aws-cdk/toolkit-lib/lib/api/io/private/message-maker.ts
+++ b/packages/@aws-cdk/toolkit-lib/lib/api/io/private/message-maker.ts
@@ -1,4 +1,4 @@
-import type { IoMessage, IoMessageCode, IoMessageLevel } from '../io-message';
+import type { IoMessage, IoRequest, IMessageMatcher, IRequestMatcher, IoMessageCode, IoMessageLevel } from '../io-message';
 import type { ActionLessMessage, ActionLessRequest } from './io-helper';
 
 /**
@@ -40,7 +40,7 @@ interface MessageInfo extends CodeInfo {
 /**
  * An interface that can produce messages for a specific code.
  */
-export interface IoMessageMaker<T> extends MessageInfo {
+export interface IoMessageMaker<T> extends MessageInfo, IMessageMatcher<T> {
   /**
    * Create a message for this code, with or without payload.
    */
@@ -109,7 +109,7 @@ interface RequestInfo<U> extends CodeInfo {
 /**
  * An interface that can produce requests for a specific code.
  */
-export interface IoRequestMaker<T, U> extends MessageInfo {
+export interface IoRequestMaker<T, U> extends MessageInfo, IRequestMatcher<T, U> {
   /**
    * Create a message for this code, with or without payload.
    */
@@ -122,7 +122,7 @@ export interface IoRequestMaker<T, U> extends MessageInfo {
   /**
    * Returns whether the given `IoMessage` instance matches this request definition
    */
-  is(x: IoMessage<unknown>): x is IoMessage<T>;
+  is(x: IoMessage<unknown>): x is IoRequest<T, U>;
 }
 
 /**
@@ -142,7 +142,7 @@ function request<T = AbsentData, U = ImpossibleType>(level: IoMessageLevel, deta
     ...details,
     level,
     req: maker as any,
-    is: (m): m is IoMessage<T> => m.code === details.code,
+    is: (m): m is IoRequest<T, U> => m.code === details.code,
   };
 }
 
@@ -172,6 +172,6 @@ export function question<T>(details: CodeInfo): IoRequestMaker<T, string> {
     ...details,
     level,
     req: maker as any,
-    is: (m): m is IoMessage<T> => m.code === details.code,
+    is: (m): m is IoRequest<T, string> => m.code === details.code,
   };
 }
diff --git a/packages/@aws-cdk/toolkit-lib/test/api/io/listeners.test.ts b/packages/@aws-cdk/toolkit-lib/test/api/io/listeners.test.ts
new file mode 100644
index 000000000..ec501a69b
--- /dev/null
+++ b/packages/@aws-cdk/toolkit-lib/test/api/io/listeners.test.ts
@@ -0,0 +1,381 @@
+import type { IIoHost, IMessageMatcher, IoMessage, IoMessageCode, IoRequest } from '../../../lib/api/io';
+import { withListeners } from '../../../lib/api/io';
+
+/**
+ * A minimal `IIoHost` that records what it is asked to handle, so we can assert
+ * on what a wrapped host forwards to it (after listeners ran).
+ */
+class RecordingIoHost implements IIoHost {
+  public readonly notified: Array<IoMessage<unknown>> = [];
+  public readonly requested: Array<IoRequest<unknown, any>> = [];
+
+  /** Response the inner host resolves a request with, standing in for a prompt. */
+  public prompted: any = 'PROMPTED';
+
+  public async notify(msg: IoMessage<unknown>): Promise<void> {
+    this.notified.push(msg);
+  }
+
+  public async requestResponse<T>(msg: IoRequest<unknown, T>): Promise<T> {
+    this.requested.push(msg);
+    return this.prompted;
+  }
+}
+
+const I2901: IoMessageCode = 'CDK_TOOLKIT_I2901'; // list result, payload has `stacks`
+const I7010: IoMessageCode = 'CDK_TOOLKIT_I7010'; // destroy confirmation request (boolean)
+
+function notification(over: Partial<IoMessage<any>> = {}): IoMessage<any> {
+  return {
+    time: new Date('2024-01-01T12:00:00'),
+    level: 'info',
+    action: 'synth',
+    code: 'CDK_TOOLKIT_I2901',
+    message: 'the original text',
+    data: { stacks: [] },
+    ...over,
+  };
+}
+
+function request(over: Partial<IoRequest<any, any>> = {}): IoRequest<any, any> {
+  return {
+    time: new Date('2024-01-01T12:00:00'),
+    level: 'info',
+    action: 'destroy',
+    code: 'CDK_TOOLKIT_I7010',
+    message: 'Are you sure?',
+    data: {},
+    defaultResponse: true,
+    ...over,
+  };
+}
+
+describe('withListeners', () => {
+  let inner: RecordingIoHost;
+
+  beforeEach(() => {
+    inner = new RecordingIoHost();
+  });
+
+  test('the wrapped host is still an IIoHost that forwards to the inner host', async () => {
+    const host = withListeners(inner);
+    const msg = notification();
+
+    await host.notify(msg);
+
+    expect(inner.notified).toEqual([msg]);
+  });
+
+  describe('on', () => {
+    test('runs the listener for a matching code and forwards the message', async () => {
+      const host = withListeners(inner);
+      const seen: Array<unknown> = [];
+      host.on(I2901, (m) => {
+        seen.push(m.data);
+      });
+
+      await host.notify(notification());
+
+      expect(seen).toHaveLength(1);
+      expect(inner.notified).toHaveLength(1);
+    });
+
+    test('does not run the listener for a non-matching code', async () => {
+      const host = withListeners(inner);
+      const fn = jest.fn();
+      host.on(I2901, fn);
+
+      await host.notify(notification({ code: 'CDK_TOOLKIT_I0001' }));
+
+      expect(fn).not.toHaveBeenCalled();
+      expect(inner.notified).toHaveLength(1);
+    });
+
+    test('awaits async listeners before forwarding', async () => {
+      const host = withListeners(inner);
+      const order: string[] = [];
+      host.on(I2901, async () => {
+        await new Promise((r) => setTimeout(r, 5));
+        order.push('listener');
+      });
+
+      await host.notify(notification());
+      order.push('forwarded');
+
+      expect(order).toEqual(['listener', 'forwarded']);
+    });
+
+    test('runs matching listeners in registration order', async () => {
+      const host = withListeners(inner);
+      const order: number[] = [];
+      host.on(I2901, () => {
+        order.push(1);
+      });
+      host.on(I2901, () => {
+        order.push(2);
+      });
+      host.on(I2901, () => {
+        order.push(3);
+      });
+
+      await host.notify(notification());
+
+      expect(order).toEqual([1, 2, 3]);
+    });
+
+    test('the disposer removes the listener', async () => {
+      const host = withListeners(inner);
+      const fn = jest.fn();
+      const dispose = host.on(I2901, fn);
+
+      await host.notify(notification());
+      dispose();
+      await host.notify(notification());
+
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+
+    test('matches on a predicate selector, firing only for matching messages', async () => {
+      const host = withListeners(inner);
+      const fn = jest.fn();
+      host.on((m) => m.level === 'warn', fn);
+
+      await host.notify(notification({ level: 'warn' }));
+      await host.notify(notification({ level: 'info' }));
+
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+
+    test('matches on a matcher and delivers the payload typed', async () => {
+      const host = withListeners(inner);
+      // A matcher carries the payload type, so `msg.data` is `{ stacks }` here
+      // without a cast (the point of the matcher overload).
+      const matcher: IMessageMatcher<{ stacks: unknown[] }> = {
+        is: (m): m is IoMessage<{ stacks: unknown[] }> => m.code === I2901,
+      };
+      const seen: number[] = [];
+      host.on(matcher, (m) => {
+        seen.push(m.data.stacks.length);
+      });
+
+      await host.notify(notification());
+      await host.notify(notification({ code: 'CDK_TOOLKIT_I0001' }));
+
+      expect(seen).toEqual([0]);
+    });
+  });
+
+  describe('once', () => {
+    test('runs only for the first matching message', async () => {
+      const host = withListeners(inner);
+      const fn = jest.fn();
+      host.once(I2901, fn);
+
+      await host.notify(notification());
+      await host.notify(notification());
+
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+
+    test('accepts a predicate selector', async () => {
+      const host = withListeners(inner);
+      const fn = jest.fn();
+      host.once((m) => m.level === 'warn', fn);
+
+      await host.notify(notification({ level: 'warn' }));
+      await host.notify(notification({ level: 'warn' }));
+
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+
+    test('fires only once even when two messages are handled concurrently', async () => {
+      const host = withListeners(inner);
+      let calls = 0;
+      // An async listener ahead of the `once` makes both notifies overlap.
+      host.on(I2901, async () => {
+        await new Promise((r) => setTimeout(r, 5));
+      });
+      host.once(I2901, () => {
+        calls++;
+      });
+
+      // eslint-disable-next-line @cdklabs/promiseall-no-unbounded-parallelism -- fixed pair, to force overlap
+      await Promise.all([host.notify(notification()), host.notify(notification())]);
+
+      expect(calls).toBe(1);
+    });
+  });
+
+  describe('rewrite', () => {
+    test('replaces the forwarded message text, leaving the code intact', async () => {
+      const host = withListeners(inner);
+      host.rewrite(I2901, (m) => `rewritten: ${(m.data as { stacks: unknown[] }).stacks.length}`);
+
+      await host.notify(notification());
+
+      expect(inner.notified[0].message).toBe('rewritten: 0');
+      expect(inner.notified[0].code).toBe('CDK_TOOLKIT_I2901');
+    });
+
+    test('can also override the level', async () => {
+      const host = withListeners(inner);
+      host.rewrite(I2901, (m) => m.message, 'debug');
+
+      await host.notify(notification());
+
+      expect(inner.notified[0].level).toBe('debug');
+    });
+
+    test('does not mutate the caller-provided message', async () => {
+      const host = withListeners(inner);
+      host.rewrite(I2901, () => 'changed');
+      const msg = notification();
+
+      await host.notify(msg);
+
+      expect(msg.message).toBe('the original text');
+    });
+
+    test('rewriteOnce applies only once', async () => {
+      const host = withListeners(inner);
+      host.rewriteOnce(I2901, () => 'changed');
+
+      await host.notify(notification());
+      await host.notify(notification());
+
+      expect(inner.notified[0].message).toBe('changed');
+      expect(inner.notified[1].message).toBe('the original text');
+    });
+
+    test('rewrites accumulate across listeners', async () => {
+      const host = withListeners(inner);
+      host.rewrite(I2901, (m) => `${m.message}-a`);
+      host.rewrite(I2901, (m) => `${m.message}-b`);
+
+      await host.notify(notification());
+
+      expect(inner.notified[0].message).toBe('the original text-a-b');
+    });
+
+    test('matching is decided against the emitted message, not an earlier rewrite', async () => {
+      const host = withListeners(inner);
+      // First listener rewrites the text; a later predicate that keys on the old
+      // text must still fire, because matching sees the emitted message.
+      host.rewrite(I2901, () => 'changed');
+      const fn = jest.fn();
+      host.on((m) => m.message === 'the original text', fn);
+
+      await host.notify(notification());
+
+      expect(fn).toHaveBeenCalledTimes(1);
+    });
+  });
+
+  describe('preventDefault', () => {
+    test('suppresses the forward to the inner host', async () => {
+      const host = withListeners(inner);
+      host.on(I2901, () => ({ preventDefault: true }));
+
+      await host.notify(notification());
+
+      expect(inner.notified).toHaveLength(0);
+    });
+  });
+
+  describe('requestResponse', () => {
+    test('forwards to the inner host when no listener answers', async () => {
+      const host = withListeners(inner);
+      inner.prompted = false;
+
+      const answer = await host.requestResponse(request());
+
+      expect(inner.requested).toHaveLength(1);
+      expect(answer).toBe(false);
+    });
+
+    test('respond answers without asking the inner host, suppressing the question', async () => {
+      const host = withListeners(inner);
+      host.respond(I7010, true);
+
+      const answer = await host.requestResponse(request({ defaultResponse: false }));
+
+      expect(answer).toBe(true);
+      expect(inner.requested).toHaveLength(0);
+      expect(inner.notified).toHaveLength(0);
+    });
+
+    test('respond with suppressQuestion=false surfaces the question but still answers', async () => {
+      const host = withListeners(inner);
+      host.respond(I7010, true, { suppressQuestion: false });
+
+      const answer = await host.requestResponse(request({ defaultResponse: false }));
+
+      expect(answer).toBe(true);
+      // The inner host is asked to show the question (as a notification), not to prompt.
+      expect(inner.notified).toHaveLength(1);
+      expect(inner.requested).toHaveLength(0);
+    });
+
+    test('respond treats presence of the value as the answer, so false is a valid answer', async () => {
+      const host = withListeners(inner);
+      host.respond(I7010, false);
+
+      const answer = await host.requestResponse(request({ defaultResponse: true }));
+
+      expect(answer).toBe(false);
+      expect(inner.requested).toHaveLength(0);
+    });
+
+    test('respondOnce answers only the first request', async () => {
+      const host = withListeners(inner);
+      inner.prompted = 'PROMPTED';
+      host.respondOnce(I7010, false);
+
+      const first = await host.requestResponse(request({ defaultResponse: 'default' }));
+      const second = await host.requestResponse(request({ defaultResponse: 'default' }));
+
+      expect(first).toBe(false);
+      expect(second).toBe('PROMPTED');
+      expect(inner.requested).toHaveLength(1);
+    });
+
+    test('a listener can reword a prompt before it reaches the inner host', async () => {
+      const host = withListeners(inner);
+      host.rewrite(I7010, () => 'reworded question');
+
+      await host.requestResponse(request());
+
+      expect(inner.requested[0].message).toBe('reworded question');
+    });
+
+    test('preventDefault on a request resolves with the default response without asking', async () => {
+      const host = withListeners(inner);
+      host.on(I7010, () => ({ preventDefault: true }));
+
+      const answer = await host.requestResponse(request({ defaultResponse: 'the-default' }));
+
+      expect(answer).toBe('the-default');
+      expect(inner.requested).toHaveLength(0);
+    });
+
+    test('respond on a notification code leaves the message alone instead of suppressing it', async () => {
+      const host = withListeners(inner);
+      // I2901 is a notification, not a request: there is nothing to answer, so
+      // respond must not drop the message.
+      host.respond(I2901, true);
+
+      await host.notify(notification());
+
+      expect(inner.notified).toHaveLength(1);
+    });
+
+    test('respondOnce on a notification code leaves the message alone instead of suppressing it', async () => {
+      const host = withListeners(inner);
+      host.respondOnce(I2901, true);
+
+      await host.notify(notification());
+
+      expect(inner.notified).toHaveLength(1);
+    });
+  });
+});
diff --git a/packages/aws-cdk/lib/cli/io-host/cli-io-host.ts b/packages/aws-cdk/lib/cli/io-host/cli-io-host.ts
index 0cb712b75..7a9229b6d 100644
--- a/packages/aws-cdk/lib/cli/io-host/cli-io-host.ts
+++ b/packages/aws-cdk/lib/cli/io-host/cli-io-host.ts
@@ -5,8 +5,8 @@ import { ToolkitError } from '@aws-cdk/toolkit-lib';
 import type { HotswapResult, IIoHost, IoMessage, IoMessageCode, IoMessageLevel, IoRequest, ToolkitAction } from '@aws-cdk/toolkit-lib';
 import chalk from 'chalk';
 import * as promptly from 'promptly';
-import type { IoHelper, ActivityPrinterProps, IActivityPrinter, IoMessageMaker, IoRequestMaker, IoDefaultMessages } from '../../../lib/api-private';
-import { asIoHelper, IO, isMessageRelevantForLevel, CurrentActivityPrinter, HistoryActivityPrinter } from '../../../lib/api-private';
+import type { IoHelper, ActivityPrinterProps, IActivityPrinter, IoMessageMaker, IoRequestMaker, IoDefaultMessages, MessageListenerResult, MessageListenerResultOrPromise, MessageSelector } from '../../../lib/api-private';
+import { asIoHelper, IO, isMessageRelevantForLevel, CurrentActivityPrinter, HistoryActivityPrinter, ListenerRegistry, matchAny } from '../../../lib/api-private';
 import type { Context } from '../../api/context';
 import { StackActivityProgress } from '../../commands/deploy';
 import { canCollectTelemetry } from '../telemetry/collect-telemetry';
@@ -22,6 +22,8 @@ import type { ITelemetrySink } from '../telemetry/sink/sink-interface';
 import { isCI } from '../util/ci';
 
 export type { IIoHost, IoMessage, IoMessageCode, IoMessageLevel, IoRequest };
+export { matchAny };
+export type { MessageListenerResult, MessageListenerResultOrPromise, MessageSelector } from '../../../lib/api-private';
 
 /**
  * The current action being performed by the CLI. 'none' represents the absence of an action.
@@ -104,105 +106,6 @@ export interface CliIoHostProps {
  */
 export type TargetStream = 'stdout' | 'stderr' | 'drop';
 
-/**
- * The result a message listener may return to influence how a message is handled.
- *
- * A listener may update the message _text_ and/or its _level_; it cannot change
- * any other field of the message (such as its `code`), which keeps the
- * code-keyed listener registry valid.
- */
-export interface MessageListenerResult {
-  /**
-   * Replace the text that is printed for this message.
-   *
-   * @default - the message text is left unchanged
-   */
-  readonly message?: string;
-
-  /**
-   * Override the level of this message.
-   *
-   * The new level is used for both verbosity filtering and stream selection, so
-   * this can move a message between stdout/stderr (e.g. downgrade a `result` to
-   * `info`). The `code` is intentionally left unchanged.
-   *
-   * @default - the message level is left unchanged
-   */
-  readonly level?: IoMessageLevel;
-
-  /**
-   * Skip the default handling of the message.
-   *
-   * For a notification this means it is not written to a stream. For a request
-   * it stops processing entirely: the user is not prompted, nothing is written,
-   * and the request resolves with its (possibly `respond`-overridden) default
-   * response.
-   *
-   * @default false
-   */
-  readonly preventDefault?: boolean;
-
-  /**
-   * For requests only: the value to resolve the request with. It is folded into
-   * the request's default response and skips the prompt (the request is treated
-   * as not promptable). The question is still written unless `preventDefault` is
-   * also set. Ignored for plain notifications.
-   *
-   * The presence of the key is what matters, so `false`/`0`/`''` are valid
-   * answers. Use the `respond`/`respondOnce` helpers for the common case.
-   *
-   * @default - this listener does not supply a response
-   */
-  readonly respond?: unknown;
-}
-
-/**
- * What a message listener may return: nothing, a `MessageListenerResult`, or a
- * `Promise` of either.
- *
- * Listeners may be async. The host awaits each listener before running the
- * next, so registration order — and the cumulative effect on the message — is
- * preserved regardless of whether listeners are sync or async.
- */
-export type MessageListenerResultOrPromise = void | MessageListenerResult | Promise<void | MessageListenerResult>;
-
-/**
- * A registered message listener. Its return value (if any) may update the
- * message text and/or prevent the default processing. It may be async.
- */
-type MessageListenerFn = (msg: IoMessage<any>) => MessageListenerResultOrPromise;
-interface MessageListener {
-  readonly once: boolean;
-  readonly fn: MessageListenerFn;
-  /**
-   * Decides which messages this listener applies to. For a listener registered
-   * with a maker this matches by `code`; for one registered with a predicate it
-   * is the predicate itself.
-   */
-  readonly matches: (msg: IoMessage<unknown>) => boolean;
-  /**
-   * Whether this is one of the host's own internal listeners (e.g. stack-activity
-   * routing). Internal listeners are not removed by `removeAllListeners`.
-   *
-   * @default false - a user listener registered via `on`/`once`/`rewrite`/`respond`
-   */
-  readonly internal?: boolean;
-}
-
-/**
- * Selects which messages a listener applies to.
- *
- * Either a message/request *maker* — the listener fires for messages with that
- * maker's `code` (the original behavior) — or a custom *predicate* over the
- * message. A maker's `.is` type guard (e.g. `IO.CDK_TOOLKIT_I7010.is`) is a
- * convenient predicate, but any `(msg) => boolean` works (e.g. to match a family
- * of codes, or on the message level).
- */
-export type MessageSelector<T> =
-  | IoMessageMaker<T>
-  | IoRequestMaker<T, any>
-  | ((msg: IoMessage<any>) => boolean);
-
 /**
  * How an IoHost processed a single message or request.
  *
@@ -331,9 +234,10 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
   private corkedCounter = 0;
   private readonly corkedLoggingBuffer: IoMessage<unknown>[] = [];
 
-  // Message listeners in registration order. Each carries a matcher (by code,
-  // or a custom predicate). See `on`/`once`/`rewrite`/`respond`.
-  private readonly messageListeners: MessageListener[] = [];
+  // The shared listener engine. Registration and message transformation live
+  // here; this host does its own I/O (writing, prompting, telemetry, observers)
+  // around `registry.apply`. See `on`/`once`/`rewrite`/`respond`.
+  private readonly registry = new ListenerRegistry();
 
   // Observers of how messages are handled (see ObservableIoHost / observeMessages).
   private readonly messageObservers = new Set<(observation: IoMessageObservation) => void>();
@@ -526,8 +430,8 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
     predicate: (msg: IoMessage<any>) => boolean,
     listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
   ): () => void;
-  public on(selector: MessageSelector<any>, listener: MessageListenerFn): () => void {
-    return this.addMessageListener({ once: false, fn: listener, matches: messageMatcher(selector) });
+  public on(selector: MessageSelector<any>, listener: (msg: IoMessage<any>) => MessageListenerResultOrPromise): () => void {
+    return this.registry.on(selector as any, listener as any);
   }
 
   /**
@@ -558,8 +462,8 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
     predicate: (msg: IoMessage<any>) => boolean,
     listener: (msg: IoMessage<unknown>) => MessageListenerResultOrPromise,
   ): () => void;
-  public once(selector: MessageSelector<any>, listener: MessageListenerFn): () => void {
-    return this.addMessageListener({ once: true, fn: listener, matches: messageMatcher(selector) });
+  public once(selector: MessageSelector<any>, listener: (msg: IoMessage<any>) => MessageListenerResultOrPromise): () => void {
+    return this.registry.once(selector as any, listener as any);
   }
 
   /**
@@ -574,13 +478,7 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
    * leak into the next test).
    */
   public removeAllListeners(): void {
-    // Drop user listeners in place (preserving array identity for any
-    // outstanding dispose closures); keep the host's internal listeners.
-    for (let i = this.messageListeners.length - 1; i >= 0; i--) {
-      if (!this.messageListeners[i].internal) {
-        this.messageListeners.splice(i, 1);
-      }
-    }
+    this.registry.removeUserListeners();
   }
 
   /**
@@ -599,14 +497,14 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
    * const dispose = ioHost.respond(IO.CDK_TOOLKIT_I7010, true);
    */
   public respond<T, U>(code: IoRequestMaker<T, U>, value: U, suppressQuestion = true): () => void {
-    return this.addMessageListener({ once: false, fn: () => ({ respond: value, preventDefault: suppressQuestion }), matches: messageMatcher(code) });
+    return this.registry.respond(code, value, suppressQuestion);
   }
 
   /**
    * Like `respond`, but the answer is given only once and then removed.
    */
   public respondOnce<T, U>(code: IoRequestMaker<T, U>, value: U, suppressQuestion = true): () => void {
-    return this.addMessageListener({ once: true, fn: () => ({ respond: value, preventDefault: suppressQuestion }), matches: messageMatcher(code) });
+    return this.registry.respondOnce(code, value, suppressQuestion);
   }
 
   /**
@@ -630,7 +528,7 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
     formatter: (msg: IoMessage<T>) => string,
     level?: IoMessageLevel,
   ): () => void {
-    return this.on(code, (msg) => ({ message: formatter(msg), ...(level !== undefined ? { level } : {}) }));
+    return this.registry.rewrite(code, formatter, level);
   }
 
   /**
@@ -642,83 +540,7 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
     formatter: (msg: IoMessage<T>) => string,
     level?: IoMessageLevel,
   ): () => void {
-    return this.once(code, (msg) => ({ message: formatter(msg), ...(level !== undefined ? { level } : {}) }));
-  }
-
-  /**
-   * Add a listener to the registry and return a function that removes it.
-   */
-  private addMessageListener(listener: MessageListener): () => void {
-    this.messageListeners.push(listener);
-
-    return () => {
-      const index = this.messageListeners.indexOf(listener);
-      if (index >= 0) {
-        this.messageListeners.splice(index, 1);
-      }
-    };
-  }
-
-  /**
-   * Run every registered listener that matches the message, in registration
-   * order. A listener matches by its code (maker) or its custom predicate.
-   *
-   * A listener may update the message text/level (passed on to subsequent
-   * listeners and the rest of the pipeline), prevent the default processing, or
-   * (for requests) answer it. `once` listeners are removed after they have run.
-   * Matching is decided against the message as emitted, so a rewrite by an
-   * earlier listener does not change which later listeners apply.
-   *
-   * Returns the (possibly updated) message, whether the default processing was
-   * prevented, and whether a listener answered the request (and with what).
-   */
-  private async applyMessageListeners<T extends IoMessage<unknown>>(msg: T): Promise<{
-    message: T;
-    preventDefault: boolean;
-    responded: boolean;
-  }> {
-    let current = msg;
-    let preventDefault = false;
-    let responded = false;
-    // Iterate over a copy so that `once` listeners can remove themselves safely.
-    for (const listener of [...this.messageListeners]) {
-      // Match against the emitted message; a listener receives the cumulatively
-      // transformed `current` message.
-      if (!listener.matches(msg)) {
-        continue;
-      }
-
-      // Listeners may be async; await each one before running the next so the
-      // cumulative effect on the message stays order-deterministic.
-      const result = await listener.fn(current);
-
-      if (listener.once) {
-        const index = this.messageListeners.indexOf(listener);
-        if (index >= 0) {
-          this.messageListeners.splice(index, 1);
-        }
-      }
-
-      if (result) {
-        if (result.message !== undefined) {
-          current = { ...current, message: result.message };
-        }
-        if (result.level !== undefined) {
-          current = { ...current, level: result.level };
-        }
-        if (result.preventDefault) {
-          preventDefault = true;
-        }
-        if ('respond' in result && 'defaultResponse' in msg) {
-          // Fold the answer into the request's default response and mark it
-          // answered, so we skip prompting and resolve with this value.
-          current = { ...current, defaultResponse: result.respond };
-          responded = true;
-        }
-      }
-    }
-
-    return { message: current, preventDefault, responded };
+    return this.registry.rewriteOnce(code, formatter, level);
   }
 
   /**
@@ -737,7 +559,7 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
     // already-transformed message.
     const { message, preventDefault } = this.corkReplaying
       ? { message: msg, preventDefault: false }
-      : await this.applyMessageListeners(msg);
+      : await this.registry.apply(msg);
 
     // Tell observers how this message was handled (its effective form and
     // whether it was dropped). Skipped while replaying corked messages so each
@@ -817,12 +639,10 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
 
     // A single internal listener (so it survives `removeAllListeners()` and the
     // host keeps routing stack activity) matching any of the activity codes.
-    this.addMessageListener({
-      once: false,
-      internal: true,
-      fn: route,
-      matches: matchAny(IO.CDK_TOOLKIT_I5501, IO.CDK_TOOLKIT_I5502, IO.CDK_TOOLKIT_I5503),
-    });
+    this.registry.addInternal(
+      matchAny(IO.CDK_TOOLKIT_I5501, IO.CDK_TOOLKIT_I5502, IO.CDK_TOOLKIT_I5503),
+      route,
+    );
   }
 
   /**
@@ -891,7 +711,7 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
   public async requestResponse<DataType, ResponseType>(msg: IoRequest<DataType, ResponseType>): Promise<ResponseType> {
     // Listeners run exactly once here (so we don't go back through `notify`):
     // they may answer the request, or reword/relevel the question shown below.
-    const { message, ...listenerResult } = await this.applyMessageListeners(msg);
+    const { message, ...listenerResult } = await this.registry.apply(msg);
 
     const response = await this.resolveRequest(message, listenerResult);
 
@@ -1049,32 +869,6 @@ export class CliIoHost implements IIoHost, ObservableIoHost {
   }
 }
 
-/**
- * Convert a `MessageSelector` into a predicate that decides whether a listener
- * applies to a message. A maker matches messages carrying its `code`; a
- * predicate (e.g. a maker's `.is`, or any `(msg) => boolean`) is used as-is.
- */
-function messageMatcher(selector: MessageSelector<any>): (msg: IoMessage<unknown>) => boolean {
-  if (typeof selector === 'function') {
-    return selector;
-  }
-  const { code } = selector;
-  return (msg) => msg.code === code;
-}
-
-/**
- * Combine several selectors into a single predicate that matches a message when
- * *any* of them matches. Each selector may be a maker (matched by its `code`)
- * or a predicate.
- *
- * Useful for one listener that spans multiple codes, e.g.
- * `ioHost.on(matchAny(IO.CDK_TOOLKIT_I5501, IO.CDK_TOOLKIT_I5502), listener)`.
- */
-export function matchAny(...selectors: MessageSelector<any>[]): (msg: IoMessage<unknown>) => boolean {
-  const matchers = selectors.map(messageMatcher);
-  return (msg) => matchers.some((matches) => matches(msg));
-}
-
 /**
  * This IoHost implementation considers a request promptable, if:
  * - it's a yes/no confirmation
diff --git a/packages/aws-cdk/test/_helpers/io-recorder.test.ts b/packages/aws-cdk/test/_helpers/io-recorder.test.ts
index 6282cadd6..a26c66782 100644
--- a/packages/aws-cdk/test/_helpers/io-recorder.test.ts
+++ b/packages/aws-cdk/test/_helpers/io-recorder.test.ts
@@ -1,5 +1,6 @@
 import { IoHostRecorder } from './io-recorder';
 import { asIoHelper, IO } from '../../lib/api-private';
+import type { IoMessage } from '../../lib/cli/io-host';
 import { CliIoHost } from '../../lib/cli/io-host';
 
 describe('IoHostRecorder', () => {
@@ -46,7 +47,7 @@ describe('IoHostRecorder', () => {
     // Answer the request through a listener so the real `requestResponse` runs
     // (and is therefore observed) — the recorder never spies on it. The question
     // is not suppressed, so it stays in the recorded stream.
-    const dispose = ioHost.on({ code: 'CDK_TOOLKIT_I0000' } as any, () => ({ respond: true }));
+    const dispose = ioHost.on((m) => m.code === 'CDK_TOOLKIT_I0000', () => ({ respond: true }));
 
     await ioHelper.defaults.info('before');
     await ioHelper.requestResponse({
@@ -122,7 +123,7 @@ describe('IoHostRecorder', () => {
 
     // Suppress a specific coded message, the way the CLI drops the synth/destroy
     // time lines on the destroy path.
-    const dispose = ioHost.on({ code: 'CDK_TOOLKIT_I9999' } as any, () => ({ preventDefault: true }));
+    const dispose = ioHost.on((m) => m.code === 'CDK_TOOLKIT_I9999', () => ({ preventDefault: true }));
 
     await ioHelper.notify({ time: new Date(), level: 'info', code: 'CDK_TOOLKIT_I9999', message: 'suppressed', data: undefined });
     await ioHelper.defaults.info('shown');
@@ -143,7 +144,7 @@ describe('IoHostRecorder', () => {
     const recorder = IoHostRecorder.create(ioHost);
     const ioHelper = asIoHelper(ioHost, 'destroy');
 
-    const dispose = ioHost.rewrite({ code: 'CDK_TOOLKIT_I9998' } as any, () => 'rewritten by listener');
+    const dispose = ioHost.rewrite({ is: (m: IoMessage<unknown>) => m.code === 'CDK_TOOLKIT_I9998' } as any, () => 'rewritten by listener');
 
     await ioHelper.notify({ time: new Date(), level: 'info', code: 'CDK_TOOLKIT_I9998', message: 'original', data: undefined });