From 48ab04063589b0b0dc9bed09b1a4bd5076616930 Mon Sep 17 00:00:00 2001 From: Ha02hen <824702599@qq.com> Date: Mon, 21 Sep 2026 23:21:25 +0800 Subject: [PATCH 1/2] Keep exporting spans after OpenCode instance disposal --- src/runtime.ts | 58 ++++++++++++++++++++++++++++++++----- src/v1.ts | 18 +++++------- src/v2.ts | 7 +++-- test/integration/v1.test.ts | 27 +++++++++++++++++ 4 files changed, 88 insertions(+), 22 deletions(-) diff --git a/src/runtime.ts b/src/runtime.ts index dde4670..1b3e918 100644 --- a/src/runtime.ts +++ b/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 | undefined; - - return () => { - return (shutdownPromise ??= Effect.runPromise(langfuse.shutdown)); - }; -}; diff --git a/src/v1.ts b/src/v1.ts index 52567ca..bdb4d26 100644 --- a/src/v1.ts +++ b/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) => +const eventHook = (event: OpencodeEvent) => Effect.gen(function* () { const langfuse = yield* LangfuseClientService; @@ -73,12 +73,9 @@ const eventHook = (event: OpencodeEvent, shutdown?: () => Promise) => 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>(); 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( diff --git a/src/v2.ts b/src/v2.ts index 27b7f49..8712f67 100644 --- a/src/v2.ts +++ b/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 }[] = []; - const shutdown = createShutdownOnce(langfuse); const userMessageIDs = new Map(); 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; diff --git a/test/integration/v1.test.ts b/test/integration/v1.test.ts index d87fe97..11c3578 100644 --- a/test/integration/v1.test.ts +++ b/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 From a8d699e9e349225be2b30f46ae3b64005162770e Mon Sep 17 00:00:00 2001 From: Ben Bachem <10088265+bezbac@users.noreply.github.com> Date: Tue, 22 Sep 2026 14:26:24 +0200 Subject: [PATCH 2/2] Improve code --- src/langfuse.ts | 20 +----- src/runtime.ts | 105 ++++++++++++++++++------------- src/v1.ts | 19 ++---- src/v2.ts | 11 ++-- test/integration/runtime.test.ts | 48 ++++++++++++++ test/integration/v1.test.ts | 15 +++++ test/integration/v2.test.ts | 36 ++++++++++- tsconfig.eslint.json | 7 ++- vitest.config.ts | 7 +++ 9 files changed, 182 insertions(+), 86 deletions(-) create mode 100644 vitest.config.ts diff --git a/src/langfuse.ts b/src/langfuse.ts index e1fc62f..b909fd2 100644 --- a/src/langfuse.ts +++ b/src/langfuse.ts @@ -17,19 +17,16 @@ import { PLUGIN_VERSION } from "./version.js"; export class LangfuseClient { readonly baseUrl: string; readonly forceFlush: Effect.Effect; - readonly shutdown: Effect.Effect; private readonly traceState: LangfuseTraceState; constructor(input: { baseUrl: string; traceState: LangfuseTraceState; forceFlush: Effect.Effect; - shutdown: Effect.Effect; }) { this.baseUrl = input.baseUrl; this.traceState = input.traceState; this.forceFlush = input.forceFlush; - this.shutdown = input.shutdown; } clearTraceState() { @@ -1797,8 +1794,6 @@ export const createLangfuseClient = (input: { makeAppRootSpanProcessor(traceState.tracerName), ], }); - let isShutdown = false; - yield* Effect.sync(() => { provider.register(); }); @@ -1806,19 +1801,6 @@ export const createLangfuseClient = (input: { 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()), }); }); diff --git a/src/runtime.ts b/src/runtime.ts index 1b3e918..dd842ad 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -1,9 +1,14 @@ -import { trace } from "@opentelemetry/api"; 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"; @@ -22,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; @@ -86,45 +95,26 @@ const loadLangfuseCredentials = Effect.gen(function* () { * 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()). + * Symbol.for keeps the owner stable if the package is loaded more than once. */ -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; +type SharedClientState = + | { readonly _tag: "Empty" } + | { + readonly _tag: "Ready"; + readonly client: LangfuseClient; + readonly key: string; + }; + +declare global { + var langfuseOpencodeRuntimeState: + | SynchronizedRefType.SynchronizedRef + | undefined; +} 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({ + const clientInput = { publicKey: credentials.publicKey, secretKey: credentials.secretKey, baseUrl: @@ -139,13 +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, - }); - - if (registeredProvider() !== providerBefore) { - sharedClient = client; - sharedClientKey = cacheKey; - sharedClientProvider = registeredProvider(); - } - - return client; + } satisfies Parameters[0]; + const cacheKey = JSON.stringify(clientInput); + const sharedClientState = + globalThis.langfuseOpencodeRuntimeState ?? + Effect.runSync( + SynchronizedRef.make({ _tag: "Empty" }), + ); + globalThis.langfuseOpencodeRuntimeState = sharedClientState; + + 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", + }), + ); + } + + return [state.client, state] as const; + } + + const client = yield* createLangfuseClient(clientInput); + const nextState = { + _tag: "Ready", + client, + key: cacheKey, + } as const satisfies SharedClientState; + + return [client, nextState] as const; + }), + ); }); diff --git a/src/v1.ts b/src/v1.ts index bdb4d26..c58763b 100644 --- a/src/v1.ts +++ b/src/v1.ts @@ -338,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) { @@ -397,14 +397,7 @@ const main = Effect.gen(function* () { dispose: () => runHook( "dispose", - finalizeTracing.pipe( - Effect.zipRight( - Effect.tryPromise({ - try: () => Effect.runPromise(langfuse.forceFlush), - catch: (error) => error, - }), - ), - ), + finalizeTracing.pipe(Effect.zipRight(langfuse.forceFlush)), ), event: ({ event }) => runHook("event", eventHook(event)), diff --git a/src/v2.ts b/src/v2.ts index 8712f67..4a088fa 100644 --- a/src/v2.ts +++ b/src/v2.ts @@ -7,13 +7,14 @@ 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) { diff --git a/test/integration/runtime.test.ts b/test/integration/runtime.test.ts index d27cd58..f847d3a 100644 --- a/test/integration/runtime.test.ts +++ b/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"); diff --git a/test/integration/v1.test.ts b/test/integration/v1.test.ts index 11c3578..d5010d8 100644 --- a/test/integration/v1.test.ts +++ b/test/integration/v1.test.ts @@ -515,6 +515,7 @@ const disposeHooks = async () => { await hooks.dispose?.(); } finally { trace.disable(); + globalThis.langfuseOpencodeRuntimeState = undefined; } }; @@ -2590,6 +2591,19 @@ describe("built plugin", { concurrent: false }, () => { expect(await runTurn("dispose-keep-export-after")).not.toEqual([]); }, 15_000); + test("rejects a re-created instance when its configuration changed", async () => { + await hooks.dispose?.(); + process.env.LANGFUSE_SECRET_KEY = "sk-changed"; + + try { + await expect(createHooks(collectorBaseUrl)).rejects.toThrow( + "Langfuse configuration changed while the process-wide OpenTelemetry provider is active; restart OpenCode to apply it", + ); + } finally { + process.env.LANGFUSE_SECRET_KEY = "sk-test"; + } + }); + 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 @@ -2778,5 +2792,6 @@ describe("built plugin", { concurrent: false }, () => { await expect(hooks.dispose?.()).resolves.toBeUndefined(); hooksDisposed = true; trace.disable(); + globalThis.langfuseOpencodeRuntimeState = undefined; }); }); diff --git a/test/integration/v2.test.ts b/test/integration/v2.test.ts index 89e9b72..a67fa05 100644 --- a/test/integration/v2.test.ts +++ b/test/integration/v2.test.ts @@ -26,7 +26,7 @@ const runtime = vi.hoisted(() => ({ endActiveTurnObservations: vi.fn(), clearSessionTraceState: vi.fn(), clearTraceState: vi.fn(), - shutdown: vi.fn(), + forceFlush: vi.fn(), })); vi.mock("../../src/runtime.js", async () => { @@ -37,10 +37,9 @@ vi.mock("../../src/runtime.js", async () => { runtime.createLangfuseRuntime(input); return Effect.succeed({ ...runtime, - forceFlush: Effect.void, + forceFlush: Effect.sync(runtime.forceFlush), }); }, - createShutdownOnce: () => runtime.shutdown, }; }); @@ -139,6 +138,37 @@ describe("OpenCode 2 package entrypoint", () => { expect(runtime.traceGeneration).not.toHaveBeenCalled(); }); + test("keeps the shared runtime alive when an instance is disposed and re-created", async () => { + const registration = { dispose: vi.fn(() => Promise.resolve()) }; + const contextInput: unknown = { + app: { version: "2.0.4" }, + session: { hook: vi.fn(() => Promise.resolve(registration)) }, + tool: { hook: vi.fn(() => Promise.resolve(registration)) }, + event: { + subscribe: () => ({ + async *[Symbol.asyncIterator]() { + await Promise.resolve(); + yield* []; + }, + }), + }, + }; + const context = Schema.decodeUnknownSync( + Schema.declare( + (input): input is Parameters[0] => + typeof input === "object" && input !== null, + ), + )(contextInput); + + const firstCleanup = await SourcePlugin.setup(context); + await firstCleanup?.(); + const secondCleanup = await SourcePlugin.setup(context); + await secondCleanup?.(); + + expect(runtime.createLangfuseRuntime).toHaveBeenCalledTimes(2); + expect(runtime.forceFlush).toHaveBeenCalledTimes(2); + }); + test("traces a complete session with prompt, text, reasoning, and tools", async () => { let prompt: | ((input: { diff --git a/tsconfig.eslint.json b/tsconfig.eslint.json index 899ad5d..90755bd 100644 --- a/tsconfig.eslint.json +++ b/tsconfig.eslint.json @@ -1,4 +1,9 @@ { "extends": "./tsconfig.test.json", - "include": ["src/**/*.ts", "test/**/*.ts", "tsdown.config.ts"] + "include": [ + "src/**/*.ts", + "test/**/*.ts", + "tsdown.config.ts", + "vitest.config.ts" + ] } diff --git a/vitest.config.ts b/vitest.config.ts new file mode 100644 index 0000000..51a17b5 --- /dev/null +++ b/vitest.config.ts @@ -0,0 +1,7 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + fileParallelism: false, + }, +});