diff --git a/plugins/tracing/src/index.ts b/plugins/tracing/src/index.ts
index b32a583..fddf944 100644
--- a/plugins/tracing/src/index.ts
+++ b/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);
   }
 }
 
diff --git a/plugins/tracing/src/parse.ts b/plugins/tracing/src/parse.ts
index 29f6173..3f07941 100644
--- a/plugins/tracing/src/parse.ts
+++ b/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.
diff --git a/plugins/tracing/src/sidecar.ts b/plugins/tracing/src/sidecar.ts
index e0efd87..f09a6f6 100644
--- a/plugins/tracing/src/sidecar.ts
+++ b/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 {
diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts
index 24b4f0e..19ea22f 100644
--- a/plugins/tracing/src/trace.ts
+++ b/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;
 }
diff --git a/plugins/tracing/src/turn-lifecycle.ts b/plugins/tracing/src/turn-lifecycle.ts
new file mode 100644
index 0000000..3552b1c
--- /dev/null
+++ b/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 ?? "");
+}
diff --git a/plugins/tracing/test/hook-command.test.ts b/plugins/tracing/test/hook-command.test.ts
index 0df8a63..838ba7d 100644
--- a/plugins/tracing/test/hook-command.test.ts
+++ b/plugins/tracing/test/hook-command.test.ts
@@ -1,17 +1,110 @@
 import { spawn } from "node:child_process";
 import * as fs from "node:fs";
+import * as http from "node:http";
 import * as os from "node:os";
 import * as path from "node:path";
 import { fileURLToPath } from "node:url";
+import { gunzipSync } from "node:zlib";
 
 import { afterEach, describe, expect, it } from "vitest";
 
 const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../../..");
-const hookConfigFile = path.join(repoRoot, "plugins/tracing/hooks/hooks.json");
-const pluginRootDir = path.join(repoRoot, "plugins/tracing");
+const pluginRootDir = process.env.TEST_PLUGIN_ROOT ?? path.join(repoRoot, "plugins/tracing");
+const hookConfigFile = path.join(pluginRootDir, "hooks/hooks.json");
 
 const tmpDirs: string[] = [];
 
+type ProtobufField = { number: number; wireType: number; value: Buffer | bigint };
+
+function protobufFields(message: Buffer): ProtobufField[] {
+  const fields: ProtobufField[] = [];
+  let offset = 0;
+  const readVarint = (): bigint => {
+    let value = 0n;
+    let shift = 0n;
+    while (offset < message.length) {
+      const byte = message[offset++];
+      value |= BigInt(byte & 0x7f) << shift;
+      if ((byte & 0x80) === 0) return value;
+      shift += 7n;
+    }
+    throw new Error("truncated protobuf varint");
+  };
+
+  while (offset < message.length) {
+    const tag = Number(readVarint());
+    const number = tag >>> 3;
+    const wireType = tag & 7;
+    if (wireType === 0) {
+      fields.push({ number, wireType, value: readVarint() });
+    } else if (wireType === 1) {
+      const value = message.readBigUInt64LE(offset);
+      offset += 8;
+      fields.push({ number, wireType, value });
+    } else if (wireType === 2) {
+      const length = Number(readVarint());
+      const value = message.subarray(offset, offset + length);
+      offset += length;
+      fields.push({ number, wireType, value });
+    } else if (wireType === 5) {
+      offset += 4;
+    } else {
+      throw new Error(`unsupported protobuf wire type ${wireType}`);
+    }
+  }
+  return fields;
+}
+
+type ExportedSpan = { traceId: string; parentSpanId: string; endTimeUnixNano: bigint };
+
+function exportedSpans(requestBody: Buffer): ExportedSpan[] {
+  if (requestBody[0] === "{".charCodeAt(0)) {
+    const payload = JSON.parse(requestBody.toString("utf-8")) as {
+      resourceSpans?: Array<{
+        scopeSpans?: Array<{
+          spans?: Array<{
+            traceId?: string;
+            parentSpanId?: string;
+            endTimeUnixNano?: string;
+          }>;
+        }>;
+      }>;
+    };
+    return (payload.resourceSpans ?? []).flatMap((resource) =>
+      (resource.scopeSpans ?? []).flatMap((scope) =>
+        (scope.spans ?? []).map((span) => ({
+          traceId: span.traceId ?? "",
+          parentSpanId: span.parentSpanId ?? "",
+          endTimeUnixNano: BigInt(span.endTimeUnixNano ?? "0"),
+        })),
+      ),
+    );
+  }
+
+  const nested = (message: Buffer, fieldNumber: number): Buffer[] =>
+    protobufFields(message)
+      .filter((field) => field.number === fieldNumber && Buffer.isBuffer(field.value))
+      .map((field) => field.value as Buffer);
+
+  return nested(requestBody, 1).flatMap((resourceSpans) =>
+    nested(resourceSpans, 2).flatMap((scopeSpans) =>
+      nested(scopeSpans, 2).map((span) => {
+        const fields = protobufFields(span);
+        const bytes = (number: number): Buffer => {
+          const value = fields.find((field) => field.number === number)?.value;
+          return Buffer.isBuffer(value) ? value : Buffer.alloc(0);
+        };
+        const endTime = fields.find((field) => field.number === 8)?.value;
+        return {
+          traceId: bytes(1).toString("hex"),
+          parentSpanId: bytes(4).toString("hex"),
+          endTimeUnixNano: typeof endTime === "bigint" ? endTime : 0n,
+        };
+      }),
+    ),
+  );
+}
+
 function makeTempDir(prefix: string): string {
   const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
   tmpDirs.push(dir);
@@ -59,6 +152,36 @@ function runShellCommand(
   });
 }
 
