fix: finalize the Stop hook target turn

plugins/tracing/src/index.ts

@@ -1,6 +1,7 @@
 import { getConfig } from "./config.js";
 import { setupInstrumentation } from "./instrumentation.js";
 import { convertRollout } from "./trace.js";
+import { markTurnUploaded } from "./sidecar.js";
 import type { HookInput } from "./types.js";
 import { debugLog, readStdin, setDebug } from "./utils.js";
 
@@ -45,18 +46,32 @@ export async function runHook(): Promise<void> {
   }
 
   const instrumentation = setupInstrumentation(config);
+  let uploadedTurnIds: string[] = [];
+  let failure: unknown;
   try {
-    await convertRollout(hookInput.transcript_path, { config });
+    uploadedTurnIds = await convertRollout(hookInput.transcript_path, {
+      config,
+      finalizeTurnId: hookInput.turn_id ?? undefined,
+    });
   } catch (error) {
-    debugLog("failed to convert rollout:", error);
-    if (config.fail_on_error) throw error;
-  } finally {
-    try {
-      await instrumentation.shutdown();
-    } catch (error) {
-      debugLog("error during flush/shutdown:", error);
-      if (config.fail_on_error) throw error;
-    }
+    failure = error;
+  }
+  try {
+    await instrumentation.shutdown();
+  } catch (error) {
+    failure ??= error;
+  }
+  if (failure) {
+    // Fail-open must still be observable, and without a sidecar entry a later
+    // hook or manual replay can retry the turn.
+    // eslint-disable-next-line no-console
+    console.error("[langfuse-codex] telemetry export failed; turn remains retryable");
+    debugLog("telemetry export failure:", failure);
+    if (config.fail_on_error) throw failure;
+    return;
+  }
+  for (const turnId of uploadedTurnIds) {
+    await markTurnUploaded(hookInput.transcript_path, turnId);
   }
 }
 

plugins/tracing/src/parse.ts

@@ -14,6 +14,7 @@ import type {
   ToolCall,
   Turn,
 } from "./types.js";
+import { isTerminalTurnEvent } from "./turn-lifecycle.js";
 import { isPrimitive, toText } from "./utils.js";
 
 /** Extract printable text from a Codex message `content` array. */
@@ -314,10 +315,8 @@ export function parseSession(lines: RolloutLine[]): {
       } else if (et === "token_count") {
         if (p.info?.total_token_usage) turn!.totalUsage = p.info.total_token_usage;
         closeStep(ts, p.info?.last_token_usage ?? undefined);
-      } else if (et === "task_complete") {
-        finishTurn(ts, { completed: true, aborted: false });
-      } else if (et === "turn_aborted") {
-        finishTurn(ts, { completed: true, aborted: true });
+      } else if (isTerminalTurnEvent(et)) {
+        finishTurn(ts, { completed: true, aborted: et === "turn_aborted" });
       } else {
         // A subagent spawn records the child thread *and* (since it carries a
         // call_id ending in "_end") enriches the spawning tool call below.

plugins/tracing/src/sidecar.ts

@@ -1,13 +1,13 @@
 import * as fs from "node:fs/promises";
 
 /**
- * Per-rollout dedup ledger.
+ * Per-rollout delivery ledger.
  *
  * The `Stop` hook fires after every Codex turn and re-reads the whole rollout
  * file, so completed turns would be re-uploaded each time. We record uploaded
  * turn ids in a sidecar file (`<rolloutFile>.langfuse`) and skip them on
- * subsequent invocations. In-progress (not-yet-completed) turns are uploaded
- * but intentionally not recorded, so they finalize on the next hook run.
+ * subsequent invocations. The hook writes this ledger only after telemetry
+ * flush succeeds, so an export failure remains retryable.
  */
 export async function loadUploadedTurnIds(rolloutFile: string): Promise<Set<string>> {
   try {

plugins/tracing/src/trace.ts

@@ -13,7 +13,7 @@ import { TraceFlags, type SpanContext } from "@opentelemetry/api";
 
 import type { Config } from "./config.js";
 import { parseSession } from "./parse.js";
-import { loadUploadedTurnIds, markTurnUploaded } from "./sidecar.js";
+import { loadUploadedTurnIds } from "./sidecar.js";
 import type { ModelStep, RolloutLine, SessionMeta, TokenUsage, ToolCall, Turn } from "./types.js";
 import { debugLog, toText, truncate } from "./utils.js";
 
@@ -329,27 +329,44 @@ function emitToolCall(
  */
 export async function convertRollout(
   rolloutFile: string,
-  options: { config: Config; parentObservation?: LangfuseObservation },
-): Promise<void> {
+  options: {
+    config: Config;
+    parentObservation?: LangfuseObservation;
+    finalizeTurnId?: string;
+  },
+): Promise<string[]> {
   const { sessionMeta, turns } = parseSession(await loadSession(rolloutFile));
   debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`);
 
   // Subagent rollout: nest everything under the parent turn, no dedup/session wrapping.
   if (options.parentObservation) {
     for (const turn of turns) {
+      if (!turn.completed) {
+        debugLog(`skipping in-progress subagent turn ${turn.turnId ?? "(unknown)"}`);
+        continue;
+      }
       await emitTurn(turn, sessionMeta, {
         config: options.config,
         rolloutFile,
         parentObservation: options.parentObservation,
       });
     }
-    return;
+    return [];
   }
 
   const uploaded = await loadUploadedTurnIds(rolloutFile);
+  const uploadedTurnIds: string[] = [];
 
   for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) {
-    const turn = turns[turnIndex];
+    const parsedTurn = turns[turnIndex];
+    const turn =
+      !parsedTurn.completed && parsedTurn.turnId === options.finalizeTurnId
+        ? { ...parsedTurn, completed: true }
+        : parsedTurn;
+    if (!turn.completed) {
+      debugLog(`skipping in-progress turn ${turn.turnId ?? "(unknown)"} not named by Stop hook`);
+      continue;
+    }
     if (turn.completed && turn.turnId && uploaded.has(turn.turnId)) {
       continue; // already uploaded in a previous hook invocation
     }
@@ -375,15 +392,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) {
+    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`,
-      );
+      uploadedTurnIds.push(turn.turnId);
     }
   }
+
+  return uploadedTurnIds;
 }

plugins/tracing/src/turn-lifecycle.ts

@@ -0,0 +1,5 @@
+const TERMINAL_EVENT_TYPES = new Set(["task_complete", "turn_aborted"]);
+
+export function isTerminalTurnEvent(eventType: string | undefined): boolean {
+  return TERMINAL_EVENT_TYPES.has(eventType ?? "");
+}