feat(telemetry): align client with wire contract — chunking, deterministic batchId, batched retry, bounded store (#9946)

packages/shared/src/telemetry/client-contract.test.ts

@@ -0,0 +1,205 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { TelemetryClient } from "./client.js";
+import type { TelemetryConfig, TelemetryState } from "./types.js";
+
+/**
+ * Wire-contract suite for the telemetry client.
+ *
+ * These tests assert the client honors the backend wire contract using ONLY
+ * relative invariants derived from injected config — never the server's literal
+ * threshold values. There are intentionally no literal `50` / `512` / `524288`
+ * / `2` assertion targets in this file: caps are injected small and every bound
+ * is checked against the injected config value. This keeps the client
+ * "compatible, not a mirror" — server numbers live in config, not test logic.
+ */
+
+// The server envelope allow-set (`validators.mjs` ENVELOPE_FIELDS). Any extra
+// top-level key would be rejected, so every emitted envelope must be a subset.
+const CONTRACT_KEYS = new Set(["app", "schemaVersion", "installId", "version", "events", "batchId"]);
+
+const TEST_STATE: TelemetryState = {
+  installId: "contract-install",
+  salt: "contract-salt",
+  createdAt: "2026-01-01T00:00:00Z",
+  firstSeenVersion: "0.0.0",
+};
+
+function makeClient(
+  config?: Partial<TelemetryConfig>,
+  stateFactory: () => TelemetryState = () => TEST_STATE,
+  // 0.5 => zero jitter; deterministic backoff.
+  random: () => number = () => 0.5,
+) {
+  return new TelemetryClient(
+    { enabled: true, endpoint: "http://localhost:9999/ingest", ...config },
+    stateFactory,
+    "0.0.0-test",
+    random,
+  );
+}
+
+function sentBodies(): Array<Record<string, unknown>> {
+  return vi.mocked(fetch).mock.calls.map((call) => {
+    const requestInit = call[1] as RequestInit | undefined;
+    return JSON.parse(String(requestInit?.body ?? "{}"));
+  });
+}
+
+function sentRawBodies(): string[] {
+  return vi.mocked(fetch).mock.calls.map((call) => String((call[1] as RequestInit).body));
+}
+
+describe("telemetry client wire contract", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it("emits only allow-set keys with schemaVersion '1' on every POST", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const maxEventsPerBatch = 3;
+    const client = makeClient({ maxEventsPerBatch });
+
+    for (let i = 0; i < 7; i++) client.trackDynamic("plugin.telemetry.evt", { i });
+    await client.flush();
+
+    for (const body of sentBodies()) {
+      for (const key of Object.keys(body)) {
+        expect(CONTRACT_KEYS.has(key)).toBe(true);
+      }
+      expect(body.schemaVersion).toBe("1");
+    }
+  });
+
+  it("never exceeds the injected maxEventsPerBatch on any POST", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const maxEventsPerBatch = 4;
+    const client = makeClient({ maxEventsPerBatch });
+
+    for (let i = 0; i < 10; i++) client.trackDynamic("plugin.telemetry.evt", { i });
+    await client.flush();
+
+    const bodies = sentBodies();
+    expect(bodies.length).toBeGreaterThan(1); // proves it actually chunked
+    for (const body of bodies) {
+      expect((body.events as unknown[]).length).toBeLessThanOrEqual(maxEventsPerBatch);
+    }
+  });
+
+  it("never exceeds the injected maxBodyBytes on any POST", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const maxBodyBytes = 1500;
+    const client = makeClient({ maxEventsPerBatch: 1000, maxBodyBytes });
+
+    const blob = "y".repeat(300);
+    for (let i = 0; i < 12; i++) client.trackDynamic("plugin.telemetry.evt", { i, blob });
+    await client.flush();
+
+    const raw = sentRawBodies();
+    expect(raw.length).toBeGreaterThan(1); // proves byte-splitting happened
+    for (const body of raw) {
+      expect(Buffer.byteLength(body)).toBeLessThanOrEqual(maxBodyBytes);
+    }
+  });
+
+  it("a large flush splits into N envelopes that are each count- and byte-compliant", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const maxEventsPerBatch = 5;
+    const maxBodyBytes = 4096;
+    const client = makeClient({ maxEventsPerBatch, maxBodyBytes });
+
+    for (let i = 0; i < 23; i++) client.trackDynamic("plugin.telemetry.evt", { i });
+    await client.flush();
+
+    const bodies = sentBodies();
+    const raw = sentRawBodies();
+    const total = bodies.reduce((sum, b) => sum + (b.events as unknown[]).length, 0);
+    expect(total).toBe(23); // no events lost
+    bodies.forEach((body, idx) => {
+      expect((body.events as unknown[]).length).toBeLessThanOrEqual(maxEventsPerBatch);
+      expect(Buffer.byteLength(raw[idx] as string)).toBeLessThanOrEqual(maxBodyBytes);
+    });
+  });
+});
+
+describe("telemetry client wire contract — retry semantics", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("retries a 429 and re-sends the identical batchId (server idempotency)", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const client = makeClient({
+      backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 },
+    });
+
+    client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await client.flush();
+    await vi.advanceTimersByTimeAsync(1_000);
+
+    const bodies = sentBodies();
+    expect(bodies).toHaveLength(2);
+    expect(bodies[0]?.batchId).toBe(bodies[1]?.batchId);
+    expect(bodies[1]?.events).toEqual(bodies[0]?.events); // no re-mix
+    client.stop();
+  });
+
+  it("does not retry a terminal 400 or 413", async () => {
+    for (const status of [400, 413]) {
+      const fetchMock = vi.fn().mockResolvedValue({ ok: false, status });
+      vi.stubGlobal("fetch", fetchMock);
+      const client = makeClient();
+
+      client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+      await client.flush();
+      await vi.advanceTimersByTimeAsync(120_000);
+
+      expect(fetchMock).toHaveBeenCalledTimes(1);
+      client.stop();
+      vi.restoreAllMocks();
+    }
+  });
+});
+
+describe("telemetry client wire contract — batchId identity", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it("derives a different batchId for the same events under a different installId", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+
+    const a = makeClient(undefined, () => ({ ...TEST_STATE, installId: "install-a" }));
+    a.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await a.flush();
+    const idA = sentBodies().at(-1)?.batchId;
+
+    vi.mocked(fetch).mockClear();
+
+    const b = makeClient(undefined, () => ({ ...TEST_STATE, installId: "install-b" }));
+    b.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await b.flush();
+    const idB = sentBodies().at(-1)?.batchId;
+
+    expect(idA).not.toBe(idB);
+  });
+
+  it("emits a batchId of at least 32 hex chars — 128-bit collision floor", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const client = makeClient();
+
+    client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await client.flush();
+
+    const id = sentBodies().at(-1)?.batchId as string;
+    expect(id.length).toBeGreaterThanOrEqual(32);
+    expect(id).toMatch(/^[0-9a-f]+$/);
+  });
+});

