From 06a86494acd804057d30ce54e2ae3fa55bee27eb Mon Sep 17 00:00:00 2001 From: Tuan Tran Date: Tue, 14 Jul 2026 19:02:34 +0700 Subject: [PATCH] feat(tracing): add trace_scope to group a Codex session into one trace Upstream emits one Langfuse trace per Codex turn, linked only by session id, so an active session floods the traces list with near-identical "Codex Turn" rows. trace_scope="session" pins every turn of a thread to one deterministic trace id derived from the thread id, so the session becomes a single "Codex Session" trace whose top-level spans are its turns (generations, tool calls and subagents still nested underneath). Defaults to "turn", so existing behaviour is unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) --- README.md | 46 +++++++++----- plugins/tracing/dist/index.mjs | 53 +++++++++++----- plugins/tracing/src/config.ts | 10 ++- plugins/tracing/src/trace.ts | 50 ++++++++++++--- .../tracing/test/trace-seed-failure.test.ts | 1 + plugins/tracing/test/trace.test.ts | 61 +++++++++++++++++++ 6 files changed, 182 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index 2074583..cc028fb 100644 --- a/README.md +++ b/README.md @@ -83,20 +83,21 @@ Run a Codex turn, then open your Langfuse project to see the trace. ## 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_TRACE_SEED` | No | — | Derive deterministic trace ids ([details](#deterministic-trace-ids)) | +| `LANGFUSE_CODEX_TRACE_SCOPE` | No | `turn` | `session` groups a whole thread into one trace ([details](#trace-scope)) | +| `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 @@ -107,6 +108,22 @@ Run a Codex turn, then open your Langfuse project to see the trace. | 🇯🇵 Japan | `https://jp.cloud.langfuse.com` | | ⚕️ HIPAA | `https://hipaa.cloud.langfuse.com` | +## Trace scope + +By default (`trace_scope: "turn"`) every Codex turn becomes its own trace, and turns of the same thread are linked only through the Langfuse **session** id. Long sessions therefore produce a long run of near-identical `Codex Turn` rows in the traces list. + +Set `trace_scope` to `"session"` (or `LANGFUSE_CODEX_TRACE_SCOPE=session`) to collapse a thread into **one trace per Codex session** instead: + +```json +{ "enabled": true, "trace_scope": "session" } +``` + +Each turn stays a top-level `Codex Turn` span inside that trace — with its generations, tool calls, and subagents nested underneath, exactly as before — so nothing is lost; the turns are grouped rather than scattered. The trace is named `Codex Session`, and its cost, token usage, and latency aggregate over the whole session. + +The trace id is derived from the Codex thread id (`hex(sha256("codex-session:")).slice(0, 32)`), so every Stop-hook invocation of the same session appends to the same trace. Combined with `trace_seed`, the id becomes `hex(sha256(":")).slice(0, 32)`. + +Trade-offs worth knowing: a session trace's latency spans the wall-clock life of the session (including the time you spent thinking between turns), and switching scope mid-session splits it — turns already uploaded keep their old traces, because the plugin never re-uploads a turn. + ## 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: @@ -159,6 +176,7 @@ The same works from JavaScript with the Langfuse SDK: `await createTraceId(`${se | `tags` | `LANGFUSE_CODEX_TAGS` | — | Tags for all traces | | `metadata` | `LANGFUSE_CODEX_METADATA` | — | Metadata object for all traces | | `trace_seed` | `LANGFUSE_CODEX_TRACE_SEED` | — | Deterministic trace-id seed | +| `trace_scope` | `LANGFUSE_CODEX_TRACE_SCOPE` | `turn` | `turn` or `session` | | `max_chars` | `LANGFUSE_CODEX_MAX_CHARS` | `20000` | Input/output truncation threshold | | `debug` | `LANGFUSE_CODEX_DEBUG` | `false` | Verbose logging | | `fail_on_error` | `LANGFUSE_CODEX_FAIL_ON_ERROR` | `false` | Fail the hook on upload errors | diff --git a/plugins/tracing/dist/index.mjs b/plugins/tracing/dist/index.mjs index cdc80c5..3e9df48 100644 --- a/plugins/tracing/dist/index.mjs +++ b/plugins/tracing/dist/index.mjs @@ -4275,6 +4275,7 @@ const ConfigSchema = object({ tags: array(string()).optional(), metadata: record(string(), string()).optional(), trace_seed: string().optional(), + trace_scope: _enum(["turn", "session"]), max_chars: number().int().positive(), debug: boolean(), fail_on_error: boolean() @@ -4283,6 +4284,7 @@ const PartialConfigSchema = ConfigSchema.partial(); const DEFAULTS = { enabled: false, base_url: "https://cloud.langfuse.com", + trace_scope: "turn", max_chars: 2e4, debug: false, fail_on_error: false @@ -4393,6 +4395,7 @@ function readEnvConfig(env) { tags: parseTags(env.LANGFUSE_CODEX_TAGS), metadata: parseMetadata(env.LANGFUSE_CODEX_METADATA), trace_seed: env.LANGFUSE_CODEX_TRACE_SEED, + trace_scope: env.LANGFUSE_CODEX_TRACE_SCOPE, max_chars: parseInteger(env.LANGFUSE_CODEX_MAX_CHARS), debug: parseBoolean(env.LANGFUSE_CODEX_DEBUG), fail_on_error: parseBoolean(env.LANGFUSE_CODEX_FAIL_ON_ERROR) @@ -15269,7 +15272,7 @@ function propagateAttributes(params, fn) { let context$1 = context.active(); const span = trace.getActiveSpan(); const asBaggage = (_a2 = params.asBaggage) != null ? _a2 : false; - const { userId, sessionId, metadata, version: version$1, tags, traceName, _internalExperiment } = params; + const { userId, sessionId, metadata, version: version$1, tags, traceName: traceName$1, _internalExperiment } = params; if (userId) { if (isValidPropagatedString({ value: userId, @@ -15306,13 +15309,13 @@ function propagateAttributes(params, fn) { asBaggage }); } - if (traceName) { + if (traceName$1) { if (isValidPropagatedString({ - value: traceName, + value: traceName$1, attributeName: "traceName" })) context$1 = setPropagatedAttribute({ key: "traceName", - value: traceName, + value: traceName$1, context: context$1, span, asBaggage @@ -15381,10 +15384,10 @@ function getPropagatedAttributesFromContext(context$1) { const spanKey = getSpanKeyForPropagatedKey("version"); propagatedAttributes[spanKey] = version$1; } - const traceName = context$1.getValue(LangfuseOtelContextKeys["traceName"]); - if (traceName && typeof traceName === "string") { + const traceName$1 = context$1.getValue(LangfuseOtelContextKeys["traceName"]); + if (traceName$1 && typeof traceName$1 === "string") { const spanKey = getSpanKeyForPropagatedKey("traceName"); - propagatedAttributes[spanKey] = traceName; + propagatedAttributes[spanKey] = traceName$1; } const tags = context$1.getValue(LangfuseOtelContextKeys["tags"]); if (tags && Array.isArray(tags)) { @@ -46956,21 +46959,29 @@ async function findSubagentRollout(parentFile, threadId) { */ const SEED_PARENT_SPAN_ID = "0123456789abcdef"; /** -* Derive the deterministic trace id for a turn from `config.trace_seed`. +* Derive the deterministic trace id a turn should be pinned to. * -* Main-thread turn N (1-based, rollout order): createTraceId(`${seed}:${N}`) -* Subagent-thread turn N: createTraceId(`${seed}:${threadId}:${N}`) +* `trace_scope: "session"` — every turn of a thread shares one trace id, so the +* session is a single trace whose top-level spans are its turns: +* +* createTraceId(`codex-session:${threadId}`) (`${seed}:${threadId}` with a seed) +* +* `trace_scope: "turn"` (default) — each turn is its own trace, and ids are only +* pinned when `trace_seed` is set: +* +* Main-thread turn N (1-based, rollout order): createTraceId(`${seed}:${N}`) +* Subagent-thread turn N: createTraceId(`${seed}:${threadId}:${N}`) * * The main-thread form deliberately excludes the thread id so external systems * can precompute trace ids (hex(sha256(seed)).slice(0, 32)) before the Codex -* thread exists. Returns `undefined` (auto-generated ids) when no seed is set +* thread exists. Returns `undefined` (auto-generated ids) when no id is pinned * or derivation fails — the hook must never block an upload. */ async function seededTraceParent(config$1, sessionMeta, turnNumber) { - if (!config$1.trace_seed) return void 0; + if (!(config$1.trace_scope === "session") && !config$1.trace_seed) return void 0; try { return { - traceId: await createTraceId(sessionMeta.isSubagentThread ? `${config$1.trace_seed}:${sessionMeta.sessionId}:${turnNumber}` : `${config$1.trace_seed}:${turnNumber}`), + traceId: await createTraceId(traceSeed(config$1, sessionMeta, turnNumber)), spanId: SEED_PARENT_SPAN_ID, traceFlags: TraceFlags.SAMPLED, isRemote: true @@ -46981,6 +46992,20 @@ async function seededTraceParent(config$1, sessionMeta, turnNumber) { return; } } +/** +* Trace name. Under `trace_scope: "session"` the trace spans the whole thread, +* so it is named for the session rather than for whichever turn triggered the +* upload; each turn keeps its own "Codex Turn" span inside it. +*/ +function traceName(config$1, sessionMeta) { + const subagent = sessionMeta.isSubagentThread === true; + if (config$1.trace_scope === "session") return subagent ? "Codex Subagent Session" : "Codex Session"; + return subagent ? "Codex Subagent Turn" : "Codex Turn"; +} +function traceSeed(config$1, sessionMeta, turnNumber) { + if (config$1.trace_scope === "session") return config$1.trace_seed ? `${config$1.trace_seed}:${sessionMeta.sessionId}` : `codex-session:${sessionMeta.sessionId}`; + return sessionMeta.isSubagentThread ? `${config$1.trace_seed}:${sessionMeta.sessionId}:${turnNumber}` : `${config$1.trace_seed}:${turnNumber}`; +} function toUsageDetails(usage) { if (!usage) return void 0; const details = {}; @@ -47121,7 +47146,7 @@ async function convertRollout(rolloutFile, options) { const seededParent = await seededTraceParent(options.config, sessionMeta, turnIndex + 1); await propagateAttributes({ sessionId: sessionMeta.sessionId, - traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", + traceName: traceName(options.config, sessionMeta), ...options.config.user_id ? { userId: options.config.user_id } : {}, ...options.config.tags ? { tags: options.config.tags } : {}, ...options.config.metadata ? { metadata: options.config.metadata } : {} diff --git a/plugins/tracing/src/config.ts b/plugins/tracing/src/config.ts index eb2ed69..0654bc5 100644 --- a/plugins/tracing/src/config.ts +++ b/plugins/tracing/src/config.ts @@ -33,6 +33,9 @@ export const ConfigSchema = z.object({ metadata: z.record(z.string(), z.string()).optional(), // LANGFUSE_CODEX_TRACE_SEED — deterministic trace ids derived from this seed trace_seed: z.string().optional(), + // LANGFUSE_CODEX_TRACE_SCOPE — "turn": one trace per turn; "session": one + // trace per Codex thread, with each turn as a top-level span inside it + trace_scope: z.enum(["turn", "session"]), // LANGFUSE_CODEX_MAX_CHARS — truncate large inputs/outputs max_chars: z.number().int().positive(), // LANGFUSE_CODEX_DEBUG @@ -45,9 +48,13 @@ export type Config = z.infer; const PartialConfigSchema = ConfigSchema.partial(); -const DEFAULTS: Pick = { +const DEFAULTS: Pick< + Config, + "enabled" | "base_url" | "trace_scope" | "max_chars" | "debug" | "fail_on_error" +> = { enabled: false, base_url: "https://cloud.langfuse.com", + trace_scope: "turn", max_chars: 20_000, debug: false, fail_on_error: false, @@ -187,6 +194,7 @@ function readEnvConfig(env: Record): Partial tags: parseTags(env.LANGFUSE_CODEX_TAGS), metadata: parseMetadata(env.LANGFUSE_CODEX_METADATA), trace_seed: env.LANGFUSE_CODEX_TRACE_SEED, + trace_scope: env.LANGFUSE_CODEX_TRACE_SCOPE, max_chars: parseInteger(env.LANGFUSE_CODEX_MAX_CHARS), debug: parseBoolean(env.LANGFUSE_CODEX_DEBUG), fail_on_error: parseBoolean(env.LANGFUSE_CODEX_FAIL_ON_ERROR), diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 3f5e2d0..731fbbc 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -76,14 +76,22 @@ async function findSubagentRollout( const SEED_PARENT_SPAN_ID = "0123456789abcdef"; /** - * Derive the deterministic trace id for a turn from `config.trace_seed`. + * Derive the deterministic trace id a turn should be pinned to. * - * Main-thread turn N (1-based, rollout order): createTraceId(`${seed}:${N}`) - * Subagent-thread turn N: createTraceId(`${seed}:${threadId}:${N}`) + * `trace_scope: "session"` — every turn of a thread shares one trace id, so the + * session is a single trace whose top-level spans are its turns: + * + * createTraceId(`codex-session:${threadId}`) (`${seed}:${threadId}` with a seed) + * + * `trace_scope: "turn"` (default) — each turn is its own trace, and ids are only + * pinned when `trace_seed` is set: + * + * Main-thread turn N (1-based, rollout order): createTraceId(`${seed}:${N}`) + * Subagent-thread turn N: createTraceId(`${seed}:${threadId}:${N}`) * * The main-thread form deliberately excludes the thread id so external systems * can precompute trace ids (hex(sha256(seed)).slice(0, 32)) before the Codex - * thread exists. Returns `undefined` (auto-generated ids) when no seed is set + * thread exists. Returns `undefined` (auto-generated ids) when no id is pinned * or derivation fails — the hook must never block an upload. */ async function seededTraceParent( @@ -91,13 +99,11 @@ async function seededTraceParent( sessionMeta: SessionMeta, turnNumber: number, ): Promise { - if (!config.trace_seed) return undefined; + const sessionScoped = config.trace_scope === "session"; + if (!sessionScoped && !config.trace_seed) return undefined; try { - const seed = sessionMeta.isSubagentThread - ? `${config.trace_seed}:${sessionMeta.sessionId}:${turnNumber}` - : `${config.trace_seed}:${turnNumber}`; return { - traceId: await createTraceId(seed), + traceId: await createTraceId(traceSeed(config, sessionMeta, turnNumber)), spanId: SEED_PARENT_SPAN_ID, traceFlags: TraceFlags.SAMPLED, isRemote: true, @@ -109,6 +115,30 @@ async function seededTraceParent( } } +/** + * Trace name. Under `trace_scope: "session"` the trace spans the whole thread, + * so it is named for the session rather than for whichever turn triggered the + * upload; each turn keeps its own "Codex Turn" span inside it. + */ +function traceName(config: Config, sessionMeta: SessionMeta): string { + const subagent = sessionMeta.isSubagentThread === true; + if (config.trace_scope === "session") { + return subagent ? "Codex Subagent Session" : "Codex Session"; + } + return subagent ? "Codex Subagent Turn" : "Codex Turn"; +} + +function traceSeed(config: Config, sessionMeta: SessionMeta, turnNumber: number): string { + if (config.trace_scope === "session") { + return config.trace_seed + ? `${config.trace_seed}:${sessionMeta.sessionId}` + : `codex-session:${sessionMeta.sessionId}`; + } + return sessionMeta.isSubagentThread + ? `${config.trace_seed}:${sessionMeta.sessionId}:${turnNumber}` + : `${config.trace_seed}:${turnNumber}`; +} + function toUsageDetails(usage: TokenUsage | undefined): Record | undefined { if (!usage) return undefined; const details: Record = {}; @@ -328,7 +358,7 @@ export async function convertRollout( await propagateAttributes( { sessionId: sessionMeta.sessionId, - traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", + traceName: traceName(options.config, sessionMeta), ...(options.config.user_id ? { userId: options.config.user_id } : {}), ...(options.config.tags ? { tags: options.config.tags } : {}), ...(options.config.metadata ? { metadata: options.config.metadata } : {}), diff --git a/plugins/tracing/test/trace-seed-failure.test.ts b/plugins/tracing/test/trace-seed-failure.test.ts index cef4b1e..3238e9d 100644 --- a/plugins/tracing/test/trace-seed-failure.test.ts +++ b/plugins/tracing/test/trace-seed-failure.test.ts @@ -34,6 +34,7 @@ const baseConfig: Config = { debug: false, fail_on_error: false, trace_seed: "ci-run-42", + trace_scope: "turn", }; const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index e3c0c95..d2f4667 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -26,8 +26,15 @@ const baseConfig: Config = { max_chars: 20_000, debug: false, fail_on_error: false, + trace_scope: "turn", }; +/** + * Placeholder parent span id the plugin pins seeded/session traces to — mirrored + * here on purpose so the test breaks if that contract silently changes. + */ +const SEED_PARENT_SPAN_ID = "0123456789abcdef"; + const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); /** Copy the fixture session tree to a fresh temp dir (isolates sidecar writes). */ @@ -174,6 +181,60 @@ describe("convertRollout", () => { }); }); +describe("session-scoped traces (trace_scope: session)", () => { + const sessionConfig: Config = { ...baseConfig, trace_scope: "session" }; + + const turnRoots = () => + exporter + .getFinishedSpans() + .filter((s) => s.name === "Codex Turn" || s.name === "Codex Subagent Turn") + .sort((a, b) => startMs(a) - startMs(b)); + + it("puts every turn of a thread in one trace keyed by the session id", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-two-turns-main.jsonl"), { + config: sessionConfig, + }); + + const roots = turnRoots(); + expect(roots).toHaveLength(2); + + // Both turns — and every span beneath them — share the session's trace id. + const expected = seededTraceId("codex-session:sess-two-turns"); + const traceIds = new Set(exporter.getFinishedSpans().map((s) => s.spanContext().traceId)); + expect([...traceIds]).toEqual([expected]); + + // Each turn is still its own top-level span inside that trace, not nested + // in the preceding turn. + for (const root of roots) { + expect(obsType(root)).toBe("agent"); + expect(parentId(root)).toBe(SEED_PARENT_SPAN_ID); + } + expect(roots[0].spanContext().spanId).not.toBe(parentId(roots[1])); + }); + + it("gives each thread its own trace, so sessions never collide", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-two-turns-main.jsonl"), { + config: sessionConfig, + }); + await convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { config: sessionConfig }); + + const traceIds = new Set(turnRoots().map((s) => s.spanContext().traceId)); + expect(traceIds.size).toBe(2); + }); + + it("combines with trace_seed so external systems can precompute the id", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-two-turns-main.jsonl"), { + config: { ...sessionConfig, trace_seed: "ci-run-42" }, + }); + + const traceIds = new Set(turnRoots().map((s) => s.spanContext().traceId)); + expect([...traceIds]).toEqual([seededTraceId("ci-run-42:sess-two-turns")]); + }); +}); + describe("deterministic trace ids (trace_seed)", () => { const seed = "ci-run-42"; const seededConfig: Config = { ...baseConfig, trace_seed: seed };