diff --git a/README.md b/README.md index 00573e5..3d367b2 100644 --- a/README.md +++ b/README.md @@ -83,19 +83,20 @@ 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_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_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 @@ -106,6 +107,45 @@ 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` | +## 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: + +- **Turn N of the main thread** (1-based, in rollout order) gets the trace id `hex(sha256(":")).slice(0, 32)`. +- **Turn N of a subagent thread** gets `hex(sha256("::")).slice(0, 32)`, scoped by the subagent's thread id so it cannot collide with main-thread ids. (Subagent turns spawned _within_ a main-thread turn are nested inside that turn's trace as usual and don't get their own trace id.) + +The main-thread formula deliberately excludes the Codex thread id, so you can compute the trace id **before** the run starts — no thread id, no polling. The derivation matches the Langfuse SDKs' `createTraceId(seed)` helper and always yields a valid W3C trace id. + +**Use a unique seed per session** (e.g. a UUID or your job/run id). Reusing a seed across sessions produces colliding trace ids, and the second upload would merge into (and overwrite parts of) the first trace. + +If derivation ever fails, the hook falls back to auto-generated ids and still uploads — it never blocks the session (set `LANGFUSE_CODEX_FAIL_ON_ERROR=true` while testing to surface such errors). + +### Example: link a Codex run to a dataset run item + +A harness can compute the trace id up front and register it with a [dataset run](https://langfuse.com/docs/evaluation/dataset-runs/native-run) — without ever fetching traces: + +```bash +SEED="$(uuidgen)" # unique per codex exec invocation + +# Trace id of the first main-thread turn: hex(sha256(":1")).slice(0, 32) +TRACE_ID=$(printf '%s:1' "$SEED" | shasum -a 256 | cut -c1-32) + +# Link the precomputed trace id to a dataset run item before (or after) the run. +curl -s -X POST "$LANGFUSE_BASE_URL/api/public/dataset-run-items" \ + -u "$LANGFUSE_PUBLIC_KEY:$LANGFUSE_SECRET_KEY" \ + -H "Content-Type: application/json" \ + -d "{ + \"runName\": \"codex-benchmark-2026-07-13\", + \"datasetItemId\": \"$DATASET_ITEM_ID\", + \"traceId\": \"$TRACE_ID\" + }" + +# Run Codex; the Stop hook uploads the turn with exactly $TRACE_ID. +LANGFUSE_CODEX_TRACE_SEED="$SEED" codex exec "your prompt" +``` + +The same works from JavaScript with the Langfuse SDK: `await createTraceId(`${seed}:1`)` (from `@langfuse/tracing`) returns the identical id. + ## JSON config reference | Config key | Environment variable | Default | Description | @@ -118,6 +158,7 @@ Run a Codex turn, then open your Langfuse project to see the trace. | `user_id` | `LANGFUSE_CODEX_USER_ID` | Codex auth email, if found | User id for all traces | | `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 | | `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 0460e13..83fc783 100644 --- a/plugins/tracing/dist/index.mjs +++ b/plugins/tracing/dist/index.mjs @@ -4274,6 +4274,7 @@ const ConfigSchema = object({ user_id: string().optional(), tags: array(string()).optional(), metadata: record(string(), string()).optional(), + trace_seed: string().optional(), max_chars: number().int().positive(), debug: boolean(), fail_on_error: boolean() @@ -4391,6 +4392,7 @@ function readEnvConfig(env) { user_id: env.LANGFUSE_CODEX_USER_ID, tags: parseTags(env.LANGFUSE_CODEX_TAGS), metadata: parseMetadata(env.LANGFUSE_CODEX_METADATA), + trace_seed: env.LANGFUSE_CODEX_TRACE_SEED, max_chars: parseInteger(env.LANGFUSE_CODEX_MAX_CHARS), debug: parseBoolean(env.LANGFUSE_CODEX_DEBUG), fail_on_error: parseBoolean(env.LANGFUSE_CODEX_FAIL_ON_ERROR) @@ -46540,6 +46542,17 @@ function startObservation(name, attributes, options) { }); } } +async function createTraceId(seed) { + if (seed) { + const data = new TextEncoder().encode(seed); + const hashBuffer = await crypto.subtle.digest("SHA-256", data); + return uint8ArrayToHex(new Uint8Array(hashBuffer)).slice(0, 32); + } + return uint8ArrayToHex(crypto.getRandomValues(new Uint8Array(16))); +} +function uint8ArrayToHex(array$1) { + return Array.from(array$1).map((b) => b.toString(16).padStart(2, "0")).join(""); +} //#endregion //#region src/utils.ts @@ -46704,7 +46717,8 @@ function parseSession(lines) { sessionId: typeof p.id === "string" ? p.id : sessionMeta.sessionId, cliVersion: p.cli_version, modelProvider: p.model_provider ?? void 0, - baseInstructions: p.base_instructions?.text + baseInstructions: p.base_instructions?.text, + isSubagentThread: typeof p.parent_thread_id === "string" || p.thread_source === "subagent" }; continue; } @@ -46843,6 +46857,7 @@ async function markTurnUploaded(rolloutFile, turnId) { //#endregion //#region src/trace.ts +init_esm$2(); async function loadSession(file) { const data = await fs.readFile(file, "utf-8"); const lines = []; @@ -46882,6 +46897,39 @@ async function findSubagentRollout(parentFile, threadId) { } return walk(root); } +/** +* Placeholder parent span id used to pin a deterministic trace id on a root +* span (the pattern the Langfuse SDK documents for custom trace ids). The id +* never exists as a real span, so Langfuse still renders the turn as the +* trace root. +*/ +const SEED_PARENT_SPAN_ID = "0123456789abcdef"; +/** +* Derive the deterministic trace id for a turn from `config.trace_seed`. +* +* 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 +* 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; + try { + return { + traceId: await createTraceId(sessionMeta.isSubagentThread ? `${config$1.trace_seed}:${sessionMeta.sessionId}:${turnNumber}` : `${config$1.trace_seed}:${turnNumber}`), + spanId: SEED_PARENT_SPAN_ID, + traceFlags: TraceFlags.SAMPLED, + isRemote: true + }; + } catch (error) { + debugLog("failed to derive seeded trace id; falling back to auto-generated:", error); + if (config$1.fail_on_error) throw error; + return; + } +} function toUsageDetails(usage) { if (!usage) return void 0; const details = {}; @@ -46932,7 +46980,7 @@ async function emitTurn(turn, sessionMeta, ctx) { }, { asType: "agent", startTime: new Date(turn.startTime), - parentSpanContext: ctx.parentObservation?.otelSpan.spanContext() + parentSpanContext: ctx.parentObservation?.otelSpan.spanContext() ?? ctx.seededParent }); let previousToolResults = void 0; for (let i = 0; i < turn.steps.length; i++) { @@ -47001,8 +47049,10 @@ async function convertRollout(rolloutFile, options) { return; } const uploaded = await loadUploadedTurnIds(rolloutFile); - for (const turn of turns) { + for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { + const turn = turns[turnIndex]; if (turn.completed && turn.turnId && uploaded.has(turn.turnId)) continue; + const seededParent = await seededTraceParent(options.config, sessionMeta, turnIndex + 1); await propagateAttributes({ sessionId: sessionMeta.sessionId, traceName: "Codex Turn", @@ -47012,7 +47062,8 @@ async function convertRollout(rolloutFile, options) { }, async () => { await emitTurn(turn, sessionMeta, { config: options.config, - rolloutFile + rolloutFile, + seededParent }); }); if (turn.completed && turn.turnId) { diff --git a/plugins/tracing/src/config.ts b/plugins/tracing/src/config.ts index 23844fd..eb2ed69 100644 --- a/plugins/tracing/src/config.ts +++ b/plugins/tracing/src/config.ts @@ -31,6 +31,8 @@ export const ConfigSchema = z.object({ tags: z.array(z.string()).optional(), // LANGFUSE_CODEX_METADATA (JSON object; values coerced to strings) 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_MAX_CHARS — truncate large inputs/outputs max_chars: z.number().int().positive(), // LANGFUSE_CODEX_DEBUG @@ -184,6 +186,7 @@ function readEnvConfig(env: Record): Partial user_id: env.LANGFUSE_CODEX_USER_ID, tags: parseTags(env.LANGFUSE_CODEX_TAGS), metadata: parseMetadata(env.LANGFUSE_CODEX_METADATA), + trace_seed: env.LANGFUSE_CODEX_TRACE_SEED, 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/parse.ts b/plugins/tracing/src/parse.ts index c56c513..9dbdda7 100644 --- a/plugins/tracing/src/parse.ts +++ b/plugins/tracing/src/parse.ts @@ -158,12 +158,15 @@ export function parseSession(lines: RolloutLine[]): { cli_version?: string; model_provider?: string | null; base_instructions?: { text?: string } | null; + parent_thread_id?: string | null; + thread_source?: string | null; }; sessionMeta = { sessionId: typeof p.id === "string" ? p.id : sessionMeta.sessionId, cliVersion: p.cli_version, modelProvider: p.model_provider ?? undefined, baseInstructions: p.base_instructions?.text, + isSubagentThread: typeof p.parent_thread_id === "string" || p.thread_source === "subagent", }; continue; } diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 2c949ab..89e04cd 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -2,7 +2,13 @@ import type { Dirent } from "node:fs"; import * as fs from "node:fs/promises"; import * as path from "node:path"; -import { propagateAttributes, startObservation, type LangfuseObservation } from "@langfuse/tracing"; +import { + createTraceId, + propagateAttributes, + startObservation, + type LangfuseObservation, +} from "@langfuse/tracing"; +import { TraceFlags, type SpanContext } from "@opentelemetry/api"; import type { Config } from "./config.js"; import { parseSession } from "./parse.js"; @@ -61,6 +67,48 @@ async function findSubagentRollout( return walk(root); } +/** + * Placeholder parent span id used to pin a deterministic trace id on a root + * span (the pattern the Langfuse SDK documents for custom trace ids). The id + * never exists as a real span, so Langfuse still renders the turn as the + * trace root. + */ +const SEED_PARENT_SPAN_ID = "0123456789abcdef"; + +/** + * Derive the deterministic trace id for a turn from `config.trace_seed`. + * + * 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 + * or derivation fails — the hook must never block an upload. + */ +async function seededTraceParent( + config: Config, + sessionMeta: SessionMeta, + turnNumber: number, +): Promise { + if (!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), + spanId: SEED_PARENT_SPAN_ID, + traceFlags: TraceFlags.SAMPLED, + isRemote: true, + }; + } catch (error) { + debugLog("failed to derive seeded trace id; falling back to auto-generated:", error); + if (config.fail_on_error) throw error; + return undefined; + } +} + function toUsageDetails(usage: TokenUsage | undefined): Record | undefined { if (!usage) return undefined; const details: Record = {}; @@ -115,6 +163,8 @@ async function emitTurn( config: Config; rolloutFile: string; parentObservation?: LangfuseObservation; + /** Pre-derived trace id for top-level turns (see seededTraceParent). */ + seededParent?: SpanContext; }, ): Promise { const clip = makeClip(ctx.config.max_chars); @@ -139,7 +189,7 @@ async function emitTurn( { asType: "agent", startTime: new Date(turn.startTime), - parentSpanContext: ctx.parentObservation?.otelSpan.spanContext(), + parentSpanContext: ctx.parentObservation?.otelSpan.spanContext() ?? ctx.seededParent, }, ); @@ -249,11 +299,16 @@ export async function convertRollout( const uploaded = await loadUploadedTurnIds(rolloutFile); - for (const turn of turns) { + for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { + const turn = turns[turnIndex]; if (turn.completed && turn.turnId && uploaded.has(turn.turnId)) { continue; // already uploaded in a previous hook invocation } + // 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); + await propagateAttributes( { sessionId: sessionMeta.sessionId, @@ -266,6 +321,7 @@ export async function convertRollout( await emitTurn(turn, sessionMeta, { config: options.config, rolloutFile, + seededParent, }); }, ); diff --git a/plugins/tracing/src/types.ts b/plugins/tracing/src/types.ts index 8649714..f54e1d1 100644 --- a/plugins/tracing/src/types.ts +++ b/plugins/tracing/src/types.ts @@ -141,6 +141,12 @@ export type SessionMeta = { cliVersion?: string; modelProvider?: string; baseInstructions?: string; + /** + * Whether this rollout belongs to a subagent thread rather than the main + * session. Codex marks subagent rollouts with `parent_thread_id` and/or + * `thread_source: "subagent"` in `session_meta`. + */ + isSubagentThread?: boolean; }; /** A single tool invocation, assembled from response items + event_msg. */ diff --git a/plugins/tracing/test/config.test.ts b/plugins/tracing/test/config.test.ts index 2835256..123100b 100644 --- a/plugins/tracing/test/config.test.ts +++ b/plugins/tracing/test/config.test.ts @@ -193,6 +193,26 @@ describe("getConfig", () => { expect(config.enabled).toBe(false); }); + it("leaves trace_seed unset by default and reads it from config files and env", async () => { + const unset = await getConfig({ home: emptyHome(), cwd: emptyHome(), env: {} }); + expect(unset.trace_seed).toBeUndefined(); + + const home = makeTmpHome({ + rel: ".codex/langfuse.json", + contents: { trace_seed: "seed-from-file" }, + }); + + const fromFile = await getConfig({ home, cwd: emptyHome(), env: {} }); + expect(fromFile.trace_seed).toBe("seed-from-file"); + + const fromEnv = await getConfig({ + home, + cwd: emptyHome(), + env: { LANGFUSE_CODEX_TRACE_SEED: "seed-from-env" }, + }); + expect(fromEnv.trace_seed).toBe("seed-from-env"); + }); + it("parses fail-on-error from config and environment", async () => { const home = makeTmpHome({ rel: ".codex/langfuse.json", diff --git a/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-child-thread-child.jsonl b/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-child-thread-child.jsonl index cbbcf09..0842168 100644 --- a/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-child-thread-child.jsonl +++ b/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-child-thread-child.jsonl @@ -1,4 +1,4 @@ -{"timestamp":"2026-06-03T11:00:03.000Z","type":"session_meta","payload":{"id":"thread-child","cli_version":"0.123.0","model_provider":"openai"}} +{"timestamp":"2026-06-03T11:00:03.000Z","type":"session_meta","payload":{"id":"thread-child","cli_version":"0.123.0","model_provider":"openai","parent_thread_id":"sess-parent","thread_source":"subagent"}} {"timestamp":"2026-06-03T11:00:03.100Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-child"}} {"timestamp":"2026-06-03T11:00:03.200Z","type":"turn_context","payload":{"model":"gpt-5.4"}} {"timestamp":"2026-06-03T11:00:03.300Z","type":"event_msg","payload":{"type":"user_message","message":"tell a joke"}} diff --git a/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-two-turns-main.jsonl b/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-two-turns-main.jsonl new file mode 100644 index 0000000..46f7ac8 --- /dev/null +++ b/plugins/tracing/test/fixtures/sessions/2026/06/03/rollout-two-turns-main.jsonl @@ -0,0 +1,15 @@ +{"timestamp":"2026-06-03T12:00:00.000Z","type":"session_meta","payload":{"id":"sess-two-turns","cli_version":"0.123.0","model_provider":"openai"}} +{"timestamp":"2026-06-03T12:00:01.000Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-a"}} +{"timestamp":"2026-06-03T12:00:01.100Z","type":"turn_context","payload":{"model":"gpt-5.4"}} +{"timestamp":"2026-06-03T12:00:01.200Z","type":"event_msg","payload":{"type":"user_message","message":"What is 1 + 1?"}} +{"timestamp":"2026-06-03T12:00:02.000Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"1 + 1 = 2."}]}} +{"timestamp":"2026-06-03T12:00:02.100Z","type":"event_msg","payload":{"type":"agent_message","message":"1 + 1 = 2."}} +{"timestamp":"2026-06-03T12:00:02.200Z","type":"event_msg","payload":{"type":"token_count","info":{"last_token_usage":{"input_tokens":20,"output_tokens":8,"total_tokens":28},"total_token_usage":{"input_tokens":20,"output_tokens":8,"total_tokens":28}}}} +{"timestamp":"2026-06-03T12:00:02.300Z","type":"event_msg","payload":{"type":"task_complete","turn_id":"turn-a"}} +{"timestamp":"2026-06-03T12:01:00.000Z","type":"event_msg","payload":{"type":"task_started","turn_id":"turn-b"}} +{"timestamp":"2026-06-03T12:01:00.100Z","type":"turn_context","payload":{"model":"gpt-5.4"}} +{"timestamp":"2026-06-03T12:01:00.200Z","type":"event_msg","payload":{"type":"user_message","message":"And 2 + 2?"}} +{"timestamp":"2026-06-03T12:01:01.000Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"2 + 2 = 4."}]}} +{"timestamp":"2026-06-03T12:01:01.100Z","type":"event_msg","payload":{"type":"agent_message","message":"2 + 2 = 4."}} +{"timestamp":"2026-06-03T12:01:01.200Z","type":"event_msg","payload":{"type":"token_count","info":{"last_token_usage":{"input_tokens":32,"output_tokens":8,"total_tokens":40},"total_token_usage":{"input_tokens":52,"output_tokens":16,"total_tokens":68}}}} +{"timestamp":"2026-06-03T12:01:01.300Z","type":"event_msg","payload":{"type":"task_complete","turn_id":"turn-b"}} diff --git a/plugins/tracing/test/trace-seed-failure.test.ts b/plugins/tracing/test/trace-seed-failure.test.ts new file mode 100644 index 0000000..cef4b1e --- /dev/null +++ b/plugins/tracing/test/trace-seed-failure.test.ts @@ -0,0 +1,80 @@ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import { fileURLToPath } from "node:url"; + +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"; + +import type { Config } from "../src/config.js"; +import { convertRollout } from "../src/trace.js"; + +// Force seeded trace-id derivation to fail so we can assert the hook fails +// open (uploads with auto-generated ids) unless fail_on_error is set. +vi.mock("@langfuse/tracing", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createTraceId: vi.fn(async () => { + throw new Error("derivation boom"); + }), + }; +}); + +const exporter = new InMemorySpanExporter(); +let provider: NodeTracerProvider; + +const baseConfig: 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, + trace_seed: "ci-run-42", +}; + +const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); + +function stageFixtures(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-seed-fail-")); + fs.cpSync(fixturesRoot, path.join(dir, "sessions"), { recursive: true }); + return path.join(dir, "sessions", "2026", "06", "03"); +} + +beforeAll(() => { + provider = new NodeTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + provider.register(); +}); + +afterAll(async () => { + await provider.shutdown(); +}); + +beforeEach(() => { + exporter.reset(); +}); + +describe("trace seed derivation failure", () => { + it("falls back to auto-generated trace ids and still uploads", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { config: baseConfig }); + + const root = exporter.getFinishedSpans().find((s) => s.name === "Codex Turn"); + expect(root, "expected the turn to upload despite the derivation failure").toBeDefined(); + expect(root!.spanContext().traceId).toMatch(/^[0-9a-f]{32}$/); + }); + + it("propagates the derivation error when fail_on_error is set", async () => { + const dir = stageFixtures(); + await expect( + convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { + config: { ...baseConfig, fail_on_error: true }, + }), + ).rejects.toThrow("derivation boom"); + }); +}); diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index ec429d1..2c2c003 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; @@ -36,6 +37,13 @@ function stageFixtures(): string { return path.join(dir, "sessions", "2026", "06", "03"); } +/** + * The derivation external systems use to precompute a seeded trace id — + * intentionally independent of the Langfuse SDK helper the plugin calls. + */ +const seededTraceId = (seed: string): string => + createHash("sha256").update(seed).digest("hex").slice(0, 32); + const attr = (span: ReadableSpan, key: string): string => span.attributes[key] == null ? "" : String(span.attributes[key]); const obsType = (span: ReadableSpan): string => attr(span, "langfuse.observation.type"); @@ -139,3 +147,117 @@ describe("convertRollout", () => { expect(exporter.getFinishedSpans()).toHaveLength(0); }); }); + +describe("deterministic trace ids (trace_seed)", () => { + const seed = "ci-run-42"; + const seededConfig: Config = { ...baseConfig, trace_seed: seed }; + + const turnRoots = () => + exporter + .getFinishedSpans() + .filter((s) => s.name === "Codex Turn") + .sort((a, b) => startMs(a) - startMs(b)); + + it("derives the N-th main-thread turn's trace id from `${seed}:${N}`", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-two-turns-main.jsonl"), { + config: seededConfig, + }); + + const roots = turnRoots(); + expect(roots).toHaveLength(2); + expect(roots[0].spanContext().traceId).toBe(seededTraceId(`${seed}:1`)); + expect(roots[1].spanContext().traceId).toBe(seededTraceId(`${seed}:2`)); + + // Every span (generations included) lands in one of the two seeded traces. + const traceIds = new Set(exporter.getFinishedSpans().map((s) => s.spanContext().traceId)); + expect([...traceIds].sort()).toEqual( + [seededTraceId(`${seed}:1`), seededTraceId(`${seed}:2`)].sort(), + ); + }); + + it("keeps generations and tool spans in the seeded trace", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { config: seededConfig }); + + const spans = exporter.getFinishedSpans(); + const expected = seededTraceId(`${seed}:1`); + expect(spans.length).toBeGreaterThan(2); // root + generations + tool + for (const span of spans) { + expect(span.spanContext().traceId).toBe(expected); + } + // 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"); + const generations = spans.filter((s) => obsType(s) === "generation"); + expect(generations).toHaveLength(2); + for (const gen of generations) { + expect(parentId(gen)).toBe(root.spanContext().spanId); + } + }); + + it("scopes subagent-thread rollouts by thread id so they don't collide", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-child-thread-child.jsonl"), { + config: seededConfig, + }); + + const roots = turnRoots(); + expect(roots).toHaveLength(1); + expect(roots[0].spanContext().traceId).toBe(seededTraceId(`${seed}:thread-child:1`)); + expect(roots[0].spanContext().traceId).not.toBe(seededTraceId(`${seed}:1`)); + }); + + it("nests subagent turns inside the parent's seeded trace", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-parent.jsonl"), { config: seededConfig }); + + const roots = turnRoots(); + expect(roots).toHaveLength(2); // parent turn + nested subagent turn + const expected = seededTraceId(`${seed}:1`); + for (const root of roots) { + expect(root.spanContext().traceId).toBe(expected); + } + }); + + it("leaves trace ids auto-generated when the seed is unset", async () => { + const dir = stageFixtures(); + await convertRollout(path.join(dir, "rollout-two-turns-main.jsonl"), { config: baseConfig }); + + const roots = turnRoots(); + expect(roots).toHaveLength(2); + for (const root of roots) { + // Same shape as before the feature: true root span, random trace id. + expect(parentId(root)).toBeUndefined(); + expect(root.spanContext().traceId).not.toBe(seededTraceId(`${seed}:1`)); + expect(root.spanContext().traceId).not.toBe(seededTraceId(`${seed}:2`)); + } + expect(roots[0].spanContext().traceId).not.toBe(roots[1].spanContext().traceId); + }); + + it("keeps sidecar dedup working when a seed is set", async () => { + const dir = stageFixtures(); + const file = path.join(dir, "rollout-two-turns-main.jsonl"); + + await convertRollout(file, { config: seededConfig }); + expect(turnRoots()).toHaveLength(2); + expect(fs.existsSync(`${file}.langfuse`)).toBe(true); + + exporter.reset(); + await convertRollout(file, { config: seededConfig }); + expect(exporter.getFinishedSpans()).toHaveLength(0); + }); + + it("numbers turns over the full rollout even when earlier turns are deduped", async () => { + const dir = stageFixtures(); + const file = path.join(dir, "rollout-two-turns-main.jsonl"); + + // Pretend turn 1 was uploaded by a previous hook invocation. + fs.writeFileSync(`${file}.langfuse`, "turn-a\n"); + await convertRollout(file, { config: seededConfig }); + + const roots = turnRoots(); + expect(roots).toHaveLength(1); + expect(roots[0].spanContext().traceId).toBe(seededTraceId(`${seed}:2`)); + }); +});