fix(plugins): stop the TypeScript telemetry losing events and misattributing accounts

The Python plugin telemetry was hardened across #7322 to #7326. The TypeScript
side has the same defect classes and was not touched, because the two share no
code: agent-plugin-core/python generates into six bundles, agent-plugin-core/
typescript is a separate core each plugin wraps. Fixing one surfaced nothing
about the other, which is how this survived.

Three fixes, all confirmed by running the code rather than reading it.

A failed delivery deleted the batch. The core detached the queue before the
await and swallowed the error, so one blip destroyed the events with nothing
recording that it happened. Probed: two events in, delivery throws, queue goes to
zero, no retry ever. The batch is now put back, bounded by maxQueueSize and
biased to the newest so a long outage costs the oldest events rather than
unbounded memory, and repeated failures back off to a ceiling instead of retrying
every flush against a host that is blocking us. Every event now carries a uuid
stamped at capture, which is what makes the retry safe: PostHog collapses
anything it already accepted.

Deliberately no disk spool, and that is written into the code so it reads as a
decision. Python spools because its hooks are per-tool-call processes that exit
immediately. These plugins live inside a host for a whole session, so
re-queueing covers the same transient failures without the claim and lease
machinery that took three review rounds to get right on the Python side. What it
leaves uncovered is narrow: a session that both starts and ends offline.

openclaw used a cached email forever. The refresh was guarded by !hasEmail, so
after an API key change every event kept reporting under the previous account.
The email is now bound to a fingerprint of the key it was resolved for and only
used while those agree; a mismatch forgets the account and re-resolves. The
resolution latch is per key rather than once per process, so a key changed
mid-session is actually looked up. A row with an email and no fingerprint, which
is what an upgrade from the current version looks like, is verified rather than
adopted, matching the decision reached on #7325.

opencode hashed the project id unsalted, which is reversible for anyone who can
enumerate project ids. Salted with the API key rather than a stored per-install
value: it is already in play, it is high entropy, and it needs no new file and so
no write race to get wrong. Per account rather than per machine, which also keeps
joins working across machines, and it resets on key rotation consistently with
distinctId, which already did.

Also wired opencode's tests into CI. They existed and nothing ran them, so the
regression test asked for on #7322 would not have gated anything.

Verified: agent-plugin-core/ts 30, openclaw 431, opencode 35, pi-agent 89,
deepseek 47. The new tests were each checked against the unfixed code first; the
core ones fail 3 of 3 and the openclaw one fails without the fingerprint gate.

Claude-Session: https://claude.ai/code/session_01C7tEmH86HAr7GoAAKCEHZb

integrations/agent-plugin-core/typescript/src/telemetry.ts

@@ -1,3 +1,5 @@
+import { randomUUID } from "node:crypto";
+
 import { redactSecrets } from "./lifecycle.ts";
 
 const POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX";
@@ -74,9 +76,20 @@ export function errorKind(error: unknown): string {
   return error instanceof Error ? error.constructor.name : "other";
 }
 
