Fix plugin lifecycle

src/langfuse.ts

@@ -17,19 +17,16 @@ import { PLUGIN_VERSION } from "./version.js";
 export class LangfuseClient {
   readonly baseUrl: string;
   readonly forceFlush: Effect.Effect<void, unknown>;
-  readonly shutdown: Effect.Effect<void, unknown>;
   private readonly traceState: LangfuseTraceState;
 
   constructor(input: {
     baseUrl: string;
     traceState: LangfuseTraceState;
     forceFlush: Effect.Effect<void, unknown>;
-    shutdown: Effect.Effect<void, unknown>;
   }) {
     this.baseUrl = input.baseUrl;
     this.traceState = input.traceState;
     this.forceFlush = input.forceFlush;
-    this.shutdown = input.shutdown;
   }
 
   clearTraceState() {
@@ -1797,28 +1794,13 @@ export const createLangfuseClient = (input: {
         makeAppRootSpanProcessor(traceState.tracerName),
       ],
     });
-    let isShutdown = false;
-
     yield* Effect.sync(() => {
       provider.register();
     });
 
     return new LangfuseClient({
       baseUrl: input.baseUrl,
       traceState,
-      forceFlush: Effect.tryPromise(() =>
-        isShutdown ? Promise.resolve() : processor.forceFlush(),
-      ),
-      shutdown: Effect.gen(function* () {
-        if (isShutdown) {
-          return;
-        }
-
-        isShutdown = true;
-        yield* Effect.tryPromise(() => processor.forceFlush()).pipe(
-          Effect.catchAll(() => Effect.void),
-        );
-        yield* Effect.tryPromise(() => provider.shutdown());
-      }),
+      forceFlush: Effect.tryPromise(() => processor.forceFlush()),
     });
   });

src/runtime.ts

@@ -2,7 +2,13 @@ import { readFile } from "node:fs/promises";
 import { homedir } from "node:os";
 import { join } from "node:path";
 
-import { Data, Effect, Schema } from "effect";
+import {
+  Data,
+  Effect,
+  Schema,
+  SynchronizedRef,
+  type SynchronizedRef as SynchronizedRefType,
+} from "effect";
 
 import { createLangfuseClient, type LangfuseClient } from "./langfuse.js";
 
@@ -21,6 +27,10 @@ class MissingLangfuseCredentials extends Data.TaggedError(
   "MissingLangfuseCredentials",
 )<{ readonly message: string }> {}
 
+class ChangedLangfuseConfiguration extends Data.TaggedError(
+  "ChangedLangfuseConfiguration",
+)<{ readonly message: string }> {}
+
 const loadLangfuseCredentials = Effect.gen(function* () {
   const publicKey = process.env.LANGFUSE_PUBLIC_KEY;
   const secretKey = process.env.LANGFUSE_SECRET_KEY;
@@ -77,10 +87,34 @@ const loadLangfuseCredentials = Effect.gen(function* () {
   return credentials;
 });
 
+/**
+ * opencode disposes and re-creates plugin instances inside the same process
+ * (e.g. when the effective config changes) while the OTel tracer provider is
+ * registered process-wide and cannot be registered a second time. Spans are
+ * exported by that first provider, so a re-created instance has to keep using
+ * the client that owns it: flushing or shutting down a second provider would
+ * silently drop the session.
+ *
+ * Symbol.for keeps the owner stable if the package is loaded more than once.
+ */
+type SharedClientState =
+  | { readonly _tag: "Empty" }
+  | {
+      readonly _tag: "Ready";
+      readonly client: LangfuseClient;
+      readonly key: string;
+    };
+
+declare global {
+  var langfuseOpencodeRuntimeState:
+    | SynchronizedRefType.SynchronizedRef<SharedClientState>
+    | undefined;
+}
+
 export const createLangfuseRuntime = (input: { opencodeVersion?: string }) =>
   Effect.gen(function* () {
     const credentials = yield* loadLangfuseCredentials;
-    const client = yield* createLangfuseClient({
+    const clientInput = {
       publicKey: credentials.publicKey,
       secretKey: credentials.secretKey,
       baseUrl:
@@ -95,15 +129,38 @@ export const createLangfuseRuntime = (input: { opencodeVersion?: string }) =>
       userId: credentials.userId ?? process.env.LANGFUSE_USER_ID,
       serviceName: credentials.serviceName ?? process.env.LANGFUSE_SERVICE_NAME,
       opencodeVersion: input.opencodeVersion,
-    });
+    } satisfies Parameters<typeof createLangfuseClient>[0];
+    const cacheKey = JSON.stringify(clientInput);
+    const sharedClientState =
+      globalThis.langfuseOpencodeRuntimeState ??
+      Effect.runSync(
+        SynchronizedRef.make<SharedClientState>({ _tag: "Empty" }),
+      );
+    globalThis.langfuseOpencodeRuntimeState = sharedClientState;
 
-    return client;
-  });
+    return yield* SynchronizedRef.modifyEffect(sharedClientState, (state) =>
+      Effect.gen(function* () {
+        if (state._tag === "Ready") {
+          if (state.key !== cacheKey) {
+            return yield* Effect.fail(
+              new ChangedLangfuseConfiguration({
+                message:
+                  "Langfuse configuration changed while the process-wide OpenTelemetry provider is active; restart OpenCode to apply it",
+              }),
+            );
+          }
 
-export const createShutdownOnce = (langfuse: LangfuseClient) => {
-  let shutdownPromise: Promise<void> | undefined;
+          return [state.client, state] as const;
+        }
 
-  return () => {
-    return (shutdownPromise ??= Effect.runPromise(langfuse.shutdown));
-  };
-};
+        const client = yield* createLangfuseClient(clientInput);
+        const nextState = {
+          _tag: "Ready",
+          client,
+          key: cacheKey,
+        } as const satisfies SharedClientState;
+
+        return [client, nextState] as const;
+      }),
+    );
+  });

