From 6419741d109f0a31115cdab84e1ad5ec508c6b1e Mon Sep 17 00:00:00 2001 From: brsbl <57682038+brsbl@users.noreply.github.com> Date: Wed, 21 Oct 2026 04:31:10 +0000 Subject: [PATCH] Validate canonical work events and preserve initiator telemetry --- apps/server/src/routes/system.ts | 2 + .../src/services/system/provider-telemetry.ts | 23 + apps/server/src/services/system/telemetry.ts | 87 ++- .../services/system/work-telemetry-schema.ts | 133 +++++ .../src/services/system/work-telemetry.ts | 532 ++++++++++++++++++ apps/server/test/services/telemetry.test.ts | 34 ++ .../services/work-telemetry-schema.test.ts | 70 +++ docs/configuration.md | 22 + packages/server-contract/src/api/system.ts | 11 + 9 files changed, 905 insertions(+), 9 deletions(-) create mode 100644 apps/server/src/services/system/provider-telemetry.ts create mode 100644 apps/server/src/services/system/work-telemetry-schema.ts create mode 100644 apps/server/src/services/system/work-telemetry.ts create mode 100644 apps/server/test/services/work-telemetry-schema.test.ts diff --git a/apps/server/src/routes/system.ts b/apps/server/src/routes/system.ts index 66eca6f9541..14a7b983d90 100644 --- a/apps/server/src/routes/system.ts +++ b/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"]); diff --git a/apps/server/src/services/system/provider-telemetry.ts b/apps/server/src/services/system/provider-telemetry.ts new file mode 100644 index 00000000000..5271a9bd843 --- /dev/null +++ b/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(); +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"; +} diff --git a/apps/server/src/services/system/telemetry.ts b/apps/server/src/services/system/telemetry.ts index 416995b7846..7db264894a4 100644 --- a/apps/server/src/services/system/telemetry.ts +++ b/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,14 +28,43 @@ 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 & { + 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 & { is_child_thread: boolean; provider: string; initiator?: TelemetryInitiator; @@ -37,13 +72,24 @@ export type TelemetryEvent = } | { name: "user_message_sent"; - properties: { + properties: Partial & { + 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( 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, diff --git a/apps/server/src/services/system/work-telemetry-schema.ts b/apps/server/src/services/system/work-telemetry-schema.ts new file mode 100644 index 00000000000..7157af36d4b --- /dev/null +++ b/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; +} diff --git a/apps/server/src/services/system/work-telemetry.ts b/apps/server/src/services/system/work-telemetry.ts new file mode 100644 index 00000000000..47573fdbe84 --- /dev/null +++ b/apps/server/src/services/system/work-telemetry.ts @@ -0,0 +1,532 @@ +import { captureFeatureUsed } from "./feature-telemetry.js"; +import { AsyncLocalStorage } from "node:async_hooks"; +import { randomBytes } from "node:crypto"; +import { and, eq, inArray, isNotNull, sql } from "drizzle-orm"; +import { + afterCommit, + events, + pendingInteractions, + getThread, + telemetryThreads, + telemetryMessages, + telemetryOrigins, + type DbQueryConnection, +} from "@bb/db"; +import type { + PendingInteraction, + Thread, + TurnRequestEventData, +} from "@bb/domain"; +import { z } from "zod"; +import { turnRequestEventDataSchema } from "@bb/domain"; +import { + initiatorTelemetryProperty, + resolveTelemetryInitiator, +} from "../threads/telemetry-initiator.js"; +import { + currentTelemetryInitiator, + type TelemetryInitiator, +} from "./telemetry.js"; +import type { TelemetryEvent, TelemetryService } from "./telemetry.js"; + +export interface AnalyticsThreadProperties { + thread_analytics_id: string; + parent_analytics_id?: string; + parent_link: "root" | "child" | "unknown"; +} + +export type MessageSource = "thread_create" | "thread_send" | "queued_message"; +export type WorkOriginProperties = { + request_link: "single" | "grouped" | "none" | "unavailable"; + request_analytics_id?: string; + request_analytics_ids?: string[]; + attempt_kind: "initial" | "retry" | "unknown"; + initiator: TelemetryInitiator; + message_source?: MessageSource; +}; + +type WorkDeps = { + db: DbQueryConnection; + telemetry: TelemetryService; + config: { featureFlags: import("@bb/domain").FeatureFlags }; +}; +let workDeps: WorkDeps | undefined; + +export function setWorkTelemetryDeps(deps: WorkDeps | undefined): void { + workDeps = deps; +} + +interface MessageOrigin { + threadId: string; + ids: string[]; + source: MessageSource; + initiator: TelemetryInitiator; + captureMessage: boolean; + fresh: boolean; +} + +const originStorage = new AsyncLocalStorage(); +const tokenListSchema = z + .array(z.string().regex(/^[0-9a-f]{32}$/)) + .min(1) + .max(100); + +function enabled(): boolean { + return workDeps?.telemetry.isEnabled() === true; +} + +function token(db: DbQueryConnection, threadId: string): string { + const existing = db + .select() + .from(telemetryThreads) + .where(eq(telemetryThreads.threadId, threadId)) + .get(); + if (existing) return existing.analyticsId; + const id = randomBytes(16).toString("hex"); + db.insert(telemetryThreads) + .values({ threadId, analyticsId: id }) + .onConflictDoNothing({ target: telemetryThreads.threadId }) + .run(); + return db + .select() + .from(telemetryThreads) + .where(eq(telemetryThreads.threadId, threadId)) + .get()!.analyticsId; +} + +export function analyticsThreadProperties( + db: DbQueryConnection, + thread: Pick, +): AnalyticsThreadProperties { + const id = token(db, thread.id); + if (thread.parentThreadId === null) + return { thread_analytics_id: id, parent_link: "root" }; + const parent = getThread(db, thread.parentThreadId); + return parent && parent.deletedAt === null + ? { + thread_analytics_id: id, + parent_analytics_id: token(db, parent.id), + parent_link: "child", + } + : { thread_analytics_id: id, parent_link: "unknown" }; +} + +export function captureWorkEvent( + db: DbQueryConnection, + event: TelemetryEvent, + occurredAt: number, +): void { + const deps = workDeps; + if (!deps?.telemetry.isEnabled()) return; + afterCommit(db, () => { + try { + deps.telemetry.capture(event, occurredAt); + } catch {} + }); +} + +function originIds( + db: DbQueryConnection, + threadId: string, + kind: "queue" | "request", + key: string, +): string[] { + const row = db + .select() + .from(telemetryOrigins) + .where( + and( + eq(telemetryOrigins.threadId, threadId), + eq(telemetryOrigins.kind, kind), + eq(telemetryOrigins.key, key), + ), + ) + .get(); + return tokenListSchema.safeParse(row?.messageIds).data ?? []; +} + +export function withMessageOrigin( + args: { + threadId: string; + source: MessageSource; + initiator: TelemetryInitiator; + captureMessage: boolean; + queuedIds?: readonly string[]; + retryRequestId?: string; + reuse?: boolean; + }, + run: () => T, +): T { + if (args.reuse && originStorage.getStore()?.threadId === args.threadId) + return run(); + const deps = workDeps; + if (!deps || !enabled()) return run(); + let ids: string[] = []; + const fresh = + args.queuedIds === undefined && args.retryRequestId === undefined; + try { + if ( + args.retryRequestId !== undefined && + (args.queuedIds?.length ?? 1) === 1 + ) + ids = originIds(deps.db, args.threadId, "request", args.retryRequestId); + else if (args.queuedIds !== undefined) { + const groups = args.queuedIds.map((id) => + originIds(deps.db, args.threadId, "queue", id), + ); + if (groups.every((group) => group.length > 0)) + ids = [...new Set(groups.flat())]; + } else ids = [randomBytes(16).toString("hex")]; + } catch {} + return originStorage.run( + { + threadId: args.threadId, + ids, + source: args.source, + initiator: args.initiator, + captureMessage: args.captureMessage, + fresh, + }, + run, + ); +} + +function bindOrigin( + db: DbQueryConnection, + threadId: string, + kind: "queue" | "request", + key: string, + fallback?: MessageOrigin, +): void { + if (!enabled()) return; + const origin = originStorage.getStore() ?? fallback; + if ( + !origin || + origin.threadId !== threadId || + !tokenListSchema.safeParse(origin.ids).success + ) + return; + const thread = getThread(db, threadId); + if (!thread) return; + const identity = analyticsThreadProperties(db, thread); + if (origin.fresh) { + const id = origin.ids[0]!; + const acceptedAt = Date.now(); + const inserted = db + .insert(telemetryMessages) + .values({ + analyticsId: id, + threadId, + acceptedAt, + initiator: origin.initiator, + messageSource: origin.source, + }) + .onConflictDoNothing() + .returning() + .get(); + if (inserted && origin.captureMessage) + captureWorkEvent( + db, + { + name: "user_message_sent", + properties: { + ...identity, + request_analytics_id: id, + is_child_thread: thread.parentThreadId !== null, + message_source: origin.source, + provider: thread.providerId, + ...initiatorTelemetryProperty( + workDeps!.config.featureFlags, + origin.initiator, + ), + }, + }, + acceptedAt, + ); + } + const stored = db + .select({ id: telemetryMessages.analyticsId }) + .from(telemetryMessages) + .where( + and( + eq(telemetryMessages.threadId, threadId), + inArray(telemetryMessages.analyticsId, origin.ids), + ), + ) + .all(); + if (stored.length !== origin.ids.length) return; + db.insert(telemetryOrigins) + .values({ threadId, kind, key, messageIds: origin.ids }) + .onConflictDoNothing() + .run(); +} + +export function recordMessageRequest( + db: DbQueryConnection, + threadId: string, + request: TurnRequestEventData, +): void { + try { + if (!enabled()) return; + const fallback: MessageOrigin = { + threadId, + ids: request.retryOfRequestId + ? originIds(db, threadId, "request", request.retryOfRequestId) + : [], + source: request.source === "spawn" ? "thread_create" : "thread_send", + initiator: resolveTelemetryInitiator({ + origin: null, + originPluginId: null, + initiator: request.initiator, + }), + captureMessage: false, + fresh: false, + }; + if (request.input.length > 0) + bindOrigin(db, threadId, "request", request.requestId, fallback); + if ( + request.target.kind === "thread-start" || + request.target.kind === "new-turn" || + (request.target.kind === "auto" && request.target.expectedTurnId === null) + ) { + token(db, threadId); + db.update(telemetryThreads) + .set({ pendingRequestId: request.requestId }) + .where(eq(telemetryThreads.threadId, threadId)) + .run(); + } + } catch {} +} + +export function recordQueuedMessageOrigin( + db: DbQueryConnection, + threadId: string, + queueId: string, +): void { + try { + bindOrigin(db, threadId, "queue", queueId); + } catch {} +} + +export function workOriginProperties( + db: DbQueryConnection, + threadId: string, + requestId: string | null, +): WorkOriginProperties { + if (requestId === null) + return { + request_link: "unavailable", + attempt_kind: "unknown", + initiator: currentTelemetryInitiator(), + }; + const row = db + .select({ data: events.data }) + .from(events) + .where( + and( + eq(events.threadId, threadId), + eq(events.type, "client/turn/requested"), + sql`json_extract(${events.data}, '$.requestId') = ${requestId}`, + ), + ) + .get(); + const request = row + ? turnRequestEventDataSchema.safeParse(JSON.parse(row.data)).data + : undefined; + const attempt = + request === undefined + ? "unknown" + : request.retryOfRequestId === undefined + ? "initial" + : "retry"; + if (request?.input.length === 0) + return { + request_link: "none", + attempt_kind: attempt, + initiator: currentTelemetryInitiator(), + }; + const ids = originIds(db, threadId, "request", requestId); + const messages = + ids.length === 0 + ? [] + : db + .select() + .from(telemetryMessages) + .where( + and( + eq(telemetryMessages.threadId, threadId), + inArray(telemetryMessages.analyticsId, ids), + ), + ) + .all(); + const first = messages[0]; + const consistent = first !== undefined && messages.length === ids.length; + const initiator = + consistent && + messages.every((message) => message.initiator === first.initiator) + ? first.initiator + : resolveTelemetryInitiator({ + origin: null, + originPluginId: null, + initiator: request?.initiator ?? "system", + }); + const source = + consistent && + messages.every((message) => message.messageSource === first.messageSource) + ? first.messageSource + : undefined; + return { + request_link: !consistent + ? "unavailable" + : ids.length === 1 + ? "single" + : "grouped", + ...(consistent + ? ids.length === 1 + ? { request_analytics_id: ids[0]! } + : { request_analytics_ids: ids } + : {}), + attempt_kind: attempt, + initiator, + ...(source === undefined ? {} : { message_source: source }), + }; +} + +export function captureThreadCreated( + thread: Thread, + initiator: TelemetryInitiator, +): void { + const deps = workDeps; + if (!deps || !enabled()) return; + try { + if (thread.parentThreadId !== null) + afterCommit(deps.db, () => + captureFeatureUsed(deps.telemetry, { + kind: "child_thread", + feature_scope: "core", + interaction: "invoked", + initiator, + }), + ); + captureWorkEvent( + deps.db, + { + name: "thread_created", + properties: { + ...analyticsThreadProperties(deps.db, thread), + is_child_thread: thread.parentThreadId !== null, + provider: thread.providerId, + ...initiatorTelemetryProperty(deps.config.featureFlags, initiator), + }, + }, + thread.createdAt, + ); + } catch {} +} + +type ThreadTransition = + | "thread_stopped" + | "thread_archived" + | "thread_unarchived" + | "thread_deleted" + | "parent_changed"; +export function captureThreadTransition( + thread: Thread, + transition: ThreadTransition, +): void { + const deps = workDeps; + if (!deps || !enabled()) return; + try { + const properties = { + ...analyticsThreadProperties(deps.db, thread), + transition, + initiator: currentTelemetryInitiator(), + }; + captureWorkEvent( + deps.db, + { name: "thread_lifecycle_changed", properties }, + thread.updatedAt, + ); + if (transition === "thread_deleted") { + deps.db + .delete(telemetryThreads) + .where(eq(telemetryThreads.threadId, thread.id)) + .run(); + deps.db + .delete(telemetryOrigins) + .where(eq(telemetryOrigins.threadId, thread.id)) + .run(); + deps.db + .delete(telemetryMessages) + .where(eq(telemetryMessages.threadId, thread.id)) + .run(); + } + } catch {} +} + +export function captureInteractionWait( + db: DbQueryConnection, + interaction: PendingInteraction, + entering: boolean, +): void { + if (!enabled() || interaction.turnId === null) return; + if ( + !entering && + interaction.status !== "resolved" && + interaction.status !== "interrupted" + ) + return; + try { + const active = db + .select({ id: pendingInteractions.id }) + .from(pendingInteractions) + .where( + and( + eq(pendingInteractions.threadId, interaction.threadId), + isNotNull(pendingInteractions.turnId), + inArray(pendingInteractions.status, ["pending", "resolving"]), + ), + ) + .all(); + const waiting = active.length > 0; + token(db, interaction.threadId); + const previous = db + .select() + .from(telemetryThreads) + .where(eq(telemetryThreads.threadId, interaction.threadId)) + .get(); + if (previous?.waiting === waiting) return; + db.update(telemetryThreads) + .set({ waiting }) + .where(eq(telemetryThreads.threadId, interaction.threadId)) + .run(); + const thread = getThread(db, interaction.threadId); + if (!thread) return; + captureWorkEvent( + db, + { + name: "thread_lifecycle_changed", + properties: { + ...analyticsThreadProperties(db, thread), + initiator: waiting ? "agent" : currentTelemetryInitiator(), + transition: waiting ? "waiting_on_user" : "waiting_cleared", + wait_kind: + interaction.payload.kind === "user_question" + ? "question" + : interaction.payload.kind === "approval" + ? "approval" + : "other", + }, + }, + Date.now(), + ); + } catch {} +} + +export function clearActiveWorkTelemetry(db: DbQueryConnection): void { + try { + db.update(telemetryThreads) + .set({ activeRequestId: null, pendingRequestId: null, waiting: false }) + .run(); + } catch {} +} diff --git a/apps/server/test/services/telemetry.test.ts b/apps/server/test/services/telemetry.test.ts index 0e24ff2d755..4caf167d2bb 100644 --- a/apps/server/test/services/telemetry.test.ts +++ b/apps/server/test/services/telemetry.test.ts @@ -115,6 +115,40 @@ describe("telemetry service", () => { }); }); + it("blocks custom provider identifiers at the turn outcome sender", async () => { + const telemetry = await createTelemetryService({ + apiKey: "phc_test", + appInstall: null, + appSurface: "web", + appVersion: "1.2.3", + dataDir, + telemetryEnabled: true, + enabled: true, + workTelemetry: true, + logger: createTestLogger(), + }); + for (const provider of ["acp-private-team-agent", "codex", "other"]) { + telemetry.capture({ + name: "turn_finished", + properties: { + thread_analytics_id: "a".repeat(32), + parent_link: "root", + request_link: "unavailable", + attempt_kind: "unknown", + initiator: "system", + outcome: "succeeded", + error_category: "none", + install_turn_number: 1, + provider, + }, + }); + } + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(JSON.stringify(fetchMock.mock.calls)).not.toContain( + "acp-private-team-agent", + ); + }); + it("tags events with how bb was installed without sending any commits", async () => { const commit = "a".repeat(40); const installs = [ diff --git a/apps/server/test/services/work-telemetry-schema.test.ts b/apps/server/test/services/work-telemetry-schema.test.ts new file mode 100644 index 00000000000..7f6ec48b781 --- /dev/null +++ b/apps/server/test/services/work-telemetry-schema.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; +import { validWorkTelemetry } from "../../src/services/system/work-telemetry-schema.js"; + +describe("work telemetry vocabulary", () => { + it.each(["user", "agent", "automation", "plugin", "system"])( + "accepts the canonical %s initiator on existing events", + (initiator) => { + expect( + validWorkTelemetry("thread_created", { + is_child_thread: false, + provider: "codex", + initiator, + }), + ).toBe(true); + expect( + validWorkTelemetry("user_message_sent", { + is_child_thread: false, + provider: "codex", + message_source: "thread_send", + initiator, + }), + ).toBe(true); + }, + ); + + it("rejects private providers and retired initiators", () => { + expect( + validWorkTelemetry("thread_created", { + is_child_thread: false, + provider: "my-private-agent", + initiator: "user", + }), + ).toBe(false); + for (const initiator of ["person", "schedule", "unknown"]) { + expect( + validWorkTelemetry("user_message_sent", { + is_child_thread: false, + provider: "codex", + message_source: "thread_create", + initiator, + }), + ).toBe(false); + } + }); + + it("accepts canonical finishes and rejects retired fields and values", () => { + const properties = { + thread_analytics_id: "a".repeat(32), + parent_link: "root", + request_link: "none", + attempt_kind: "initial", + initiator: "system", + outcome: "succeeded", + error_category: "none", + provider: "codex", + install_turn_number: 1, + }; + expect(validWorkTelemetry("turn_finished", properties)).toBe(true); + for (const extra of [ + { outcome: "completed" }, + { first_turn: true }, + { install_turn_number: 0 }, + { trigger: "person" }, + ]) { + expect( + validWorkTelemetry("turn_finished", { ...properties, ...extra }), + ).toBe(false); + } + }); +}); diff --git a/docs/configuration.md b/docs/configuration.md index 042e68f1372..313691fe639 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -308,6 +308,28 @@ writes and layout restoration are excluded. The telemetry endpoint rejects unknown fields, feature codes and caller-supplied plugin provenance. Legacy client event variants remain accepted for compatibility. +Work events add random analytics-only thread IDs and the parent's analytics ID +where applicable. Accepted messages have independently random request analytics +IDs; a grouped dispatch carries the complete bounded list. Local lookup records +keep these separate from real thread, queue, message and provider IDs. They are +never PostHog person IDs and are not accepted as bb API routing keys. + +`thread_lifecycle_changed` reports applied turn starts, explicit user-blocking waits and +clears, settled stops, archives, unarchives, deletion and parent changes. +`turn_finished` reports `completed`, `failed` or `stopped`, the bundled provider +(or `other`), a coarse error category and original message linkage. Explicit +retries retain the original message identities and carry `attempt_kind=retry`; +steering cannot replace a turn's initiating messages. Missing origin records +remain `unavailable`. Queue acceptance and turn events retain their original +occurrence timestamps. Capture runs only after a successful transaction. + +These tokens permit per-thread timelines, parent-child graphs, message-to-work +and retry joins, and observed queue delays within an installation. They do not +encode content or replace the anonymous install ID. The local database contains +the mapping; PostHog does not receive it. Opt-out prevents future capture and +backfill, but does not erase prior events. Forked/recreated threads get fresh +identities; copying a whole database can copy analytics identities as well. + ## AI services Thread titles (and the branch names built from them), commit messages, and diff --git a/packages/server-contract/src/api/system.ts b/packages/server-contract/src/api/system.ts index a4bc88cf706..8eac735b8fd 100644 --- a/packages/server-contract/src/api/system.ts +++ b/packages/server-contract/src/api/system.ts @@ -314,6 +314,17 @@ export const recordTelemetryEventRequestSchema = z.discriminatedUnion("name", [ .strict(), }) .strict(), + z + .object({ + name: z.literal("device_paired"), + properties: z + .object({ + kind: z.enum(["mobile_ios", "mobile_android", "machine"]), + is_first_of_kind: z.boolean(), + }) + .strict(), + }) + .strict(), z .object({ name: z.literal("split_view_opened"),