feat: group traces by langfuse session id

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 |
 
 ---
 

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<string, unknown> = {};
 
 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<string, unknown>) => {
+    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",
+        },
+      });
+    });
+  });
 });

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<void> {
+    return Promise.resolve();
+  }
+
+  shutdown(): Promise<void> {
+    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();