packages/shared/src/telemetry/client.test.ts

@@ -1,7 +1,7 @@
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { TelemetryClient } from "./client.js";
 import { resolveTelemetryConfig } from "./config.js";
-import type { TelemetryConfig, TelemetryState } from "./types.js";
+import type { TelemetryConfig, TelemetryDimensions, TelemetryState } from "./types.js";
 
 const TEST_STATE: TelemetryState = {
   installId: "test-install",
@@ -10,12 +10,19 @@ const TEST_STATE: TelemetryState = {
   firstSeenVersion: "0.0.0",
 };
 
-function makeClient(stateFactory = vi.fn(() => TEST_STATE), config?: Partial<TelemetryConfig>) {
+function makeClient(
+  stateFactory = vi.fn(() => TEST_STATE),
+  config?: Partial<TelemetryConfig>,
+  // Seeded RNG for deterministic backoff jitter. 0.5 => zero jitter (the
+  // symmetric midpoint), so retry delays equal the un-jittered exponential base.
+  random: () => number = () => 0.5,
+) {
   return {
     client: new TelemetryClient(
       { enabled: true, endpoint: "http://localhost:9999/ingest", ...config },
       stateFactory,
       "0.0.0-test",
+      random,
     ),
     stateFactory,
   };
@@ -26,6 +33,14 @@ function sentBody() {
   return JSON.parse(String(requestInit?.body ?? "{}"));
 }
 
+// Parsed request bodies for every POST the client made this test, in call order.
+function sentBodies(): Array<Record<string, unknown>> {
+  return vi.mocked(fetch).mock.calls.map((call) => {
+    const requestInit = call[1] as RequestInit | undefined;
+    return JSON.parse(String(requestInit?.body ?? "{}"));
+  });
+}
+
 describe("TelemetryClient runtime event gate", () => {
   afterEach(() => {
     vi.restoreAllMocks();
@@ -103,89 +118,9 @@ describe("TelemetryClient runtime event gate", () => {
   });
 });
 
-// Stubs `fetch` to reject the batch with a given non-OK HTTP status for every
-// endpoint the client may try. Returns the mock so call counts can be asserted.
-function stubFetchStatus(status: number) {
-  const fetchMock = vi.fn().mockResolvedValue({ ok: false, status });
-  vi.stubGlobal("fetch", fetchMock);
-  return fetchMock;
-}
-
-// Phase 1 (PAP-2862): characterization pins for today's best-effort, silent-drop
-// flush. On ANY non-OK response or network error the drained batch is dropped
-// with no re-queue and no second attempt, and no `batchId` is emitted. These pins
-// lock the current baseline; Impl-2 (PAP-2853) replaces them when retry lands.
-describe("TelemetryClient silent-drop baseline (characterization)", () => {
-  afterEach(() => {
-    vi.restoreAllMocks();
-  });
-
-  it("drops the batch on a 429 with no re-queue", async () => {
-    const fetchMock = stubFetchStatus(429);
-    const { client } = makeClient();
-
-    client.track("install.started", {});
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-
-    // Queue was drained despite the failure: a second flush sends nothing.
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-  });
-
-  it("drops the batch on a 413 with no re-queue", async () => {
-    const fetchMock = stubFetchStatus(413);
-    const { client } = makeClient();
-
-    client.track("install.started", {});
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-  });
-
-  it("drops the batch on a 400 with no re-queue", async () => {
-    const fetchMock = stubFetchStatus(400);
-    const { client } = makeClient();
-
-    client.track("install.started", {});
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-  });
-
-  it("drops the batch on network error with no re-queue", async () => {
-    const fetchMock = vi.fn().mockRejectedValue(new Error("network down"));
-    vi.stubGlobal("fetch", fetchMock);
-    const { client } = makeClient();
-
-    client.track("install.started", {});
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-
-    await client.flush();
-    expect(fetchMock).toHaveBeenCalledTimes(1);
-  });
-
-  it("emits no batchId today", async () => {
-    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
-    const { client } = makeClient();
-
-    client.track("install.started", {});
-    await client.flush();
-
-    expect(fetch).toHaveBeenCalledTimes(1);
-    expect(sentBody()).not.toHaveProperty("batchId");
-  });
-});
-
-// Phase 2 (PAP-2862): config surface for soft caps + backoff. Fields are optional
-// and additive; `resolveTelemetryConfig` fills documented defaults centrally so no
-// existing caller changes behavior. Nothing reads these yet — Impl-2 is the first
-// consumer.
+// Config surface for soft caps + backoff. Fields are optional and additive;
+// `resolveTelemetryConfig` fills documented defaults centrally so no existing
+// caller changes behavior.
 describe("resolveTelemetryConfig caps + backoff surface", () => {
   it("resolveTelemetryConfig returns default caps and backoff", () => {
     const config = resolveTelemetryConfig();
@@ -225,3 +160,456 @@ describe("resolveTelemetryConfig caps + backoff surface", () => {
     });
   });
 });
+
+// flush() must never emit an oversized batch. The drained
+// queue is sub-divided into envelopes of <= config.maxEventsPerBatch events AND
+// <= config.maxBodyBytes serialized bytes, one POST per chunk. Caps are injected
+// (small) so assertions are RELATIVE invariants, never server literals.
+describe("TelemetryClient chunking (count + bytes)", () => {
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it("splits more than maxEventsPerBatch events into multiple compliant POSTs", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const cap = 2;
+    const { client } = makeClient(undefined, { maxEventsPerBatch: cap });
+
+    for (let i = 0; i < 5; i++) client.track("install.started", {});
+    await client.flush();
+
+    // 5 events / cap 2 => 3 POSTs (2 + 2 + 1)
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    const bodies = sentBodies();
+    expect(bodies.map((b) => (b.events as unknown[]).length)).toEqual([2, 2, 1]);
+    for (const body of bodies) {
+      expect((body.events as unknown[]).length).toBeLessThanOrEqual(cap);
+    }
+  });
+
+  it("splits a chunk whose serialized bytes exceed maxBodyBytes", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    // Big enough to hold a single fat event but force >1 event to split.
+    const maxBodyBytes = 900;
+    const { client } = makeClient(undefined, { maxEventsPerBatch: 100, maxBodyBytes });
+
+    const blob = "x".repeat(300);
+    for (let i = 0; i < 6; i++) client.trackDynamic("plugin.telemetry.blob", { blob });
+    await client.flush();
+
+    expect(fetchMock.mock.calls.length).toBeGreaterThan(1);
+    for (const call of fetchMock.mock.calls) {
+      const body = String((call[1] as RequestInit).body);
+      expect(Buffer.byteLength(body)).toBeLessThanOrEqual(maxBodyBytes);
+    }
+  });
+
+  it("drops a single event larger than maxBodyBytes and logs, sending nothing", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const { client } = makeClient(undefined, { maxBodyBytes: 200 });
+
+    client.trackDynamic("plugin.telemetry.blob", { blob: "x".repeat(5000) });
+    await client.flush();
+
+    expect(fetchMock).not.toHaveBeenCalled();
+    expect(warn).toHaveBeenCalled();
+  });
+});
+
+// Every emitted chunk carries a deterministic, salt-free content-hash `batchId`
+// so server-side retries de-dupe (202) instead of double-counting. Two
+// invariants hold: the hash input includes `installId` (so two installs sending
+// identical events get distinct ids), and the id is >= 32 hex chars (128-bit
+// collision floor).
+describe("TelemetryClient deterministic batchId", () => {
+  // `deriveBatchId` intentionally hashes the full event objects, INCLUDING each
+  // event's `occurredAt` wall-clock stamp, so the id stays a faithful content
+  // hash: two genuinely distinct sends get distinct ids and the server ledger
+  // (keyed on `batchId`) counts both. Dropping `occurredAt` from the hash would
+  // collapse same-shape sends at different times onto one id and make the server
+  // silently drop the later batch as a replay — a silent-loss failure mode.
+  // These cross-instance determinism checks therefore freeze the clock so
+  // both clients stamp an identical `occurredAt`, isolating the hash's structural
+  // determinism from wall-clock skew.
+  beforeEach(() => {
+    vi.useFakeTimers({ now: new Date("2026-01-01T00:00:00.000Z") });
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("emits a batchId on every envelope", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const { client } = makeClient();
+
+    client.track("install.started", {});
+    await client.flush();
+
+    expect(typeof sentBody().batchId).toBe("string");
+  });
+
+  it("derives a stable batchId for identical events (idempotent retry key)", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+
+    const first = makeClient();
+    first.client.trackDynamic("plugin.telemetry.evt", { a: 1, b: "two" });
+    await first.client.flush();
+    const idA = sentBody().batchId;
+
+    vi.mocked(fetch).mockClear();
+
+    const second = makeClient();
+    second.client.trackDynamic("plugin.telemetry.evt", { a: 1, b: "two" });
+    await second.client.flush();
+    const idB = sentBody().batchId;
+
+    // Same installId + same event content => identical id.
+    expect(idA).toBe(idB);
+  });
+
+  it("derives a different batchId for different events", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+
+    const first = makeClient();
+    first.client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await first.client.flush();
+    const idA = sentBody().batchId;
+
+    vi.mocked(fetch).mockClear();
+
+    const second = makeClient();
+    second.client.trackDynamic("plugin.telemetry.evt", { a: 2 });
+    await second.client.flush();
+    const idB = sentBody().batchId;
+
+    expect(idA).not.toBe(idB);
+  });
+
+  it("derives a different batchId for the same events under a different installId", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+
+    const stateA = vi.fn(() => ({ ...TEST_STATE, installId: "install-a" }));
+    const first = makeClient(stateA);
+    first.client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await first.client.flush();
+    const idA = sentBody().batchId;
+
+    vi.mocked(fetch).mockClear();
+
+    const stateB = vi.fn(() => ({ ...TEST_STATE, installId: "install-b" }));
+    const second = makeClient(stateB);
+    second.client.trackDynamic("plugin.telemetry.evt", { a: 1 });
+    await second.client.flush();
+    const idB = sentBody().batchId;
+
+    expect(idA).not.toBe(idB);
+  });
+
+  it("emits a batchId of at least 32 hex chars (128-bit collision floor)", async () => {
+    vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+    const { client } = makeClient();
+
+    client.track("install.started", {});
+    await client.flush();
+
+    const id = sentBody().batchId as string;
+    expect(id.length).toBeGreaterThanOrEqual(32);
+    expect(id).toMatch(/^[0-9a-f]+$/);
+  });
+
+  it("does not crash flush on a circular dimension; drops-and-logs it and still sends valid events", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const { client } = makeClient();
+
+    // A plugin passing a circular `dimensions` object. `enqueue` shallow-copies
+    // the top level, so the cycle survives one level down and would drive both
+    // `stableStringify` (deriveBatchId) and `JSON.stringify` (serializedBytes /
+    // wire body) into an unhandled throw without the guards.
+    const circular: Record<string, unknown> = { a: 1 };
+    circular.self = circular;
+    // Cast models an untyped third-party plugin passing a malformed object at
+    // runtime — TypeScript would reject the cycle, but the wire path cannot.
+    client.trackDynamic("plugin.telemetry.circular", circular as unknown as TelemetryDimensions);
+    // A well-formed event queued alongside it must still be delivered.
+    client.trackDynamic("plugin.telemetry.ok", { a: 1 });
+
+    // flush() resolves instead of rejecting with a RangeError/TypeError.
+    await expect(client.flush()).resolves.toBeUndefined();
+
+    // The circular event was dropped-and-logged (fail loudly), not sent.
+    expect(warn).toHaveBeenCalledWith(
+      expect.stringContaining("non-serializable dimension (circular reference?)"),
+    );
+
+    // The valid event still went out on the wire with a well-formed batchId.
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const body = sentBody();
+    expect(body.events).toHaveLength(1);
+    expect((body.events as Array<{ name: string }>)[0]?.name).toBe("plugin.telemetry.ok");
+    const id = body.batchId as string;
+    expect(id.length).toBeGreaterThanOrEqual(32);
+    expect(id).toMatch(/^[0-9a-f]+$/);
+  });
+});
+
+// Batch-grouped retry with capped, jittered exponential
+// backoff. Retryable statuses (429/502/503/504 + network) re-send the EXACT
+// same events + batchId (no re-mix — preserves server idempotency); terminal
+// statuses (400/405/409/413) never retry. Backoff uses the injected seeded RNG
+// and retries are driven off fake timers so tests are deterministic.
+describe("TelemetryClient batched retry + backoff", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("retries a 429 with the same batchId and eventually succeeds", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const { client } = makeClient(undefined, {
+      backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 },
+    });
+
+    client.track("install.started", {});
+    await client.flush();
+    expect(fetchMock).toHaveBeenCalledTimes(1); // attempt 1 -> 429, queued for retry
+
+    // Un-jittered delay for attempt 1 == baseDelayMs.
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(fetchMock).toHaveBeenCalledTimes(2); // attempt 2 -> 200
+
+    const bodies = sentBodies();
+    expect(bodies[0]?.batchId).toBe(bodies[1]?.batchId); // identical id on retry
+    client.stop();
+  });
+
+  it("does not retry a terminal 400", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 400 });
+    vi.stubGlobal("fetch", fetchMock);
+    const { client } = makeClient();
+
+    client.track("install.started", {});
+    await client.flush();
+    await vi.advanceTimersByTimeAsync(120_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    client.stop();
+  });
+
+  it("does not retry a terminal 413", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 413 });
+    vi.stubGlobal("fetch", fetchMock);
+    const { client } = makeClient();
+
+    client.track("install.started", {});
+    await client.flush();
+    await vi.advanceTimersByTimeAsync(120_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    client.stop();
+  });
+
+  it("stops after maxAttempts on a persistent 429 and drops-and-logs", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 });
+    vi.stubGlobal("fetch", fetchMock);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const maxAttempts = 3;
+    const { client } = makeClient(undefined, {
+      backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts, jitterRatio: 0.25 },
+    });
+
+    client.track("install.started", {});
+    await client.flush();
+    // Drive all remaining scheduled retries.
+    await vi.advanceTimersByTimeAsync(600_000);
+
+    expect(fetchMock).toHaveBeenCalledTimes(maxAttempts);
+    expect(warn).toHaveBeenCalled();
+    client.stop();
+  });
+
+  // Issue 1 (PR #9946): a transient upstream 5xx on the primary endpoint must
+  // fall through to the healthy secondary endpoint instead of returning early.
+  it("falls through to the secondary endpoint on a transient 5xx", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 503 }) // primary endpoint: transient
+      .mockResolvedValueOnce({ ok: true }); // secondary endpoint: healthy
+    vi.stubGlobal("fetch", fetchMock);
+    // Empty endpoint => the two built-in DEFAULT_ENDPOINTS are used.
+    const { client } = makeClient(undefined, { endpoint: "" });
+
+    client.track("install.started", {});
+    await client.flush();
+
+    // Both endpoints tried within a single attempt; delivered on the secondary,
+    // so no retry is queued.
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    await vi.advanceTimersByTimeAsync(120_000);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    client.stop();
+  });
+
+  // Issue 1 (PR #9946): when every endpoint returns a transient 5xx the status
+  // is still surfaced as retryable (not swallowed).
+  it("surfaces the transient status for retry when all endpoints 5xx", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 502 }) // primary
+      .mockResolvedValueOnce({ ok: false, status: 502 }) // secondary
+      .mockResolvedValue({ ok: true }); // retry succeeds
+    vi.stubGlobal("fetch", fetchMock);
+    const { client } = makeClient(undefined, {
+      endpoint: "",
+      backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 },
+    });
+
+    client.track("install.started", {});
+    await client.flush();
+    expect(fetchMock).toHaveBeenCalledTimes(2); // both endpoints 502 -> queued for retry
+
+    await vi.advanceTimersByTimeAsync(1_000); // attempt 2 -> ok on primary
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    client.stop();
+  });
+
+  // Issue 3 (PR #9946): an out-of-range `Retry-After` hint is clamped to
+  // maxDelayMs rather than overflowing the timer range into a near-immediate
+  // (Node-clamped ~1ms) retry.
+  it("caps a large Retry-After hint at maxDelayMs", async () => {
+    const retryAfterResponse = {
+      ok: false,
+      status: 429,
+      headers: { get: (name: string) => (name === "retry-after" ? "999999999" : null) },
+    };
+    const fetchMock = vi.fn().mockResolvedValueOnce(retryAfterResponse).mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const { client } = makeClient(undefined, {
+      backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 },
+    });
+
+    client.track("install.started", {});
+    await client.flush();
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    // Well before the cap: without clamping the huge hint overflows the timer
+    // range and Node fires it near-immediately, so this would already be 2.
+    await vi.advanceTimersByTimeAsync(5_000);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+
+    // At the cap the retry fires.
+    await vi.advanceTimersByTimeAsync(25_000);
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    client.stop();
+  });
+});
+
+// The pending-retry store is bounded at
+// config.maxPendingRetryBatches. On overflow the OLDEST batch is evicted
+// (newest prioritized) and each eviction is logged (no silent loss). In-memory
+// only. Caps are injected; assertions are relative.
+describe("TelemetryClient bounded pending-retry store", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("evicts the oldest batch and retries only the newest within the bound", async () => {
+    // First 3 POSTs (one per single-event chunk) fail 429; retries then succeed.
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 2 });
+
+    client.trackDynamic("plugin.telemetry.evt", { n: 1 });
+    client.trackDynamic("plugin.telemetry.evt", { n: 2 });
+    client.trackDynamic("plugin.telemetry.evt", { n: 3 });
+    await client.flush();
+
+    // 3 initial attempts (one per chunk), each 429 -> enqueued; the 3rd enqueue
+    // overflows the bound (2) and evicts the oldest.
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    const initial = sentBodies();
+    const [oldestId, midId, newestId] = initial.map((b) => b.batchId as string);
+    expect(warn).toHaveBeenCalled(); // eviction logged
+
+    await vi.advanceTimersByTimeAsync(120_000);
+
+    // Only the 2 newest were retained and retried; the oldest was dropped.
+    const retriedIds = sentBodies()
+      .slice(3)
+      .map((b) => b.batchId as string);
+    expect(retriedIds.sort()).toEqual([midId, newestId].sort());
+    expect(retriedIds).not.toContain(oldestId);
+    client.stop();
+  });
+
+  // Issue 2 (PR #9946): a batch that is immediately evicted by the bound must
+  // NOT leave a retry timer behind. With maxPendingRetryBatches: 0 every failed
+  // batch is discarded on enqueue, so no timer should ever fire.
+  it("schedules no retry timer for a batch evicted on enqueue (bound 0)", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 });
+    vi.stubGlobal("fetch", fetchMock);
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
+    const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 0 });
+
+    client.trackDynamic("plugin.telemetry.evt", { n: 1 });
+    client.trackDynamic("plugin.telemetry.evt", { n: 2 });
+    await client.flush();
+
+    // 2 initial attempts (one per chunk); each 429 is enqueued then immediately
+    // evicted by the 0 bound, so nothing is retried.
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(warn).toHaveBeenCalled(); // eviction logged
+
+    await vi.advanceTimersByTimeAsync(120_000);
+    expect(fetchMock).toHaveBeenCalledTimes(2); // no timer fired
+    client.stop();
+  });
+
+  // A batch evicted to keep `pending` within the bound must also have its
+  // already-scheduled retry timer cancelled — otherwise each overflow strands a
+  // live timer for a batch that no longer exists, and the timer set grows
+  // unbounded even though `pending` stays bounded.
+  it("cancels the retry timer of a batch evicted from the bounded store", async () => {
+    const fetchMock = vi.fn().mockResolvedValue({ ok: false, status: 429 });
+    vi.stubGlobal("fetch", fetchMock);
+    vi.spyOn(console, "warn").mockImplementation(() => {});
+    const { client } = makeClient(undefined, { maxEventsPerBatch: 1, maxPendingRetryBatches: 1 });
+
+    client.trackDynamic("plugin.telemetry.evt", { n: 1 });
+    client.trackDynamic("plugin.telemetry.evt", { n: 2 });
+    client.trackDynamic("plugin.telemetry.evt", { n: 3 });
+    await client.flush();
+
+    // 3 chunks each 429 -> enqueued; every enqueue past the first overflows the
+    // bound (1) and evicts the previous batch. Only the newest batch survives, so
+    // exactly one retry timer should remain live (the two evicted timers were
+    // cancelled), not one per failed chunk.
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(vi.getTimerCount()).toBe(1);
+    client.stop();
+  });
+});

