From 7d47e6ac6709c152966d7b7d5a4b4c9f35141ff1 Mon Sep 17 00:00:00 2001 From: zenor0 Date: Sun, 6 Sep 2026 14:14:10 +0800 Subject: [PATCH 1/2] feat(tracing): attach Codex turns to external parent spans Accept LANGFUSE_CODEX_TRACEPARENT to attach the complete Codex observation tree to a process-scoped W3C parent span. Honor upstream sampling and preserve application trace ownership. Preserve standalone sampling and deterministic trace IDs. Reuse the existing fixtures with a real Langfuse processor for local regression coverage. Keep npm scripts unchanged. Validation: 51 local tests, full lint, and an npm pack dry run pass. Generated bundles remain untracked. --- README.md | 63 ++++++++--- package.json | 1 + plugins/tracing/src/index.ts | 8 +- plugins/tracing/src/instrumentation.ts | 20 +++- plugins/tracing/src/parent-context.ts | 27 +++++ plugins/tracing/src/trace.ts | 63 +++++++---- plugins/tracing/test/instrumentation.test.ts | 73 ++++++++++++ plugins/tracing/test/parent-context.test.ts | 40 +++++++ .../tracing/test/trace-seed-failure.test.ts | 21 ++++ plugins/tracing/test/trace.test.ts | 105 +++++++++++++++++- pnpm-lock.yaml | 3 + 11 files changed, 378 insertions(+), 46 deletions(-) create mode 100644 plugins/tracing/src/parent-context.ts create mode 100644 plugins/tracing/test/instrumentation.test.ts create mode 100644 plugins/tracing/test/parent-context.test.ts diff --git a/README.md b/README.md index 0cbf536..5f368be 100644 --- a/README.md +++ b/README.md @@ -8,11 +8,11 @@ Once enabled, every Codex turn shows up in Langfuse as a trace you can inspect, After each Codex turn, the plugin reads the session's rollout transcript and uploads it to Langfuse as a [trace](https://langfuse.com/docs/observability/data-model). The structure mirrors how Codex actually works: -- **Turn** (`Codex Turn`, an [agent observation](https://langfuse.com/docs/observability/features/observation-types)) — one trace per turn, from your prompt to the final answer. +- **Turn** (`Codex Turn`, an [agent observation](https://langfuse.com/docs/observability/features/observation-types)) — one trace per turn by default, or a child of an existing application span in [attached mode](#attach-to-an-existing-trace), from your prompt to the final answer. - **Generations** — one per model response within the turn, named `LLM` (or `LLM Subagent` inside subagent threads), with the model recorded on the observation plus reasoning, assistant text, the tool calls it requested, and token usage. - **Tool calls** — shell commands, `apply_patch`, `spawn_agent`, MCP tools, web searches, etc., each with its input, output, and error status. MCP calls are named `server.tool`, and failed commands are flagged as errors. - **Subagents** — subagent threads are resolved from their own rollout files and nested under the spawning turn as `Codex Subagent Turn`. -- **Sessions** — all turns from one Codex session are grouped via the Codex thread id, so you can replay the whole session in Langfuse's [Sessions](https://langfuse.com/docs/observability/features/sessions) view. +- **Sessions** — in standalone mode, all turns from one Codex session are grouped via the Codex thread id, so you can replay the whole session in Langfuse's [Sessions](https://langfuse.com/docs/observability/features/sessions) view. Interrupted turns (where you cancel mid-response) are still uploaded and flagged as interrupted. @@ -101,20 +101,21 @@ codex plugin list ## Environment variables -| Variable | Required | Default | Description | -| ------------------------------------------------------------- | -------- | ---------------------------- | -------------------------------------------------------------------- | -| `TRACE_TO_LANGFUSE` | Yes | `false` | Set to `"true"` to enable tracing | -| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_CODEX_PUBLIC_KEY` | Yes | — | Langfuse public key (`pk-lf-...`) | -| `LANGFUSE_SECRET_KEY` / `LANGFUSE_CODEX_SECRET_KEY` | Yes | — | Langfuse secret key (`sk-lf-...`) | -| `LANGFUSE_BASE_URL` / `LANGFUSE_CODEX_BASE_URL` | No | `https://cloud.langfuse.com` | Langfuse host / data region | -| `LANGFUSE_TRACING_ENVIRONMENT` / `LANGFUSE_CODEX_ENVIRONMENT` | No | — | Environment label for the traces (e.g. `production`) | -| `LANGFUSE_CODEX_USER_ID` | No | Codex auth email, if found | Attach a user id to all traces | -| `LANGFUSE_CODEX_TAGS` | No | — | Tags for all traces (JSON array or comma-separated) | -| `LANGFUSE_CODEX_METADATA` | No | — | JSON object of metadata to attach to all traces | -| `LANGFUSE_CODEX_TRACE_SEED` | No | — | Derive deterministic trace ids ([details](#deterministic-trace-ids)) | -| `LANGFUSE_CODEX_MAX_CHARS` | No | `20000` | Truncate inputs/outputs longer than this many characters | -| `LANGFUSE_CODEX_DEBUG` | No | `false` | Set to `"true"` for verbose logging to stderr | -| `LANGFUSE_CODEX_FAIL_ON_ERROR` | No | `false` | Set to `"true"` to make hook upload errors fail the hook | +| Variable | Required | Default | Description | +| ------------------------------------------------------------- | -------- | ---------------------------- | ------------------------------------------------------------------------------------- | +| `TRACE_TO_LANGFUSE` | Yes | `false` | Set to `"true"` to enable tracing | +| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_CODEX_PUBLIC_KEY` | Yes | — | Langfuse public key (`pk-lf-...`) | +| `LANGFUSE_SECRET_KEY` / `LANGFUSE_CODEX_SECRET_KEY` | Yes | — | Langfuse secret key (`sk-lf-...`) | +| `LANGFUSE_BASE_URL` / `LANGFUSE_CODEX_BASE_URL` | No | `https://cloud.langfuse.com` | Langfuse host / data region | +| `LANGFUSE_TRACING_ENVIRONMENT` / `LANGFUSE_CODEX_ENVIRONMENT` | No | — | Environment label for the traces (e.g. `production`) | +| `LANGFUSE_CODEX_USER_ID` | No | Codex auth email, if found | Attach a user id to all traces | +| `LANGFUSE_CODEX_TAGS` | No | — | Tags for all traces (JSON array or comma-separated) | +| `LANGFUSE_CODEX_METADATA` | No | — | JSON object of metadata to attach to all traces | +| `LANGFUSE_CODEX_TRACEPARENT` | No | — | Attach turns beneath an existing W3C parent ([details](#attach-to-an-existing-trace)) | +| `LANGFUSE_CODEX_TRACE_SEED` | No | — | Derive deterministic trace ids ([details](#deterministic-trace-ids)) | +| `LANGFUSE_CODEX_MAX_CHARS` | No | `20000` | Truncate inputs/outputs longer than this many characters | +| `LANGFUSE_CODEX_DEBUG` | No | `false` | Set to `"true"` for verbose logging to stderr | +| `LANGFUSE_CODEX_FAIL_ON_ERROR` | No | `false` | Set to `"true"` to make hook upload errors fail the hook | ### Data regions @@ -125,6 +126,36 @@ codex plugin list | 🇯🇵 Japan | `https://jp.cloud.langfuse.com` | | ⚕️ HIPAA | `https://hipaa.cloud.langfuse.com` | +## Attach to an existing trace + +If another application launches a dedicated Codex process as part of an existing agent run, pass that run's [W3C Trace Context](https://www.w3.org/TR/trace-context/) in `LANGFUSE_CODEX_TRACEPARENT`: + +```bash +TRACE_ID="0af7651916cd43dd8448eb211c80319c" +AGENT_RUN_SPAN_ID="b7ad6b7169203331" + +LANGFUSE_CODEX_TRACEPARENT="00-${TRACE_ID}-${AGENT_RUN_SPAN_ID}-01" \ + codex exec "your prompt" +``` + +The plugin then preserves its detailed observation tree while attaching it to the real application span: + +```text +Master Agent Run +`-- Codex Turn + |-- LLM + | `-- exec_command + `-- Codex Subagent Turn +``` + +Use this mode only when one Master Agent Run owns one Codex/App Server process. The environment variable is process-scoped, so every top-level Codex turn in that process is attached directly to the same parent span. A long-lived App Server shared by unrelated runs requires request-level context propagation, which Codex Stop hooks do not currently expose. + +`LANGFUSE_CODEX_TRACEPARENT` is intentionally runtime-only: set it on the child Codex process, not in `langfuse.json` or a persistent shell profile. The plugin does not read the unscoped `TRACEPARENT` variable or `tracestate`. If both `LANGFUSE_CODEX_TRACEPARENT` and `LANGFUSE_CODEX_TRACE_SEED` are set, the explicit parent takes precedence. + +The Master application and this plugin must export to the same Langfuse project for the observations to appear in one trace. In attached mode, the Master application owns trace-level name, session, user, tags, and metadata; the plugin still records all Codex observation metadata, inputs, outputs, reasoning, usage, and errors. + +The final trace flags are honored. A parent ending in `-01` exports the Codex observations; a parent ending in `-00` exports none of them, while completed turns are still written to the dedup sidecar as processed. Attached mode treats this upstream decision as authoritative even when `OTEL_TRACES_SAMPLER` is set. Standalone mode continues to honor the standard OpenTelemetry sampler environment variables. Invalid values fall back to `LANGFUSE_CODEX_TRACE_SEED` or an auto-generated trace; set `LANGFUSE_CODEX_FAIL_ON_ERROR=true` to reject them instead. + ## Deterministic trace ids By default, trace ids are auto-generated, and an external system (a CI harness, benchmark runner, or dataset-experiment service) that runs `codex exec` headlessly has to poll the Langfuse API to discover the trace a run produced. Set `LANGFUSE_CODEX_TRACE_SEED` (or `trace_seed` in `langfuse.json`) to make trace ids predictable instead: diff --git a/package.json b/package.json index 34db8db..1ba3c90 100644 --- a/package.json +++ b/package.json @@ -32,6 +32,7 @@ "lint": "pnpm run format:check && pnpm run lint:tsc && pnpm run build" }, "dependencies": { + "@langfuse/core": "^5.4.1", "@langfuse/otel": "^5.4.1", "@langfuse/tracing": "^5.4.1", "@opentelemetry/api": "^1.9.0", diff --git a/plugins/tracing/src/index.ts b/plugins/tracing/src/index.ts index b32a583..67187be 100644 --- a/plugins/tracing/src/index.ts +++ b/plugins/tracing/src/index.ts @@ -1,5 +1,6 @@ import { getConfig } from "./config.js"; import { setupInstrumentation } from "./instrumentation.js"; +import { readExternalParentSpanContext } from "./parent-context.js"; import { convertRollout } from "./trace.js"; import type { HookInput } from "./types.js"; import { debugLog, readStdin, setDebug } from "./utils.js"; @@ -44,9 +45,12 @@ export async function runHook(): Promise { return; } - const instrumentation = setupInstrumentation(config); + const parentSpanContext = readExternalParentSpanContext(process.env, config.fail_on_error); + const instrumentation = setupInstrumentation(config, { + attached: parentSpanContext != null, + }); try { - await convertRollout(hookInput.transcript_path, { config }); + await convertRollout(hookInput.transcript_path, { config, parentSpanContext }); } catch (error) { debugLog("failed to convert rollout:", error); if (config.fail_on_error) throw error; diff --git a/plugins/tracing/src/instrumentation.ts b/plugins/tracing/src/instrumentation.ts index 7ad72dd..83db63c 100644 --- a/plugins/tracing/src/instrumentation.ts +++ b/plugins/tracing/src/instrumentation.ts @@ -1,4 +1,5 @@ import { LangfuseSpanProcessor } from "@langfuse/otel"; +import { AlwaysOnSampler, ParentBasedSampler } from "@opentelemetry/sdk-trace-base"; import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; import type { Config } from "./config.js"; @@ -8,6 +9,11 @@ export type Instrumentation = { shutdown: () => Promise; }; +type InstrumentationOptions = { + /** Whether spans are children of a process-level parent supplied by the launcher. */ + attached?: boolean; +}; + /** * Configure an isolated OpenTelemetry tracer provider wired to Langfuse. * @@ -21,19 +27,29 @@ export type Instrumentation = { * is far faster than one request per span — important for the hook's timeout * budget. `shutdown()` below calls `forceFlush()` before the process exits. */ -export function setupInstrumentation(config: Config): Instrumentation { +export function setupInstrumentation( + config: Config, + options: InstrumentationOptions = {}, +): Instrumentation { const spanProcessor = new LangfuseSpanProcessor({ publicKey: config.public_key, secretKey: config.secret_key, baseUrl: config.base_url, environment: config.environment, exportMode: "batched", - // The hook only ever creates Langfuse spans, so export all of them. + // The hook only creates Langfuse spans, so export every recorded span. + // Parent-based sampling below decides whether a span is recorded at all. shouldExportSpan: () => true, }); const provider = new NodeTracerProvider({ spanProcessors: [spanProcessor], + // Attached mode treats the launcher's sampled bit as authoritative. + // Standalone mode leaves this unset so standard OTEL_TRACES_SAMPLER + // configuration keeps working exactly as it did before attached mode. + ...(options.attached + ? { sampler: new ParentBasedSampler({ root: new AlwaysOnSampler() }) } + : {}), }); provider.register(); diff --git a/plugins/tracing/src/parent-context.ts b/plugins/tracing/src/parent-context.ts new file mode 100644 index 0000000..b884a28 --- /dev/null +++ b/plugins/tracing/src/parent-context.ts @@ -0,0 +1,27 @@ +import type { SpanContext } from "@opentelemetry/api"; +import { parseTraceParent } from "@opentelemetry/core"; + +import { debugLog } from "./utils.js"; + +export const EXTERNAL_TRACEPARENT_ENV_VAR = "LANGFUSE_CODEX_TRACEPARENT"; + +/** Read the transient W3C parent context supplied by the process owner. */ +export function readExternalParentSpanContext( + env: Readonly>, + failOnError: boolean, +): SpanContext | undefined { + const value = env[EXTERNAL_TRACEPARENT_ENV_VAR]; + if (value === undefined) return undefined; + + const parsed = parseTraceParent(value); + if (parsed) { + return { ...parsed, isRemote: true }; + } + + const error = new Error(`${EXTERNAL_TRACEPARENT_ENV_VAR} must be a valid W3C traceparent value`); + debugLog( + `invalid ${EXTERNAL_TRACEPARENT_ENV_VAR}; falling back to trace_seed or an auto-generated trace`, + ); + if (failOnError) throw error; + return undefined; +} diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 24b4f0e..a397500 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -2,6 +2,7 @@ import type { Dirent } from "node:fs"; import * as fs from "node:fs/promises"; import * as path from "node:path"; +import { setLangfuseTraceIdInBaggage } from "@langfuse/core"; import { createTraceId, propagateAttributes, @@ -9,7 +10,7 @@ import { type LangfuseGenerationAttributes, type LangfuseObservation, } from "@langfuse/tracing"; -import { TraceFlags, type SpanContext } from "@opentelemetry/api"; +import { context, TraceFlags, type SpanContext } from "@opentelemetry/api"; import type { Config } from "./config.js"; import { parseSession } from "./parse.js"; @@ -208,8 +209,8 @@ async function emitTurn( config: Config; rolloutFile: string; parentObservation?: LangfuseObservation; - /** Pre-derived trace id for top-level turns (see seededTraceParent). */ - seededParent?: SpanContext; + /** External or seed-derived parent context for top-level turns. */ + parentSpanContext?: SpanContext; }, ): Promise { const clip = makeClip(ctx.config.max_chars); @@ -238,7 +239,7 @@ async function emitTurn( { asType: "agent", startTime: new Date(turn.startTime), - parentSpanContext: ctx.parentObservation?.otelSpan.spanContext() ?? ctx.seededParent, + parentSpanContext: ctx.parentObservation?.otelSpan.spanContext() ?? ctx.parentSpanContext, }, ); @@ -329,7 +330,12 @@ function emitToolCall( */ export async function convertRollout( rolloutFile: string, - options: { config: Config; parentObservation?: LangfuseObservation }, + options: { + config: Config; + parentObservation?: LangfuseObservation; + /** Process-level parent owned by the application that launched Codex. */ + parentSpanContext?: SpanContext; + }, ): Promise { const { sessionMeta, turns } = parseSession(await loadSession(rolloutFile)); debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`); @@ -356,24 +362,37 @@ export async function convertRollout( // Turn numbering stays 1-based over the full rollout (including turns // skipped by dedup above) so the derived id is stable across hook runs. - const seededParent = await seededTraceParent(options.config, sessionMeta, turnIndex + 1); + const parentSpanContext = + options.parentSpanContext ?? + (await seededTraceParent(options.config, sessionMeta, turnIndex + 1)); + const emit = () => + emitTurn(turn, sessionMeta, { + config: options.config, + rolloutFile, + parentSpanContext, + }); - await propagateAttributes( - { - sessionId: sessionMeta.sessionId, - traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", - ...(options.config.user_id ? { userId: options.config.user_id } : {}), - ...(options.config.tags ? { tags: options.config.tags } : {}), - ...(options.config.metadata ? { metadata: options.config.metadata } : {}), - }, - async () => { - await emitTurn(turn, sessionMeta, { - config: options.config, - rolloutFile, - seededParent, - }); - }, - ); + if (options.parentSpanContext) { + // The external application owns trace-level name, session, user, tags, + // and metadata. Codex observation metadata is still emitted by emitTurn. + // Carry its Langfuse claim so the SDK does not mark Codex as another app root. + const parentContext = setLangfuseTraceIdInBaggage( + context.active(), + options.parentSpanContext.traceId, + ); + await context.with(parentContext, emit); + } else { + await propagateAttributes( + { + sessionId: sessionMeta.sessionId, + traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", + ...(options.config.user_id ? { userId: options.config.user_id } : {}), + ...(options.config.tags ? { tags: options.config.tags } : {}), + ...(options.config.metadata ? { metadata: options.config.metadata } : {}), + }, + emit, + ); + } // Only mark completed turns as uploaded; an in-progress trailing turn is // re-uploaded (and finalized) on the next hook invocation. diff --git a/plugins/tracing/test/instrumentation.test.ts b/plugins/tracing/test/instrumentation.test.ts new file mode 100644 index 0000000..fa31b6c --- /dev/null +++ b/plugins/tracing/test/instrumentation.test.ts @@ -0,0 +1,73 @@ +import { context, trace, TraceFlags, type SpanContext } from "@opentelemetry/api"; +import type { ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { Config } from "../src/config.js"; +import { setupInstrumentation } from "../src/instrumentation.js"; + +const { finishedSpans } = vi.hoisted(() => ({ finishedSpans: [] as ReadableSpan[] })); + +vi.mock("@langfuse/otel", () => ({ + LangfuseSpanProcessor: class { + onStart(): void {} + onEnd(span: ReadableSpan): void { + finishedSpans.push(span); + } + async forceFlush(): Promise {} + async shutdown(): Promise {} + }, +})); + +const config: Config = { + enabled: true, + public_key: "pk-lf-test", + secret_key: "sk-lf-test", + base_url: "https://cloud.langfuse.com", + max_chars: 20_000, + debug: false, + fail_on_error: false, +}; + +const parent = (traceFlags: TraceFlags): SpanContext => ({ + traceId: "0af7651916cd43dd8448eb211c80319c", + spanId: "b7ad6b7169203331", + traceFlags, + isRemote: true, +}); + +afterEach(() => { + vi.unstubAllEnvs(); + finishedSpans.length = 0; + context.disable(); + trace.disable(); +}); + +describe("setupInstrumentation", () => { + it("makes the attached parent's sampled flag authoritative", async () => { + vi.stubEnv("OTEL_TRACES_SAMPLER", "always_off"); + const instrumentation = setupInstrumentation(config, { attached: true }); + const { startObservation } = await import("@langfuse/tracing"); + + for (const flag of [TraceFlags.SAMPLED, TraceFlags.NONE]) { + startObservation( + String(flag), + {}, + { asType: "agent", parentSpanContext: parent(flag) }, + ).end(); + } + + expect(finishedSpans.map((span) => span.name)).toEqual([String(TraceFlags.SAMPLED)]); + await instrumentation.shutdown(); + }); + + it("preserves OTEL_TRACES_SAMPLER behavior in standalone mode", async () => { + vi.stubEnv("OTEL_TRACES_SAMPLER", "always_off"); + const instrumentation = setupInstrumentation(config); + const { startObservation } = await import("@langfuse/tracing"); + + startObservation("standalone", {}, { asType: "agent" }).end(); + + expect(finishedSpans).toHaveLength(0); + await instrumentation.shutdown(); + }); +}); diff --git a/plugins/tracing/test/parent-context.test.ts b/plugins/tracing/test/parent-context.test.ts new file mode 100644 index 0000000..8bd70e7 --- /dev/null +++ b/plugins/tracing/test/parent-context.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from "vitest"; + +import { + EXTERNAL_TRACEPARENT_ENV_VAR, + readExternalParentSpanContext, +} from "../src/parent-context.js"; + +const TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; +const SPAN_ID = "b7ad6b7169203331"; + +describe("readExternalParentSpanContext", () => { + it.each([0, 1])("parses a remote context with sampled flag %s", (traceFlags) => { + expect( + readExternalParentSpanContext( + { [EXTERNAL_TRACEPARENT_ENV_VAR]: `00-${TRACE_ID}-${SPAN_ID}-0${traceFlags}` }, + false, + ), + ).toEqual({ + traceId: TRACE_ID, + spanId: SPAN_ID, + traceFlags, + isRemote: true, + }); + }); + + it("does not read the unscoped TRACEPARENT variable", () => { + expect( + readExternalParentSpanContext({ TRACEPARENT: `00-${TRACE_ID}-${SPAN_ID}-01` }, false), + ).toBeUndefined(); + }); + + it("ignores an invalid value unless fail-on-error is enabled", () => { + const env = { [EXTERNAL_TRACEPARENT_ENV_VAR]: "not-a-traceparent" }; + + expect(readExternalParentSpanContext(env, false)).toBeUndefined(); + expect(() => readExternalParentSpanContext(env, true)).toThrow( + `${EXTERNAL_TRACEPARENT_ENV_VAR} must be a valid W3C traceparent value`, + ); + }); +}); diff --git a/plugins/tracing/test/trace-seed-failure.test.ts b/plugins/tracing/test/trace-seed-failure.test.ts index cef4b1e..ec5b56e 100644 --- a/plugins/tracing/test/trace-seed-failure.test.ts +++ b/plugins/tracing/test/trace-seed-failure.test.ts @@ -3,6 +3,7 @@ import * as os from "node:os"; import * as path from "node:path"; import { fileURLToPath } from "node:url"; +import { TraceFlags } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -77,4 +78,24 @@ describe("trace seed derivation failure", () => { }), ).rejects.toThrow("derivation boom"); }); + + it("does not derive a seeded id when an external parent is present", async () => { + const dir = stageFixtures(); + const traceId = "0af7651916cd43dd8448eb211c80319c"; + + await expect( + convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { + config: { ...baseConfig, fail_on_error: true }, + parentSpanContext: { + traceId, + spanId: "b7ad6b7169203331", + traceFlags: TraceFlags.SAMPLED, + isRemote: true, + }, + }), + ).resolves.toBeUndefined(); + + const root = exporter.getFinishedSpans().find((span) => span.name === "Codex Turn"); + expect(root?.spanContext().traceId).toBe(traceId); + }); }); diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index c3f66a9..85d24ce 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -4,20 +4,29 @@ import * as os from "node:os"; import * as path from "node:path"; import { fileURLToPath } from "node:url"; +import { LangfuseSpanProcessor } from "@langfuse/otel"; +import { context, trace, TraceFlags, type SpanContext } from "@opentelemetry/api"; import { + AlwaysOnSampler, InMemorySpanExporter, + ParentBasedSampler, type ReadableSpan, - SimpleSpanProcessor, } from "@opentelemetry/sdk-trace-base"; import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; -import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, onTestFinished } from "vitest"; import type { Config } from "../src/config.js"; -import { convertRollout } from "../src/trace.js"; +import { convertRollout as convert } from "../src/trace.js"; const exporter = new InMemorySpanExporter(); +const processor = new LangfuseSpanProcessor({ exporter, shouldExportSpan: () => true }); let provider: NodeTracerProvider; +async function convertRollout(...args: Parameters): Promise { + await convert(...args); + await processor.forceFlush(); +} + const baseConfig: Config = { enabled: true, public_key: "pk-lf-test", @@ -26,13 +35,24 @@ const baseConfig: Config = { max_chars: 20_000, debug: false, fail_on_error: false, + user_id: "codex-user", + tags: ["codex-tag"], + metadata: { owner: "codex" }, }; const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); +const externalParent: SpanContext = { + traceId: "0af7651916cd43dd8448eb211c80319c", + spanId: "b7ad6b7169203331", + traceFlags: TraceFlags.SAMPLED, + isRemote: true, +}; + /** Copy the fixture session tree to a fresh temp dir (isolates sidecar writes). */ function stageFixtures(): string { const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-trace-")); + onTestFinished(() => fs.rmSync(dir, { recursive: true, force: true })); fs.cpSync(fixturesRoot, path.join(dir, "sessions"), { recursive: true }); return path.join(dir, "sessions", "2026", "06", "03"); } @@ -52,15 +72,30 @@ const parentId = (span: ReadableSpan): string | undefined => (span as unknown as { parentSpanContext?: { spanId?: string } }).parentSpanContext?.spanId ?? (span as unknown as { parentSpanId?: string }).parentSpanId; +function expectStandaloneOwnership(spans: ReadableSpan[]): void { + const roots = spans.filter((span) => span.attributes["langfuse.internal.is_app_root"] === true); + expect(roots.map((span) => span.name)).toEqual(["Codex Turn"]); + expect(roots[0].attributes).toMatchObject({ + "langfuse.trace.name": "Codex Turn", + "langfuse.trace.metadata.owner": "codex", + "user.id": "codex-user", + "session.id": "sess-basic", + "langfuse.trace.tags": ["codex-tag"], + }); +} + beforeAll(() => { provider = new NodeTracerProvider({ - spanProcessors: [new SimpleSpanProcessor(exporter)], + spanProcessors: [processor], + sampler: new ParentBasedSampler({ root: new AlwaysOnSampler() }), }); provider.register(); }); afterAll(async () => { await provider.shutdown(); + context.disable(); + trace.disable(); }); beforeEach(() => { @@ -79,6 +114,7 @@ describe("convertRollout", () => { expect(parentId(root!)).toBeUndefined(); // top-level turn = its own trace expect(attr(root!, "langfuse.observation.input")).toContain("List the files"); expect(attr(root!, "langfuse.observation.output")).toContain("two files"); + expectStandaloneOwnership(spans); // Backdated to the turn's task_started timestamp. expect(startMs(root!)).toBe(Date.parse("2026-06-03T10:00:01.000Z")); @@ -217,6 +253,66 @@ describe("convertRollout", () => { }); }); +describe("external parent context", () => { + it.each([ + ["rollout-basic-main.jsonl", 1, 0, 4], + ["rollout-two-turns-main.jsonl", 2, 0, 4], + ["rollout-parent.jsonl", 1, 1, 6], + ])( + "attaches %s without taking trace ownership or using trace_seed", + async (file, turns, subagents, totalSpans) => { + const previousContext = context.active(); + await convertRollout(path.join(stageFixtures(), file), { + config: { ...baseConfig, trace_seed: "unused-seed" }, + parentSpanContext: externalParent, + }); + + const spans = exporter.getFinishedSpans(); + expect(spans).toHaveLength(totalSpans); + const roots = spans.filter((span) => span.name === "Codex Turn"); + expect(roots).toHaveLength(turns); + expect(spans.filter((span) => span.name === "Codex Subagent Turn")).toHaveLength(subagents); + for (const span of spans) { + expect(span.spanContext().traceId).toBe(externalParent.traceId); + expect(span.attributes["langfuse.internal.is_app_root"]).not.toBe(true); + expect(span.attributes["user.id"]).toBeUndefined(); + expect(span.attributes["session.id"]).toBeUndefined(); + expect( + Object.keys(span.attributes).filter((key) => key.startsWith("langfuse.trace.")), + ).toEqual([]); + if (span.name === "Codex Turn") { + expect(parentId(span)).toBe(externalParent.spanId); + expect(span.attributes["langfuse.observation.metadata.codex.turn_id"]).toBeDefined(); + } else { + const parent = spans.find( + (candidate) => candidate.spanContext().spanId === parentId(span), + ); + expect(parent).toBeDefined(); + if (span.name === "Codex Subagent Turn") { + expect(parent!.name).toBe("Codex Turn"); + } else { + expect(obsType(parent!)).toBe(obsType(span) === "tool" ? "generation" : "agent"); + } + } + } + expect(context.active()).toBe(previousContext); + }, + ); + + it("exports no spans for an unsampled parent but still records completed turns", async () => { + const dir = stageFixtures(); + const file = path.join(dir, "rollout-basic-main.jsonl"); + + await convertRollout(file, { + config: baseConfig, + parentSpanContext: { ...externalParent, traceFlags: TraceFlags.NONE }, + }); + + expect(exporter.getFinishedSpans()).toHaveLength(0); + expect(fs.readFileSync(`${file}.langfuse`, "utf-8")).toBe("turn-1\n"); + }); +}); + describe("deterministic trace ids (trace_seed)", () => { const seed = "ci-run-42"; const seededConfig: Config = { ...baseConfig, trace_seed: seed }; @@ -258,6 +354,7 @@ describe("deterministic trace ids (trace_seed)", () => { // Structure is unchanged: root agent span with its generations beneath it. const root = spans.find((s) => s.name === "Codex Turn")!; expect(obsType(root)).toBe("agent"); + expectStandaloneOwnership(spans); const generations = spans.filter((s) => obsType(s) === "generation"); expect(generations).toHaveLength(2); for (const gen of generations) { diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 74b9a7a..176d437 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8,6 +8,9 @@ importers: .: dependencies: + '@langfuse/core': + specifier: ^5.4.1 + version: 5.4.1(@opentelemetry/api@1.9.1) '@langfuse/otel': specifier: ^5.4.1 version: 5.4.1(@opentelemetry/api@1.9.1)(@opentelemetry/core@2.7.1(@opentelemetry/api@1.9.1))(@opentelemetry/exporter-trace-otlp-http@0.205.0(@opentelemetry/api@1.9.1))(@opentelemetry/sdk-trace-base@2.7.1(@opentelemetry/api@1.9.1)) From 31a152f3872409c5762239059ed2bb422d1f86cb Mon Sep 17 00:00:00 2001 From: zenor0 Date: Sun, 6 Sep 2026 14:15:39 +0800 Subject: [PATCH 2/2] test(tracing): add optional backend e2e coverage Add an opt-in credentialed test for a real application span and bundled Codex hook, with a dedicated Vitest config, test:e2e script, and usage documentation. This commit can be dropped without affecting the tracing feature or regular test suite. Validation: 51 local tests, full lint, and an npm pack dry run pass. The real-backend E2E test was not run. --- README.md | 3 + package.json | 1 + .../tracing/test/e2e/attached-backend.e2e.ts | 148 ++++++++++++++++++ vitest.e2e.config.ts | 9 ++ 4 files changed, 161 insertions(+) create mode 100644 plugins/tracing/test/e2e/attached-backend.e2e.ts create mode 100644 vitest.e2e.config.ts diff --git a/README.md b/README.md index 5f368be..8814a2a 100644 --- a/README.md +++ b/README.md @@ -243,10 +243,13 @@ The hook fails open: any tracing error is logged and swallowed so it never block ```bash pnpm install pnpm test # build, then run the test suite +pnpm run test:e2e # test a real Master span + bundled hook against Langfuse pnpm run lint # prettier + tsc + build pnpm run build # bundle the hook to plugins/tracing/dist/index.mjs ``` +The opt-in E2E test requires `LANGFUSE_PUBLIC_KEY` and `LANGFUSE_SECRET_KEY` (plus `LANGFUSE_BASE_URL` for a non-EU or self-hosted instance). It creates one uniquely identified trace, validates the persisted parent-child tree through the public observations API, and deletes that trace before exiting. The regular test suite never sends data to Langfuse. + The hook ships as a single self-contained `plugins/tracing/dist/index.mjs`, because Codex runs the plugin without an install step and never installs its dependencies. The bundle is a build output and is not committed: `prepack` builds it when the npm package is published, so it travels in the tarball instead of in Git. `pnpm test` builds first, since the hook-command test executes the bundled hook. ### Releasing diff --git a/package.json b/package.json index 1ba3c90..e15b0e8 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,7 @@ "scripts": { "build": "tsdown --config plugins/tracing/tsdown.config.ts", "test": "pnpm run build && vitest run", + "test:e2e": "pnpm run build && vitest run --config vitest.e2e.config.ts", "test:watch": "pnpm run build && vitest", "format": "prettier --write .", "format:check": "prettier --check .", diff --git a/plugins/tracing/test/e2e/attached-backend.e2e.ts b/plugins/tracing/test/e2e/attached-backend.e2e.ts new file mode 100644 index 0000000..39232e6 --- /dev/null +++ b/plugins/tracing/test/e2e/attached-backend.e2e.ts @@ -0,0 +1,148 @@ +import { spawnSync } from "node:child_process"; +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { fileURLToPath } from "node:url"; + +import { LangfuseAPIClient } from "@langfuse/core"; +import { LangfuseSpanProcessor } from "@langfuse/otel"; +import { startObservation } from "@langfuse/tracing"; +import { AlwaysOnSampler } from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import { expect, it, vi } from "vitest"; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../../../.."); +const bundle = path.join(repoRoot, "plugins/tracing/dist/index.mjs"); +const fixture = path.join( + repoRoot, + "plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-basic-main.jsonl", +); + +function getCredentials(): { publicKey: string; secretKey: string; baseUrl: string } { + const publicKey = process.env.LANGFUSE_CODEX_PUBLIC_KEY ?? process.env.LANGFUSE_PUBLIC_KEY; + const secretKey = process.env.LANGFUSE_CODEX_SECRET_KEY ?? process.env.LANGFUSE_SECRET_KEY; + const baseUrl = + process.env.LANGFUSE_CODEX_BASE_URL ?? + process.env.LANGFUSE_BASE_URL ?? + "https://cloud.langfuse.com"; + + if (!publicKey || !secretKey) { + throw new Error( + "test:e2e requires LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY (or their LANGFUSE_CODEX_* variants)", + ); + } + // Validate before either process attempts to export. + new URL(baseUrl); + return { publicKey, secretKey, baseUrl }; +} + +/** Copy the fixture with current timestamps so backend retention filters include it. */ +function stageFixture(tempDir: string): string { + const lines = fs + .readFileSync(fixture, "utf-8") + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { timestamp: string; [key: string]: unknown }); + const firstTimestamp = Date.parse(lines[0].timestamp); + const stagedStart = Date.now() - 10_000; + + for (const line of lines) { + line.timestamp = new Date( + stagedStart + Date.parse(line.timestamp) - firstTimestamp, + ).toISOString(); + } + + const transcript = path.join(tempDir, "rollout-e2e.jsonl"); + fs.writeFileSync(transcript, `${lines.map((line) => JSON.stringify(line)).join("\n")}\n`); + return transcript; +} + +it("joins a real Master span and bundled Codex observations in Langfuse", async ({ + onTestFinished, +}) => { + const credentials = getCredentials(); + const api = new LangfuseAPIClient({ + environment: credentials.baseUrl, + username: credentials.publicKey, + password: credentials.secretKey, + }); + const requestOptions = { timeoutInSeconds: 10, maxRetries: 0 }; + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-attached-e2e-")); + onTestFinished(() => fs.rmSync(tempDir, { recursive: true, force: true })); + const transcript = stageFixture(tempDir); + const processor = new LangfuseSpanProcessor({ + ...credentials, + environment: "codex-plugin-e2e", + shouldExportSpan: () => true, + }); + const provider = new NodeTracerProvider({ + spanProcessors: [processor], + sampler: new AlwaysOnSampler(), + }); + provider.register(); + + const master = startObservation( + "Master Agent Run", + { metadata: { "e2e.source": "codex-observability-plugin" } }, + { asType: "agent", startTime: new Date(Date.now() - 15_000) }, + ); + const { traceId, spanId } = master.otelSpan.spanContext(); + master.end(); + onTestFinished(async () => { + try { + await provider.shutdown(); + } finally { + await api.trace.delete(traceId, requestOptions); + } + }); + await processor.forceFlush(); + + const hook = spawnSync(process.execPath, [bundle], { + cwd: repoRoot, + input: JSON.stringify({ hook_event_name: "Stop", transcript_path: transcript }), + encoding: "utf-8", + timeout: 45_000, + env: { + ...process.env, + CODEX_HOME: tempDir, + TRACE_TO_LANGFUSE: "true", + LANGFUSE_CODEX_PUBLIC_KEY: credentials.publicKey, + LANGFUSE_CODEX_SECRET_KEY: credentials.secretKey, + LANGFUSE_CODEX_BASE_URL: credentials.baseUrl, + LANGFUSE_CODEX_ENVIRONMENT: "codex-plugin-e2e", + LANGFUSE_CODEX_TRACEPARENT: `00-${traceId}-${spanId}-01`, + LANGFUSE_CODEX_FAIL_ON_ERROR: "true", + OTEL_TRACES_SAMPLER: "always_off", + }, + }); + if (hook.error) throw hook.error; + expect(hook.status, hook.stderr).toBe(0); + + const observations = await vi.waitUntil( + async () => { + const { data } = await api.legacy.observationsV1.getMany( + { traceId, limit: 100 }, + requestOptions, + ); + return data.length >= 5 && data; // Master, Codex turn, two generations, and tool. + }, + { timeout: 60_000, interval: 1_000 }, + ); + const masterObservation = observations.find((observation) => observation.id === master.id)!; + const codexTurn = observations.find((observation) => observation.name === "Codex Turn")!; + const generations = observations.filter((observation) => observation.name === "LLM"); + const tool = observations.find((observation) => observation.name === "exec_command")!; + + expect(masterObservation.type).toBe("AGENT"); + expect(masterObservation.parentObservationId).toBeNull(); + expect(codexTurn.type).toBe("AGENT"); + expect(codexTurn.parentObservationId).toBe(master.id); + expect(generations).toHaveLength(2); + expect(generations.every((generation) => generation.type === "GENERATION")).toBe(true); + expect(generations.every((generation) => generation.parentObservationId === codexTurn.id)).toBe( + true, + ); + expect(tool.type).toBe("TOOL"); + expect(generations.map((generation) => generation.id)).toContain(tool.parentObservationId); + expect(observations.every((observation) => observation.traceId === traceId)).toBe(true); +}); diff --git a/vitest.e2e.config.ts b/vitest.e2e.config.ts new file mode 100644 index 0000000..c3aaf6a --- /dev/null +++ b/vitest.e2e.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["plugins/tracing/test/e2e/**/*.e2e.ts"], + hookTimeout: 120_000, + testTimeout: 120_000, + }, +});