fix: bound generation history and flush per turn to stop Stop-hook OOM on long sessions

README.md

@@ -65,6 +65,7 @@ The remaining options are settable both ways, as a `langfuse.json` key or as the
 - `metadata` attaches a JSON object to every trace.
 - `skill_tags` tags traces with `skill:<name>` for every skill invoked in the turn, and defaults to `true`.
 - `trace_seed` derives deterministic trace ids, so a headless caller knows a run's trace id up front. Use a unique seed per session.
+- `max_history_chars` caps the conversation history attached to each model generation, keeping the most recent messages up to this many serialized characters and replacing older ones with a single "earlier message(s) omitted" marker. It defaults to `200000`, and `0` disables the cap. Tool spans always keep their full output. Raising it or disabling it can exhaust memory on long sessions.
 - `debug` logs verbosely to stderr, and defaults to `false`.
 - `fail_on_error` fails the hook on upload errors instead of failing open, and defaults to `false`.
 

plugins/tracing/src/config.ts

@@ -25,6 +25,7 @@ export const ConfigSchema = z.object({
   metadata: z.record(z.string(), z.string()).optional(),
   skill_tags: z.boolean(),
   trace_seed: z.string().optional(),
+  max_history_chars: z.number().int().nonnegative(),
   debug: z.boolean(),
   fail_on_error: z.boolean(),
 });
@@ -33,10 +34,14 @@ export type Config = z.infer<typeof ConfigSchema>;
 
 const PartialConfigSchema = ConfigSchema.partial();
 
-const DEFAULTS: Pick<Config, "enabled" | "base_url" | "skill_tags" | "debug" | "fail_on_error"> = {
+const DEFAULTS: Pick<
+  Config,
+  "enabled" | "base_url" | "skill_tags" | "max_history_chars" | "debug" | "fail_on_error"
+> = {
   enabled: false,
   base_url: "https://cloud.langfuse.com",
   skill_tags: true,
+  max_history_chars: 200_000,
   debug: false,
   fail_on_error: false,
 };
@@ -60,6 +65,16 @@ function parseBoolean(value: unknown): boolean | undefined {
   return undefined;
 }
 
+function parseNonNegativeInt(value: unknown): number | undefined {
+  const parsed =
+    typeof value === "number"
+      ? value
+      : typeof value === "string" && value.trim()
+        ? Number(value)
+        : NaN;
+  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : undefined;
+}
+
 function parseTags(value: unknown): string[] | undefined {
   if (Array.isArray(value)) return value.map(String);
   if (typeof value !== "string" || value.trim().length === 0) return undefined;
@@ -113,6 +128,8 @@ async function readConfigFile(file: string): Promise<Partial<Config> | undefined
         tags: raw.tags != null ? parseTags(raw.tags) : undefined,
         metadata: raw.metadata != null ? parseMetadata(raw.metadata) : undefined,
         skill_tags: raw.skill_tags != null ? parseBoolean(raw.skill_tags) : undefined,
+        max_history_chars:
+          raw.max_history_chars != null ? parseNonNegativeInt(raw.max_history_chars) : undefined,
         debug: raw.debug != null ? parseBoolean(raw.debug) : undefined,
         fail_on_error: raw.fail_on_error != null ? parseBoolean(raw.fail_on_error) : undefined,
       }),
@@ -169,6 +186,7 @@ function readEnvConfig(env: Record<string, string | undefined>): Partial<Config>
       metadata: parseMetadata(env.LANGFUSE_CODEX_METADATA),
       skill_tags: parseBoolean(env.LANGFUSE_CODEX_SKILL_TAGS),
       trace_seed: env.LANGFUSE_CODEX_TRACE_SEED,
+      max_history_chars: parseNonNegativeInt(env.LANGFUSE_CODEX_MAX_HISTORY_CHARS),
       debug: parseBoolean(env.LANGFUSE_CODEX_DEBUG),
       fail_on_error: parseBoolean(env.LANGFUSE_CODEX_FAIL_ON_ERROR),
     }),

plugins/tracing/src/index.ts

@@ -51,14 +51,23 @@ export async function runHook(): Promise<void> {
   }
 
   const instrumentation = setupInstrumentation(config);