packages/shared/src/telemetry/client.ts

@@ -4,35 +4,111 @@ import type {
   TelemetryDimensions,
   TelemetryEvent,
   TelemetryEventDimensions,
+  TelemetryEventEnvelope,
   TelemetryEventName,
   TelemetryState,
 } from "./types.js";
+import { type ResolvedTelemetryCaps, resolveCaps } from "./config.js";
 import { PAPERCLIP_EVENTS } from "./generated/paperclip-telemetry.js";
 
 const DEFAULT_ENDPOINTS = [
   "https://telemetry.paperclip.ing/ingest",
   "https://rusqrrg391.execute-api.us-east-1.amazonaws.com/ingest",
 ] as const;
+// Queue-pressure valve: auto-flush once this many events are buffered. This is
+// an in-memory backpressure trigger, independent of the wire caps that
+// `chunkForSend` enforces on each POST.
 const BATCH_SIZE = 50;
 const SEND_TIMEOUT_MS = 5_000;
 
+/**
+ * Deterministic, key-stable JSON serialization used to derive a content-stable
+ * `batchId`. Object keys are sorted so two structurally-equal event sets always
+ * produce the same string regardless of insertion order. Mirrors the
+ * `stableStringify` exemplar in `external-objects-server.ts`.
+ */
+function stableStringify(value: unknown): string {
+  if (Array.isArray(value)) {
+    return `[${value.map((entry) => stableStringify(entry)).join(",")}]`;
+  }
+  if (value && typeof value === "object") {
+    const record = value as Record<string, unknown>;
+    return `{${Object.keys(record)
+      .sort()
+      .map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`)
+      .join(",")}}`;
+  }
+  return JSON.stringify(value);
+}
+
+/**
+ * Parses a `Retry-After` header into milliseconds — defensively, only when
+ * present and a non-negative number of seconds. The telemetry backend emits no
+ * `Retry-After` today, so this never fires in practice; it exists so a future
+ * server hint is honored rather than ignored. The HTTP-date form is
+ * intentionally not parsed.
+ */
+function parseRetryAfterMs(response: { headers?: { get?(name: string): string | null } }): number | undefined {
+  const raw = response.headers?.get?.("retry-after");
+  if (!raw) return undefined;
+  const seconds = Number(raw);
+  if (!Number.isFinite(seconds) || seconds < 0) return undefined;
+  return Math.round(seconds * 1000);
+}
+
 type TrackArgs<K extends TelemetryEventName> =
   keyof TelemetryEventDimensions<K> extends never
     ? [dimensions?: TelemetryEventDimensions<K>]
     : [dimensions: TelemetryEventDimensions<K>];
 
+// Length of the truncated hex `batchId`. 32 hex chars = 128 bits — the
+// collision-safe floor. Do not lower below 32.
+const BATCH_ID_HEX_LENGTH = 32;
+
+/**
+ * A chunk awaiting retry. `events` + `batchId` are frozen at first send and
+ * re-sent verbatim on every attempt — re-mixing events would change the content
+ * the server hashed under this id and trigger a 409. `attempt` is 1-based (the
+ * value of the attempt about to be made); `nextAttemptAt` is an epoch-ms gate.
+ * `timerId` is this batch's scheduled retry wake-up (if any) — held on the record
+ * so it can be cancelled when the batch is evicted from the bounded store,
+ * otherwise the timer would keep firing for a batch that no longer exists.
+ */
+interface PendingBatch {
+  events: TelemetryEvent[];
+  batchId: string;
+  attempt: number;
+  nextAttemptAt: number;
+  timerId?: ReturnType<typeof setTimeout>;
+}
+
 export class TelemetryClient {
   private queue: TelemetryEvent[] = [];
   private readonly config: TelemetryConfig;
+  private readonly caps: ResolvedTelemetryCaps;
   private readonly stateFactory: () => TelemetryState;
   private readonly version: string;
+  private readonly random: () => number;
   private state: TelemetryState | null = null;
   private flushInterval: ReturnType<typeof setInterval> | null = null;
+  // In-memory pending-retry store (best-effort; never persisted). Bounded by
+  // `maxPendingRetryBatches`. Insertion order == age (oldest at the front).
+  private pending: PendingBatch[] = [];
+  private readonly retryTimers = new Set<ReturnType<typeof setTimeout>>();
 
-  constructor(config: TelemetryConfig, stateFactory: () => TelemetryState, version: string) {
+  constructor(
+    config: TelemetryConfig,
+    stateFactory: () => TelemetryState,
+    version: string,
+    // Injectable RNG for backoff jitter — defaults to `Math.random`; tests pass
+    // a seeded function for deterministic backoff. Callers keep the 3-arg form.
+    random: () => number = Math.random,
+  ) {
     this.config = config;
+    this.caps = resolveCaps(config);
     this.stateFactory = stateFactory;
     this.version = version;
+    this.random = random;
   }
 
   /**
@@ -70,21 +146,263 @@ export class TelemetryClient {
   }
 
   async flush(): Promise<void> {
-    if (!this.config.enabled || this.queue.length === 0) return;
+    if (!this.config.enabled) return;
+
+    // Re-send any due retries first, then send freshly-queued events.
+    await this.drainPending();
+    if (this.queue.length === 0) return;
 
     const events = this.queue.splice(0);
+    for (const chunk of this.chunkForSend(events)) {
+      await this.attemptSend({
+        events: chunk,
+        batchId: this.deriveBatchId(this.getState().installId, chunk),
+        attempt: 1,
+        nextAttemptAt: 0,
+      });
+    }
+  }
+
+  /**
+   * Partitions a drained event list into wire-compliant chunks: first by
+   * `maxEventsPerBatch` (count), then recursively byte-splits any chunk whose
+   * serialized envelope still exceeds `maxBodyBytes` (halving). A single event
+   * that alone exceeds `maxBodyBytes` is dropped-and-logged (fail loudly) rather
+   * than sent over-limit.
+   */
+  private chunkForSend(events: TelemetryEvent[]): TelemetryEvent[][] {
+    const maxCount = Math.max(1, this.caps.maxEventsPerBatch);
+    const out: TelemetryEvent[][] = [];
+    for (let i = 0; i < events.length; i += maxCount) {
+      this.splitByBytes(events.slice(i, i + maxCount), out);
+    }
+    return out;
+  }
+
+  private splitByBytes(chunk: TelemetryEvent[], out: TelemetryEvent[][]): void {
+    if (chunk.length === 0) return;
+    const bytes = this.serializedBytes(this.buildEnvelope(chunk));
+    if (bytes <= this.caps.maxBodyBytes) {
+      out.push(chunk);
+      return;
+    }
+    if (chunk.length === 1) {
+      this.warn(
+        Number.isFinite(bytes)
+          ? `dropping 1 event whose serialized envelope exceeds maxBodyBytes (${this.caps.maxBodyBytes} bytes); event="${chunk[0]?.name}"`
+          : `dropping 1 event with a non-serializable dimension (circular reference?); event="${chunk[0]?.name}"`,
+      );
+      return;
+    }
+    const mid = Math.ceil(chunk.length / 2);
+    this.splitByBytes(chunk.slice(0, mid), out);
+    this.splitByBytes(chunk.slice(mid), out);
+  }
+
+  private buildEnvelope(events: TelemetryEvent[], batchId?: string): TelemetryEventEnvelope {
     const state = this.getState();
-    const endpoints = this.resolveEndpoints();
-    const app = this.config.app ?? "paperclip";
-    const schemaVersion = this.config.schemaVersion ?? "1";
-    const body = JSON.stringify({
-      app,
-      schemaVersion,
+    return {
+      app: this.config.app ?? "paperclip",
+      schemaVersion: this.config.schemaVersion ?? "1",
       installId: state.installId,
       version: this.version,
       events,
-    });
+      batchId: batchId ?? this.deriveBatchId(state.installId, events),
+    };
+  }
 
+  /**
+   * Deterministic, salt-free content-hash idempotency key. Hashes
+   * `{installId, events}` (scopes the key per install, matching the
+   * server's `contentSha256` scope so cross-install batches never collide on the
+   * server ledger key) and truncates to 128 bits. Identical input always
+   * yields the same id, so a retried batch replays idempotently (202) instead of
+   * double-counting; different events/install yield a different id.
+   */
+  private deriveBatchId(installId: string, events: TelemetryEvent[]): string {
+    try {
+      return createHash("sha256")
+        .update(stableStringify({ installId, events }))
+        .digest("hex")
+        .slice(0, BATCH_ID_HEX_LENGTH);
+    } catch {
+      // A plugin can pass a circular (or otherwise non-serializable) `dimensions`
+      // object via `trackDynamic`; `stableStringify` would then recurse until it
+      // throws a `RangeError`. Fall back to a count-based hash so this call never
+      // crashes the flush. Such an event can't be JSON-serialized for the wire
+      // either, so it is dropped-and-logged in `splitByBytes` — this fallback id
+      // is only ever attached to a batch that is about to be dropped, so its
+      // weakened idempotency is moot. The single operator-facing signal is the
+      // drop-and-log warning, not a (recursively-repeated) warning here.
+      return createHash("sha256")
+        .update(JSON.stringify({ installId, count: events.length }))
+        .digest("hex")
+        .slice(0, BATCH_ID_HEX_LENGTH);
+    }
+  }
+
+  private serializedBytes(envelope: TelemetryEventEnvelope): number {
+    try {
+      return Buffer.byteLength(JSON.stringify(envelope));
+    } catch {
+      // A non-serializable event (e.g. a circular `dimensions` object from a
+      // plugin) can neither be byte-measured nor sent over the wire. Report it
+      // as effectively unbounded so `splitByBytes` routes it to the existing
+      // over-limit drop-and-log path instead of throwing out of the flush.
+      return Number.POSITIVE_INFINITY;
+    }
+  }
+
+  /**
+   * Sends one batch (its current `attempt`). On a retryable failure the EXACT
+   * same events + `batchId` are re-queued with capped, jittered backoff; on a
+   * terminal failure or after `maxAttempts` the batch is dropped-and-logged.
+   */
+  private async attemptSend(batch: PendingBatch): Promise<void> {
+    const body = JSON.stringify(this.buildEnvelope(batch.events, batch.batchId));
+    const outcome = this.classifyOutcome(await this.postEnvelope(body));
+
+    if (outcome.kind === "ok") return;
+
+    if (outcome.kind === "terminal") {
+      this.warn(
+        `dropping batch ${batch.batchId} on terminal response (HTTP ${outcome.status}); ${batch.events.length} event(s) lost`,
+      );
+      return;
+    }
+
+    // Retryable (429/502/503/504 or network/timeout).
+    if (batch.attempt >= this.caps.backoff.maxAttempts) {
+      this.warn(
+        `dropping batch ${batch.batchId} after ${batch.attempt} attempt(s); ${batch.events.length} event(s) lost`,
+      );
+      return;
+    }
+    // Cap the delay at maxDelayMs. `computeBackoffMs` is already capped, but a
+    // server `Retry-After` hint is not — an out-of-range value could otherwise
+    // overflow the runtime timer range and be clamped by Node to a near-immediate
+    // timeout, causing rapid retries. The cap keeps every retry within the
+    // configured backoff ceiling.
+    const delayMs = Math.min(
+      outcome.retryAfterMs ?? this.computeBackoffMs(batch.attempt),
+      this.caps.backoff.maxDelayMs,
+    );
+    this.enqueuePending({
+      events: batch.events,
+      batchId: batch.batchId,
+      attempt: batch.attempt + 1,
+      nextAttemptAt: Date.now() + delayMs,
+    });
+  }
+
+  private classifyOutcome(
+    result: { kind: "ok" } | { kind: "status"; status: number; retryAfterMs?: number } | { kind: "network" },
+  ): { kind: "ok" } | { kind: "retry"; retryAfterMs?: number } | { kind: "terminal"; status: number } {
+    if (result.kind === "ok") return { kind: "ok" };
+    if (result.kind === "network") return { kind: "retry" };
+    if (this.isRetryableStatus(result.status)) {
+      return { kind: "retry", retryAfterMs: result.retryAfterMs };
+    }
+    return { kind: "terminal", status: result.status };
+  }
+
+  private isRetryableStatus(status: number): boolean {
+    return status === 429 || status === 502 || status === 503 || status === 504;
+  }
+
+  /**
+   * Capped exponential backoff with symmetric jitter:
+   * `min(maxDelayMs, baseDelayMs * 2^(attempt-1)) * (1 ± jitterRatio)`, using the
+   * injected RNG. `attempt` is the failed attempt (1-based).
+   */
+  private computeBackoffMs(attempt: number): number {
+    const { baseDelayMs, maxDelayMs, jitterRatio } = this.caps.backoff;
+    const base = Math.min(maxDelayMs, baseDelayMs * 2 ** (attempt - 1));
+    const jitter = base * jitterRatio * (this.random() * 2 - 1);
+    return Math.max(0, Math.min(maxDelayMs, Math.round(base + jitter)));
+  }
+
+  /**
+   * Pushes a batch onto the pending store (bounded), then schedules its retry
+   * wake-up. On overflow the OLDEST batches (front) are evicted first — newest
+   * prioritized — and each eviction is logged so no batch is lost silently.
+   */
+  private enqueuePending(batch: PendingBatch): void {
+    this.pending.push(batch);
+    const bound = Math.max(0, this.caps.maxPendingRetryBatches);
+    while (this.pending.length > bound) {
+      const evicted = this.pending.shift();
+      // Cancel the evicted batch's scheduled retry so its timer doesn't keep the
+      // wake-up alive for a batch that is no longer in the store. Without this a
+      // flush that overflows the bound would strand one live timer per evicted
+      // batch even though `pending` itself stays bounded.
+      if (evicted) this.cancelRetryTimer(evicted);
+      this.warn(
+        `pending-retry store full (bound=${bound}); evicted oldest batch ${evicted?.batchId}; ${evicted?.events.length ?? 0} event(s) lost`,
+      );
+    }
+    // Only schedule a wake-up if this batch actually survived eviction. When the
+    // batch is immediately evicted by the bound (e.g. a large failed flush, or
+    // `maxPendingRetryBatches: 0`) it has no pending work, so scheduling a timer
+    // for it would strand thousands of no-op timers behind a small bound.
+    if (this.pending.includes(batch)) {
+      this.scheduleDrain(batch);
+    }
+  }
+
+  /** Cancels a pending batch's scheduled retry wake-up, if it has one. */
+  private cancelRetryTimer(batch: PendingBatch): void {
+    if (batch.timerId === undefined) return;
+    clearTimeout(batch.timerId);
+    this.retryTimers.delete(batch.timerId);
+    batch.timerId = undefined;
+  }
+
+  private scheduleDrain(batch: PendingBatch): void {
+    const delay = Math.max(0, batch.nextAttemptAt - Date.now());
+    const timer = setTimeout(() => {
+      this.retryTimers.delete(timer);
+      batch.timerId = undefined;
+      void this.drainPending();
+    }, delay);
+    // Don't keep the process alive for a best-effort retry (CLI exits promptly).
+    if (typeof timer === "object" && timer !== null && "unref" in timer) {
+      (timer as { unref(): void }).unref();
+    }
+    this.retryTimers.add(timer);
+    batch.timerId = timer;
+  }
+
+  /** Re-sends every pending batch whose `nextAttemptAt` is due. */
+  private async drainPending(): Promise<void> {
+    if (this.pending.length === 0) return;
+    const now = Date.now();
+    const due: PendingBatch[] = [];
+    const waiting: PendingBatch[] = [];
+    for (const batch of this.pending) {
+      (batch.nextAttemptAt <= now ? due : waiting).push(batch);
+    }
+    this.pending = waiting;
+    for (const batch of due) {
+      await this.attemptSend(batch);
+    }
+  }
+
+  /**
+   * POSTs a serialized envelope, trying each endpoint in order. Returns the
+   * definitive outcome: `ok` on a 2xx; the HTTP `status` on a definitive non-2xx
+   * response (4xx incl. 429, which the shared API gateway fronts for every
+   * endpoint, so it is authoritative — stop here); or `network` when every
+   * endpoint threw. A transient upstream 5xx (502/503/504) does NOT stop the
+   * loop: a sibling endpoint may be healthy, so we fall through to it and only
+   * surface the last transient status if every endpoint returns one — matching
+   * the pre-retry loop's endpoint-fallback behavior.
+   */
+  private async postEnvelope(
+    body: string,
+  ): Promise<{ kind: "ok" } | { kind: "status"; status: number; retryAfterMs?: number } | { kind: "network" }> {
+    const endpoints = this.resolveEndpoints();
+    let lastTransient: { kind: "status"; status: number; retryAfterMs?: number } | undefined;
     for (const endpoint of endpoints) {
       const controller = new AbortController();
       const timer = setTimeout(() => controller.abort(), SEND_TIMEOUT_MS);
@@ -95,15 +413,32 @@ export class TelemetryClient {
           body,
           signal: controller.signal,
         });
-        if (response.ok) {
-          return;
+        if (response.ok) return { kind: "ok" };
+        const status = { kind: "status" as const, status: response.status, retryAfterMs: parseRetryAfterMs(response) };
+        // Transient upstream 5xx: remember it and try the next endpoint.
+        if (this.isTransientServerStatus(response.status)) {
+          lastTransient = status;
+          continue;
         }
+        return status;
       } catch {
-        // Try the next built-in endpoint before dropping the batch.
+        // Network/timeout on this endpoint — try the next built-in endpoint.
       } finally {
         clearTimeout(timer);
       }
     }
+    // Every endpoint failed: surface the last transient status (still retryable)
+    // if we saw one, otherwise report a pure network failure.
+    return lastTransient ?? { kind: "network" };
+  }
+
+  /** Upstream 5xx that a healthy sibling endpoint may still be able to serve. */
+  private isTransientServerStatus(status: number): boolean {
+    return status === 502 || status === 503 || status === 504;
+  }
+
+  private warn(message: string): void {
+    console.warn(`[telemetry] ${message}`);
   }
 
   startPeriodicFlush(intervalMs: number = 60_000): void {
@@ -122,6 +457,10 @@ export class TelemetryClient {
       clearInterval(this.flushInterval);
       this.flushInterval = null;
     }
+    for (const timer of this.retryTimers) {
+      clearTimeout(timer);
+    }
+    this.retryTimers.clear();
   }
 
   hashPrivateRef(value: string): string {

packages/shared/src/telemetry/config.ts

@@ -5,7 +5,7 @@ const CI_ENV_VARS = ["CI", "CONTINUOUS_INTEGRATION", "BUILD_NUMBER", "GITHUB_ACT
 /**
  * Single source of truth for telemetry soft caps + backoff. Kept as config
  * *defaults* (not hardcoded flush logic) so later work reads config, not
- * literals. Exported so Impl-2's `client.ts` consumer resolves the same values.
+ * literals. Exported so the `client.ts` consumer resolves the same values.
  */
 export const TELEMETRY_DEFAULTS: {
   readonly maxEventsPerBatch: number;
@@ -32,17 +32,24 @@ export type TelemetryConfigOverrides = Partial<
   >
 >;
 
-type ResolvedCaps = Pick<
-  TelemetryConfig,
-  "maxEventsPerBatch" | "maxBodyBytes" | "maxPendingRetryBatches" | "backoff"
->;
+/**
+ * Fully-resolved caps + backoff — every field is present (defaults applied), so
+ * `client.ts` can consume them without re-defaulting. `TelemetryConfig`'s own
+ * cap fields stay optional (additive wire surface); this is the resolved view.
+ */
+export interface ResolvedTelemetryCaps {
+  maxEventsPerBatch: number;
+  maxBodyBytes: number;
+  maxPendingRetryBatches: number;
+  backoff: TelemetryBackoffConfig;
+}
 
 /**
  * Resolves soft caps + backoff, applying `TELEMETRY_DEFAULTS` for any field the
  * caller did not override. One source of truth for both the config surface and
- * Impl-2's future `client.ts` consumer.
+ * the `client.ts` consumer.
  */
-export function resolveCaps(overrides?: TelemetryConfigOverrides): ResolvedCaps {
+export function resolveCaps(overrides?: TelemetryConfigOverrides): ResolvedTelemetryCaps {
   return {
     maxEventsPerBatch: overrides?.maxEventsPerBatch ?? TELEMETRY_DEFAULTS.maxEventsPerBatch,
     maxBodyBytes: overrides?.maxBodyBytes ?? TELEMETRY_DEFAULTS.maxBodyBytes,

packages/shared/src/telemetry/types.ts

@@ -13,8 +13,8 @@ export interface TelemetryState {
 /**
  * Exponential-backoff-with-jitter parameters for the (future) batched-retry
  * sender. Shape mirrors the plugin worker crash-recovery backoff
- * (`server/src/services/plugin-worker-manager.ts`). Consumed by Impl-2; nothing
- * reads it yet.
+ * (`server/src/services/plugin-worker-manager.ts`). Consumed by the
+ * batched-retry sender; nothing reads it yet.
  */
 export interface TelemetryBackoffConfig {
   baseDelayMs: number;
@@ -31,7 +31,7 @@ export interface TelemetryConfig {
   /**
    * Optional, additive soft caps + backoff. Defaulted centrally in
    * `resolveTelemetryConfig`; no wire/envelope change and no consumer today —
-   * Impl-2 (PAP-2853) is the first reader.
+   * the batched-retry sender is the first reader.
    */
   maxEventsPerBatch?: number;
   maxBodyBytes?: number;
@@ -56,6 +56,14 @@ export interface TelemetryEventEnvelope {
   installId: string;
   version: string;
   events: TelemetryEvent[];
+  /**
+   * Deterministic, salt-free content-hash of `{installId, events}` used as the
+   * server idempotency key so a retried batch de-dupes (202 replay) instead of
+   * double-counting. Derived in `client.ts`; the server allow-set already
+   * accepts it. The hash is salt-free so it stays stable across installs and
+   * never leaks the per-install salt.
+   */
+  batchId: string;
 }
 
 export type RegisteredPluginEventName = never;

server/src/__tests__/telemetry-client-flush.test.ts

@@ -98,3 +98,51 @@ describe("TelemetryClient periodic flush", () => {
     client.stop();
   });
 });
+
+describe("TelemetryClient retry integration", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.restoreAllMocks();
+  });
+
+  it("retries a 429 batch on the same batchId until it succeeds", async () => {
+    const fetchMock = vi
+      .fn()
+      .mockResolvedValueOnce({ ok: false, status: 429 })
+      .mockResolvedValue({ ok: true });
+    vi.stubGlobal("fetch", fetchMock);
+
+    const state: TelemetryState = {
+      installId: "test-install",
+      salt: "test-salt",
+      createdAt: "2026-01-01T00:00:00Z",
+      firstSeenVersion: "0.0.0",
+    };
+    const client = new TelemetryClient(
+      {
+        enabled: true,
+        endpoint: "http://localhost:9999/ingest",
+        backoff: { baseDelayMs: 1_000, maxDelayMs: 30_000, maxAttempts: 5, jitterRatio: 0.25 },
+      },
+      () => state,
+      "0.0.0-test",
+      () => 0.5, // seeded RNG -> zero jitter -> delay == baseDelayMs
+    );
+
+    client.track("install.started");
+    await client.flush();
+    expect(fetchMock).toHaveBeenCalledTimes(1); // attempt 1 -> 429
+
+    await vi.advanceTimersByTimeAsync(1_000);
+    expect(fetchMock).toHaveBeenCalledTimes(2); // attempt 2 -> 200
+
+    const first = JSON.parse(String((fetchMock.mock.calls[0]?.[1] as RequestInit).body));
+    const second = JSON.parse(String((fetchMock.mock.calls[1]?.[1] as RequestInit).body));
+    expect(second.batchId).toBe(first.batchId);
+    expect(second.events).toEqual(first.events);
+    client.stop();
+  });
+});