+function runStopHook(options: {
+  rollout: string;
+  baseUrl: string;
+  codexHome: string;
+  sessionCwd: string;
+  failOnError?: boolean;
+}): Promise<{ code: number | null; stderr: string; stdout: string }> {
+  return runShellCommand(readHookCommand(), {
+    cwd: options.sessionCwd,
+    env: {
+      ...process.env,
+      PLUGIN_ROOT: pluginRootDir,
+      CODEX_HOME: options.codexHome,
+      HOME: options.codexHome,
+      TRACE_TO_LANGFUSE: "true",
+      LANGFUSE_PUBLIC_KEY: "pk-lf-test",
+      LANGFUSE_SECRET_KEY: "sk-lf-test",
+      LANGFUSE_BASE_URL: options.baseUrl,
+      ...(options.failOnError === undefined
+        ? {}
+        : { LANGFUSE_CODEX_FAIL_ON_ERROR: String(options.failOnError) }),
+    },
+    input: JSON.stringify({
+      hook_event_name: "Stop",
+      turn_id: "turn-1",
+      transcript_path: options.rollout,
+    }),
+  });
+}
+
 afterEach(() => {
   while (tmpDirs.length) {
     fs.rmSync(tmpDirs.pop()!, { recursive: true, force: true });
@@ -66,6 +189,112 @@ afterEach(() => {
 });
 
 describe("bundled Stop hook command", () => {
+  it("exports the payload turn before Codex appends task_complete", async () => {
+    const codexHome = makeTempDir("lf-codex-home-");
+    const sessionCwd = makeTempDir("lf-codex-cwd-");
+    const rollout = path.join(sessionCwd, "rollout.jsonl");
+    const completed = fs.readFileSync(
+      path.join(pluginRootDir, "test/fixtures/sessions/2026/06/03/rollout-basic-main.jsonl"),
+      "utf-8",
+    );
+    const completedLines = completed.trimEnd().split("\n");
+    const preTerminalLines = completedLines.slice(0, -1);
+    const lastPersisted = JSON.parse(preTerminalLines.at(-1)!) as { timestamp: string };
+    fs.writeFileSync(rollout, `${preTerminalLines.join("\n")}\n`);
+
+    const exportBodies: Buffer[] = [];
+    const server = http.createServer((request, response) => {
+      const chunks: Buffer[] = [];
+      request.on("data", (chunk) => chunks.push(Buffer.from(chunk)));
+      request.on("end", () => {
+        const body = Buffer.concat(chunks);
+        exportBodies.push(request.headers["content-encoding"] === "gzip" ? gunzipSync(body) : body);
+        response.writeHead(200, { "content-type": "application/json" });
+        response.end("{}");
+      });
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    const address = server.address();
+    if (!address || typeof address === "string") throw new Error("expected a TCP address");
+
+    const baseUrl = `http://127.0.0.1:${address.port}`;
+    const result = await runStopHook({
+      rollout,
+      baseUrl,
+      codexHome,
+      sessionCwd,
+    });
+
+    expect(result.code).toBe(0);
+    expect(exportBodies).toHaveLength(1);
+    const exported = exportBodies.flatMap(exportedSpans);
+    const roots = exported.filter((span) => span.parentSpanId === "");
+    expect(roots).toHaveLength(1);
+    expect(roots[0].endTimeUnixNano).toBe(BigInt(Date.parse(lastPersisted.timestamp)) * 1_000_000n);
+    const exportedPayload = Buffer.concat(exportBodies).toString("utf-8");
+    expect(exportedPayload).toContain("turn-1");
+    expect(exportedPayload).toContain("sess-basic");
+    expect(exportedPayload).toContain("There are two files: file1.txt and file2.txt.");
+    expect(fs.readFileSync(`${rollout}.langfuse`, "utf-8")).toBe("turn-1\n");
+
+    // Codex persists this only after the Stop hook process has returned.
+    fs.appendFileSync(rollout, `${completedLines.at(-1)}\n`);
+    const requestsAfterFirstRun = exportBodies.length;
+    const replay = await runStopHook({
+      rollout,
+      baseUrl,
+      codexHome,
+      sessionCwd,
+    });
+    expect(replay.code).toBe(0);
+    expect(exportBodies).toHaveLength(requestsAfterFirstRun);
+    expect(fs.readFileSync(`${rollout}.langfuse`, "utf-8")).toBe("turn-1\n");
+    await new Promise<void>((resolve, reject) =>
+      server.close((error) => (error ? reject(error) : resolve())),
+    );
+  });
+
+  it.each([
+    { failOnError: false, expectedCode: 0 },
+    { failOnError: true, expectedCode: 1 },
+  ])(
+    "keeps a failed flush retryable when fail_on_error=$failOnError",
+    async ({ failOnError, expectedCode }) => {
+      const codexHome = makeTempDir("lf-codex-home-");
+      const sessionCwd = makeTempDir("lf-codex-cwd-");
+      const rollout = path.join(sessionCwd, "rollout.jsonl");
+      const completed = fs.readFileSync(
+        path.join(pluginRootDir, "test/fixtures/sessions/2026/06/03/rollout-basic-main.jsonl"),
+        "utf-8",
+      );
+      fs.writeFileSync(rollout, `${completed.trimEnd().split("\n").slice(0, -1).join("\n")}\n`);
+
+      const server = http.createServer((_request, response) => {
+        response.writeHead(503, { "content-type": "application/json" });
+        response.end('{"error":"unavailable"}');
+      });
+      await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+      const address = server.address();
+      if (!address || typeof address === "string") throw new Error("expected a TCP address");
+
+      const result = await runStopHook({
+        rollout,
+        baseUrl: `http://127.0.0.1:${address.port}`,
+        codexHome,
+        sessionCwd,
+        failOnError,
+      });
+
+      expect(result.code).toBe(expectedCode);
+      expect(result.stderr).toContain("telemetry export failed; turn remains retryable");
+      expect(fs.existsSync(`${rollout}.langfuse`)).toBe(false);
+      await new Promise<void>((resolve, reject) =>
+        server.close((error) => (error ? reject(error) : resolve())),
+      );
+    },
+    10_000,
+  );
+
   it("runs from an arbitrary session cwd via PLUGIN_ROOT instead of a relative repo path", async () => {
     const codexHome = makeTempDir("lf-codex-home-");
     const sessionCwd = makeTempDir("lf-codex-cwd-");
@@ -77,6 +306,7 @@ describe("bundled Stop hook command", () => {
         PLUGIN_ROOT: pluginRootDir,
         CODEX_HOME: codexHome,
         HOME: codexHome,
+        TRACE_TO_LANGFUSE: "false",
       },
       input: JSON.stringify({
         hook_event_name: "Stop",
diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts
index c3f66a9..ade2440 100644
--- a/plugins/tracing/test/trace.test.ts
+++ b/plugins/tracing/test/trace.test.ts
@@ -13,6 +13,7 @@ import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
 
 import type { Config } from "../src/config.js";
+import { markTurnUploaded } from "../src/sidecar.js";
 import { convertRollout } from "../src/trace.js";
 
 const exporter = new InMemorySpanExporter();
@@ -48,6 +49,7 @@ 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");
 const startMs = (span: ReadableSpan): number => span.startTime[0] * 1000 + span.startTime[1] / 1e6;
+const endMs = (span: ReadableSpan): number => span.endTime[0] * 1000 + span.endTime[1] / 1e6;
 const parentId = (span: ReadableSpan): string | undefined =>
   (span as unknown as { parentSpanContext?: { spanId?: string } }).parentSpanContext?.spanId ??
   (span as unknown as { parentSpanId?: string }).parentSpanId;
@@ -68,6 +70,56 @@ beforeEach(() => {
 });
 
 describe("convertRollout", () => {
+  it("does not emit an incomplete top-level turn", async () => {
+    const dir = stageFixtures();
+    const file = path.join(dir, "rollout-basic-main.jsonl");
+    const lines = fs.readFileSync(file, "utf-8").trimEnd().split("\n");
+    fs.writeFileSync(file, `${lines.slice(0, -1).join("\n")}\n`);
+
+    await convertRollout(file, { config: baseConfig });
+
+    expect(exporter.getFinishedSpans()).toHaveLength(0);
+    expect(fs.existsSync(`${file}.langfuse`)).toBe(false);
+  });
+
+  it("finalizes the incomplete turn identified by the Stop hook", async () => {
+    const dir = stageFixtures();
+    const file = path.join(dir, "rollout-basic-main.jsonl");
+    const lines = fs.readFileSync(file, "utf-8").trimEnd().split("\n");
+    const preTerminalLines = lines.slice(0, -1);
+    const lastPersisted = JSON.parse(preTerminalLines.at(-1)!) as { timestamp: string };
+    fs.writeFileSync(file, `${preTerminalLines.join("\n")}\n`);
+
+    const uploadedTurnIds = await convertRollout(file, {
+      config: baseConfig,
+      finalizeTurnId: "turn-1",
+    });
+
+    const roots = exporter.getFinishedSpans().filter((span) => span.name === "Codex Turn");
+    expect(roots).toHaveLength(1);
+    expect(attr(roots[0], "langfuse.observation.metadata.codex.turn_id")).toBe("turn-1");
+    expect(attr(roots[0], "langfuse.observation.metadata.codex.thread_id")).toBe("sess-basic");
+    expect(attr(roots[0], "langfuse.observation.output")).toContain("two files");
+    expect(endMs(roots[0])).toBe(Date.parse(lastPersisted.timestamp));
+    expect(uploadedTurnIds).toEqual(["turn-1"]);
+    expect(fs.existsSync(`${file}.langfuse`)).toBe(false);
+  });
+
+  it("does not finalize an incomplete turn that differs from the Stop payload", async () => {
+    const dir = stageFixtures();
+    const file = path.join(dir, "rollout-basic-main.jsonl");
+    const lines = fs.readFileSync(file, "utf-8").trimEnd().split("\n");
+    fs.writeFileSync(file, `${lines.slice(0, -1).join("\n")}\n`);
+
+    const uploadedTurnIds = await convertRollout(file, {
+      config: baseConfig,
+      finalizeTurnId: "turn-other",
+    });
+
+    expect(exporter.getFinishedSpans()).toHaveLength(0);
+    expect(uploadedTurnIds).toEqual([]);
+  });
+
   it("emits an agent → generation → tool tree with backdated timestamps", async () => {
     const dir = stageFixtures();
     await convertRollout(path.join(dir, "rollout-basic-main.jsonl"), { config: baseConfig });
@@ -150,6 +202,8 @@ describe("convertRollout", () => {
 
     // Aborted turn is flagged on the parent root.
     expect(attr(parent!, "langfuse.observation.level")).toBe("WARNING");
+    expect(attr(parent!, "langfuse.observation.metadata.codex.aborted")).toBe("true");
+    expect(attr(parent!, "langfuse.observation.status_message")).toBe("Turn interrupted by user");
 
     // The failing exec is recorded as an ERROR-level tool span.
     const failedTool = spans.find(
@@ -159,6 +213,20 @@ describe("convertRollout", () => {
     expect(attr(failedTool!, "langfuse.observation.status_message")).toContain("command failed");
   });
 
+  it("does not emit an incomplete subagent turn", async () => {
+    const dir = stageFixtures();
+    const childFile = path.join(dir, "rollout-child-thread-child.jsonl");
+    const childLines = fs.readFileSync(childFile, "utf-8").trimEnd().split("\n");
+    fs.writeFileSync(childFile, `${childLines.slice(0, -1).join("\n")}\n`);
+
+    await convertRollout(path.join(dir, "rollout-parent.jsonl"), { config: baseConfig });
+
+    const childTurns = exporter
+      .getFinishedSpans()
+      .filter((span) => span.name === "Codex Subagent Turn" && obsType(span) === "agent");
+    expect(childTurns).toHaveLength(0);
+  });
+
   it("nests subagent turns discovered via sub_agent_activity events", async () => {
     const dir = stageFixtures();
     await convertRollout(path.join(dir, "rollout-activity-main.jsonl"), { config: baseConfig });
@@ -206,9 +274,10 @@ describe("convertRollout", () => {
     const dir = stageFixtures();
     const file = path.join(dir, "rollout-basic-main.jsonl");
 
-    await convertRollout(file, { config: baseConfig });
+    const uploadedTurnIds = await convertRollout(file, { config: baseConfig });
     const firstCount = exporter.getFinishedSpans().length;
     expect(firstCount).toBeGreaterThan(0);
+    for (const turnId of uploadedTurnIds) await markTurnUploaded(file, turnId);
     expect(fs.existsSync(`${file}.langfuse`)).toBe(true);
 
     exporter.reset();
@@ -308,8 +377,9 @@ describe("deterministic trace ids (trace_seed)", () => {
     const dir = stageFixtures();
     const file = path.join(dir, "rollout-two-turns-main.jsonl");
 
-    await convertRollout(file, { config: seededConfig });
+    const uploadedTurnIds = await convertRollout(file, { config: seededConfig });
     expect(turnRoots()).toHaveLength(2);
+    for (const turnId of uploadedTurnIds) await markTurnUploaded(file, turnId);
     expect(fs.existsSync(`${file}.langfuse`)).toBe(true);
 
     exporter.reset();