Skip to content

Commit fce40fe

Browse files
feat: 加上 userId 的传递
1 parent a7e03a5 commit fce40fe

3 files changed

Lines changed: 109 additions & 3 deletions

File tree

src/entrypoints/init.ts

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ import { isBetaTracingEnabled } from '../utils/telemetry/betaSessionTracing.js'
4949
import { getTelemetryAttributes } from '../utils/telemetryAttributes.js'
5050
import { setShellIfWindows } from '../utils/windowsPaths.js'
5151
import { initSentry } from '../utils/sentry.js'
52+
import { initUser } from '../utils/user.js'
5253
import { initLangfuse, shutdownLangfuse } from '../services/langfuse/index.js'
5354

5455
// initialize1PEventLogging is dynamically imported to defer OpenTelemetry sdk-logs/resources
@@ -156,6 +157,8 @@ export const init = memoize(async (): Promise<void> => {
156157
initSentry()
157158

158159
// Initialize Langfuse tracing (no-op if keys not configured)
160+
// Pre-warm user email cache so Langfuse traces include userId
161+
await initUser()
159162
initLangfuse()
160163
registerCleanup(shutdownLangfuse)
161164

src/services/langfuse/__tests__/langfuse.test.ts

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ const mockRootEnd = mock(() => {})
2929
// Mock LangfuseOtelSpanAttributes (re-exported from @langfuse/core)
3030
const mockLangfuseOtelSpanAttributes: Record<string, string> = {
3131
TRACE_SESSION_ID: 'session.id',
32+
TRACE_USER_ID: 'user.id',
3233
OBSERVATION_TYPE: 'observation.type',
3334
OBSERVATION_INPUT: 'observation.input',
3435
OBSERVATION_OUTPUT: 'observation.output',
@@ -74,6 +75,14 @@ mock.module('src/utils/debug.js', () => ({
7475
logForDebugging: mock(() => {}),
7576
}))
7677

78+
// Mock user data — resolveLangfuseUserId uses getCoreUserData().email and .deviceId
79+
mock.module('src/utils/user.js', () => ({
80+
getCoreUserData: mock(() => ({
81+
email: 'test-device-id',
82+
deviceId: 'test-device-id',
83+
})),
84+
}))
85+
7786
describe('Langfuse integration', () => {
7887
beforeEach(() => {
7988
// Reset env
@@ -477,6 +486,70 @@ describe('Langfuse integration', () => {
477486
})
478487
})
479488

489+
describe('createTrace with username', () => {
490+
test('sets user.id attribute when username is provided', async () => {
491+
process.env.LANGFUSE_PUBLIC_KEY = 'pk-test'
492+
process.env.LANGFUSE_SECRET_KEY = 'sk-test'
493+
mockSetAttribute.mockClear()
494+
const { createTrace } = await import('../tracing.js')
495+
const span = createTrace({
496+
sessionId: 's1',
497+
model: 'claude-3',
498+
provider: 'firstParty',
499+
username: 'user@example.com',
500+
})
501+
expect(span).not.toBeNull()
502+
expect(mockSetAttribute).toHaveBeenCalledWith('user.id', 'user@example.com')
503+
})
504+
505+
test('falls back to LANGFUSE_USER_ID env when username not provided', async () => {
506+
process.env.LANGFUSE_PUBLIC_KEY = 'pk-test'
507+
process.env.LANGFUSE_SECRET_KEY = 'sk-test'
508+
process.env.LANGFUSE_USER_ID = 'env-user@test.com'
509+
mockSetAttribute.mockClear()
510+
const { createTrace } = await import('../tracing.js')
511+
const span = createTrace({
512+
sessionId: 's1',
513+
model: 'claude-3',
514+
provider: 'firstParty',
515+
})
516+
expect(span).not.toBeNull()
517+
expect(mockSetAttribute).toHaveBeenCalledWith('user.id', 'env-user@test.com')
518+
delete process.env.LANGFUSE_USER_ID
519+
})
520+
521+
test('falls back to deviceId when neither username nor env is provided', async () => {
522+
process.env.LANGFUSE_PUBLIC_KEY = 'pk-test'
523+
process.env.LANGFUSE_SECRET_KEY = 'sk-test'
524+
delete process.env.LANGFUSE_USER_ID
525+
mockSetAttribute.mockClear()
526+
const { createTrace } = await import('../tracing.js')
527+
createTrace({ sessionId: 's1', model: 'claude-3', provider: 'firstParty' })
528+
// Falls back to getCoreUserData().deviceId (mocked as 'test-device-id')
529+
expect(mockSetAttribute).toHaveBeenCalledWith('user.id', 'test-device-id')
530+
})
531+
532+
test('username takes precedence over LANGFUSE_USER_ID env', async () => {
533+
process.env.LANGFUSE_PUBLIC_KEY = 'pk-test'
534+
process.env.LANGFUSE_SECRET_KEY = 'sk-test'
535+
process.env.LANGFUSE_USER_ID = 'env-user@test.com'
536+
mockSetAttribute.mockClear()
537+
const { createTrace } = await import('../tracing.js')
538+
createTrace({
539+
sessionId: 's1',
540+
model: 'claude-3',
541+
provider: 'firstParty',
542+
username: 'param-user@test.com',
543+
})
544+
const userIdCalls = mockSetAttribute.mock.calls.filter(
545+
(call: unknown[]) => Array.isArray(call) && call[0] === 'user.id',
546+
)
547+
expect(userIdCalls.length).toBe(1)
548+
expect((userIdCalls[0] as unknown[])[1]).toBe('param-user@test.com')
549+
delete process.env.LANGFUSE_USER_ID
550+
})
551+
})
552+
480553
describe('nested agent scenario', () => {
481554
test('sub-agent trace shares sessionId with parent', async () => {
482555
process.env.LANGFUSE_PUBLIC_KEY = 'pk-test'

src/services/langfuse/tracing.ts

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,11 +3,17 @@ import type { LangfuseSpan, LangfuseGeneration, LangfuseAgent } from '@langfuse/
33
import { isLangfuseEnabled } from './client.js'
44
import { sanitizeToolInput, sanitizeToolOutput } from './sanitize.js'
55
import { logForDebugging } from 'src/utils/debug.js'
6+
import { getCoreUserData } from 'src/utils/user.js'
67

78
export type { LangfuseSpan }
89

910
// Root trace is an agent observation — represents one full agentic turn/session
10-
type RootTrace = LangfuseAgent & { _sessionId?: string }
11+
type RootTrace = LangfuseAgent & { _sessionId?: string; _userId?: string }
12+
13+
/** Resolve the user ID for Langfuse traces: explicit param > env var > email > deviceId */
14+
function resolveLangfuseUserId(username?: string): string | undefined {
15+
return username ?? process.env.LANGFUSE_USER_ID ?? getCoreUserData().email ?? getCoreUserData().deviceId
16+
}
1117

1218
export function createTrace(params: {
1319
sessionId: string
@@ -16,6 +22,7 @@ export function createTrace(params: {
1622
input?: unknown
1723
name?: string
1824
querySource?: string
25+
username?: string
1926
}): LangfuseSpan | null {
2027
if (!isLangfuseEnabled()) return null
2128
try {
@@ -31,6 +38,11 @@ export function createTrace(params: {
3138
}, { asType: 'agent' }) as RootTrace
3239
rootSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_SESSION_ID, params.sessionId)
3340
rootSpan._sessionId = params.sessionId
41+
const userId = resolveLangfuseUserId(params.username)
42+
if (userId) {
43+
rootSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_USER_ID, userId)
44+
rootSpan._userId = userId
45+
}
3446
logForDebugging(`[langfuse] Trace created: ${rootSpan.id}`)
3547
return rootSpan as unknown as LangfuseSpan
3648
} catch (e) {
@@ -87,11 +99,15 @@ export function recordLLMObservation(
8799
},
88100
)
89101

90-
// Propagate session ID to generation span so Langfuse links it correctly
102+
// Propagate session ID and user ID to generation span so Langfuse links it correctly
91103
const sessionId = (rootSpan as unknown as RootTrace)._sessionId
92104
if (sessionId) {
93105
gen.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_SESSION_ID, sessionId)
94106
}
107+
const userId = (rootSpan as unknown as RootTrace)._userId
108+
if (userId) {
109+
gen.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_USER_ID, userId)
110+
}
95111

96112
gen.update({
97113
output: params.output,
@@ -142,11 +158,15 @@ export function recordToolObservation(
142158
},
143159
)
144160

145-
// Propagate session ID to tool span so Langfuse links it correctly
161+
// Propagate session ID and user ID to tool span so Langfuse links it correctly
146162
const sessionId = (rootSpan as unknown as RootTrace)._sessionId
147163
if (sessionId) {
148164
toolObs.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_SESSION_ID, sessionId)
149165
}
166+
const userId = (rootSpan as unknown as RootTrace)._userId
167+
if (userId) {
168+
toolObs.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_USER_ID, userId)
169+
}
150170

151171
toolObs.update({
152172
output: sanitizeToolOutput(params.toolName, params.output),
@@ -190,6 +210,10 @@ export function createToolBatchSpan(
190210
if (sessionId) {
191211
batchSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_SESSION_ID, sessionId)
192212
}
213+
const userId = (rootSpan as unknown as RootTrace)._userId
214+
if (userId) {
215+
batchSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_USER_ID, userId)
216+
}
193217

194218
logForDebugging(`[langfuse] Tool batch span created: ${batchSpan.id} (tools=${params.toolNames.join(',')})`)
195219
return batchSpan
@@ -216,6 +240,7 @@ export function createSubagentTrace(params: {
216240
model: string
217241
provider: string
218242
input?: unknown
243+
username?: string
219244
}): LangfuseSpan | null {
220245
if (!isLangfuseEnabled()) return null
221246
try {
@@ -230,6 +255,11 @@ export function createSubagentTrace(params: {
230255
}, { asType: 'agent' }) as RootTrace
231256
rootSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_SESSION_ID, params.sessionId)
232257
rootSpan._sessionId = params.sessionId
258+
const userId = resolveLangfuseUserId(params.username)
259+
if (userId) {
260+
rootSpan.otelSpan.setAttribute(LangfuseOtelSpanAttributes.TRACE_USER_ID, userId)
261+
rootSpan._userId = userId
262+
}
233263
logForDebugging(`[langfuse] Sub-agent trace created: ${rootSpan.id} (type=${params.agentType})`)
234264
return rootSpan as unknown as LangfuseSpan
235265
} catch (e) {

0 commit comments

Comments
 (0)