diff --git a/README.md b/README.md index 5c4652e..0277fb2 100644 --- a/README.md +++ b/README.md @@ -33,6 +33,7 @@ Go to **Settings → API Keys** and copy your keys. export LANGFUSE_PUBLIC_KEY="pk-lf-..." export LANGFUSE_SECRET_KEY="sk-lf-..." export LANGFUSE_BASEURL="https://cloud.langfuse.com" # Optional +export LANGFUSE_SESSION_ID="workflow-run-123" # Optional ``` ### 3. Enable Plugin + OTEL @@ -52,6 +53,15 @@ In `.opencode/opencode.json`: That's it! All traces appear automatically in your Langfuse dashboard. +To group several OpenCode runs into one Langfuse session, reuse the same +`LANGFUSE_SESSION_ID` for every process in your workflow: + +```bash +WORKFLOW_ID="workflow-$(date +%s)" +LANGFUSE_SESSION_ID="$WORKFLOW_ID" opencode +LANGFUSE_SESSION_ID="$WORKFLOW_ID" opencode +``` + --- ## How It Works @@ -71,6 +81,7 @@ OpenCode (OTEL spans) → LangfuseSpanProcessor → Langfuse Dashboard | `LANGFUSE_PUBLIC_KEY` | Yes | - | Langfuse public key | | `LANGFUSE_SECRET_KEY` | Yes | - | Langfuse secret key | | `LANGFUSE_BASEURL` | No | `https://cloud.langfuse.com` | Self-hosted instance | +| `LANGFUSE_SESSION_ID` | No | - | Group traces by session | --- diff --git a/src/index.test.ts b/src/index.test.ts index c08b77c..d107cf5 100644 --- a/src/index.test.ts +++ b/src/index.test.ts @@ -3,6 +3,9 @@ import { LangfusePlugin } from "./index"; const mockForceFlush = mock(() => Promise.resolve()); const mockStart = mock(() => {}); +const mockShutdown = mock(() => Promise.resolve()); + +let capturedNodeSDKOptions: Record = {}; mock.module("@langfuse/otel", () => ({ LangfuseSpanProcessor: mock(() => ({ @@ -11,9 +14,13 @@ mock.module("@langfuse/otel", () => ({ })); mock.module("@opentelemetry/sdk-node", () => ({ - NodeSDK: mock(() => ({ - start: mockStart, - })), + NodeSDK: mock((options: Record) => { + capturedNodeSDKOptions = options; + return { + start: mockStart, + shutdown: mockShutdown, + }; + }), })); const mockLog = mock(() => {}); @@ -40,7 +47,9 @@ describe("LangfusePlugin", () => { beforeEach(() => { mockForceFlush.mockClear(); mockStart.mockClear(); + mockShutdown.mockClear(); mockLog.mockClear(); + capturedNodeSDKOptions = {}; }); afterEach(() => { @@ -197,4 +206,46 @@ describe("LangfusePlugin", () => { }); }); }); + + describe("workflow session grouping", () => { + it("preserves current span processors when LANGFUSE_SESSION_ID is missing", async () => { + setupEnv(); + + await LangfusePlugin(mockPluginInput()); + + expect(capturedNodeSDKOptions.spanProcessors).toHaveLength(1); + }); + + it("adds session.id to spans when LANGFUSE_SESSION_ID is set", async () => { + setupEnv({ LANGFUSE_SESSION_ID: "workflow-123" }); + + await LangfusePlugin(mockPluginInput()); + + const spanProcessors = capturedNodeSDKOptions.spanProcessors as Array<{ + onStart?: (span: unknown) => void; + }>; + const setAttribute = mock(() => {}); + + expect(spanProcessors).toHaveLength(2); + spanProcessors[0].onStart!({ setAttribute }); + + expect(setAttribute).toHaveBeenCalledWith("session.id", "workflow-123"); + }); + + it("ignores overlong LANGFUSE_SESSION_ID values", async () => { + setupEnv({ LANGFUSE_SESSION_ID: "x".repeat(201) }); + + await LangfusePlugin(mockPluginInput()); + + expect(capturedNodeSDKOptions.spanProcessors).toHaveLength(1); + expect(mockLog).toHaveBeenCalledWith({ + body: { + service: "langfuse-otel", + level: "warn", + message: + "LANGFUSE_SESSION_ID is longer than 200 characters - session grouping disabled", + }, + }); + }); + }); }); diff --git a/src/index.ts b/src/index.ts index e2cbb45..472887d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,6 +1,31 @@ import { LangfuseSpanProcessor } from "@langfuse/otel"; import type { Plugin } from "@opencode-ai/plugin"; import { NodeSDK } from "@opentelemetry/sdk-node"; +import type { + ReadableSpan, + Span, + SpanProcessor, +} from "@opentelemetry/sdk-trace-base"; + +const LANGFUSE_SESSION_ID_MAX_LENGTH = 200; + +class LangfuseSessionSpanProcessor implements SpanProcessor { + constructor(private readonly sessionId: string) {} + + onStart(span: Span): void { + span.setAttribute("session.id", this.sessionId); + } + + onEnd(_span: ReadableSpan): void {} + + forceFlush(): Promise { + return Promise.resolve(); + } + + shutdown(): Promise { + return Promise.resolve(); + } +} export const LangfusePlugin: Plugin = async ({ client }) => { const publicKey = process.env.LANGFUSE_PUBLIC_KEY; @@ -29,8 +54,23 @@ export const LangfusePlugin: Plugin = async ({ client }) => { environment, }); + const sessionId = process.env.LANGFUSE_SESSION_ID?.trim(); + const sessionProcessor = + sessionId && sessionId.length <= LANGFUSE_SESSION_ID_MAX_LENGTH + ? new LangfuseSessionSpanProcessor(sessionId) + : undefined; + + if (sessionId && !sessionProcessor) { + log( + "warn", + `LANGFUSE_SESSION_ID is longer than ${LANGFUSE_SESSION_ID_MAX_LENGTH} characters - session grouping disabled` + ); + } + const sdk = new NodeSDK({ - spanProcessors: [processor], + spanProcessors: sessionProcessor + ? [sessionProcessor, processor] + : [processor], }); sdk.start();