diff --git a/src/html/styles-builder/plugin-loader.ts b/src/html/styles-builder/plugin-loader.ts
index df85199b4d..cd2dfe0d13 100644
--- a/src/html/styles-builder/plugin-loader.ts
+++ b/src/html/styles-builder/plugin-loader.ts
@@ -7,7 +7,7 @@
  * @module html/styles-builder/plugin-loader
  */
 
-import { encodeBase64, serverLogger } from "#veryfront/utils";
+import { encodeBase64Bytes, serverLogger } from "#veryfront/utils";
 import {
   type ErrorSlug,
   getErrorBySlug,
@@ -99,7 +99,11 @@ export function rewriteEsmShRootRelativeImports(code: string): string {
 
 async function importBundledModule(code: string): Promise<unknown> {
   if (!isDeno) {
-    const dataUrl = `data:text/javascript;base64,${encodeBase64(code)}`;
+    // Encode as UTF-8 bytes: the data: URL importer decodes UTF-8, and btoa on
+    // the raw string would emit Latin-1 bytes for chars in [0x80, 0xFF].
+    const dataUrl = `data:text/javascript;base64,${
+      encodeBase64Bytes(new TextEncoder().encode(code))
+    }`;
     return await import(dataUrl);
   }
 
diff --git a/src/modules/server/fs-probe.ts b/src/modules/server/fs-probe.ts
new file mode 100644
index 0000000000..e732d077c5
--- /dev/null
+++ b/src/modules/server/fs-probe.ts
@@ -0,0 +1,25 @@
+/** Minimal stat surface shared by the secure and platform filesystems. */
+interface StatCapableFs {
+  stat(path: string): Promise<{ isFile: boolean }>;
+}
+
+/**
+ * Resolve the first path in `paths` order that exists as a file, or null.
+ * All candidates are stat-probed in parallel; order of `paths` decides the
+ * winner, not which probe resolves first.
+ */
+export async function findFirstExistingFile(
+  fs: StatCapableFs,
+  paths: string[],
+): Promise<string | null> {
+  const results = await Promise.all(paths.map(async (path) => {
+    try {
+      const stat = await fs.stat(path);
+      return stat.isFile ? path : null;
+    } catch {
+      return null;
+    }
+  }));
+
+  return results.find((path): path is string => path !== null) ?? null;
+}
diff --git a/src/modules/server/module-batch-handler.ts b/src/modules/server/module-batch-handler.ts
index 0d8bffec3b..c3d0a79b00 100644
--- a/src/modules/server/module-batch-handler.ts
+++ b/src/modules/server/module-batch-handler.ts
@@ -46,6 +46,7 @@ import {
   hasSourceMiss,
   rememberSourceMiss,
 } from "./module-source-resolution-cache.ts";
+import { findFirstExistingFile } from "./fs-probe.ts";
 
 const logger = serverLogger.component("module-batch");
 
@@ -83,38 +84,6 @@ const FRAMEWORK_EXTENSIONS = [
   ".js", // Regular sources for dev mode
 ] as const;
 
-async function findFirstSecureFile(
-  secureFs: ReturnType<typeof createSecureFs>,
-  paths: string[],
-): Promise<string | null> {
-  const results = await Promise.all(paths.map(async (path) => {
-    try {
-      const stat = await secureFs.stat(path);
-      return stat.isFile ? path : null;
-    } catch {
-      return null;
-    }
-  }));
-
-  return results.find((path): path is string => path !== null) ?? null;
-}
-
-async function findFirstPlatformFile(
-  platformFs: ReturnType<typeof createFileSystem>,
-  paths: string[],
-): Promise<string | null> {
-  const results = await Promise.all(paths.map(async (path) => {
-    try {
-      const stat = await platformFs.stat(path);
-      return stat.isFile ? path : null;
-    } catch {
-      return null;
-    }
-  }));
-
-  return results.find((path): path is string => path !== null) ?? null;
-}
-
 export interface BatchHandlerOptions {
   projectDir: string;
   adapter: RuntimeAdapter;
@@ -341,7 +310,7 @@ async function loadAndTransformModule(
   });
   if (hasSourceMiss(missCacheKey)) return null;
 
-  const sourcePath = await findFirstSecureFile(
+  const sourcePath = await findFirstExistingFile(
     secureFs,
     EXTENSIONS.map((ext) => join(projectDir, basePath + ext)),
   );
@@ -359,7 +328,7 @@ async function loadAndTransformModule(
 
   const platformFs = createFileSystem();
   for (const lookupDir of frameworkLookupDirs) {
-    const frameworkPath = await findFirstPlatformFile(
+    const frameworkPath = await findFirstExistingFile(
       platformFs,
       FRAMEWORK_EXTENSIONS.map((ext) => join(lookupDir, basePath + ext)),
     );
diff --git a/src/modules/server/module-server.ts b/src/modules/server/module-server.ts
index c75ef5d76c..db7edb2875 100644
--- a/src/modules/server/module-server.ts
+++ b/src/modules/server/module-server.ts
@@ -55,6 +55,7 @@ import {
   getReleaseModuleResponse,
   rememberReleaseModuleResponse,
 } from "./module-response-cache.ts";
+import { findFirstExistingFile } from "./fs-probe.ts";
 import { ensureFilenameDefaultExport } from "#veryfront/modules/loader-shared/filename-default-export.ts";
 
 const logger = serverLogger.component("module-server");
@@ -791,44 +792,12 @@ async function findFrameworkPackageAssetFile(
 ): Promise<string | null> {
   if (hasUnsafePackageAssetPath(basePathWithoutExt)) return null;
 
-  return await findFirstPlatformFile(
+  return await findFirstExistingFile(
     fs,
     extensions.map((ext) => join(FRAMEWORK_ROOT, basePathWithoutExt + ext)),
   );
 }
 
-async function findFirstPlatformFile(
-  fs: ReturnType<typeof createFileSystem>,
-  paths: string[],
-): Promise<string | null> {
-  const results = await Promise.all(paths.map(async (path) => {
-    try {
-      const stat = await fs.stat(path);
-      return stat.isFile ? path : null;
-    } catch {
-      return null;
-    }
-  }));
-
-  return results.find((path): path is string => path !== null) ?? null;
-}
-
-async function findFirstSecureFile(
-  secureFs: ReturnType<typeof createSecureFs>,
-  paths: string[],
-): Promise<string | null> {
-  const results = await Promise.all(paths.map(async (path) => {
-    try {
-      const stat = await secureFs.stat(path);
-      return stat.isFile ? path : null;
-    } catch {
-      return null;
-    }
-  }));
-
-  return results.find((path): path is string => path !== null) ?? null;
-}
-
 async function findSourceFile(
   secureFs: ReturnType<typeof createSecureFs>,
   projectDir: string,
@@ -970,7 +939,7 @@ async function findSourceFile(
     : extensions;
 
   // Project file lookups (using secureFs which may go through FSAdapter in proxy mode)
-  const projectFilePath = await findFirstSecureFile(
+  const projectFilePath = await findFirstExistingFile(
     secureFs,
     projectLookupExtensions.map((ext) => join(projectDir, basePathWithoutExt + ext)),
   );
@@ -984,7 +953,7 @@ async function findSourceFile(
     if (!basePathWithoutExt.startsWith(prefix)) continue;
 
     const strippedPath = basePathWithoutExt.slice(prefix.length);
-    const strippedFilePath = await findFirstSecureFile(
+    const strippedFilePath = await findFirstExistingFile(
       secureFs,
       projectLookupExtensions.map((ext) => join(projectDir, strippedPath + ext)),
     );
@@ -998,7 +967,7 @@ async function findSourceFile(
     }
   }
 
-  const indexFilePath = await findFirstSecureFile(
+  const indexFilePath = await findFirstExistingFile(
     secureFs,
     projectLookupExtensions.map((ext) => join(projectDir, basePathWithoutExt, `index${ext}`)),
   );
@@ -1013,7 +982,7 @@ async function findSourceFile(
   // Try looking in common project directories
   const commonDirs = ["components", "app", "pages", "lib", "src"];
   for (const dir of commonDirs) {
-    const commonDirFilePath = await findFirstSecureFile(
+    const commonDirFilePath = await findFirstExistingFile(
       secureFs,
       projectLookupExtensions.map((ext) => join(projectDir, dir, basePathWithoutExt + ext)),
     );
diff --git a/src/observability/error-collector.ts b/src/observability/error-collector.ts
index db9874bf47..6fb5202104 100644
--- a/src/observability/error-collector.ts
+++ b/src/observability/error-collector.ts
@@ -6,6 +6,7 @@
  **************************/
 
 import { type ErrorCategory, INVALID_ARGUMENT } from "#veryfront/errors";
+import { createSubscriberSet } from "#veryfront/utils/subscriber-set.ts";
 
 /** Public API contract for error type. */
 export type ErrorType = "compile" | "runtime" | "bundle" | "hmr" | "module";
@@ -65,7 +66,7 @@ export type ErrorSubscriber = (error: DevError) => void;
 /** Implement error collector. */
 export class ErrorCollector {
   private errors = new Map<string, DevError>();
-  private subscribers = new Set<ErrorSubscriber>();
+  private subscribers = createSubscriberSet<[DevError]>();
   private idCounter = 0;
   private maxErrors: number;
 
@@ -99,13 +100,7 @@ export class ErrorCollector {
 
     this.errors.set(fullError.id, fullError);
 
-    for (const subscriber of this.subscribers) {
-      try {
-        subscriber(fullError);
-      } catch (_) {
-        /* expected: subscriber errors must not break error collection */
-      }
-    }
+    this.subscribers.notify(fullError);
 
     return fullError;
   }
@@ -320,8 +315,7 @@ export class ErrorCollector {
   }
 
   subscribe(callback: ErrorSubscriber): () => void {
-    this.subscribers.add(callback);
-    return () => this.subscribers.delete(callback);
+    return this.subscribers.subscribe(callback);
   }
 
   toJSON(): DevError[] {
diff --git a/src/observability/log-buffer.ts b/src/observability/log-buffer.ts
index 4d8a844b85..a247aa0c7f 100644
--- a/src/observability/log-buffer.ts
+++ b/src/observability/log-buffer.ts
@@ -1,4 +1,5 @@
 import { redactSensitive } from "#veryfront/utils/logger/redact.ts";
+import { createSubscriberSet } from "#veryfront/utils/subscriber-set.ts";
 
 /** Public API contract for log level. */
 export type LogLevel = "debug" | "info" | "warn" | "error";
@@ -27,7 +28,7 @@ export type LogSubscriber = (entry: LogEntry) => void;
 /** Implement log buffer. */
 export class LogBuffer {
   private entries: LogEntry[] = [];
-  private subscribers = new Set<LogSubscriber>();
+  private subscribers = createSubscriberSet<[LogEntry]>();
   private idCounter = 0;
   private maxSize: number;
 
@@ -55,13 +56,7 @@ export class LogBuffer {
       this.entries.shift();
     }
 
-    for (const subscriber of this.subscribers) {
-      try {
-        subscriber(fullEntry);
-      } catch (_) {
-        /* expected: subscriber errors must not break log buffering */
-      }
-    }
+    this.subscribers.notify(fullEntry);
 
     return fullEntry;
   }
@@ -146,8 +141,7 @@ export class LogBuffer {
   }
 
   subscribe(callback: LogSubscriber): () => void {
-    this.subscribers.add(callback);
-    return () => this.subscribers.delete(callback);
+    return this.subscribers.subscribe(callback);
   }
 
   toJSON(): LogEntry[] {
diff --git a/src/observability/request-profiler.ts b/src/observability/request-profiler.ts
index cc29651662..e258f60075 100644
--- a/src/observability/request-profiler.ts
+++ b/src/observability/request-profiler.ts
@@ -35,7 +35,8 @@ const records: RequestProfileRecord[] = [];
 const MAX_RECORDS = 200;
 let sequence = 0;
 
-function roundMs(value: number): number {
+/** Round to 2 decimal places (Server-Timing millisecond precision). */
+export function roundMs(value: number): number {
   return Math.round(value * 100) / 100;
 }
 
@@ -171,16 +172,27 @@ function formatDuration(value: number): string {
   return Math.max(0, roundMs(value)).toFixed(2);
 }
 
-export function buildServerTimingHeader(record: RequestProfileRecord): string {
-  const metrics = [`total;dur=${formatDuration(record.totalMs)}`];
-
-  for (const [name, duration] of Object.entries(record.phases).slice(0, 20)) {
+/** Build a Server-Timing header value from a total plus named phase durations. */
+export function buildServerTimingValue(
+  totalLabel: string,
+  totalMs: number,
+  phases: Iterable<[string, number]>,
+): string {
+  const metrics = [`${sanitizeMetricName(totalLabel)};dur=${formatDuration(totalMs)}`];
+  for (const [name, duration] of phases) {
     metrics.push(`${sanitizeMetricName(name)};dur=${formatDuration(duration)}`);
   }
-
   return metrics.join(", ");
 }
 
+export function buildServerTimingHeader(record: RequestProfileRecord): string {
+  return buildServerTimingValue(
+    "total",
+    record.totalMs,
+    Object.entries(record.phases).slice(0, 20),
+  );
+}
+
 export function withServerTimingHeader(
   response: Response,
   record: RequestProfileRecord | null,
diff --git a/src/proxy/server-timing.ts b/src/proxy/server-timing.ts
index 42747aece3..1432d4932a 100644
--- a/src/proxy/server-timing.ts
+++ b/src/proxy/server-timing.ts
@@ -1,4 +1,5 @@
 import { getEnv } from "#veryfront/platform/compat/process.ts";
+import { buildServerTimingValue, roundMs } from "#veryfront/observability/request-profiler.ts";
 
 export interface ProxyServerTiming {
   enabled: boolean;
@@ -6,18 +7,6 @@ export interface ProxyServerTiming {
   phases: Map<string, number>;
 }
 
-function roundMs(value: number): number {
-  return Math.round(value * 100) / 100;
-}
-
-function formatDuration(value: number): string {
-  return Math.max(0, roundMs(value)).toFixed(2);
-}
-
-function sanitizeMetricName(name: string): string {
-  return name.replace(/[^A-Za-z0-9!#$%&'*+\-.^_`|~]/g, "_");
-}
-
 export function shouldEnableProxyServerTiming(): boolean {
   return getEnv("VERYFRONT_ENABLE_PROXY_SERVER_TIMING") === "1" ||
     getEnv("VERYFRONT_ENABLE_SERVER_TIMING") === "1";
@@ -64,12 +53,7 @@ export function withProxyServerTimingHeader(
 ): Response {
   if (!timing.enabled) return response;
 
-  const metrics = [`proxy.total;dur=${formatDuration(totalMs)}`];
-  for (const [name, duration] of timing.phases.entries()) {
-    metrics.push(`${sanitizeMetricName(name)};dur=${formatDuration(duration)}`);
-  }
-
-  const value = metrics.join(", ");
+  const value = buildServerTimingValue("proxy.total", totalMs, timing.phases.entries());
 
   try {
     const existing = response.headers.get("Server-Timing");
diff --git a/src/server/reload-notifier.ts b/src/server/reload-notifier.ts
index f59f2fb483..d8fef80e44 100644
--- a/src/server/reload-notifier.ts
+++ b/src/server/reload-notifier.ts
@@ -1,4 +1,5 @@
 import { serverLogger } from "#veryfront/utils";
+import { createSubscriberSet } from "#veryfront/utils/subscriber-set.ts";
 
 const logger = serverLogger.component("reload-notifier");
 
@@ -20,8 +21,12 @@ type ReloadProjectInput = ReloadProjectInfo | string | undefined;
 const DEBOUNCE_MS = 300;
 
 class ReloadNotifierImpl {
-  private listeners = new Set<ReloadListener>();
-  private invalidateListeners = new Set<InvalidateListener>();
+  private listeners = createSubscriberSet<[string[] | undefined, ReloadProjectInfo | undefined]>(
+    (error) => logger.error("Listener error:", error),
+  );
+  private invalidateListeners = createSubscriberSet(
+    (error) => logger.error("Invalidate listener error:", error),
+  );
   private debounceTimer: ReturnType<typeof setTimeout> | null = null;
   private pendingChangedPaths = new Set<string>();
   private pendingProject?: ReloadProjectInfo;
@@ -32,13 +37,11 @@ class ReloadNotifierImpl {
   };
 
   subscribe(listener: ReloadListener): () => void {
-    this.listeners.add(listener);
-    return () => this.listeners.delete(listener);
+    return this.listeners.subscribe(listener);
   }
 
   subscribeInvalidate(listener: InvalidateListener): () => void {
-    this.invalidateListeners.add(listener);
-    return () => this.invalidateListeners.delete(listener);
+    return this.invalidateListeners.subscribe(listener);
   }
 
   triggerReload(changedPaths?: string[], project?: ReloadProjectInput): void {
@@ -87,13 +90,7 @@ class ReloadNotifierImpl {
       count: this.invalidateListeners.size,
     });
 
-    for (const listener of this.invalidateListeners) {
-      try {
-        listener();
-      } catch (error) {
-        logger.error("Invalidate listener error:", error);
-      }
-    }
+    this.invalidateListeners.notify();
   }
 
   private notifyListeners(changedPaths?: string[], project?: ReloadProjectInfo): void {
@@ -105,23 +102,13 @@ class ReloadNotifierImpl {
       project,
     });
 
-    for (const listener of this.listeners) {
-      try {
-        listener(changedPaths, project);
-      } catch (error) {
-        logger.error("Listener error:", error);
-      }
-    }
+    this.listeners.notify(changedPaths, project);
   }
 
   getListenerCount(): number {
     return this.listeners.size;
   }
 
-  getInvalidateListenerCount(): number {
-    return this.invalidateListeners.size;
-  }
-
   getMetrics(): {
     triggerCalls: number;
     broadcastsSent: number;
diff --git a/src/utils/base64url.ts b/src/utils/base64url.ts
index 81191407f5..fbd0f41328 100644
--- a/src/utils/base64url.ts
+++ b/src/utils/base64url.ts
@@ -8,17 +8,20 @@ function toBase64Url(b64: string): string {
   return b64.replaceAll("+", "-").replaceAll("/", "_").replaceAll("=", "");
 }
 
-/** Encode a UTF-8 string as standard base64 (handles non-Latin1 input). */
+/**
+ * Encode a string as standard base64. Latin-1 input (all code points <= 0xFF)
+ * is encoded with btoa's binary-string semantics; input outside Latin-1 falls
+ * back to UTF-8 bytes. Callers that need guaranteed UTF-8 bytes regardless of
+ * input (e.g. data: URLs decoded as UTF-8) should use
+ * `encodeBase64Bytes(new TextEncoder().encode(value))` instead.
+ */
 export function encodeBase64(value: string): string {
   if (typeof globalThis.btoa === "function") {
     try {
       return globalThis.btoa(value);
     } catch (_) {
-      /* expected: non-Latin1 string — fall back to TextEncoder */
-      const bytes = new TextEncoder().encode(value);
-      let binary = "";
-      for (const byte of bytes) binary += String.fromCharCode(byte);
-      return globalThis.btoa(binary);
+      /* expected: non-Latin1 string — fall back to UTF-8 bytes */
+      return encodeBase64Bytes(new TextEncoder().encode(value));
     }
   }
 
@@ -29,6 +32,22 @@ export function encodeBase64(value: string): string {
   throw new Error("Base64 encoding is not supported in this runtime");
 }
 
+/** Encode raw bytes as standard base64. */
+export function encodeBase64Bytes(bytes: Uint8Array): string {
+  // Prefer Buffer where available (Node): avoids building a large intermediate
+  // binary string, which is slower and can hit maximum-string-size limits.
+  const bufferCtor = (globalThis as { Buffer?: typeof Buffer }).Buffer;
+  if (bufferCtor) return bufferCtor.from(bytes).toString("base64");
+
+  if (typeof globalThis.btoa === "function") {
+    let binary = "";
+    for (const byte of bytes) binary += String.fromCharCode(byte);
+    return globalThis.btoa(binary);
+  }
+
+  throw new Error("Base64 encoding is not supported in this runtime");
+}
+
 /** Encode a string as unpadded base64url. */
 export function base64urlEncode(input: string): string {
   return toBase64Url(encodeBase64(input));
@@ -36,7 +55,5 @@ export function base64urlEncode(input: string): string {
 
 /** Encode raw bytes as unpadded base64url. */
 export function base64urlEncodeBytes(bytes: Uint8Array): string {
-  let binary = "";
-  for (const byte of bytes) binary += String.fromCharCode(byte);
-  return toBase64Url(btoa(binary));
+  return toBase64Url(encodeBase64Bytes(bytes));
 }
diff --git a/src/utils/index.ts b/src/utils/index.ts
index fc738de901..41b98d2f68 100644
--- a/src/utils/index.ts
+++ b/src/utils/index.ts
@@ -113,10 +113,17 @@ export {
   simpleHash,
 } from "./hash-utils.ts";
 
-export { base64urlEncode, base64urlEncodeBytes, encodeBase64 } from "./base64url.ts";
+export {
+  base64urlEncode,
+  base64urlEncodeBytes,
+  encodeBase64,
+  encodeBase64Bytes,
+} from "./base64url.ts";
 
 export { sleep } from "./sleep.ts";
 
+export { createSubscriberSet, type SubscriberSet } from "./subscriber-set.ts";
+
 export { MemoCache, memoize, memoizeAsync, simpleHash as memoizeHash } from "./memoize.ts";
 
 export { normalizePath } from "./path-utils.ts";
diff --git a/src/utils/subscriber-set.test.ts b/src/utils/subscriber-set.test.ts
new file mode 100644
index 0000000000..3ac31455eb
--- /dev/null
+++ b/src/utils/subscriber-set.test.ts
@@ -0,0 +1,86 @@
+import { describe, it } from "#veryfront/testing/bdd";
+import { assertEquals } from "#veryfront/testing/assert";
+import { createSubscriberSet } from "./subscriber-set.ts";
+
+describe("utils/subscriber-set", () => {
+  it("notifies all listeners with the given arguments", () => {
+    const set = createSubscriberSet<[string]>();
+    const seen: string[] = [];
+    set.subscribe((value) => seen.push(`a:${value}`));
+    set.subscribe((value) => seen.push(`b:${value}`));
+
+    set.notify("x");
+    assertEquals(seen, ["a:x", "b:x"]);
+  });
+
+  it("still notifies a listener that was unsubscribed earlier in the same notify (snapshot)", () => {
+    const set = createSubscriberSet();
+    const calls: string[] = [];
+    // Register the remover FIRST so it removes a listener that has not run
+    // yet. Live Set iteration would skip b; the snapshot must still call it.
+    let unsubscribeB = () => {};
+    set.subscribe(() => {
+      calls.push("a");
+      unsubscribeB();
+    });
+    unsubscribeB = set.subscribe(() => calls.push("b"));
+
+    set.notify();
+    assertEquals(calls, ["a", "b"]);
+
+    // The removal still takes effect for subsequent notifies.
+    set.notify();
+    assertEquals(calls, ["a", "b", "a"]);
+  });
+
+  it("isolates a throwing listener so the rest still run", () => {
+    const set = createSubscriberSet();
+    const calls: string[] = [];
+    set.subscribe(() => {
+      throw new Error("bad listener");
+    });
+    set.subscribe(() => calls.push("survivor"));
+
+    set.notify();
+    assertEquals(calls, ["survivor"]);
+  });
+
+  it("routes listener errors to onListenerError", () => {
+    const errors: unknown[] = [];
+    const set = createSubscriberSet((error) => errors.push(error));
+    set.subscribe(() => {
+      throw new Error("routed");
+    });
+
+    set.notify();
+    assertEquals((errors[0] as Error).message, "routed");
+  });
+
+  it("survives a throwing onListenerError and keeps notifying", () => {
+    const calls: string[] = [];
+    const set = createSubscriberSet(() => {
+      throw new Error("bad error handler");
+    });
+    set.subscribe(() => {
+      throw new Error("boom");
+    });
+    set.subscribe(() => calls.push("survivor"));
+
+    set.notify();
+    assertEquals(calls, ["survivor"]);
+  });
+
+  it("tracks size, ignores double-unsubscribe, and clears", () => {
+    const set = createSubscriberSet();
+    const unsubscribe = set.subscribe(() => {});
+    set.subscribe(() => {});
+    assertEquals(set.size, 2);
+
+    unsubscribe();
+    unsubscribe();
+    assertEquals(set.size, 1);
+
+    set.clear();
+    assertEquals(set.size, 0);
+  });
+});
diff --git a/src/utils/subscriber-set.ts b/src/utils/subscriber-set.ts
new file mode 100644
index 0000000000..cfefbac972
--- /dev/null
+++ b/src/utils/subscriber-set.ts
@@ -0,0 +1,51 @@
+/** Listener registry returned by {@link createSubscriberSet}. */
+export interface SubscriberSet<Args extends unknown[] = []> {
+  /** Register a listener; returns its unsubscribe function. */
+  subscribe(listener: (...args: Args) => void): () => void;
+  /** Invoke every listener; a throwing listener never stops the others. */
+  notify(...args: Args): void;
+  /** Number of registered listeners. */
+  readonly size: number;
+  /** Remove all listeners. */
+  clear(): void;
+}
+
+/**
+ * Create a subscriber set: the canonical subscribe/notify observable used
+ * across modules. Notification iterates a snapshot, so a listener that
+ * unsubscribes (itself or others) mid-notify is safe, and listener errors are
+ * isolated (routed to `onListenerError` when provided, otherwise swallowed).
+ */
+export function createSubscriberSet<Args extends unknown[] = []>(
+  onListenerError?: (error: unknown) => void,
+): SubscriberSet<Args> {
+  const listeners = new Set<(...args: Args) => void>();
+
+  return {
+    subscribe(listener) {
+      listeners.add(listener);
+      return () => {
+        listeners.delete(listener);
+      };
+    },
+    notify(...args) {
+      for (const listener of [...listeners]) {
+        try {
+          listener(...args);
+        } catch (error) {
+          try {
+            onListenerError?.(error);
+          } catch {
+            // A throwing error handler must not break notification either.
+          }
+        }
+      }
+    },
+    get size() {
+      return listeners.size;
+    },
+    clear() {
+      listeners.clear();
+    },
+  };
+}
diff --git a/src/workflow/claude-code/websocket-publisher.ts b/src/workflow/claude-code/websocket-publisher.ts
index 78de659c4b..568ace9bd8 100644
--- a/src/workflow/claude-code/websocket-publisher.ts
+++ b/src/workflow/claude-code/websocket-publisher.ts
@@ -5,6 +5,7 @@
  */
 
 import { logger as baseLogger } from "#veryfront/utils";
+import { createSubscriberSet } from "#veryfront/utils/subscriber-set.ts";
 import type {
   BidirectionalPublisher,
   CancelledEvent,
@@ -55,7 +56,11 @@ export class WebSocketPublisher implements BidirectionalPublisher {
   private config: Required<Omit<WebSocketPublisherConfig, "socket">> & {
     socket: WebSocket;
   };
-  private commandHandlers = new Set<ClientCommandHandler>();
+  private commandHandlers = createSubscriberSet<[ClientCommand]>((error) => {
+    if (this.config.debug) {
+      logger.error("Handler error", error);
+    }
+  });
   private closed = false;
   private pingTimer: number | null = null;
 
@@ -114,15 +119,7 @@ export class WebSocketPublisher implements BidirectionalPublisher {
     }
 
     // Dispatch to handlers
-    for (const handler of this.commandHandlers) {
-      try {
-        handler(command);
-      } catch (error) {
-        if (this.config.debug) {
-          logger.error("Handler error", error);
-        }
-      }
-    }
+    this.commandHandlers.notify(command);
   }
 
   private sendPong(): void {
@@ -165,10 +162,7 @@ export class WebSocketPublisher implements BidirectionalPublisher {
    * Subscribe to client commands
    */
   onCommand(handler: ClientCommandHandler): () => void {
-    this.commandHandlers.add(handler);
-    return () => {
-      this.commandHandlers.delete(handler);
-    };
+    return this.commandHandlers.subscribe(handler);
   }
 
   /**
diff --git a/tests/integration/server/modules/reload-notifier.test.ts b/tests/integration/server/modules/reload-notifier.test.ts
index e5bac1a67e..19a40435c6 100644
--- a/tests/integration/server/modules/reload-notifier.test.ts
+++ b/tests/integration/server/modules/reload-notifier.test.ts
@@ -24,7 +24,7 @@ describe("ReloadNotifier Tests", { sanitizeOps: false, sanitizeResources: false
   describe("ReloadNotifier - Subscription Management", () => {
     it("starts with zero listeners", (): void => {
       assertEquals(ReloadNotifier.getListenerCount(), 0);
-      assertEquals(ReloadNotifier.getInvalidateListenerCount(), 0);
+      assertEquals(ReloadNotifier.getMetrics().activeInvalidateListeners, 0);
     });
 
     it("can subscribe and unsubscribe reload listeners", (): void => {
@@ -42,13 +42,13 @@ describe("ReloadNotifier Tests", { sanitizeOps: false, sanitizeResources: false
     it("can subscribe and unsubscribe invalidate listeners", (): void => {
       const listener = (): void => {};
 
-      assertEquals(ReloadNotifier.getInvalidateListenerCount(), 0);
+      assertEquals(ReloadNotifier.getMetrics().activeInvalidateListeners, 0);
 
       const unsubscribe = ReloadNotifier.subscribeInvalidate(listener);
-      assertEquals(ReloadNotifier.getInvalidateListenerCount(), 1);
+      assertEquals(ReloadNotifier.getMetrics().activeInvalidateListeners, 1);
 
       unsubscribe();
-      assertEquals(ReloadNotifier.getInvalidateListenerCount(), 0);
+      assertEquals(ReloadNotifier.getMetrics().activeInvalidateListeners, 0);
     });
 
     it("supports multiple listeners", (): void => {