From b31147f7517339076ec46ae6df3ebf8a294e7cdc Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 21 Oct 2026 09:12:06 +0000 Subject: [PATCH] fix(tracing): bound earlier-turn history in generation inputs (max_history_chars) Every generation's input carried the full history of all earlier turns, so memory grew with steps x session size and long sessions hit a V8 heap OOM in the Stop hook, which aborts before turns are marked uploaded and fails on every later Stop. Keep the most recent earlier-turn messages within a character budget (default 200k, LANGFUSE_CODEX_MAX_HISTORY_CHARS), replace the rest with an omission marker, and record the count in codex.history_omitted_messages. Fixes #119 Co-authored-by: Hassieb Pakzad --- README.md | 1 + plugins/tracing/src/config.ts | 15 +++- plugins/tracing/src/trace.ts | 82 ++++++++++++++++--- plugins/tracing/test/config.test.ts | 30 +++++++ plugins/tracing/test/instrumentation.test.ts | 1 + .../test/trace-attached-ownership.test.ts | 1 + .../tracing/test/trace-seed-failure.test.ts | 1 + plugins/tracing/test/trace.test.ts | 70 ++++++++++++++++ .../tracing/test/turn-emit-failure.test.ts | 1 + 9 files changed, 188 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 774f3e7..2a9071e 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ The remaining options are settable both ways, as a `langfuse.json` key or as the - `trace_seed` derives deterministic trace ids, so a headless caller knows a run's trace id up front. Use a unique seed per session. - `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`. +- `max_history_chars` caps how many characters of earlier turns each generation's input carries, and defaults to `200000`. The most recent messages are kept and older ones are replaced by a single "earlier message(s) omitted" marker, with the count recorded in the generation's `codex.history_omitted_messages` metadata. The current turn's own messages are never dropped. Raising it keeps more context in traces but costs memory in the hook on long sessions; `0` drops earlier turns entirely. Everything the plugin traces is uploaded to Langfuse, including prompts and tool inputs and outputs, so do not enable it for sessions containing data you do not want stored there. diff --git a/plugins/tracing/src/config.ts b/plugins/tracing/src/config.ts index 1a82fc3..c035c35 100644 --- a/plugins/tracing/src/config.ts +++ b/plugins/tracing/src/config.ts @@ -27,18 +27,23 @@ export const ConfigSchema = z.object({ trace_seed: z.string().optional(), debug: z.boolean(), fail_on_error: z.boolean(), + max_history_chars: z.number().int().nonnegative(), }); export type Config = z.infer; const PartialConfigSchema = ConfigSchema.partial(); -const DEFAULTS: Pick = { +const DEFAULTS: Pick< + Config, + "enabled" | "base_url" | "skill_tags" | "debug" | "fail_on_error" | "max_history_chars" +> = { enabled: false, base_url: "https://cloud.langfuse.com", skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, }; const CodexAuthSchema = z @@ -60,6 +65,11 @@ function parseBoolean(value: unknown): boolean | undefined { return undefined; } +function parseNonNegativeInt(value: unknown): number | undefined { + const n = typeof value === "string" && value.trim() !== "" ? Number(value) : value; + return typeof n === "number" && Number.isInteger(n) && n >= 0 ? n : 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; @@ -115,6 +125,8 @@ async function readConfigFile(file: string): Promise | undefined skill_tags: raw.skill_tags != null ? parseBoolean(raw.skill_tags) : undefined, debug: raw.debug != null ? parseBoolean(raw.debug) : undefined, fail_on_error: raw.fail_on_error != null ? parseBoolean(raw.fail_on_error) : undefined, + max_history_chars: + raw.max_history_chars != null ? parseNonNegativeInt(raw.max_history_chars) : undefined, }), ); } catch { @@ -171,6 +183,7 @@ function readEnvConfig(env: Record): Partial trace_seed: env.LANGFUSE_CODEX_TRACE_SEED, debug: parseBoolean(env.LANGFUSE_CODEX_DEBUG), fail_on_error: parseBoolean(env.LANGFUSE_CODEX_FAIL_ON_ERROR), + max_history_chars: parseNonNegativeInt(env.LANGFUSE_CODEX_MAX_HISTORY_CHARS), }), ); } diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index b67396a..8857ef1 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -367,9 +367,68 @@ function turnHistoryMessages(turn: Turn): ChatMlMessage[] { return messages; } +function messageChars(message: ChatMlMessage): number { + let n = 0; + if (typeof message.content === "string") { + n += message.content.length; + } else if (Array.isArray(message.content)) { + for (const part of message.content) { + n += part.type === "text" ? part.text.length : part.image_url.url.length; + } + } + if (message.role === "assistant") { + for (const part of message.thinking ?? []) n += part.content.length; + for (const tc of message.tool_calls ?? []) { + n += tc.function.name.length + (tc.function.arguments?.length ?? 0); + } + } + return n; +} + +type History = { messages: ChatMlMessage[]; omitted: number }; + +/** + * The messages of the turns before `turnIndex`, newest first until `budget` + * characters are used, with a marker standing in for everything older. Every + * generation's input is serialized onto its own span, so an unbounded history + * grows memory with steps × session size and long sessions exhaust the heap. + */ +function boundedHistory( + turnMessages: readonly ChatMlMessage[][], + turnIndex: number, + budget: number, +): History { + const kept: ChatMlMessage[] = []; + let total = 0; + let used = 0; + let full = false; + for (let t = turnIndex - 1; t >= 0; t--) { + const messages = turnMessages[t]; + total += messages.length; + for (let m = messages.length - 1; m >= 0 && !full; m--) { + const size = messageChars(messages[m]); + if (used + size > budget) { + full = true; + } else { + used += size; + kept.push(messages[m]); + } + } + } + const omitted = total - kept.length; + kept.reverse(); + if (omitted > 0) { + kept.unshift({ + role: "system", + content: `[${omitted} earlier message(s) omitted from this trace to stay within max_history_chars]`, + }); + } + return { messages: kept, omitted }; +} + function generationInput( systemMessage: string | undefined, - historyPrefix: ChatMlMessage[], + historyPrefix: readonly ChatMlMessage[], turn: Turn, stepIndex: number, ): ChatMlMessage[] | undefined { @@ -434,7 +493,7 @@ async function emitTurn( seenThreadIds: Set; unannouncedSubagents?: SubagentRollout[]; inheritableTurnIds?: ReadonlySet; - historyPrefix?: ChatMlMessage[]; + history?: History; }, ): Promise { const isSubagent = sessionMeta.isSubagentThread === true || ctx.parentObservation != null; @@ -485,7 +544,8 @@ async function emitTurn( let failure: unknown; try { const systemMessage = systemPromptText(turn.systemPrompt); - const historyPrefix = ctx.historyPrefix ?? []; + const historyPrefix = ctx.history?.messages ?? []; + const omittedHistory = ctx.history?.omitted ?? 0; const spawnObservations = new Map(); for (let i = 0; i < turn.steps.length; i++) { @@ -506,6 +566,7 @@ async function emitTurn( metadata: { "codex.step_index": i, "codex.reasoning_effort": turn.reasoningEffort, + ...(omittedHistory > 0 ? { "codex.history_omitted_messages": omittedHistory } : {}), }, }, { @@ -643,14 +704,9 @@ export async function convertRollout( ): Promise { const { sessionMeta, turns } = parseSession(await loadSession(rolloutFile)); - const historyPrefixes: ChatMlMessage[][] = []; - { - const seen: ChatMlMessage[] = []; - for (const turn of turns) { - historyPrefixes.push([...seen]); - seen.push(...turnHistoryMessages(turn)); - } - } + const turnMessages = turns.map(turnHistoryMessages); + const historyFor = (turnIndex: number): History => + boundedHistory(turnMessages, turnIndex, options.config.max_history_chars); debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`); const subagentIndex = @@ -716,7 +772,7 @@ export async function convertRollout( seenThreadIds, unannouncedSubagents: subagentsFor(turnIndex), inheritableTurnIds, - historyPrefix: historyPrefixes[turnIndex], + history: historyFor(turnIndex), }); } return []; @@ -753,7 +809,7 @@ export async function convertRollout( seenThreadIds, unannouncedSubagents: subagentsFor(turnIndex), inheritableTurnIds, - historyPrefix: historyPrefixes[turnIndex], + history: historyFor(turnIndex), }); if (attached) { diff --git a/plugins/tracing/test/config.test.ts b/plugins/tracing/test/config.test.ts index 741a158..ebda282 100644 --- a/plugins/tracing/test/config.test.ts +++ b/plugins/tracing/test/config.test.ts @@ -249,4 +249,34 @@ describe("getConfig", () => { }); expect(fromEnv.fail_on_error).toBe(false); }); + + it("parses max_history_chars from config and environment, ignoring invalid values", async () => { + const defaults = await getConfig({ home: emptyHome(), cwd: emptyHome(), env: {} }); + expect(defaults.max_history_chars).toBe(200_000); + + const home = makeTmpHome({ + rel: ".codex/langfuse.json", + contents: { public_key: "pk-lf-kept", 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); + + const invalid = await getConfig({ + home: makeTmpHome({ + rel: ".codex/langfuse.json", + contents: { public_key: "pk-lf-kept", max_history_chars: -1 }, + }), + cwd: emptyHome(), + env: { LANGFUSE_CODEX_MAX_HISTORY_CHARS: "lots" }, + }); + expect(invalid.max_history_chars).toBe(200_000); + expect(invalid.public_key).toBe("pk-lf-kept"); + }); }); diff --git a/plugins/tracing/test/instrumentation.test.ts b/plugins/tracing/test/instrumentation.test.ts index 444335d..e028d7f 100644 --- a/plugins/tracing/test/instrumentation.test.ts +++ b/plugins/tracing/test/instrumentation.test.ts @@ -36,6 +36,7 @@ const baseConfig: Config = { skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, }; const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); diff --git a/plugins/tracing/test/trace-attached-ownership.test.ts b/plugins/tracing/test/trace-attached-ownership.test.ts index 1519a76..dde6236 100644 --- a/plugins/tracing/test/trace-attached-ownership.test.ts +++ b/plugins/tracing/test/trace-attached-ownership.test.ts @@ -30,6 +30,7 @@ const baseConfig: Config = { skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, }; const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); diff --git a/plugins/tracing/test/trace-seed-failure.test.ts b/plugins/tracing/test/trace-seed-failure.test.ts index 2673ec9..463f15d 100644 --- a/plugins/tracing/test/trace-seed-failure.test.ts +++ b/plugins/tracing/test/trace-seed-failure.test.ts @@ -33,6 +33,7 @@ const baseConfig: Config = { skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, trace_seed: "ci-run-42", }; diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index 36e015e..ea2748e 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -28,6 +28,7 @@ const baseConfig: Config = { skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, }; const fixturesRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures/sessions"); @@ -271,6 +272,75 @@ describe("convertRollout", () => { fs.rmSync(dir, { recursive: true, force: true }); }); + it("keeps only the most recent history within max_history_chars", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-hist-cap-")); + const day = path.join(dir, "sessions", "2026", "06", "03"); + fs.mkdirSync(day, { recursive: true }); + const file = path.join(day, "rollout-history-cap.jsonl"); + const ts = (n: number): string => `2026-06-03T09:00:${String(n).padStart(2, "0")}.000Z`; + const turn = (n: number, id: string, prompt: string, answer: string): unknown[] => [ + { timestamp: ts(n), type: "event_msg", payload: { type: "task_started", turn_id: id } }, + { timestamp: ts(n), type: "event_msg", payload: { type: "user_message", message: prompt } }, + { + timestamp: ts(n + 1), + type: "response_item", + payload: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: answer }], + }, + }, + { timestamp: ts(n + 1), type: "event_msg", payload: { type: "token_count", info: {} } }, + { timestamp: ts(n + 2), type: "event_msg", payload: { type: "task_complete", turn_id: id } }, + ]; + fs.writeFileSync( + file, + [ + { timestamp: ts(0), type: "session_meta", payload: { id: "sess-hist-cap" } }, + ...turn(1, "t1", "First question", "First answer"), + ...turn(4, "t2", "Second question", "Second answer"), + ...turn(7, "t3", "Third question", "Third answer"), + ] + .map((l) => JSON.stringify(l)) + .join("\n"), + ); + + await convertAndMark(file, { + config: { ...baseConfig, max_history_chars: 30 }, + stoppedTurnId: "t3", + }); + + const generations = exporter + .getFinishedSpans() + .filter((s) => obsType(s) === "generation") + .sort((a, b) => startMs(a) - startMs(b)); + expect(generations).toHaveLength(3); + + expect(JSON.parse(attr(generations[1], "langfuse.observation.input"))).toEqual([ + { role: "user", content: "First question" }, + { role: "assistant", content: "First answer" }, + { role: "user", content: "Second question" }, + ]); + expect( + attr(generations[1], "langfuse.observation.metadata.codex.history_omitted_messages"), + ).toBe(""); + + expect(JSON.parse(attr(generations[2], "langfuse.observation.input"))).toEqual([ + { + role: "system", + content: "[2 earlier message(s) omitted from this trace to stay within max_history_chars]", + }, + { role: "user", content: "Second question" }, + { role: "assistant", content: "Second answer" }, + { role: "user", content: "Third question" }, + ]); + expect( + attr(generations[2], "langfuse.observation.metadata.codex.history_omitted_messages"), + ).toBe("2"); + + fs.rmSync(dir, { recursive: true, force: true }); + }); + it("sends attached images as multimodal content and the loaded tools on the call", async () => { const URI = "data:image/png;base64,iVBORw0KGgo="; const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-media-")); diff --git a/plugins/tracing/test/turn-emit-failure.test.ts b/plugins/tracing/test/turn-emit-failure.test.ts index 6c81fa9..b9e477f 100644 --- a/plugins/tracing/test/turn-emit-failure.test.ts +++ b/plugins/tracing/test/turn-emit-failure.test.ts @@ -32,6 +32,7 @@ const baseConfig: Config = { skill_tags: true, debug: false, fail_on_error: false, + max_history_chars: 200_000, }; const line = (ts: string, type: string, payload: Record) =>