fix: defer incomplete Codex turn exports

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 `<rollout>.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 `<rollout>.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 <recent ISO> --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.

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",

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",

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`);
+		}
 	}
 }
 

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"
           }

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`,
-      );
     }
   }
 }

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);

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)", () => {