Batch the telemetry transport

docs/TELEMETRY.md

@@ -46,6 +46,13 @@ Any of the following disables telemetry entirely:
 - `CORBITS_TELEMETRY` set to any falsy value: `0`, `false`, `off`, `no`, or empty
 - `DO_NOT_TRACK=1` (the standard [Console Do Not Track](https://consoledonottrack.com/) convention)
 
+Turning telemetry off also discards whatever is still queued and unsent.
+Events captured earlier in the session but not yet transmitted are thrown
+away at the moment you opt out, not sent on the way out — opting out covers
+the activity you have already generated, not just the activity still to
+come. A batch already in flight to the server at the moment you opt out is
+not recalled; discarding only reaches events still held in memory.
+
 Re-enable from the same Telemetry tab or by removing the env var / settings
 override. While an env kill is active the Telemetry tab cannot re-enable —
 the env override always wins, and the attempt is refused rather than
@@ -92,6 +99,27 @@ Events are sent to PostHog. PostHog derives an approximate country from the
 request IP server-side; the client sends no location data itself. No
 self-hosted or third-party analytics beyond PostHog are used.
 
+## On the wire
+
+Events are not sent one at a time. Each captured event is stamped with its
+capture time and held in an in-memory queue, which is posted to PostHog's
+`/batch/` endpoint when it reaches the batch size or when the batch
+interval elapses, whichever comes first. At most one request is ever in
+flight: events captured while a request is open wait for it rather than
+opening another connection. Exit paths flush the queue, bounded by a short
+deadline so a slow endpoint cannot delay quitting.
+
+The queue has a hard depth limit. Once it is full — which in practice means
+the endpoint is unreachable, as on a captive portal or behind a hung proxy
+— the oldest queued events are dropped to make room for new ones. Telemetry
+is therefore lossy by design: it never grows memory without bound, never
+retries indefinitely, and never blocks or reports failures to the user.
+Nothing is written to disk, so dropped events are gone rather than deferred
+to a later run.
+
+See `src/telemetry/index.ts` for the batch size, interval, and queue limit
+in force.
+
 ## Not this document
 
 Local performance tracing and optional OpenTelemetry export to an operator-owned

src/telemetry/index.ts

@@ -17,8 +17,25 @@ export const POSTHOG_HOST = process.env[TELEMETRY_HOST_ENV] ?? DEFAULT_POSTHOG_H
 export const POSTHOG_API_KEY = process.env[TELEMETRY_KEY_ENV] ?? DEFAULT_POSTHOG_API_KEY;
 
 // Upper bound on how long flush() may hold up process exit; anything still
-// in flight past this is dropped.
-const FLUSH_DEADLINE_MS = 500;
+// in flight past this is dropped. Exported so tests can assert against the
+// deadline itself rather than a duplicated magic number.
+export const FLUSH_DEADLINE_MS = 500;
+
+// Batching defaults. A busy turn can emit an event per tool call, so events
+// accumulate until either trigger fires rather than opening a socket each
+// time. The queue limit bounds memory when the endpoint is unreachable —
+// a captive portal or hung proxy would otherwise grow the queue for the
+// whole session behind a single stuck request.
+const DEFAULT_BATCH_SIZE = 20;
+const DEFAULT_BATCH_INTERVAL_MS = 10_000;
+const DEFAULT_QUEUE_LIMIT = 500;
+const REQUEST_TIMEOUT_MS = 3000;
+
+export type BatchTuning = {
+  size?: number;
+  intervalMs?: number;
+  queueLimit?: number;
+};
 
 // Shown once per installation, in whichever surface a new user reaches
 // first: the onboarding panel on a fresh install (so disclosure accompanies
@@ -113,19 +130,32 @@ export type CreateTelemetryOptions = {
   fetchFn?: typeof fetch;
   host?: string;
   apiKey?: string;
+  batch?: BatchTuning;
+};
+
+type QueuedEvent = {
+  event: TelemetryEvent;
+  properties: Record<string, unknown>;
+  timestamp: string;
 };
 
 export type Telemetry = {
   enabled: boolean;
   capture(event: TelemetryEvent, properties?: Record<string, unknown>): void;
-  // Waits briefly for captures currently in flight to settle, giving up
+  // Sends whatever is queued and waits briefly for it to settle, giving up
   // after a short deadline so a slow endpoint can never hold up process
   // exit. Callers use this to bound exit against dropped fire-and-forget
   // requests without ever making capture() itself blocking.
   flush(): Promise<void>;
+  // Throws away everything queued and disarms the batch timer, so nothing
+  // captured before this call can ever reach the network. Opting out uses
+  // this: a user who says stop mid-session is saying they do not want the
+  // activity they have already generated sent, which makes discarding the
+  // queue the honest reading of that intent and flushing it a betrayal.
+  discard(): void;
 };
 
-// Fire-and-forget PostHog capture client. Never throws, never blocks the
+// Fire-and-forget PostHog batch client. Never throws, never blocks the
 // caller — errors (including timeouts) are swallowed silently since
 // telemetry must never affect product behavior.
 export function createTelemetry(options: CreateTelemetryOptions): Telemetry {
@@ -136,18 +166,64 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry {
   const fetchFn = options.fetchFn ?? fetch;
   const installationId = options.settings?.telemetry?.installationId ?? "";
 
-  // Tracked so flush() can wait for in-flight requests without making
-  // capture() itself awaitable.
-  const pending = new Set<Promise<void>>();
+  const batchSize = options.batch?.size ?? DEFAULT_BATCH_SIZE;
+  const batchIntervalMs = options.batch?.intervalMs ?? DEFAULT_BATCH_INTERVAL_MS;
+  const queueLimit = options.batch?.queueLimit ?? DEFAULT_QUEUE_LIMIT;
+
+  const queue: QueuedEvent[] = [];
+  let inFlight: Promise<void> | null = null;
+  let timer: ReturnType<typeof setTimeout> | null = null;
+
+  function cancelTimer(): void {
+    if (timer === null) return;
+    clearTimeout(timer);
+    timer = null;
+  }
+
+  async function send(events: QueuedEvent[]): Promise<void> {
+    const body = {
+      api_key: apiKey,
+      batch: events.map((queued) => ({
+        event: queued.event,
+        timestamp: queued.timestamp,
+        properties: { ...queued.properties, distinct_id: installationId },
+      })),
+    };
+    try {
+      await fetchFn(`${host}/batch/`, {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify(body),
+        signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
+      });
+    } catch {
+      // Swallow all errors — telemetry must never surface failures.
+    }
+  }
+
+  // Returns the existing drain when one is running so at most one request is
+  // ever open: events captured mid-flight are picked up by that drain's next
+  // iteration instead of opening a second socket.
+  function drain(): Promise<void> {
+    if (inFlight !== null) return inFlight;
+    const running = (async () => {
+      while (queue.length > 0) {
+        await send(queue.splice(0, batchSize));
+      }
+    })().finally(() => {
+      inFlight = null;
+    });
+    inFlight = running;
+    return running;
+  }
 
   function capture(event: TelemetryEvent, properties?: Record<string, unknown>): void {
     if (!enabled) return;
     if (!(event === "cli_start" || event === "session_end" || event === "inference_turn")) return;
 
-    const body = {
-      api_key: apiKey,
+    queue.push({
       event,
-      distinct_id: installationId,
+      timestamp: new Date().toISOString(),
       properties: {
         ...allowedProperties(event, properties),
         service_version: pkg.version,
@@ -156,34 +232,47 @@ export function createTelemetry(options: CreateTelemetryOptions): Telemetry {
         schema_version: 1,
         session_id: SESSION_ID,
       },
-    };
+    });
 
-    const request = fetchFn(`${host}/capture/`, {
-      method: "POST",
-      headers: { "content-type": "application/json" },
-      body: JSON.stringify(body),
-      signal: AbortSignal.timeout(3000),
-    })
-      .then(() => undefined)
-      .catch(() => {
-        // Swallow all errors — telemetry must never surface failures.
-      });
-    pending.add(request);
-    void request.finally(() => pending.delete(request));
+    // Oldest first: a stuck endpoint makes the head of the queue the least
+    // likely to still be worth reporting, and unbounded growth is never an
+    // acceptable alternative.
+    if (queue.length > queueLimit) queue.splice(0, queue.length - queueLimit);
+
+    if (queue.length >= batchSize) {
+      cancelTimer();
+      void drain();
+      return;
+    }
+    if (timer === null) {
+      timer = setTimeout(() => {
+        timer = null;
+        void drain();
+      }, batchIntervalMs);
+      timer.unref?.();
+    }
   }
 
   async function flush(): Promise<void> {
-    if (pending.size === 0) return;
+    cancelTimer();
+    if (queue.length === 0 && inFlight === null) return;
     // Race against a short deadline: stragglers are dropped rather than
-    // allowed to delay exit for the full 3s per-request AbortSignal window.
+    // allowed to delay exit for the full per-request AbortSignal window.
     await Promise.race([
-      Promise.allSettled(pending),
+      drain(),
       new Promise<void>((resolve) => {
-        const timer = setTimeout(resolve, FLUSH_DEADLINE_MS);
-        timer.unref?.();
+        const deadline = setTimeout(resolve, FLUSH_DEADLINE_MS);
+        deadline.unref?.();
       }),
     ]);
   }
 
-  return { enabled, capture, flush };
+  // A request already on the wire cannot be unsent, but nothing still held
+  // in memory follows it.
+  function discard(): void {
+    cancelTimer();
+    queue.length = 0;
+  }
+
+  return { enabled, capture, flush, discard };
 }

src/telemetry/singleton.ts

@@ -5,7 +5,12 @@ import type { Telemetry } from "./index.js";
 // than threading it through every intermediate call site. Defaults to a
 // disabled no-op so any code path that runs before index.ts sets it (or in
 // tests) never throws.
-let instance: Telemetry = { enabled: false, capture: () => {}, flush: async () => {} };
+let instance: Telemetry = {
+  enabled: false,
+  capture: () => {},
+  flush: async () => {},
+  discard: () => {},
+};
 
 export function setTelemetry(telemetry: Telemetry): void {
   instance = telemetry;

src/telemetry/toggle.ts

@@ -57,10 +57,14 @@ export function createTelemetryToggleHandler(
       return;
     }
     if (!enabled) {
-      // Opt-out must be immediate and absolute: swap the in-memory singleton
-      // synchronously, before any await, so no capture in flight during the
-      // persistence step below can land on a still-enabled instance, and so
-      // an unhandled rejection from disk I/O can never leave telemetry on.
+      // Opt-out must be immediate and absolute: discard whatever the outgoing
+      // instance has queued (dropping the singleton alone would leave its
+      // batch timer armed to send it anyway), then swap the in-memory
+      // singleton synchronously, before any await, so no capture in flight
+      // during the persistence step below can land on a still-enabled
+      // instance, and so an unhandled rejection from disk I/O can never
+      // leave telemetry on.
+      deps.getTelemetry().discard();
       deps.setTelemetry(
         deps.createTelemetry({ settings: { providers: {}, telemetry: { enabled: false } } }),
       );

tests/unit/telemetry-first-run.test.ts

@@ -68,6 +68,7 @@ test("activation stamps the notice, swaps the singleton, and fires cli_start", a
   await activateHeldTelemetry("/fake/path", () => true, deps);
   expect(markCalls()).toBe(1);
   expect(getInstance()?.enabled).toBe(true);
+  await getInstance()?.flush();
   expect(fetchCalls()).toBe(1);
 });
 
@@ -99,5 +100,6 @@ test("a failed notice stamp does not block activation", async () => {
   });
   await activateHeldTelemetry("/fake/path", () => true, deps);
   expect(getInstance()?.enabled).toBe(true);
+  await getInstance()?.flush();
   expect(fetchCalls()).toBe(1);
 });

tests/unit/telemetry-toggle.test.ts

@@ -56,6 +56,43 @@ test("capture called immediately after toggle-off makes zero fetch calls", () =>
   expect(fetchCalls()).toBe(0);
 });
 
+// Opting out is a statement about activity already generated, not only about
+// activity to come: events captured before the toggle must never be sent
+// afterwards. Dropping the singleton is not enough on its own — the outgoing
+// instance's batch timer would still fire and post its queue — so this test
+// guards the explicit discard. If a future change makes opt-out flush what it
+// was holding, this fails, and that is the point.
+test("opting out discards events captured before the toggle instead of sending them", async () => {
+  let sends = 0;
+  const fetchFn = (() => {
+    sends++;
+    return Promise.resolve(new Response("1", { status: 200 }));
+  }) as unknown as typeof fetch;
+  const { deps, getInstance } = fakeDeps({
+    createTelemetry: (opts) =>
+      createTelemetry({
+        ...opts,
+        env: opts.env ?? {},
+        apiKey: opts.apiKey ?? "test-key",
+        fetchFn,
+        // Short enough that an undiscarded queue would reach the network well
+        // inside this test's wait, rather than passing by outrunning a timer.
+        batch: { intervalMs: 20 },
+      }),
+  });
+  const handler = createTelemetryToggleHandler("/fake/path", deps);
+
+  handler(true);
+  await new Promise((resolve) => setTimeout(resolve, 10));
+  getInstance().capture("cli_start");
+  expect(sends).toBe(0);
+
+  handler(false);
+  await new Promise((resolve) => setTimeout(resolve, 150));
+  expect(getInstance().enabled).toBe(false);
+  expect(sends).toBe(0);
+});
+
 test("save rejection leaves the singleton disabled with no unhandled rejection", async () => {
   const { deps, getInstance } = fakeDeps({
     saveGlobalSettings: async () => {
@@ -90,7 +127,12 @@ test("load failure skips persistence entirely and stays disabled in memory", asy
 test("toggle on while env-killed writes nothing and swaps no instance", async () => {
   let ensureCalled = false;
   let saveCalled = false;
-  const initial: Telemetry = { enabled: false, capture: () => {}, flush: async () => {} };
+  const initial: Telemetry = {
+    enabled: false,
+    capture: () => {},
+    flush: async () => {},
+    discard: () => {},
+  };
   let setInstance: Telemetry | undefined;
   const { deps } = fakeDeps({
     getTelemetry: () => initial,
@@ -180,7 +222,7 @@ test("toggle on re-enables after settings load/save resolve", async () => {
 });
 
 test("session_id on captured payloads stays constant across an enable/disable/enable toggle cycle", async () => {
-  const capturedBodies: { properties: Record<string, unknown> }[] = [];
+  const capturedBodies: { batch: { properties: Record<string, unknown> }[] }[] = [];
   const fetchFn = ((_url: string, init: RequestInit) => {
     capturedBodies.push(JSON.parse(init.body as string));
     return Promise.resolve(new Response("1", { status: 200 }));
@@ -197,6 +239,9 @@ test("session_id on captured payloads stays constant across an enable/disable/en
   handler(true);
   await new Promise((resolve) => setTimeout(resolve, 10));
   getInstance().capture("cli_start");
+  // Toggling off discards the outgoing instance and its queue, so anything
+  // captured before it has to leave the process first.
+  await getInstance().flush();
 
   handler(false);
   await new Promise((resolve) => setTimeout(resolve, 10));
@@ -205,12 +250,12 @@ test("session_id on captured payloads stays constant across an enable/disable/en
   handler(true);
   await new Promise((resolve) => setTimeout(resolve, 10));
   getInstance().capture("cli_start");
+  await getInstance().flush();
 
-  await new Promise((resolve) => setTimeout(resolve, 0));
   expect(capturedBodies.length).toBe(2);
-  const sessionId = capturedBodies[0].properties.session_id;
+  const sessionId = capturedBodies[0].batch[0].properties.session_id;
   expect(typeof sessionId).toBe("string");
   expect((sessionId as string).length).toBeGreaterThan(0);
-  expect(capturedBodies[1].properties.session_id).toBe(sessionId);
+  expect(capturedBodies[1].batch[0].properties.session_id).toBe(sessionId);
   expect(sessionId).toBe(getSessionId());
 });