feat(aiobs): dedicated AI endpoint routing

.changeset/hip-planets-visit.md

@@ -0,0 +1,6 @@
+---
+'@posthog/core': patch
+'posthog-node': patch
+---
+
+Add an internal event-channel mechanism so `$ai_*` events can be routed to a dedicated capture endpoint in their own batch, independent of analytics events. Gated behind the unstable, internal-only `_internal_dedicatedAiEndpoint` option on `posthog-node` — not ready for general use.

packages/core/src/posthog-core-stateless.ts

@@ -137,6 +137,16 @@ export enum QuotaLimitedFeature {
   Recordings = 'recordings',
 }
 
+export type EventChannel = {
+  queueKey: PostHogPersistedProperty
+  path: string
+}
+
+export const DEFAULT_EVENT_CHANNEL: EventChannel = {
+  queueKey: PostHogPersistedProperty.Queue,
+  path: '/batch/',
+}
+
 export abstract class PostHogCoreStateless {
   // options
   readonly apiKey: string
@@ -959,6 +969,23 @@ export abstract class PostHogCoreStateless {
     // Default: no-op for sync storage implementations
   }
 
+  /**
+   * The set of channels this client flushes. Override to add destinations; the
+   * returned list must include every channel {@link channelForEvent} can return,
+   * otherwise events routed to a missing channel would never flush.
+   */
+  protected eventChannels(): readonly EventChannel[] {
+    return [DEFAULT_EVENT_CHANNEL]
+  }
+
+  /**
+   * Routes an event to its ingestion channel by name. Default sends everything
+   * to the main analytics channel; override to partition (e.g. `$ai_*`).
+   */
+  protected channelForEvent(_event: unknown): EventChannel {
+    return DEFAULT_EVENT_CHANNEL
+  }
+
   protected enqueue(type: string, _message: any, options?: PostHogCaptureOptions): void {
     this.wrap(() => {
       if (this.optedOut) {
@@ -974,15 +1001,16 @@ export abstract class PostHogCoreStateless {
         return
       }
 
-      const queue = this.getPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue) || []
+      const queueKey = this.channelForEvent(message.event).queueKey
+      const queue = this.getPersistedProperty<PostHogQueueItem[]>(queueKey) || []
 
       if (queue.length >= this.maxQueueSize) {
         queue.shift()
         this._logger.info('Queue is full, the oldest event is dropped.')
       }
 
       queue.push({ message })
-      this.setPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue, queue)
+      this.setPersistedProperty<PostHogQueueItem[]>(queueKey, queue)
 
       this._events.emit(type, message)
 
@@ -1032,7 +1060,7 @@ export abstract class PostHogCoreStateless {
 
     const payload = JSON.stringify(data)
 
-    const url = `${this.host}/batch/`
+    const url = this.host + this.channelForEvent(message.event).path
 
     const gzippedPayload = !this.disableCompression ? await gzipCompress(payload, this.isDebug) : null
     const fetchOptions: PostHogFetchOptions = {
@@ -1170,7 +1198,23 @@ export abstract class PostHogCoreStateless {
     this.clearFlushTimer()
     await this._initPromise
 
-    let queue = this.getPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue) || []
+    // Guarantee all flushes run but errors are still thrown at the end
+    let firstError: unknown
+    for (const channel of this.eventChannels()) {
+      try {
+        await this._flushChannel(channel)
+      } catch (err) {
+        firstError ??= err
+      }
+    }
+
+    if (firstError !== undefined) {
+      throw firstError
+    }
+  }
+
+  private async _flushChannel(channel: EventChannel): Promise<void> {
+    let queue = this.getPersistedProperty<PostHogQueueItem[]>(channel.queueKey) || []
 
     if (!queue.length) {
       return
@@ -1184,9 +1228,9 @@ export abstract class PostHogCoreStateless {
       const batchMessages = batchItems.map((item) => item.message)
 
       const persistQueueChange = async (): Promise<void> => {
-        const refreshedQueue = this.getPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue) || []
+        const refreshedQueue = this.getPersistedProperty<PostHogQueueItem[]>(channel.queueKey) || []
         const newQueue = refreshedQueue.slice(batchItems.length)
-        this.setPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue, newQueue)
+        this.setPersistedProperty<PostHogQueueItem[]>(channel.queueKey, newQueue)
         queue = newQueue
         // Wait for storage to complete to prevent duplicate events on app crash
         await this.flushStorage()
@@ -1204,7 +1248,7 @@ export abstract class PostHogCoreStateless {
 
       const payload = JSON.stringify(data)
 
-      const url = `${this.host}/batch/`
+      const url = this.host + channel.path
 
       const gzippedPayload = !this.disableCompression ? await gzipCompress(payload, this.isDebug) : null
       const fetchOptions: PostHogFetchOptions = {
@@ -1383,9 +1427,11 @@ export abstract class PostHogCoreStateless {
         await this.promiseQueue.join()
 
         while (true) {
-          const queue = this.getPersistedProperty<PostHogQueueItem[]>(PostHogPersistedProperty.Queue) || []
+          const pending = this.eventChannels().some(
+            (channel) => (this.getPersistedProperty<PostHogQueueItem[]>(channel.queueKey) || []).length > 0
+          )
 
-          if (queue.length === 0) {
+          if (!pending) {
             break
           }
 

packages/core/src/types.ts

@@ -220,6 +220,9 @@ export enum PostHogPersistedProperty {
   BootstrapFeatureFlagPayloads = 'bootstrap_feature_flag_payloads',
   OverrideFeatureFlags = 'override_feature_flags',
   Queue = 'queue',
+  // AI events queue. `$ai_*` events are routed here so they flush to a dedicated
+  // ingestion path independently of the main analytics queue. Only used by posthog-node.
+  AiQueue = 'ai_queue',
   // Logs queue. Individual SDKs may route this key to an isolated storage
   // instance if they want to separate logs write volume from main state.
   LogsQueue = 'logs_queue',

packages/node/src/__tests__/dedicated-ai-endpoint.spec.ts

@@ -0,0 +1,79 @@
+import { PostHog, PostHogOptions } from '@/entrypoints/index.node'
+
+const mockedFetch = jest.spyOn(globalThis, 'fetch').mockImplementation()
+
+const HOST = 'http://example.com'
+const ANALYTICS_URL = `${HOST}/batch/`
+const AI_URL = `${HOST}/i/v0/ai/batch/`
+
+const okResponse = {
+  status: 200,
+  text: () => Promise.resolve('ok'),
+  json: () => Promise.resolve({ status: 'ok' }),
+} as any
+
+// Exact-match — AI_URL contains "/batch/" as a substring, so substring matching would conflate them.
+const callsTo = (url: string): any[] => mockedFetch.mock.calls.filter((c) => c[0] === url)
+
+const batchSentTo = (url: string): any[] | undefined => {
+  const call = callsTo(url).at(-1)
+  return call ? JSON.parse((call[1] as any).body).batch : undefined
+}
+
+const newClient = (options: Partial<PostHogOptions> = {}): PostHog =>
+  new PostHog('TEST_API_KEY', { host: HOST, fetchRetryCount: 0, disableCompression: true, ...options })
+
+describe('PostHog Node.js — dedicated AI endpoint (_internal_dedicatedAiEndpoint)', () => {
+  let errorSpy: jest.SpyInstance
+
+  beforeEach(() => {
+    errorSpy = jest.spyOn(console, 'error').mockImplementation(() => {})
+    mockedFetch.mockResolvedValue(okResponse)
+  })
+
+  afterEach(() => {
+    mockedFetch.mockReset()
+    errorSpy.mockRestore()
+  })
+
+  describe('when enabled', () => {
+    it('routes batched $ai_* events to the dedicated AI path', async () => {
+      const ph = newClient({ _internal_dedicatedAiEndpoint: true })
+      ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: { $ai_model: 'gpt-4' } })
+      await ph.shutdown()
+
+      expect(batchSentTo(AI_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation'])
+      expect(callsTo(ANALYTICS_URL)).toHaveLength(0)
+    })
+
+    it('keeps non-AI events on the normal path, in a separate batch from AI events', async () => {
+      const ph = newClient({ _internal_dedicatedAiEndpoint: true })
+      ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: {} })
+      ph.capture({ distinctId: 'u1', event: 'button_clicked', properties: {} })
+      await ph.shutdown()
+
+      expect(batchSentTo(AI_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation'])
+      expect(batchSentTo(ANALYTICS_URL)?.map((e: any) => e.event)).toEqual(['button_clicked'])
+    })
+
+    it('routes immediate $ai_* captures to the dedicated AI path', async () => {
+      const ph = newClient({ _internal_dedicatedAiEndpoint: true })
+      await ph.captureImmediate({ distinctId: 'u1', event: '$ai_embedding', properties: {} })
+
+      expect(batchSentTo(AI_URL)?.[0].event).toBe('$ai_embedding')
+      expect(callsTo(ANALYTICS_URL)).toHaveLength(0)
+      await ph.shutdown()
+    })
+  })
+
+  describe('when disabled (default)', () => {
+    it('routes $ai_* events to the normal batch path', async () => {
+      const ph = newClient()
+      ph.capture({ distinctId: 'u1', event: '$ai_generation', properties: {} })
+      await ph.shutdown()
+
+      expect(batchSentTo(ANALYTICS_URL)?.map((e: any) => e.event)).toEqual(['$ai_generation'])
+      expect(callsTo(AI_URL)).toHaveLength(0)
+    })
+  })
+})

packages/node/src/client.ts

@@ -1,6 +1,7 @@
 import { version } from './version'
 
 import {
+  EventChannel,
   FeatureFlagValue,
   isBlockedUA,
   isPlainObject,
@@ -56,6 +57,11 @@ const WAITUNTIL_DEBOUNCE_MS = 50
 const WAITUNTIL_MAX_WAIT_MS = 500
 const DEFAULT_NODE_HOST = 'https://us.i.posthog.com'
 
+const AI_EVENT_CHANNEL: EventChannel = {
+  queueKey: PostHogPersistedProperty.AiQueue,
+  path: '/i/v0/ai/batch/',
+}
+
 // Process-wide dedup for deprecation warnings — without this, calling a deprecated
 // method in a loop would spam logs. Matches Python's `warnings.warn` default-dedup behavior.
 const _emittedDeprecations = new Set<string>()
@@ -134,6 +140,8 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
     timer: ReturnType<typeof setTimeout> | undefined
   }
 
+  private readonly _useDedicatedAiEndpoint: boolean
+
   /**
    * Initialize a new PostHog client instance.
    *
@@ -174,6 +182,7 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
     super(normalizedApiKey, normalizedOptions)
 
     this.options = normalizedOptions
+    this._useDedicatedAiEndpoint = this.options._internal_dedicatedAiEndpoint === true
     this.context = this.initializeContext()
 
     this.options.featureFlagsPollingInterval =
@@ -224,6 +233,17 @@ export abstract class PostHogBackendClient extends PostHogCoreStateless implemen
     this.maxCacheSize = normalizedOptions.maxCacheSize || MAX_CACHE_SIZE
   }
 
+  protected override eventChannels(): readonly EventChannel[] {
+    return this._useDedicatedAiEndpoint ? [...super.eventChannels(), AI_EVENT_CHANNEL] : super.eventChannels()
+  }
+
+  protected override channelForEvent(event: unknown): EventChannel {
+    if (this._useDedicatedAiEndpoint && typeof event === 'string' && event.startsWith('$ai_')) {
+      return AI_EVENT_CHANNEL
+    }
+    return super.channelForEvent(event)
+  }
+
   protected override enqueue(type: string, message: any, options?: PostHogCaptureOptions): void {
     super.enqueue(type, message, options)
     this.scheduleDebouncedFlush()

packages/node/src/types.ts

@@ -135,6 +135,13 @@ export type PostHogOptions = Omit<PostHogCoreOptions, 'before_send'> & {
   persistence?: 'memory'
   personalApiKey?: string
   privacyMode?: boolean
+  /**
+   * Routes `$ai_*` events to a dedicated capture-ai endpoint in their own batch.
+   *
+   * @internal Not ready for use — the backend endpoint and ingress routing are still being
+   * rolled out. Do not enable; behaviour and naming may change or be removed without notice.
+   */
+  _internal_dedicatedAiEndpoint?: boolean
   enableExceptionAutocapture?: boolean
   // The interval in milliseconds between polls for refreshing feature flag definitions. Defaults to 30 seconds.
   featureFlagsPollingInterval?: number