Validate and prepare anonymous work telemetry

apps/server/src/routes/system.ts

@@ -1,3 +1,4 @@
+import { clearActiveWorkTelemetry } from "../services/system/work-telemetry.js";
 import { createMobileAppReleaseService } from "../services/install/mobile-app-releases.js";
 import {
   setMachineEnvironmentVariable,
@@ -339,6 +340,7 @@ export function registerSystemRoutes(
     setAppSettings(deps.db, updatedSettings);
     if (current.telemetryEnabled && !updatedSettings.telemetryEnabled) {
       deps.telemetry.capture({ name: "telemetry_disabled" });
+      clearActiveWorkTelemetry(deps.db);
     }
     deps.telemetry.setEnabled(updatedSettings.telemetryEnabled);
     deps.hub.notifySystem(["config-changed"]);

apps/server/src/services/system/provider-telemetry.ts

@@ -0,0 +1,23 @@
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import { pluginPackageJsonSchema } from "@bb/domain";
+import { z } from "zod";
+import { listBundledPluginRegistrations } from "../plugins/builtin-registry.js";
+
+const shippedProviderIds = new Set<string>();
+for (const { rootDir } of listBundledPluginRegistrations()) {
+  try {
+    const manifest = pluginPackageJsonSchema.parse(
+      JSON.parse(readFileSync(join(rootDir, "package.json"), "utf8")),
+    );
+    for (const provider of manifest.bb.experimental_providers ?? []) {
+      shippedProviderIds.add(provider.id);
+    }
+  } catch {}
+}
+
+export const turnProviderSchema = z.enum(["other", ...shippedProviderIds]);
+
+export function reportedProviderId(providerId: string): string {
+  return turnProviderSchema.safeParse(providerId).data ?? "other";
+}

apps/server/src/services/system/telemetry.ts

@@ -1,3 +1,9 @@
+import { reportedProviderId } from "./provider-telemetry.js";
+import { validWorkTelemetry } from "./work-telemetry-schema.js";
+import type {
+  AnalyticsThreadProperties,
+  WorkOriginProperties,
+} from "./work-telemetry.js";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { release } from "node:os";
 import type { AppInstall } from "@bb/config/app-install";
@@ -22,28 +28,68 @@ export type TelemetryInitiator =
   | "automation"
   | "plugin"
   | "system";
+export type TurnFinishedOutcome = "succeeded" | "failed" | "stopped";
+
+export type TurnErrorCategory =
+  | "none"
+  | "auth"
+  | "billing"
+  | "rate_limit"
+  | "provider_unavailable"
+  | "limit_reached"
+  | "process_exited"
+  | "start_failed"
+  | "host_lost"
+  | "other";
 
 export type TelemetryEvent =
   | { name: "feature_used"; properties: FeatureUsedProperties }
+  | {
+      name: "thread_lifecycle_changed";
+      properties: AnalyticsThreadProperties &
+        Partial<WorkOriginProperties> & {
+          transition:
+            | "turn_started"
+            | "waiting_on_user"
+            | "waiting_cleared"
+            | "thread_stopped"
+            | "thread_archived"
+            | "thread_unarchived"
+            | "thread_deleted"
+            | "parent_changed";
+          wait_kind?: "question" | "approval" | "other" | "unknown";
+        };
+    }
   | { name: "app_started" }
   | { name: "telemetry_disabled" }
   | {
       name: "thread_created";
-      properties: {
+      properties: Partial<AnalyticsThreadProperties> & {
         is_child_thread: boolean;
         provider: string;
         initiator?: TelemetryInitiator;
       };
     }
   | {
       name: "user_message_sent";
-      properties: {
+      properties: Partial<AnalyticsThreadProperties> & {
+        request_analytics_id?: string;
         is_child_thread: boolean;
         message_source: "queued_message" | "thread_create" | "thread_send";
         provider: string;
         initiator?: TelemetryInitiator;
       };
     }
+  | {
+      name: "turn_finished";
+      properties: AnalyticsThreadProperties &
+        WorkOriginProperties & {
+          outcome: TurnFinishedOutcome;
+          provider: string;
+          error_category: TurnErrorCategory;
+          install_turn_number: number;
+        };
+    }
   | {
       name: "plugin_installed";
       properties: {
@@ -56,7 +102,8 @@ export type TelemetryEvent =
   | RecordTelemetryEventRequest;
 
 export interface TelemetryService {
-  capture(event: TelemetryEvent): void;
+  capture(event: TelemetryEvent, occurrenceTime?: number): void;
+  isEnabled(): boolean;
   setEnabled(enabled: boolean): void;
 }
 
@@ -74,6 +121,7 @@ interface CreateTelemetryServiceArgs {
 
 const noopTelemetryService: TelemetryService = {
   capture: () => {},
+  isEnabled: () => false,
   setEnabled: () => {},
 };
 
@@ -117,6 +165,12 @@ function appInstallProperties(install: AppInstall | null) {
   return { install_kind: "source", source_origin: install.origin };
 }
 
+export function currentTelemetryInitiator(): TelemetryInitiator {
+  return telemetryAppSurfaceStorage.getStore() === undefined
+    ? "system"
+    : "user";
+}
+
 export function runWithTelemetryAppSurface<T>(
   appSurface: RequestAppSurface,
   callback: () => T,
@@ -157,18 +211,33 @@ export async function createTelemetryService(
     setEnabled(enabled: boolean): void {
       telemetryEnabled = enabled;
     },
-    capture(event: TelemetryEvent): void {
+    isEnabled: () => telemetryEnabled && args.workTelemetry === true,
+    capture(event: TelemetryEvent, occurrenceTime = Date.now()): void {
       if (!telemetryEnabled) return;
       if (
-        ["feature_used", "automation_created", "split_view_opened"].includes(
-          event.name,
-        ) &&
+        [
+          "feature_used",
+          "automation_created",
+          "split_view_opened",
+          "thread_lifecycle_changed",
+          "turn_finished",
+          "device_paired",
+        ].includes(event.name) &&
         !args.workTelemetry
       )
         return;
       const appSurface =
         telemetryAppSurfaceStorage.getStore() ?? args.appSurface;
-      const eventProperties = "properties" in event ? event.properties : {};
+      const rawProperties = "properties" in event ? event.properties : {};
+      const eventProperties =
+        "provider" in rawProperties &&
+        typeof rawProperties.provider === "string"
+          ? {
+              ...rawProperties,
+              provider: reportedProviderId(rawProperties.provider),
+            }
+          : rawProperties;
+      if (!validWorkTelemetry(event.name, eventProperties)) return;
       const body = JSON.stringify({
         api_key: args.apiKey,
         distinct_id: distinctId,
@@ -178,7 +247,7 @@ export async function createTelemetryService(
           ...eventProperties,
           app_surface: appSurface,
         },
-        timestamp: new Date().toISOString(),
+        timestamp: new Date(occurrenceTime).toISOString(),
       });
       fetch(POSTHOG_INGESTION_URL, {
         body,

apps/server/src/services/system/work-telemetry-schema.ts

@@ -0,0 +1,133 @@
+import { z } from "zod";
+import { turnProviderSchema } from "./provider-telemetry.js";
+import {
+  featureUsedPropertiesSchema,
+  telemetryInitiatorSchema,
+} from "@bb/server-contract";
+
+const analyticsId = z.string().regex(/^[0-9a-f]{32}$/);
+const identity = {
+  thread_analytics_id: analyticsId,
+  parent_analytics_id: analyticsId.optional(),
+  parent_link: z.enum(["root", "child", "unknown"]),
+};
+const origin = {
+  request_link: z.enum(["single", "grouped", "none", "unavailable"]),
+  request_analytics_id: analyticsId.optional(),
+  request_analytics_ids: z.array(analyticsId).min(2).max(100).optional(),
+  attempt_kind: z.enum(["initial", "retry", "unknown"]),
+  initiator: telemetryInitiatorSchema,
+  message_source: z
+    .enum(["thread_create", "thread_send", "queued_message"])
+    .optional(),
+};
+
+const lifecycleSchema = z
+  .object({
+    ...identity,
+    transition: z.enum([
+      "turn_started",
+      "waiting_on_user",
+      "waiting_cleared",
+      "thread_stopped",
+      "thread_archived",
+      "thread_unarchived",
+      "thread_deleted",
+      "parent_changed",
+    ]),
+    request_link: origin.request_link.optional(),
+    request_analytics_id: origin.request_analytics_id,
+    request_analytics_ids: origin.request_analytics_ids,
+    attempt_kind: origin.attempt_kind.optional(),
+    initiator: origin.initiator,
+    message_source: origin.message_source,
+    wait_kind: z.enum(["question", "approval", "other", "unknown"]).optional(),
+  })
+  .strict();
+
+const finishedSchema = z
+  .object({
+    ...identity,
+    ...origin,
+    outcome: z.enum(["succeeded", "failed", "stopped"]),
+    error_category: z.enum([
+      "none",
+      "auth",
+      "billing",
+      "rate_limit",
+      "provider_unavailable",
+      "limit_reached",
+      "process_exited",
+      "start_failed",
+      "host_lost",
+      "other",
+    ]),
+    provider: turnProviderSchema,
+    install_turn_number: z.number().int().positive(),
+  })
+  .strict();
+
+const createdSchema = z
+  .object({
+    thread_analytics_id: analyticsId.optional(),
+    parent_analytics_id: analyticsId.optional(),
+    parent_link: identity.parent_link.optional(),
+    is_child_thread: z.boolean(),
+    provider: turnProviderSchema,
+    initiator: telemetryInitiatorSchema.optional(),
+  })
+  .strict();
+const messageSchema = createdSchema.extend({
+  request_analytics_id: analyticsId.optional(),
+  message_source: origin.message_source.unwrap(),
+});
+
+export function validWorkTelemetry(name: string, properties: unknown): boolean {
+  if (name === "feature_used")
+    return featureUsedPropertiesSchema.safeParse(properties).success;
+  if (name === "thread_created")
+    return createdSchema.safeParse(properties).success;
+  if (name === "user_message_sent")
+    return messageSchema.safeParse(properties).success;
+  if (name !== "thread_lifecycle_changed" && name !== "turn_finished")
+    return true;
+  const result =
+    name === "thread_lifecycle_changed"
+      ? lifecycleSchema.safeParse(properties)
+      : finishedSchema.safeParse(properties);
+  if (!result.success) return false;
+  const value = result.data;
+  if (value.parent_link === "root" && value.parent_analytics_id !== undefined)
+    return false;
+  if (value.parent_link === "child" && value.parent_analytics_id === undefined)
+    return false;
+  if ("transition" in value) {
+    if (value.transition === "turn_started") {
+      if (
+        value.request_link === undefined ||
+        value.attempt_kind === undefined ||
+        value.initiator === undefined ||
+        value.wait_kind !== undefined
+      )
+        return false;
+    } else if (
+      value.request_link !== undefined ||
+      value.request_analytics_id !== undefined ||
+      value.request_analytics_ids !== undefined ||
+      value.attempt_kind !== undefined ||
+      value.message_source !== undefined
+    )
+      return false;
+  }
+  const hasSingle = value.request_analytics_id !== undefined;
+  const hasGroup = value.request_analytics_ids !== undefined;
+  if (hasSingle && hasGroup) return false;
+  if (value.request_link === "single") return hasSingle && !hasGroup;
+  if (value.request_link === "grouped")
+    return (
+      hasGroup &&
+      new Set(value.request_analytics_ids).size ===
+        value.request_analytics_ids?.length
+    );
+  return !hasSingle && !hasGroup;
+}