-  let exportedTurnIds: string[] = [];
+  const transcriptPath = hookInput.transcript_path;
   let failure: unknown;
 
   try {
-    exportedTurnIds = await convertRollout(hookInput.transcript_path, {
+    await convertRollout(transcriptPath, {
       config,
       parentSpanContext,
       stoppedTurnId: hookInput.turn_id ?? undefined,
+      // Flush and record each turn before converting the next: a backlog is
+      // delivered incrementally instead of all-or-nothing, and only one turn's
+      // spans are held in memory at a time. Record delivery only after the
+      // flush: a turn marked despite a failed export is lost silently, while
+      // an unmarked one is retried.
+      onTurnExported: async (turnId) => {
+        await instrumentation.flush();
+        await markTurnUploaded(transcriptPath, turnId);
+      },
     });
   } catch (error) {
     failure = error;
@@ -72,16 +81,9 @@ export async function runHook(): Promise<void> {
     debugLog("error during flush/shutdown:", error);
   }
 
-  // Record delivery only after the flush: a turn marked despite a failed export
-  // is lost silently, while an unmarked one is retried.
   if (failure) {
     if (config.fail_on_error) throw failure;
-    debugLog("export incomplete; leaving turns unmarked so they stay retryable");
-    return;
-  }
-
-  for (const turnId of exportedTurnIds) {
-    await markTurnUploaded(hookInput.transcript_path, turnId);
+    debugLog("export incomplete; leaving the remaining turns unmarked so they stay retryable");
   }
 }
 

plugins/tracing/src/instrumentation.ts

@@ -37,6 +37,8 @@ function seededId(kind: "trace" | "span"): string {
 }
 
 export type Instrumentation = {
+  /** Export every buffered span; rejects if the export fails. */
+  flush: () => Promise<void>;
   /** Flush buffered spans and tear down the tracer provider. */
   shutdown: () => Promise<void>;
 };
@@ -57,10 +59,11 @@ function buildResource() {
  * instrumentation loading. Registering the provider also installs the
  * AsyncLocalStorage context manager that `propagateAttributes` relies on.
  *
- * We use `exportMode: "batched"` and flush once at the end: the whole rollout
- * is converted in-process, so batching every span into one (or a few) requests
- * is far faster than one request per span — important for the hook's timeout
- * budget. `shutdown()` below calls `forceFlush()` before the process exits.
+ * We use `exportMode: "batched"` and the hook calls `flush()` once per turn:
+ * batching a turn's spans into one (or a few) requests is far faster than one
+ * request per span — important for the hook's timeout budget — while flushing
+ * between turns keeps a backlog of pending turns from piling up in memory.
+ * `shutdown()` below calls `forceFlush()` before the process exits.
  */
 export function setupInstrumentation(config: Config): Instrumentation {
   const spanProcessor = new LangfuseSpanProcessor({
@@ -86,6 +89,7 @@ export function setupInstrumentation(config: Config): Instrumentation {
   setLangfuseTracerProvider(provider);
 
   return {
+    flush: () => spanProcessor.forceFlush(),
     shutdown: async () => {
       await spanProcessor.forceFlush();
       await provider.shutdown();

plugins/tracing/src/trace.ts

@@ -367,21 +367,55 @@ function turnHistoryMessages(turn: Turn): ChatMlMessage[] {
   return messages;
 }
 
+const messageChars = new WeakMap<ChatMlMessage, number>();
+
+function serializedChars(message: ChatMlMessage): number {
+  let chars = messageChars.get(message);
+  if (chars === undefined) {
+    chars = JSON.stringify(message).length;
+    messageChars.set(message, chars);
+  }
+  return chars;
+}
+
+/**
+ * The conversation a generation saw, keeping only the most recent messages
+ * that fit in `maxChars` (0 = unbounded). Every generation serializes its
+ * input onto its span, so an unbounded history costs memory quadratic in the
+ * session length and can crash the hook on long sessions.
+ */
+function boundedConversation(
+  earlier: readonly ChatMlMessage[],
+  current: readonly ChatMlMessage[],
+  maxChars: number,
+): ChatMlMessage[] {
+  const all = [...earlier, ...current];
+  if (maxChars <= 0) return all;
+  let used = 0;
+  let start = all.length;
+  while (start > 0) {
+    const chars = serializedChars(all[start - 1]);
+    if (used + chars > maxChars) break;
+    used += chars;
+    start--;
+  }
+  if (start === 0) return all;
+  return [
+    {
+      role: "system",
+      content: `[${start} earlier message(s) omitted: the conversation exceeds max_history_chars (${maxChars})]`,
+    },
+    ...all.slice(start),
+  ];
+}
+
 function generationInput(
   systemMessage: string | undefined,
-  historyPrefix: ChatMlMessage[],
-  turn: Turn,
-  stepIndex: number,
+  conversation: ChatMlMessage[],
 ): ChatMlMessage[] | undefined {
   const messages: ChatMlMessage[] = [];
   if (systemMessage) messages.push({ role: "system", content: systemMessage });
-  messages.push(...historyPrefix);
-  const user = userMessage(turn);
-  if (user) messages.push(user);
-  for (let j = 0; j < stepIndex; j++) {
-    messages.push(assistantMessage(turn.steps[j]));
-    messages.push(...toolMessages(turn.steps[j]));
-  }
+  messages.push(...conversation);
   return messages.length > 0 ? messages : undefined;
 }
 
@@ -487,14 +521,21 @@ async function emitTurn(
     const systemMessage = systemPromptText(turn.systemPrompt);
     const historyPrefix = ctx.historyPrefix ?? [];
     const spawnObservations = new Map<string, LangfuseObservation>();
+    const user = userMessage(turn);
+    const turnMessages: ChatMlMessage[] = user ? [user] : [];
 
     for (let i = 0; i < turn.steps.length; i++) {
       const step = turn.steps[i];
+      const conversation = boundedConversation(
+        historyPrefix,
+        turnMessages,
+        ctx.config.max_history_chars,
+      );
       const generation = startObservation(
         isSubagent ? "LLM Subagent" : "LLM",
         {
           input: attachToolDefinitions(
-            generationInput(systemMessage, historyPrefix, turn, i),
+            generationInput(systemMessage, conversation),
             turn.toolDefinitions,
           ),
           output: buildGenerationOutput(step),
@@ -521,6 +562,7 @@ async function emitTurn(
       }
 
       generation.end(new Date(generationEnd(step)));
+      turnMessages.push(assistantMessage(step), ...toolMessages(step));
     }
 
     const announced: SubagentRollout[] = [];
@@ -639,6 +681,8 @@ export async function convertRollout(
     seenThreadIds?: Set<string>;
     ancestorTurnIds?: ReadonlySet<string>;
     stoppedTurnId?: string;
+    /** Awaited after each exported turn, before the next one is converted. */
+    onTurnExported?: (turnId: string) => Promise<void>;
   },
 ): Promise<string[]> {
   const { sessionMeta, turns } = parseSession(await loadSession(rolloutFile));
@@ -774,6 +818,7 @@ export async function convertRollout(
 
     uploaded.add(turn.turnId);
     exportedTurnIds.push(turn.turnId);
+    await options.onTurnExported?.(turn.turnId);
   }
 
   return exportedTurnIds;

plugins/tracing/test/config.test.ts

@@ -39,6 +39,37 @@ describe("getConfig", () => {
     expect(config.base_url).toBe("https://cloud.langfuse.com");
     expect(config.fail_on_error).toBe(false);
     expect(config.skill_tags).toBe(true);
+    expect(config.max_history_chars).toBe(200_000);
+  });
+
+  it("reads max_history_chars from the config file and the environment", async () => {
+    const home = makeTmpHome({
+      rel: ".codex/langfuse.json",
+      contents: { public_key: "pk-lf-file", max_history_chars: 5000 },
+    });
+    const fromFile = await getConfig({ home, cwd: emptyHome(), env: {} });
+    expect(fromFile.max_history_chars).toBe(5000);
+
+    const fromEnv = await getConfig({
+      home,
+      cwd: emptyHome(),
+      env: { LANGFUSE_CODEX_MAX_HISTORY_CHARS: "0" },
+    });
+    expect(fromEnv.max_history_chars).toBe(0);
+  });
+
+  it("ignores an invalid max_history_chars without discarding the rest of the file", async () => {
+    const home = makeTmpHome({
+      rel: ".codex/langfuse.json",
+      contents: { public_key: "pk-lf-file", max_history_chars: -1 },
+    });
+    const config = await getConfig({
+      home,
+      cwd: emptyHome(),
+      env: { LANGFUSE_CODEX_MAX_HISTORY_CHARS: "lots" },
+    });
+    expect(config.public_key).toBe("pk-lf-file");
+    expect(config.max_history_chars).toBe(200_000);
   });
 
   it("ignores a stale max_chars key without discarding the rest of the file", async () => {

plugins/tracing/test/hook-command.test.ts

@@ -163,6 +163,80 @@ describe("bundled Stop hook command", () => {
     }
   });
 
+  it("marks each delivered turn even when a later turn's upload fails", async () => {
+    const codexHome = makeTempDir("lf-codex-home-");
+    const sessionCwd = makeTempDir("lf-codex-cwd-");
+    const sessionsDir = path.join(sessionCwd, "sessions", "2026", "06", "03");
+    fs.mkdirSync(sessionsDir, { recursive: true });
+    const rollout = path.join(sessionsDir, "rollout.jsonl");
+
+    const event = (payload: Record<string, unknown>) =>
+      JSON.stringify({ timestamp: "2026-06-03T12:00:00.000Z", type: "event_msg", payload });
+    fs.writeFileSync(
+      rollout,
+      [
+        JSON.stringify({
+          timestamp: "2026-06-03T12:00:00.000Z",
+          type: "session_meta",
+          payload: { id: "sess-backlog", cli_version: "0.160.1" },
+        }),
+        event({ type: "task_started", turn_id: "turn-1" }),
+        event({ type: "user_message", message: "First" }),
+        event({ type: "agent_message", message: "One" }),
+        event({ type: "task_complete", turn_id: "turn-1" }),
+        event({ type: "task_started", turn_id: "turn-2" }),
+        event({ type: "user_message", message: "Second" }),
+        event({ type: "agent_message", message: "Two" }),
+        event({ type: "task_complete", turn_id: "turn-2" }),
+      ].join("\n") + "\n",
+    );
+
+    const received: string[] = [];
+    const server = http.createServer((req, res) => {
+      const chunks: Buffer[] = [];
+      req.on("data", (chunk) => chunks.push(chunk as Buffer));
+      req.on("end", () => {
+        received.push(Buffer.concat(chunks).toString("utf-8"));
+        // Accept the first turn's upload, then reject with a non-retryable status.
+        res.writeHead(received.length === 1 ? 200 : 400, { "content-type": "application/json" });
+        res.end("{}");
+      });
+    });
+    await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve));
+    const { port } = server.address() as { port: number };
+
+    try {
+      const { code } = await runShellCommand(readHookCommand(), {
+        cwd: sessionCwd,
+        env: {
+          ...process.env,
+          PLUGIN_ROOT: pluginRootDir,
+          CODEX_HOME: codexHome,
+          HOME: codexHome,
+          TRACE_TO_LANGFUSE: "true",
+          LANGFUSE_PUBLIC_KEY: "pk-lf-test",
+          LANGFUSE_SECRET_KEY: "sk-lf-test",
+          LANGFUSE_BASE_URL: `http://127.0.0.1:${port}`,
+        },
+        input: JSON.stringify({
+          hook_event_name: "Stop",
+          session_id: "sess-backlog",
+          turn_id: "turn-2",
+          transcript_path: rollout,
+        }),
+      });
+
+      expect(code).toBe(0);
+      expect(received).toHaveLength(2);
+      expect(received[0]).toContain("turn-1");
+      expect(received[0]).not.toContain("turn-2");
+      expect(received[1]).toContain("turn-2");
+      expect(fs.readFileSync(`${rollout}.langfuse`, "utf-8").trim()).toBe("turn-1");
+    } finally {
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  });
+
   it("re-exports a rollout as a byte-identical payload so Langfuse deduplicates it", async () => {
     const codexHome = makeTempDir("lf-codex-home-");
     const sessionCwd = makeTempDir("lf-codex-cwd-");

plugins/tracing/test/instrumentation.test.ts

@@ -34,6 +34,7 @@ const baseConfig: Config = {
   secret_key: "sk-lf-test",
   base_url: "https://cloud.langfuse.com",
   skill_tags: true,
+  max_history_chars: 200_000,
   debug: false,
   fail_on_error: false,
 };

plugins/tracing/test/trace-attached-ownership.test.ts

@@ -28,6 +28,7 @@ const baseConfig: Config = {
   secret_key: "sk-lf-test",
   base_url: "https://cloud.langfuse.com",
   skill_tags: true,
+  max_history_chars: 200_000,
   debug: false,
   fail_on_error: false,
 };

plugins/tracing/test/trace-seed-failure.test.ts

@@ -31,6 +31,7 @@ const baseConfig: Config = {
   secret_key: "sk-lf-test",
   base_url: "https://cloud.langfuse.com",
   skill_tags: true,
+  max_history_chars: 200_000,
   debug: false,
   fail_on_error: false,
   trace_seed: "ci-run-42",