feat: preserve workflow origin views in RUM

src/platform/telemetry/TelemetryRegistry.ts

@@ -10,6 +10,7 @@ import type {
   ShareFlowMetadata,
   ShareLinkOpenedMetadata,
   ExecutionErrorMetadata,
+  ExecutionStartMetadata,
   ExecutionSuccessMetadata,
   HelpCenterClosedMetadata,
   HelpCenterOpenedMetadata,
@@ -268,6 +269,10 @@ export class TelemetryRegistry implements TelemetryDispatcher {
     this.dispatch((provider) => provider.trackWorkflowExecution?.())
   }
 
+  trackExecutionStarted(metadata: ExecutionStartMetadata): void {
+    this.dispatch((provider) => provider.trackExecutionStarted?.(metadata))
+  }
+
   trackExecutionError(metadata: ExecutionErrorMetadata): void {
     this.dispatch((provider) => provider.trackExecutionError?.(metadata))
   }

src/platform/telemetry/initTelemetry.ts

@@ -28,7 +28,8 @@ export async function initTelemetry(): Promise<void> {
       { PostHogTelemetryProvider },
       { ClickHouseTelemetryProvider },
       { SyftTelemetryProvider },
-      { CustomerIoTelemetryProvider }
+      { CustomerIoTelemetryProvider },
+      { DatadogRumTelemetryProvider }
     ] = await Promise.all([
       import('./TelemetryRegistry'),
       import('./providers/cloud/MixpanelTelemetryProvider'),
@@ -37,7 +38,8 @@ export async function initTelemetry(): Promise<void> {
       import('./providers/cloud/PostHogTelemetryProvider'),
       import('./providers/cloud/ClickHouseTelemetryProvider'),
       import('./providers/cloud/SyftTelemetryProvider'),
-      import('./providers/cloud/CustomerIoTelemetryProvider')
+      import('./providers/cloud/CustomerIoTelemetryProvider'),
+      import('./providers/cloud/DatadogRumTelemetryProvider')
     ])
 
     const registry = new TelemetryRegistry()
@@ -48,6 +50,7 @@ export async function initTelemetry(): Promise<void> {
     registry.registerProvider(new ClickHouseTelemetryProvider())
     registry.registerProvider(new SyftTelemetryProvider())
     registry.registerProvider(new CustomerIoTelemetryProvider())
+    registry.registerProvider(new DatadogRumTelemetryProvider())
 
     setTelemetryRegistry(registry)
   })()

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.test.ts

@@ -0,0 +1,82 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import { DatadogRumTelemetryProvider } from './DatadogRumTelemetryProvider'
+
+const addAction = vi.fn()
+const getInternalContext = vi.fn()
+
+function installDatadogRum(): void {
+  Object.defineProperty(window, 'DD_RUM', {
+    configurable: true,
+    value: { addAction, getInternalContext }
+  })
+}
+
+afterEach(() => {
+  addAction.mockReset()
+  getInternalContext.mockReset()
+  Reflect.deleteProperty(window, 'DD_RUM')
+})
+
+describe('DatadogRumTelemetryProvider', () => {
+  it('tracks workflow execution starts with the originating view', () => {
+    installDatadogRum()
+    getInternalContext.mockReturnValue({ view: { id: 'view-a' } })
+
+    new DatadogRumTelemetryProvider().trackExecutionStarted({
+      jobId: 'job-a',
+      startTime: 42
+    })
+
+    expect(getInternalContext).toHaveBeenCalledWith(42)
+    expect(addAction).toHaveBeenCalledWith('workflow_execution_started', {
+      product: 'cloud_generation',
+      origin_view_id: 'view-a'
+    })
+  })
+
+  it('preserves each origin view across workflow switches', () => {
+    installDatadogRum()
+    getInternalContext
+      .mockReturnValueOnce({ view: { id: 'view-a' } })
+      .mockReturnValueOnce({ view: { id: 'view-b' } })
+    const provider = new DatadogRumTelemetryProvider()
+
+    provider.trackExecutionStarted({ jobId: 'job-a', startTime: 1 })
+    provider.trackExecutionStarted({ jobId: 'job-b', startTime: 2 })
+    provider.trackExecutionSuccess({ jobId: 'job-a' })
+    provider.trackExecutionError({ jobId: 'job-b' })
+
+    expect(addAction).toHaveBeenNthCalledWith(
+      3,
+      'workflow_execution_completed',
+      {
+        outcome: 'success',
+        origin_view_id: 'view-a',
+        product: 'cloud_generation'
+      }
+    )
+    expect(addAction).toHaveBeenNthCalledWith(
+      4,
+      'workflow_execution_completed',
+      {
+        outcome: 'failure',
+        origin_view_id: 'view-b',
+        product: 'cloud_generation'
+      }
+    )
+  })
+
+  it('does nothing when Datadog RUM is unavailable', () => {
+    const provider = new DatadogRumTelemetryProvider()
+
+    expect(() =>
+      provider.trackExecutionStarted({ jobId: 'job-a', startTime: 1 })
+    ).not.toThrow()
+    expect(() =>
+      provider.trackExecutionSuccess({ jobId: 'job-a' })
+    ).not.toThrow()
+    expect(() => provider.trackExecutionError({ jobId: 'job-a' })).not.toThrow()
+    expect(addAction).not.toHaveBeenCalled()
+  })
+})

src/platform/telemetry/providers/cloud/DatadogRumTelemetryProvider.ts

@@ -0,0 +1,65 @@
+import type {
+  ExecutionErrorMetadata,
+  ExecutionStartMetadata,
+  ExecutionSuccessMetadata,
+  TelemetryProvider
+} from '../../types'
+
+interface DatadogRumInternalContext {
+  view?: { id?: string }
+}
+
+interface DatadogRumClient {
+  addAction(name: string, context?: Record<string, unknown>): void
+  getInternalContext(startTime?: number): DatadogRumInternalContext | undefined
+}
+
+interface WindowWithDatadogRum extends Window {
+  DD_RUM?: DatadogRumClient
+}
+
+const WORKFLOW_CONTEXT = {
+  product: 'cloud_generation'
+} as const
+
+function getDatadogRum(): DatadogRumClient | undefined {
+  return (window as WindowWithDatadogRum).DD_RUM
+}
+
+export class DatadogRumTelemetryProvider implements TelemetryProvider {
+  private readonly originViewIdsByJobId = new Map<string, string>()
+
+  trackExecutionStarted(metadata: ExecutionStartMetadata): void {
+    const rum = getDatadogRum()
+    const originViewId = rum?.getInternalContext(metadata.startTime)?.view?.id
+    if (originViewId)
+      this.originViewIdsByJobId.set(metadata.jobId, originViewId)
+
+    rum?.addAction('workflow_execution_started', {
+      ...WORKFLOW_CONTEXT,
+      ...(originViewId && { origin_view_id: originViewId })
+    })
+  }
+
+  trackExecutionSuccess(metadata: ExecutionSuccessMetadata): void {
+    this.trackTerminalOutcome(metadata.jobId, 'success')
+  }
+
+  trackExecutionError(metadata: ExecutionErrorMetadata): void {
+    this.trackTerminalOutcome(metadata.jobId, 'failure')
+  }
+
+  private trackTerminalOutcome(
+    jobId: string,
+    outcome: 'success' | 'failure'
+  ): void {
+    const originViewId = this.originViewIdsByJobId.get(jobId)
+    this.originViewIdsByJobId.delete(jobId)
+
+    getDatadogRum()?.addAction('workflow_execution_completed', {
+      ...WORKFLOW_CONTEXT,
+      outcome,
+      ...(originViewId && { origin_view_id: originViewId })
+    })
+  }
+}

src/platform/telemetry/types.ts

@@ -156,6 +156,11 @@ export interface ExecutionErrorMetadata {
   error?: string
 }
 
+export interface ExecutionStartMetadata {
+  jobId: string
+  startTime: number
+}
+
 /**
  * Execution success metadata
  */
@@ -632,6 +637,7 @@ export interface TelemetryProvider {
 
   // Workflow execution events
   trackWorkflowExecution?(): void
+  trackExecutionStarted?(metadata: ExecutionStartMetadata): void
   trackExecutionError?(metadata: ExecutionErrorMetadata): void
   trackExecutionSuccess?(metadata: ExecutionSuccessMetadata): void
   trackSharedWorkflowRun?(metadata: SharedWorkflowRunMetadata): void

src/services/extensionService.ts

@@ -42,6 +42,11 @@ export const useExtensionService = () => {
     await Promise.all(
       extensions
         .filter((extension) => !extension.includes('extensions/core'))
+        .filter(
+          (extension) =>
+            __DISTRIBUTION__ !== 'cloud' ||
+            extension !== '/extensions/cloud/rum.js'
+        )
         .map(async (ext) => {
           try {
             await import(/* @vite-ignore */ api.fileURL(ext))

src/stores/executionStore.test.ts

@@ -20,6 +20,7 @@ const {
   mockActiveWorkflow,
   mockOpenWorkflows,
   mockShowTextPreview,
+  mockTrackExecutionStarted,
   mockTrackExecutionError,
   mockTrackExecutionSuccess,
   mockTrackSharedWorkflowRun
@@ -32,6 +33,7 @@ const {
     mockActiveWorkflow: shallowRef<{ path?: string } | null>(null),
     mockOpenWorkflows: shallowRef<{ path: string }[]>([]),
     mockShowTextPreview: vi.fn(),
+    mockTrackExecutionStarted: vi.fn(),
     mockTrackExecutionError: vi.fn(),
     mockTrackExecutionSuccess: vi.fn(),
     mockTrackSharedWorkflowRun: vi.fn()
@@ -91,6 +93,7 @@ vi.mock('@/platform/distribution/types', async () => ({
 
 vi.mock('@/platform/telemetry', () => ({
   useTelemetry: () => ({
+    trackExecutionStarted: mockTrackExecutionStarted,
     trackExecutionError: mockTrackExecutionError,
     trackExecutionSuccess: mockTrackExecutionSuccess,
     trackSharedWorkflowRun: mockTrackSharedWorkflowRun
@@ -631,6 +634,10 @@ describe('useExecutionStore - workflowStatus', () => {
     callStoreJob('job-1', workflowA)
     fireExecutionStart('job-1')
 
+    expect(mockTrackExecutionStarted).toHaveBeenCalledWith({
+      jobId: 'job-1',
+      startTime: expect.any(Number)
+    })
     expect(store.getWorkflowStatus(workflowA)).toBe('running')
   })
 

src/stores/executionStore.ts

@@ -748,6 +748,10 @@ export const useExecutionStore = defineStore('execution', () => {
     if (workflow?.path) {
       ensureSessionWorkflowPath(id, workflow.path)
     }
+    useTelemetry()?.trackExecutionStarted({
+      jobId: id,
+      startTime: performance.now()
+    })
     flushPendingWorkflowStatus(String(id), workflow)
   }
 

vite.config.mts

@@ -334,6 +334,27 @@ export default defineConfig({
     tailwindcss(),
     typegpuPlugin({}),
     comfyAPIPlugin(IS_DEV),
+    {
+      name: 'inject-cloud-rum',
+      apply: 'build',
+      transformIndexHtml(html) {
+        if (DISTRIBUTION !== 'cloud') return html
+
+        return {
+          html,
+          tags: [
+            {
+              tag: 'script',
+              attrs: {
+                type: 'module',
+                src: '/extensions/cloud/rum.js'
+              },
+              injectTo: 'head-prepend'
+            }
+          ]
+        }
+      }
+    },
     // Exclude proprietary ABCROM fonts from non-cloud builds
     {
       name: 'exclude-proprietary-fonts',