diff --git a/README.md b/README.md index 2074583..9b62e32 100644 --- a/README.md +++ b/README.md @@ -169,7 +169,7 @@ The same works from JavaScript with the Langfuse SDK: `await createTraceId(`${se - **Authentication fails** — check that the public/secret keys are valid and that `LANGFUSE_BASE_URL` matches the region the keys belong to. - **Traces land in the wrong project** — API keys are project-scoped in Langfuse; use the keys for the project you want. - **Testing hook failures** — set `LANGFUSE_CODEX_FAIL_ON_ERROR=true` together with `LANGFUSE_CODEX_DEBUG=true` to make Codex report upload or flush errors instead of failing open. -- **Checking dedup sidecars** — successful uploads of completed turns are recorded next to the rollout as `.jsonl.langfuse`. If a Stop hook reads the rollout before Codex has written the turn-completed marker, the trace may upload without a sidecar entry; the next Stop hook will finalize and mark it. +- **Checking dedup sidecars** — successful uploads of completed turns are recorded next to the rollout as `.jsonl.langfuse`. If a Stop hook reads the rollout before Codex has written the turn-completed marker, the exporter defers that turn until a later hook sees its finalized lifecycle event. This avoids separate provisional and finalized traces for one source turn. - **Verifying in Langfuse** — use `npx langfuse-cli api traces list --from-timestamp --limit 10 --order-by timestamp.desc --fields core,metrics,observations --json` with credentials for the same project. - **Sandboxed/network-restricted runs** — Codex sandbox or network policy can prevent exports from reaching Langfuse. Debug logging and fail-on-error mode are the quickest way to distinguish hook execution from network failure. - **Self-hosting** — the TypeScript SDK requires Langfuse platform version >= 3.95.0. diff --git a/package.json b/package.json index c426d73..d54aafd 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@langfuse/codex-observability-plugin", - "version": "0.1.0", + "version": "0.1.1", "description": "OpenAI Codex plugin that traces agent turns, tool calls, and subagents to Langfuse", "keywords": [ "codex", diff --git a/plugins/tracing/.codex-plugin/plugin.json b/plugins/tracing/.codex-plugin/plugin.json index 51019b8..0b529d3 100644 --- a/plugins/tracing/.codex-plugin/plugin.json +++ b/plugins/tracing/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "tracing", - "version": "0.1.0", + "version": "0.1.1", "description": "Trace OpenAI Codex sessions to Langfuse.", "author": { "name": "Langfuse", diff --git a/plugins/tracing/dist/index.mjs b/plugins/tracing/dist/index.mjs index cdc80c5..71f043e 100644 --- a/plugins/tracing/dist/index.mjs +++ b/plugins/tracing/dist/index.mjs @@ -47117,6 +47117,10 @@ async function convertRollout(rolloutFile, options) { const uploaded = await loadUploadedTurnIds(rolloutFile); for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { const turn = turns[turnIndex]; + if (!turn.completed) { + debugLog(`skipping incomplete turn ${turn.turnId ?? "(unknown)"}`); + continue; + } if (turn.completed && turn.turnId && uploaded.has(turn.turnId)) continue; const seededParent = await seededTraceParent(options.config, sessionMeta, turnIndex + 1); await propagateAttributes({ @@ -47132,10 +47136,10 @@ async function convertRollout(rolloutFile, options) { seededParent }); }); - if (turn.completed && turn.turnId) { + if (turn.turnId) { uploaded.add(turn.turnId); await markTurnUploaded(rolloutFile, turn.turnId); - } else if (turn.turnId) debugLog(`uploaded in-progress turn ${turn.turnId}; waiting for completion before sidecar mark`); + } } } diff --git a/plugins/tracing/hooks/hooks.json b/plugins/tracing/hooks/hooks.json index ca7598c..05ea818 100644 --- a/plugins/tracing/hooks/hooks.json +++ b/plugins/tracing/hooks/hooks.json @@ -5,7 +5,7 @@ "hooks": [ { "type": "command", - "command": "node \"${CODEX_HOME:-$HOME/.codex}/plugins/cache/codex-observability-plugin/tracing/0.1.0/dist/index.mjs\"", + "command": "node \"${CODEX_HOME:-$HOME/.codex}/plugins/cache/codex-observability-plugin/tracing/0.1.1/dist/index.mjs\"", "timeout": 30, "statusMessage": "Uploading Codex trace to Langfuse" } diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 3f5e2d0..920edb5 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -317,6 +317,16 @@ export async function convertRollout( for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { const turn = turns[turnIndex]; + // A Stop hook can observe the trailing source turn before Codex has + // appended its completion event. Exporting that provisional state and then + // exporting the completed state creates two independent OTEL observation + // trees (and therefore duplicate Langfuse cost/tool data). Wait for a + // completed or aborted lifecycle event; a later hook invocation will + // export the finalized source turn exactly once through the ledger. + if (!turn.completed) { + debugLog(`skipping incomplete turn ${turn.turnId ?? "(unknown)"}`); + continue; + } if (turn.completed && turn.turnId && uploaded.has(turn.turnId)) { continue; // already uploaded in a previous hook invocation } @@ -342,15 +352,11 @@ export async function convertRollout( }, ); - // Only mark completed turns as uploaded; an in-progress trailing turn is - // re-uploaded (and finalized) on the next hook invocation. - if (turn.completed && turn.turnId) { + // Only completed source turns reach this point and receive a durable + // ledger entry after their finalized export succeeds. + if (turn.turnId) { uploaded.add(turn.turnId); await markTurnUploaded(rolloutFile, turn.turnId); - } else if (turn.turnId) { - debugLog( - `uploaded in-progress turn ${turn.turnId}; waiting for completion before sidecar mark`, - ); } } } diff --git a/plugins/tracing/test/hook-command.test.ts b/plugins/tracing/test/hook-command.test.ts index a3bd5ef..26eb09a 100644 --- a/plugins/tracing/test/hook-command.test.ts +++ b/plugins/tracing/test/hook-command.test.ts @@ -34,7 +34,7 @@ function readPluginVersion(): string { function stageInstalledPlugin(codexHome: string): void { const installedBundle = path.join( codexHome, - "plugins/cache/codex-observability-plugin/tracing/0.1.0/dist/index.mjs", + `plugins/cache/codex-observability-plugin/tracing/${readPluginVersion()}/dist/index.mjs`, ); fs.mkdirSync(path.dirname(installedBundle), { recursive: true }); fs.copyFileSync(bundleFile, installedBundle); diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index e3c0c95..a63134a 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -172,6 +172,41 @@ describe("convertRollout", () => { await convertRollout(file, { config: baseConfig }); expect(exporter.getFinishedSpans()).toHaveLength(0); }); + + it("defers an incomplete trailing turn until its completed export", async () => { + const dir = stageFixtures(); + const file = path.join(dir, "rollout-two-turns-main.jsonl"); + const completedLines = fs.readFileSync(file, "utf-8").trimEnd().split("\n"); + const completionLine = completedLines.at(-1)!; + + // Isolate turn-b: turn-a is already finalized and should not affect either + // condition. The first pass intentionally lacks turn-b's task_complete. + fs.writeFileSync(`${file}.langfuse`, "turn-a\n"); + fs.writeFileSync(file, `${completedLines.slice(0, -1).join("\n")}\n`); + + await convertRollout(file, { config: baseConfig }); + expect( + exporter + .getFinishedSpans() + .filter((span) => span.name === "Codex Turn") + .filter((span) => attr(span, "langfuse.observation.metadata.codex.turn_id") === "turn-b"), + ).toHaveLength(0); + expect(fs.readFileSync(`${file}.langfuse`, "utf-8").trim().split("\n")).toEqual(["turn-a"]); + + exporter.reset(); + fs.appendFileSync(file, `${completionLine}\n`); + await convertRollout(file, { config: baseConfig }); + expect( + exporter + .getFinishedSpans() + .filter((span) => span.name === "Codex Turn") + .filter((span) => attr(span, "langfuse.observation.metadata.codex.turn_id") === "turn-b"), + ).toHaveLength(1); + expect(fs.readFileSync(`${file}.langfuse`, "utf-8").trim().split("\n")).toEqual([ + "turn-a", + "turn-b", + ]); + }); }); describe("deterministic trace ids (trace_seed)", () => {