+// Delivery is retried in memory, not spooled to disk, and that is a decision
+// rather than an omission. The Python core spools because its hooks are separate
+// processes that fire per tool call and exit immediately, so nothing survives
+// without a file. These plugins are loaded into a host that lives for a whole
+// session, so re-queueing covers the same transient failures without the claim
+// and lease machinery a correct cross-process spool needs. What that leaves
+// uncovered is narrow: a session that both starts and ends with no connectivity.
+const RETRY_BACKOFF_CEILING_MS = 60_000;
+
 export function createTelemetry(config: TelemetryConfig) {
   let queue: Record<string, unknown>[] = [];
   let timer: ReturnType<typeof setInterval> | undefined;
+  let consecutiveFailures = 0;
+  let retryNotBefore = 0;
   const flushThreshold = config.flushThreshold ?? 10;
   const maxQueueSize = config.maxQueueSize ?? 100;
 
@@ -91,12 +104,24 @@ export function createTelemetry(config: TelemetryConfig) {
 
   async function flush(): Promise<void> {
     if (!queue.length) return;
+    if (Date.now() < retryNotBefore) return;
     const batch = queue;
     queue = [];
     try {
       await deliver(batch);
+      consecutiveFailures = 0;
+      retryNotBefore = 0;
     } catch {
-      // Telemetry must never affect plugin behavior.
+      // Put it back. Detaching the batch and swallowing the error deleted the
+      // events outright, so any blip silently dropped telemetry with nothing
+      // recording that it had happened. Every event carries a uuid, so a retry
+      // that duplicates one PostHog already accepted is collapsed there.
+      //
+      // Bounded by maxQueueSize and biased to the newest, matching capture():
+      // a long outage costs the oldest events rather than unbounded memory.
+      queue = [...batch, ...queue].slice(-maxQueueSize);
+      consecutiveFailures += 1;
+      retryNotBefore = Date.now() + Math.min(2 ** consecutiveFailures * 1_000, RETRY_BACKOFF_CEILING_MS);
     }
   }
 
@@ -112,6 +137,10 @@ export function createTelemetry(config: TelemetryConfig) {
       return {
         event: config.eventName?.(event) ?? event,
         distinct_id: distinctId,
+        // Stamped once, at capture. This is what makes retrying safe: a batch
+        // re-sent after a failure carries the same ids, so PostHog collapses
+        // anything it already accepted instead of counting it twice.
+        uuid: randomUUID(),
         properties: {
           ...safeProperties(properties),
           ...safeProperties(config.commonProperties ?? {}),
@@ -149,6 +178,8 @@ export function createTelemetry(config: TelemetryConfig) {
 
   function resetForTesting(): void {
     queue = [];
+    consecutiveFailures = 0;
+    retryNotBefore = 0;
     if (timer) clearInterval(timer);
     timer = undefined;
     process.off("beforeExit", beforeExit);

integrations/agent-plugin-core/typescript/tests/telemetry.test.ts

@@ -122,3 +122,95 @@ test("error classification does not expose messages", () => {
   assert.equal(errorKind(new Error("request timeout")), "timeout");
   assert.equal(errorKind(new Error("fetch failed")), "network");
 });
+
+test("a failed delivery keeps the batch instead of deleting it", async () => {
+  // The defect: the queue was detached before the await and the error swallowed,
+  // so one blip destroyed the events with nothing recording that it happened.
+  const attempts: Record<string, unknown>[][] = [];
+  let failNext = true;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async (batch) => {
+      attempts.push(batch);
+      if (failNext) throw new Error("network down");
+    },
+  });
+
+  telemetry.capture("one");
+  telemetry.capture("two");
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 1);
+  assert.equal(telemetry.queueForTesting().length, 2, "events were dropped on failure");
+
+  failNext = false;
+  // Backoff is in force, so wait it out the way wall time would.
+  await new Promise((resolve) => setTimeout(resolve, 2_100));
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 2, "never retried");
+  assert.equal(telemetry.queueForTesting().length, 0);
+  telemetry.resetForTesting();
+});
+
+test("a retried event carries the same uuid so PostHog can collapse it", async () => {
+  const attempts: Record<string, unknown>[][] = [];
+  let failNext = true;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async (batch) => {
+      attempts.push(batch);
+      if (failNext) throw new Error("network down");
+    },
+  });
+
+  telemetry.capture("once");
+  await telemetry.flush();
+  failNext = false;
+  await new Promise((resolve) => setTimeout(resolve, 2_100));
+  await telemetry.flush();
+
+  assert.equal(attempts.length, 2);
+  const first = attempts[0][0].uuid;
+  assert.ok(first, "events carry no uuid, so a retry would double count");
+  assert.equal(attempts[1][0].uuid, first, "retry minted a new uuid");
+  telemetry.resetForTesting();
+});
+
+test("repeated failures back off instead of retrying every flush", async () => {
+  let calls = 0;
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000,
+    delivery: async () => { calls += 1; throw new Error("blocked"); },
+  });
+
+  telemetry.capture("one");
+  await telemetry.flush();
+  await telemetry.flush();
+  await telemetry.flush();
+
+  assert.equal(calls, 1, "a blocked host was hammered on every flush");
+  assert.equal(telemetry.queueForTesting().length, 1, "the event was lost while backing off");
+  telemetry.resetForTesting();
+});
+
+test("a long outage costs the oldest events, not unbounded memory", async () => {
+  const telemetry = createTelemetry({
+    host: "h", source: "S", version: "1", distinctId: "d",
+    flushThreshold: 1000, maxQueueSize: 3,
+    delivery: async () => { throw new Error("down"); },
+  });
+
+  telemetry.capture("a");
+  telemetry.capture("b");
+  await telemetry.flush();
+  telemetry.capture("c");
+  telemetry.capture("d");
+  telemetry.capture("e");
+
+  assert.ok(telemetry.queueForTesting().length <= 3, "queue grew past maxQueueSize");
+  telemetry.resetForTesting();
+});