diff --git a/README.md b/README.md index 774f3e7..3ac95ed 100644 --- a/README.md +++ b/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:` 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`. diff --git a/plugins/tracing/src/config.ts b/plugins/tracing/src/config.ts index 1a82fc3..2a11476 100644 --- a/plugins/tracing/src/config.ts +++ b/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; const PartialConfigSchema = ConfigSchema.partial(); -const DEFAULTS: Pick = { +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 | 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): Partial 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), }), diff --git a/plugins/tracing/src/index.ts b/plugins/tracing/src/index.ts index 2d535d8..ffe7a87 100644 --- a/plugins/tracing/src/index.ts +++ b/plugins/tracing/src/index.ts @@ -51,14 +51,23 @@ export async function runHook(): Promise { } 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 { 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"); } } diff --git a/plugins/tracing/src/instrumentation.ts b/plugins/tracing/src/instrumentation.ts index f7a3f05..715cc6b 100644 --- a/plugins/tracing/src/instrumentation.ts +++ b/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; /** Flush buffered spans and tear down the tracer provider. */ shutdown: () => Promise; }; @@ -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(); diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index b67396a..b0de00d 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -367,21 +367,55 @@ function turnHistoryMessages(turn: Turn): ChatMlMessage[] { return messages; } +const messageChars = new WeakMap(); + +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(); + 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; ancestorTurnIds?: ReadonlySet; stoppedTurnId?: string; + /** Awaited after each exported turn, before the next one is converted. */ + onTurnExported?: (turnId: string) => Promise; }, ): Promise { 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; diff --git a/plugins/tracing/test/config.test.ts b/plugins/tracing/test/config.test.ts index 741a158..3d0319a 100644 --- a/plugins/tracing/test/config.test.ts +++ b/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 () => { diff --git a/plugins/tracing/test/hook-command.test.ts b/plugins/tracing/test/hook-command.test.ts index 221b7e3..f68cb9c 100644 --- a/plugins/tracing/test/hook-command.test.ts +++ b/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) => + 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((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((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-"); diff --git a/plugins/tracing/test/instrumentation.test.ts b/plugins/tracing/test/instrumentation.test.ts index 444335d..59eb8b9 100644 --- a/plugins/tracing/test/instrumentation.test.ts +++ b/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, }; diff --git a/plugins/tracing/test/trace-attached-ownership.test.ts b/plugins/tracing/test/trace-attached-ownership.test.ts index 1519a76..7aaa37a 100644 --- a/plugins/tracing/test/trace-attached-ownership.test.ts +++ b/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, }; diff --git a/plugins/tracing/test/trace-seed-failure.test.ts b/plugins/tracing/test/trace-seed-failure.test.ts index 2673ec9..aa00781 100644 --- a/plugins/tracing/test/trace-seed-failure.test.ts +++ b/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", diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index 36e015e..fd8c1be 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -26,6 +26,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, }; @@ -271,6 +272,93 @@ describe("convertRollout", () => { fs.rmSync(dir, { recursive: true, force: true }); }); + it("keeps only the most recent messages within max_history_chars in a generation input", async () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lf-codex-bounded-")); + const day = path.join(dir, "sessions", "2026", "06", "03"); + fs.mkdirSync(day, { recursive: true }); + const file = path.join(day, "rollout-bounded.jsonl"); + const ts = (n: number): string => `2026-06-03T09:00:${String(n).padStart(2, "0")}.000Z`; + const LONG_ANSWER = "x".repeat(1000); + const lines: unknown[] = [ + { timestamp: ts(0), type: "session_meta", payload: { id: "sess-bounded" } }, + { timestamp: ts(1), type: "event_msg", payload: { type: "task_started", turn_id: "t1" } }, + { timestamp: ts(1), type: "event_msg", payload: { type: "user_message", message: "First" } }, + { + timestamp: ts(2), + type: "response_item", + payload: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: LONG_ANSWER }], + }, + }, + { timestamp: ts(2), type: "event_msg", payload: { type: "token_count", info: {} } }, + { timestamp: ts(3), type: "event_msg", payload: { type: "task_complete", turn_id: "t1" } }, + { timestamp: ts(4), type: "event_msg", payload: { type: "task_started", turn_id: "t2" } }, + { timestamp: ts(4), type: "event_msg", payload: { type: "user_message", message: "Second" } }, + { + timestamp: ts(5), + type: "response_item", + payload: { type: "function_call", name: "exec_command", call_id: "c1", arguments: "{}" }, + }, + { + timestamp: ts(6), + type: "response_item", + payload: { type: "function_call_output", call_id: "c1", output: "a.txt" }, + }, + { timestamp: ts(6), type: "event_msg", payload: { type: "token_count", info: {} } }, + { + timestamp: ts(7), + type: "response_item", + payload: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "Done" }], + }, + }, + { timestamp: ts(7), type: "event_msg", payload: { type: "token_count", info: {} } }, + { timestamp: ts(8), type: "event_msg", payload: { type: "task_complete", turn_id: "t2" } }, + ]; + fs.writeFileSync(file, lines.map((l) => JSON.stringify(l)).join("\n")); + + const generationInputs = () => + exporter + .getFinishedSpans() + .filter((s) => obsType(s) === "generation") + .sort((a, b) => startMs(a) - startMs(b)) + .map((s) => JSON.parse(attr(s, "langfuse.observation.input"))); + + await convertRollout(file, { config: { ...baseConfig, max_history_chars: 500 } }); + const bounded = generationInputs(); + expect(bounded).toHaveLength(3); + expect(bounded[0]).toEqual([{ role: "user", content: "First" }]); + // The long answer of turn 1 no longer fits, so it and everything before it + // collapse into one marker; earlier steps of the current turn are bounded too. + expect(bounded[2]).toEqual([ + { + role: "system", + content: "[2 earlier message(s) omitted: the conversation exceeds max_history_chars (500)]", + }, + { role: "user", content: "Second" }, + expect.objectContaining({ role: "assistant", tool_calls: [expect.anything()] }), + expect.objectContaining({ role: "tool", tool_call_id: "c1", content: "a.txt" }), + ]); + + exporter.reset(); + fs.rmSync(`${file}.langfuse`, { force: true }); + await convertRollout(file, { config: { ...baseConfig, max_history_chars: 0 } }); + const unbounded = generationInputs(); + expect(unbounded[2]).toEqual([ + { role: "user", content: "First" }, + { role: "assistant", content: LONG_ANSWER }, + { role: "user", content: "Second" }, + expect.objectContaining({ role: "assistant" }), + expect.objectContaining({ role: "tool", content: "a.txt" }), + ]); + + 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..3ef280a 100644 --- a/plugins/tracing/test/turn-emit-failure.test.ts +++ b/plugins/tracing/test/turn-emit-failure.test.ts @@ -30,6 +30,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, };