feat(aio): public beta captureAi with dedicated AI capture lane

.changeset/node-ai-capture-lane.md

@@ -0,0 +1,7 @@
+---
+'@posthog/core': minor
+'posthog-node': minor
+'@posthog/ai': minor
+---
+
+Public beta `captureAi()` / `captureAiImmediate()`: AI events on a dedicated isolated endpoint with the event UUID returned. New `enableFullAiCapture` option replaces the internal `_useAiLane` / `_enableMultimodalCapture`; wrappers route through the AI endpoint and skip redaction/truncation when set (privacy mode still wins).

compliance/node/adapter.js

@@ -47,9 +47,10 @@ async function discardClient() {
     try {
         client.clearFlushTimer?.()
         client.setPersistedProperty?.('queue', [])
-        // v1 mode routes $ai_* events to a separate queue; clear it too so they can't
-        // leak into the next scenario.
+        // v1 mode routes $ai_* events to a separate queue, and captureAi() events to
+        // their own dedicated queue; clear both so they can't leak into the next scenario.
         client.setPersistedProperty?.('ai_queue', [])
+        client.setPersistedProperty?.('ai_capture_queue', [])
         await client.shutdown(1)
     } catch (error) {
         // Ignore reset-time shutdown errors; the next test starts with a fresh client.
@@ -61,7 +62,10 @@ app.get('/health', (req, res) => {
         sdk_name: 'posthog-node',
         sdk_version: require('../packages/node/package.json').version,
         adapter_version: '1.0.0',
-        capabilities: CAPTURE_MODE === 'v1' ? ['capture_v1', 'encoding_gzip'] : ['capture_v0', 'encoding_gzip'],
+        capabilities:
+            CAPTURE_MODE === 'v1'
+                ? ['capture_v1', 'capture_ai_v0', 'encoding_gzip']
+                : ['capture_v0', 'capture_ai_v0', 'encoding_gzip'],
     })
 })
 
@@ -185,6 +189,46 @@ app.post('/capture', (req, res) => {
     }
 })
 
