fix: terminate telemetry work on disposal

Give the best-effort telemetry adapter a terminal ACTIVE/DISPOSED lifecycle.
dispose() now removes the pagehide listener, clears the queue, invalidates
scheduled callback generations and aborts the in-flight sink; emit after
dispose is a no-op and a sink that ignores the abort cannot reschedule or
update post-dispose state. flush() joins the active delivery instead of
resolving early, and runtime infrastructure teardown disposes telemetry first.

Telemetry and diagnostics capacities are validated at construction against a
documented ceiling, so NaN or Infinity can no longer disable eviction.

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

src/adapters/diagnostics/bounded-diagnostics.ts

@@ -6,11 +6,15 @@ import {
   type DiagnosticRecordInput,
 } from "../../contracts/diagnostics.ts";
 import { projectTelemetryEvent } from "../../contracts/telemetry.ts";
+import { assertBoundedCapacity } from "../telemetry/best-effort-telemetry.ts";
 
 export const noOpDiagnostics: DiagnosticsPort = Object.freeze({
   record() {},
 });
 
+/** N-11. Documented absolute ceiling for bounded in-memory evidence. */
+export const MAX_DIAGNOSTIC_ENTRIES = 10_000;
+
 export function createDiagnosticsAdapter(
   options: Readonly<{
     maxEntries?: number;
@@ -18,7 +22,11 @@ export function createDiagnosticsAdapter(
     sink?: (record: DiagnosticRecord) => void;
   }> = {},
 ) {
-  const maxEntries = Math.max(1, options.maxEntries ?? 100);
+  const maxEntries = assertBoundedCapacity(
+    options.maxEntries ?? 100,
+    MAX_DIAGNOSTIC_ENTRIES,
+    "diagnostics maxEntries",
+  );
   const entries: DiagnosticRecord[] = [];
   const droppedReasons = new Map<string, number>();
 

src/adapters/telemetry/best-effort-telemetry.ts

@@ -37,6 +37,31 @@ export const noOpTelemetry: TelemetryAdapter = Object.freeze({
   dispose: () => {},
 });
 
+/**
+ * N-04. Disposal is terminal: there is no durable queue and no resurrection.
+ */
+type TelemetryLifecycle = "ACTIVE" | "DISPOSED";
+
+/** N-11. Documented absolute ceiling for the in-memory best-effort queue. */
+export const MAX_TELEMETRY_QUEUE = 10_000;
+
+/**
+ * N-11. A non-finite or fractional capacity silently disables eviction, so it is
+ * a construction-time configuration error rather than a runtime drop.
+ */
+export function assertBoundedCapacity(
+  value: number,
+  ceiling: number,
+  label: string,
+): number {
+  if (!Number.isSafeInteger(value) || value < 1 || value > ceiling) {
+    throw new TypeError(
+      `${label} must be a safe integer between 1 and ${ceiling}`,
+    );
+  }
+  return value;
+}
+
 export function createTelemetryAdapter(
   options: TelemetryAdapterOptions,
 ): TelemetryAdapter {
@@ -46,11 +71,19 @@ export function createTelemetryAdapter(
 
   const endpoint = options.endpoint;
   const fetcher = options.fetcher ?? fetch;
-  const maxQueue = Math.max(1, options.maxQueue ?? 100);
+  const maxQueue = assertBoundedCapacity(
+    options.maxQueue ?? 100,
+    MAX_TELEMETRY_QUEUE,
+    "telemetry maxQueue",
+  );
   const schedule = options.schedule ?? queueMicrotask;
   const queue: TelemetryEvent[] = [];
+  let lifecycleState: TelemetryLifecycle = "ACTIVE";
+  /** Scheduled callbacks captured before disposal must not run afterwards. */
+  let scheduleGeneration = 0;
   let scheduled = false;
-  let flushing = false;
+  let activeFlush: Promise<void> | null = null;
+  let activeSink: AbortController | null = null;
   let dropped = 0;
   const dropReasons = new Map<string, number>();
   let lastDeliveryEvidence: TelemetryEvent | null = null;
@@ -93,9 +126,12 @@ export function createTelemetryAdapter(
   }
 
   function scheduleFlush(): void {
-    if (scheduled) return;
+    if (scheduled || lifecycleState === "DISPOSED") return;
     scheduled = true;
+    const generation = scheduleGeneration;
     schedule(() => {
+      // A callback captured before disposal belongs to a dead generation.
+      if (generation !== scheduleGeneration) return;
       scheduled = false;
       void flush();
     });
@@ -105,6 +141,7 @@ export function createTelemetryAdapter(
     eventName: TelemetryEventName,
     attributes: Record<string, unknown>,
   ): void {
+    if (lifecycleState === "DISPOSED") return;
     const projected = projectTelemetryEvent(
       eventName,
       attributes,
@@ -124,26 +161,44 @@ export function createTelemetryAdapter(
     scheduleFlush();
   }
 
-  async function flush(): Promise<void> {
-    if (flushing || queue.length === 0) return;
-    flushing = true;
-    const batch = queue.splice(0, queue.length);
-    try {
-      const response = await fetcher(endpoint, {
-        method: "POST",
-        headers: { "Content-Type": "application/json" },
-        body: JSON.stringify({ events: batch }),
-        keepalive: true,
-      });
-      if (!response.ok) recordDrop("sink-failure", batch.length);
-    } catch {
-      recordDrop("sink-failure", batch.length);
-    } finally {
-      flushing = false;
-      if (queue.length > 0) {
-        scheduleFlush();
-      }
+  /**
+   * `flush()` joins the active delivery instead of resolving immediately, so an
+   * awaited flush really means "the in-flight batch has settled".
+   */
+  function flush(): Promise<void> {
+    if (activeFlush) return activeFlush;
+    if (lifecycleState === "DISPOSED" || queue.length === 0) {
+      return Promise.resolve();
     }
+    const generation = scheduleGeneration;
+    const run = async () => {
+      const batch = queue.splice(0, queue.length);
+      const controller = new AbortController();
+      activeSink = controller;
+      try {
+        const response = await fetcher(endpoint, {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({ events: batch }),
+          keepalive: true,
+          signal: controller.signal,
+        });
+        // A sink that ignored the abort must not update post-dispose state.
+        if (generation !== scheduleGeneration) return;
+        if (!response.ok) recordDrop("sink-failure", batch.length);
+      } catch {
+        if (generation !== scheduleGeneration) return;
+        recordDrop("sink-failure", batch.length);
+      } finally {
+        if (activeSink === controller) activeSink = null;
+        activeFlush = null;
+        if (generation === scheduleGeneration && queue.length > 0) {
+          scheduleFlush();
+        }
+      }
+    };
+    activeFlush = run();
+    return activeFlush;
   }
 
   const flushBeforePageExit = () => {
@@ -151,8 +206,20 @@ export function createTelemetryAdapter(
   };
   lifecycle?.addEventListener("pagehide", flushBeforePageExit);
 
+  /**
+   * N-04. Terminal disposal: one state transition, no further admission, no
+   * further scheduling, and no recursive drop telemetry while shutting down.
+   */
   function dispose(): void {
+    if (lifecycleState === "DISPOSED") return;
+    lifecycleState = "DISPOSED";
+    scheduleGeneration += 1;
+    scheduled = false;
     lifecycle?.removeEventListener("pagehide", flushBeforePageExit);
+    queue.length = 0;
+    activeSink?.abort();
+    activeSink = null;
+    activeFlush = null;
   }
 
   return Object.freeze({

src/bootstrap/runtime-adapters.ts

@@ -480,6 +480,10 @@ export async function createRuntimeAdapters(
       crossContextInvalidationStatus: () =>
         serverStateGeneration.getSnapshot().crossContextStatus(),
       dispose() {
+        // N-04. Telemetry is torn down first: it must stop scheduling and
+        // delivering before the diagnostics and state dependencies it observes
+        // are destroyed.
+        telemetry.dispose();
         conditionalValidators.clear();
         serverStateScope.dispose();
         serverStateGeneration.dispose();