fix: keep exporting spans after OpenCode instance disposal

src/runtime.ts

@@ -1,3 +1,4 @@
+import { trace } from "@opentelemetry/api";
 import { readFile } from "node:fs/promises";
 import { homedir } from "node:os";
 import { join } from "node:path";
@@ -77,9 +78,52 @@ 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.
+ *
+ * The cache is keyed by the credentials and by the provider that is actually
+ * registered, so a new client is still created when the global provider is
+ * replaced (for example after trace.disable()).
+ */
+type DelegateHolder = { getDelegate: () => unknown };
+
+// The tracer provider returned by the API wraps the registered one; the type
+// does not expose the accessor, so narrow it structurally instead.
+const hasDelegate = (value: unknown): value is DelegateHolder =>
+  typeof value === "object" &&
+  value !== null &&
+  "getDelegate" in value &&
+  typeof value.getDelegate === "function";
+
+const registeredProvider = () => {
+  const global: unknown = trace.getTracerProvider();
+
+  return hasDelegate(global) ? global.getDelegate() : undefined;
+};
+
+let sharedClient: LangfuseClient | undefined;
+let sharedClientKey: string | undefined;
+let sharedClientProvider: unknown;
+
 export const createLangfuseRuntime = (input: { opencodeVersion?: string }) =>
   Effect.gen(function* () {
     const credentials = yield* loadLangfuseCredentials;
+    const cacheKey = `${credentials.publicKey}@${credentials.baseUrl ?? ""}`;
+
+    if (
+      sharedClient !== undefined &&
+      sharedClientKey === cacheKey &&
+      sharedClientProvider === registeredProvider()
+    ) {
+      return sharedClient;
+    }
+
+    const providerBefore = registeredProvider();
     const client = yield* createLangfuseClient({
       publicKey: credentials.publicKey,
       secretKey: credentials.secretKey,
@@ -97,13 +141,11 @@ export const createLangfuseRuntime = (input: { opencodeVersion?: string }) =>
       opencodeVersion: input.opencodeVersion,
     });
 
+    if (registeredProvider() !== providerBefore) {
+      sharedClient = client;
+      sharedClientKey = cacheKey;
+      sharedClientProvider = registeredProvider();
+    }
+
     return client;
   });
-
-export const createShutdownOnce = (langfuse: LangfuseClient) => {
-  let shutdownPromise: Promise<void> | undefined;
-
-  return () => {
-    return (shutdownPromise ??= Effect.runPromise(langfuse.shutdown));
-  };
-};

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") {
@@ -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 = (
@@ -404,14 +400,14 @@ const main = Effect.gen(function* () {
         finalizeTracing.pipe(
           Effect.zipRight(
             Effect.tryPromise({
-              try: () => shutdownOnce(),
+              try: () => Effect.runPromise(langfuse.forceFlush),
               catch: (error) => error,
             }),
           ),
         ),
       ),
 
-    event: ({ event }) => runHook("event", eventHook(event, shutdownOnce)),
+    event: ({ event }) => runHook("event", eventHook(event)),
 
     "chat.message": (input, output) =>
       runHook(

src/v2.ts

@@ -2,7 +2,7 @@ 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",
@@ -22,7 +22,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 +297,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/v1.test.ts

@@ -2563,6 +2563,33 @@ describe("built plugin", { concurrent: false }, () => {
     ]);
   });
 
+  test("keeps exporting spans after the instance is disposed and re-created", async () => {
+    // opencode disposes and re-creates plugin instances inside the same process
+    // (for example when the effective config changes) while the OTel tracer
+    // provider is registered process-wide. Exporting must survive that.
+    const runTurn = async (sessionID: string) => {
+      await sendUserMessage({
+        sessionID,
+        messageID: `${sessionID}-user`,
+        text: "Trace across a re-created instance",
+        started: startedAt,
+      });
+      const { requests: sessionRequests } = await flushSession(sessionID);
+      return sessionRequests;
+    };
+
+    expect(await runTurn("dispose-keep-export-before")).not.toEqual([]);
+
+    // What OpenCode does per instance on disposal. Deliberately no
+    // trace.disable() here: resetting the global provider hides the regression.
+    await hooks.dispose?.();
+
+    // OpenCode re-creates the plugin instance in the same process.
+    hooks = await createHooks(collectorBaseUrl);
+
+    expect(await runTurn("dispose-keep-export-after")).not.toEqual([]);
+  }, 15_000);
+
   test("keeps the new user message when the history refresh fails", async () => {
     // A snapshot from earlier in the same busy period does not contain a
     // request that arrives afterwards. If the refresh that would pick it up