src/v1.ts

@@ -7,7 +7,7 @@ import {
   type ToolDefinition,
 } from "./langfuse.js";
 import { OpencodeClientService } from "./opencode.js";
-import { createLangfuseRuntime, createShutdownOnce } from "./runtime.js";
+import { createLangfuseRuntime } from "./runtime.js";
 import {
   McpContentSchema,
   McpToolResultSchema,
@@ -47,7 +47,7 @@ const refreshSessionHistory = (sessionID: string) =>
     langfuse.setSessionHistory(sessionID, buildSessionHistory(response.data));
   });
 
-const eventHook = (event: OpencodeEvent, shutdown?: () => Promise<void>) =>
+const eventHook = (event: OpencodeEvent) =>
   Effect.gen(function* () {
     const langfuse = yield* LangfuseClientService;
 
@@ -73,12 +73,9 @@ const eventHook = (event: OpencodeEvent, shutdown?: () => Promise<void>) =>
     if (event.type === "server.instance.disposed") {
       finalizeSessionTracing();
 
-      if (shutdown) {
-        yield* Effect.tryPromise({
-          try: () => shutdown(),
-          catch: (error) => error,
-        });
-      }
+      // The tracer provider is process-wide and cannot be registered twice,
+      // so an instance disposal must not tear it down (see runtime.ts).
+      yield* langfuse.forceFlush;
     }
 
     if (event.type === "session.created" || event.type === "session.updated") {
@@ -341,13 +338,13 @@ const normalizeToolResult = (tool: string, output: unknown) => {
 
 const main = Effect.gen(function* () {
   const opencode = yield* OpencodeClientService;
+  const disableTracing = (error: { readonly message: string }) =>
+    log("warn", `[Tracing disabled] ${error.message}`).pipe(
+      Effect.as(undefined),
+    );
 
   const langfuse = yield* createLangfuseRuntime({}).pipe(
-    Effect.catchTag("MissingLangfuseCredentials", (error) =>
-      log("warn", `[Tracing disabled] ${error.message}`).pipe(
-        Effect.as(undefined),
-      ),
-    ),
+    Effect.catchTag("MissingLangfuseCredentials", disableTracing),
   );
 
   if (!langfuse) {
@@ -367,7 +364,6 @@ const main = Effect.gen(function* () {
     langfuse.endActiveTurnObservations();
     langfuse.clearTraceState();
   });
-  const shutdownOnce = createShutdownOnce(langfuse);
   const toolDefinitions = new Map<string, Promise<ToolDefinition[]>>();
 
   const runHook = (
@@ -401,17 +397,10 @@ const main = Effect.gen(function* () {
     dispose: () =>
       runHook(
         "dispose",
-        finalizeTracing.pipe(
-          Effect.zipRight(
-            Effect.tryPromise({
-              try: () => shutdownOnce(),
-              catch: (error) => error,
-            }),
-          ),
-        ),
+        finalizeTracing.pipe(Effect.zipRight(langfuse.forceFlush)),
       ),
 
-    event: ({ event }) => runHook("event", eventHook(event, shutdownOnce)),
+    event: ({ event }) => runHook("event", eventHook(event)),
 
     "chat.message": (input, output) =>
       runHook(

src/v2.ts

@@ -2,18 +2,19 @@ import type { Plugin } from "@opencode/plugin";
 import { Effect } from "effect";
 
 import type { ToolDefinition } from "./langfuse.js";
-import { createLangfuseRuntime, createShutdownOnce } from "./runtime.js";
+import { createLangfuseRuntime } from "./runtime.js";
 
 const LangfusePlugin = {
   id: "langfuse.observability",
   async setup(ctx) {
+    const disableTracing = (error: { readonly message: string }) =>
+      Effect.sync(() => {
+        console.warn(`[Langfuse tracing disabled] ${error.message}`);
+      }).pipe(Effect.as(undefined));
+
     const langfuse = await Effect.runPromise(
       createLangfuseRuntime({ opencodeVersion: ctx.app.version }).pipe(
-        Effect.catchTag("MissingLangfuseCredentials", (error) =>
-          Effect.sync(() => {
-            console.warn(`[Langfuse tracing disabled] ${error.message}`);
-          }).pipe(Effect.as(undefined)),
-        ),
+        Effect.catchTag("MissingLangfuseCredentials", disableTracing),
       ),
     );
     if (!langfuse) {
@@ -22,7 +23,6 @@ const LangfusePlugin = {
 
     const abort = new AbortController();
     const registrations: { dispose: () => Promise<void> }[] = [];
-    const shutdown = createShutdownOnce(langfuse);
     const userMessageIDs = new Map<string, string>();
     const generationDetails = new Map<
       string,
@@ -298,7 +298,9 @@ const LangfusePlugin = {
       langfuse.endActiveGenerationSteps();
       langfuse.endActiveTurnObservations();
       langfuse.clearTraceState();
-      await shutdown();
+      // The tracer provider is process-wide and cannot be registered twice,
+      // so an instance disposal must not tear it down (see runtime.ts).
+      await Effect.runPromise(langfuse.forceFlush);
     };
   },
 } satisfies Plugin.Plugin;

test/integration/runtime.test.ts

@@ -13,7 +13,11 @@ const originalEnvironment = {
 let temporaryHome: string | undefined;
 
 afterEach(async () => {
+  vi.doUnmock("@opentelemetry/api");
+  vi.doUnmock("../../src/langfuse.js");
+  vi.resetModules();
   vi.unstubAllGlobals();
+  globalThis.langfuseOpencodeRuntimeState = undefined;
   if (originalEnvironment.home === undefined) {
     delete process.env.HOME;
   } else {
@@ -37,6 +41,50 @@ afterEach(async () => {
 });
 
 describe("Langfuse runtime", () => {
+  test("creates one shared client for concurrent initialization", async () => {
+    const client = { id: "shared-client" };
+    let clientCreations = 0;
+
+    vi.doMock("../../src/langfuse.js", () => ({
+      createLangfuseClient: () =>
+        Effect.gen(function* () {
+          clientCreations += 1;
+          yield* Effect.sleep("10 millis");
+          return client;
+        }),
+    }));
+
+    process.env.LANGFUSE_PUBLIC_KEY = "pk-test";
+    process.env.LANGFUSE_SECRET_KEY = "sk-test";
+    const { createLangfuseRuntime } = await import("../../src/runtime.js");
+
+    const clients = await Effect.runPromise(
+      Effect.all([createLangfuseRuntime({}), createLangfuseRuntime({})], {
+        concurrency: "unbounded",
+      }),
+    );
+
+    expect(clientCreations).toBe(1);
+    expect(clients[0]).toBe(client);
+    expect(clients[1]).toBe(client);
+
+    vi.resetModules();
+    const reloadedRuntime = await import("../../src/runtime.js");
+    expect(
+      await Effect.runPromise(reloadedRuntime.createLangfuseRuntime({})),
+    ).toBe(client);
+    expect(clientCreations).toBe(1);
+
+    process.env.LANGFUSE_SECRET_KEY = "sk-changed";
+    const error = await Effect.runPromise(
+      Effect.flip(createLangfuseRuntime({})),
+    );
+
+    expect(error).toMatchObject({
+      _tag: "ChangedLangfuseConfiguration",
+    });
+  });
+
   test("does not accept empty environment credentials", async () => {
     vi.stubGlobal("__PLUGIN_VERSION__", "test");
     const { createLangfuseRuntime } = await import("../../src/runtime.js");