+app.post('/capture_ai', (req, res) => {
+    if (!state.client) {
+        return res.status(400).json({ error: 'SDK not initialized' })
+    }
+
+    const { distinct_id, event, properties, timestamp, options, uuid } = req.body
+
+    if (!distinct_id || !event) {
+        return res.status(400).json({ error: 'distinct_id and event are required' })
+    }
+
+    try {
+        const mergedProperties = { ...(properties || {}) }
+        if (options && typeof options === 'object') {
+            for (const [optionKey, sentinel] of Object.entries(OPTION_SENTINELS)) {
+                if (Object.prototype.hasOwnProperty.call(options, optionKey)) {
+                    mergedProperties[sentinel] = options[optionKey]
+                }
+            }
+        }
+
+        // Unlike /capture, forward a supplied uuid so it's echoed back to the caller.
+        const returnedUuid = state.client.captureAi({
+            distinctId: distinct_id,
+            event,
+            properties: mergedProperties,
+            timestamp: timestamp ? new Date(timestamp) : undefined,
+            uuid,
+        })
+
+        state.totalEventsCaptured++
+        state.pendingEvents++
+
+        res.json({ success: true, uuid: returnedUuid })
+    } catch (error) {
+        state.lastError = error.message
+        res.status(500).json({ error: error.message })
+    }
+})
+
 app.post('/flush', async (req, res) => {
     if (!state.client) {
         return res.status(400).json({ error: 'SDK not initialized' })

packages/ai/src/anthropic/index.ts

@@ -244,7 +244,7 @@ export class WrappedMessages extends AnthropicOriginal.Messages {
                 ...posthogParams,
                 model: anthropicParams.model,
                 provider: 'anthropic',
-                input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic')),
+                input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic'), this.phClient),
                 output: formattedOutput,
                 latency,
                 timeToFirstToken,
@@ -260,7 +260,7 @@ export class WrappedMessages extends AnthropicOriginal.Messages {
                 ...posthogParams,
                 model: anthropicParams.model,
                 provider: 'anthropic',
-                input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic')),
+                input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic'), this.phClient),
                 output: [],
                 latency: 0,
                 baseURL: this.baseURL,
@@ -298,7 +298,7 @@ export class WrappedMessages extends AnthropicOriginal.Messages {
               ...posthogParams,
               model: anthropicParams.model,
               provider: 'anthropic',
-              input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic')),
+              input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic'), this.phClient),
               output: formatResponseAnthropic(result),
               latency,
               baseURL: this.baseURL,
@@ -323,7 +323,7 @@ export class WrappedMessages extends AnthropicOriginal.Messages {
             ...posthogParams,
             model: anthropicParams.model,
             provider: 'anthropic',
-            input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic')),
+            input: sanitizeAnthropic(mergeSystemPrompt(anthropicParams, 'anthropic'), this.phClient),
             output: [],
             latency: 0,
             baseURL: this.baseURL,

packages/ai/src/captureAiEvent.ts

@@ -0,0 +1,37 @@
+import type { EventMessage } from 'posthog-node'
+
+/** @internal */
+export type FullAiCaptureGate = {
+  readonly enableFullAiCapture?: boolean
+}
+
+/** @internal */
+export interface AiLaneCapableClient extends FullAiCaptureGate {
+  capture(props: EventMessage): void
+  captureImmediate(props: EventMessage): Promise<void>
+  captureAi?(props: EventMessage): string | undefined
+  captureAiImmediate?(props: EventMessage): Promise<string | undefined>
+}
+
+/** @internal */
+export function isFullAiCaptureEnabled(client?: FullAiCaptureGate): boolean {
+  return client?.enableFullAiCapture === true
+}
+
+/** @internal */
+export function captureAiEvent(client: AiLaneCapableClient, event: EventMessage): void {
+  if (isFullAiCaptureEnabled(client) && typeof client.captureAi === 'function') {
+    client.captureAi(event)
+    return
+  }
+  client.capture(event)
+}
+
+/** @internal */
+export async function captureAiEventImmediate(client: AiLaneCapableClient, event: EventMessage): Promise<void> {
+  if (isFullAiCaptureEnabled(client) && typeof client.captureAiImmediate === 'function') {
+    await client.captureAiImmediate(event)
+    return
+  }
+  await client.captureImmediate(event)
+}

packages/ai/src/captureAiGeneration.ts

@@ -6,6 +6,7 @@ import type { TokenUsage } from './types'
 import { stringifyError } from './serializeError'
 import { AIEvent, CostOverride, getTokensSource, hasTokenOverrides, withPrivacyMode } from './utils'
 import { warnIfPostHogAiGateway } from './gatewayWarning'
+import { captureAiEvent, captureAiEventImmediate } from './captureAiEvent'
 
 /**
  * Options for `captureAiGeneration`. Mirrors the `$ai_generation` event shape
@@ -222,9 +223,9 @@ export const captureAiGeneration = async (client: PostHog, options: CaptureAiGen
     }
 
     if (options.captureImmediate) {
-      await client.captureImmediate(event)
+      await captureAiEventImmediate(client, event)
     } else {
-      client.capture(event)
+      captureAiEvent(client, event)
     }
   } catch (error) {
     // Telemetry failures must never affect the instrumented provider call.

packages/ai/src/gemini/index.ts

@@ -68,7 +68,7 @@ export class WrappedModels {
         model: geminiParams.model,
         provider: 'gemini',
         input: this.formatInputForPostHog(geminiParams),
-        output: formatResponseGemini(response),
+        output: formatResponseGemini(response, this.phClient),
         latency,
         baseURL: 'https://generativelanguage.googleapis.com',
         modelParameters: getModelParams(params as GenerateContentParameters & MonitoringParams),
@@ -424,7 +424,7 @@ export class WrappedModels {
   }
 
   private formatInputForPostHog(params: GenerateContentParameters): FormattedMessage[] {
-    const sanitized = sanitizeGemini(params.contents)
+    const sanitized = sanitizeGemini(params.contents, this.phClient)
     const messages = this.formatInput(sanitized)
 
     const systemInstruction = this.extractSystemInstruction(params)

packages/ai/src/langchain/callbacks.ts

@@ -13,6 +13,7 @@ import { sanitizeLangChain } from '../sanitization'
 import { stringifyError } from '../serializeError'
 import { warnIfPostHogAiGateway } from '../gatewayWarning'
 import { isObject } from '../typeGuards'
+import { captureAiEvent } from '../captureAiEvent'
 
 // Mirror LangGraph's isGraphBubbleUp guard without adding LangGraph as a dependency. Every
 // LangGraph control-flow exception (GraphInterrupt, NodeInterrupt, ParentCommand, GraphDrained,
@@ -338,7 +339,7 @@ export class LangChainCallbackHandler extends BaseCallbackHandler {
     const runNameFound = this._getLangchainRunName(serialized, { extraParams, runName }) || 'generation'
     const generation: GenerationMetadata = {
       name: runNameFound,
-      input: sanitizeLangChain(messages),
+      input: sanitizeLangChain(messages, this.client),
       startTime: Date.now(),
     }
     if (extraParams) {
@@ -388,7 +389,7 @@ export class LangChainCallbackHandler extends BaseCallbackHandler {
 
   private _safeCapture(message: EventMessage): void {
     try {
-      this.client.capture(message)
+      captureAiEvent(this.client, message)
     } catch {
       // Telemetry delivery must never affect the LangChain callback lifecycle.
     }
@@ -426,7 +427,7 @@ export class LangChainCallbackHandler extends BaseCallbackHandler {
       $ai_lib: 'posthog-ai',
       $ai_lib_version: version,
       $ai_trace_id: traceId,
-      $ai_input_state: withPrivacyMode(this.client, this.privacyMode, sanitizeLangChain(run.input)),
+      $ai_input_state: withPrivacyMode(this.client, this.privacyMode, sanitizeLangChain(run.input, this.client)),
       $ai_latency: latency,
       $ai_span_name: run.name,
       $ai_span_id: runId,
@@ -450,15 +451,19 @@ export class LangChainCallbackHandler extends BaseCallbackHandler {
           eventProperties['$ai_output_state'] = withPrivacyMode(
             this.client,
             this.privacyMode,
-            sanitizeLangChain({ __interrupt__: interrupts })
+            sanitizeLangChain({ __interrupt__: interrupts }, this.client)
           )
         }
       } else {
         eventProperties['$ai_error'] = stringifyError(outputs)
         eventProperties['$ai_is_error'] = true
       }
     } else if (outputs !== undefined) {
-      eventProperties['$ai_output_state'] = withPrivacyMode(this.client, this.privacyMode, sanitizeLangChain(outputs))
+      eventProperties['$ai_output_state'] = withPrivacyMode(
+        this.client,
+        this.privacyMode,
+        sanitizeLangChain(outputs, this.client)
+      )
     }
     this._safeCapture({
       distinctId: this.distinctId ? this.distinctId.toString() : runId,
@@ -685,7 +690,7 @@ export class LangChainCallbackHandler extends BaseCallbackHandler {
     }
 
     // Sanitize the message content to redact base64 images
-    return sanitizeLangChain(messageDict) as Record<string, any>
+    return sanitizeLangChain(messageDict, this.client) as Record<string, any>
   }
 
   private _extractStopReason(output: LLMResult): string | undefined {

packages/ai/src/openai-agents/processor.ts

@@ -20,6 +20,7 @@ import type {
 import { MAX_OUTPUT_SIZE, toContentString, truncate, utf8ByteLength, withPrivacyMode } from '../utils'
 import { version } from '../../package.json'
 import { warnIfPostHogAiGateway } from '../gatewayWarning'
+import { captureAiEvent, isFullAiCaptureEnabled } from '../captureAiEvent'
 
 /**
  * Normalize OpenAI Responses API input items to include a `role` field.
@@ -181,7 +182,10 @@ export class PostHogTracingProcessor implements TracingProcessor {
   private _prepareCapturedValue(value: unknown): unknown {
     const serializableValue = ensureSerializable(value)
     const serializedValue = stringifyForSizeCheck(serializableValue)
-    const boundedValue = exceedsMaxOutputSize(serializedValue) ? truncate(serializedValue) : serializableValue
+    const boundedValue =
+      isFullAiCaptureEnabled(this._client) || !exceedsMaxOutputSize(serializedValue)
+        ? serializableValue
+        : truncate(serializedValue, this._client)
     return this._withPrivacyMode(boundedValue)
   }
 
@@ -230,7 +234,7 @@ export class PostHogTracingProcessor implements TracingProcessor {
         groups: Object.keys(this._groups).length > 0 ? this._groups : undefined,
       }
 
-      this._client.capture(eventMessage)
+      captureAiEvent(this._client, eventMessage)
     } catch (error) {
       this._handleError(error, 'capture')
     }