From 548adfa7049474b588a833b291cef46db891ffe9 Mon Sep 17 00:00:00 2001 From: albin-zhu <1317911+albin-zhu@users.noreply.github.com> Date: Tue, 10 Nov 2026 19:02:27 +0800 Subject: [PATCH 1/3] perf: checkpoint acknowledged main rollout reads between Stop hooks --- README.md | 25 ++ plugins/tracing/src/parse.ts | 48 ++- plugins/tracing/src/rollout.ts | 183 +++++++++ plugins/tracing/src/trace.ts | 332 +++++++++++------ .../tracing/test/incremental-rollout.test.ts | 352 ++++++++++++++++++ 5 files changed, 813 insertions(+), 127 deletions(-) create mode 100644 plugins/tracing/src/rollout.ts create mode 100644 plugins/tracing/test/incremental-rollout.test.ts diff --git a/README.md b/README.md index 3ac95ed..bf74331 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,31 @@ The remaining options are settable both ways, as a `langfuse.json` key or as the 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. +## Incremental rollout reads + +The Stop hook checkpoints acknowledged main-thread turns in a private +`.langfuse.cursor` file. Subsequent hooks seek to that byte offset rather +than re-reading the acknowledged prefix. The checkpoint retains parser context +(instructions and discovered tools), original turn numbering, and only the +conversation suffix needed by `max_history_chars`. It contains conversation +content; protect it like the transcript. It is written atomically with mode `0600` +on POSIX systems; Windows inherits directory ACLs. + +The existing `.langfuse` delivery receipts remain authoritative. A checkpoint is +committed only after the export callback has completed and its turn receipt is +present. A stopped turn without a completion record is still delivered, but its +bytes remain retryable until completion or the next turn starts. Partial trailing +lines are deferred until a newline is appended. + +Missing/corrupt checkpoints, a replaced or truncated file, removed receipts, and +changes to `max_history_chars` rebuild from the transcript. Concurrent hooks may +replace the checkpoint with an older acknowledged position: this causes extra +reads, not lost turns. Subagent/ancestor traversal and unbounded history +(`max_history_chars: 0`) retain the existing full-read path for fidelity. Session +directory metadata discovery is unchanged. This optimization reduces repeated +main-rollout reads; it does not claim constant total I/O or eliminate the cost of +converting a single large turn. + ## Contributing See the [contributing guide](./CONTRIBUTING.md). diff --git a/plugins/tracing/src/parse.ts b/plugins/tracing/src/parse.ts index 5c70645..4e7560b 100644 --- a/plugins/tracing/src/parse.ts +++ b/plugins/tracing/src/parse.ts @@ -92,6 +92,14 @@ function extractToolError(payload: EventMsgPayload): string | undefined { return undefined; } +export type ParserContext = { + sessionMeta: SessionMeta; + developerMessages: string[]; + injectedContext: string[]; + toolDefinitions: ToolDefinition[]; + exportedSegmentCount: number; +}; + type MutableTurn = Turn & { lastAgentMessage?: string; userInputFallback?: string }; const TURN_OPENING_EVENTS = new Set(["user_message", "item_completed", "agent_message"]); @@ -131,25 +139,36 @@ function sessionMetaFrom(line: RolloutLine): SessionMeta { }; } -export function parseSession(lines: RolloutLine[]): { +export function parseSession( + lines: RolloutLine[], + options: { + context?: ParserContext; + onTurn?: ( + turn: Turn, + context: ParserContext, + boundary: { nextLineIndex: number; stable: boolean }, + ) => void; + } = {}, +): { sessionMeta: SessionMeta; turns: Turn[]; } { const ownHeader = lines.find((line) => line.type === "session_meta"); const sessionMeta: SessionMeta = ownHeader ? sessionMetaFrom(ownHeader) - : { sessionId: "unknown" }; + : (options.context?.sessionMeta ?? { sessionId: "unknown" }); const turns: Turn[] = []; let turn: MutableTurn | null = null; let step: ModelStep | null = null; let toolCallsById = new Map(); let lastTimestamp = Date.now(); + let lineIndex = 0; - const developerMessages: string[] = []; - const injectedContext: string[] = []; - const toolDefinitions: ToolDefinition[] = []; - let exportedSegmentCount = 0; + const developerMessages = [...(options.context?.developerMessages ?? [])]; + const injectedContext = [...(options.context?.injectedContext ?? [])]; + const toolDefinitions = [...(options.context?.toolDefinitions ?? [])]; + let exportedSegmentCount = options.context?.exportedSegmentCount ?? 0; const collect = (into: string[], text: string) => { if (!into.includes(text)) into.push(text); @@ -211,12 +230,27 @@ export function parseSession(lines: RolloutLine[]): { turn.systemPrompt = systemPromptFor(); turn.toolDefinitions = [...toolDefinitions]; turns.push(turn); + options.onTurn?.( + turn, + { + sessionMeta, + developerMessages: [...developerMessages], + injectedContext: [...injectedContext], + toolDefinitions: [...toolDefinitions], + exportedSegmentCount, + }, + { + nextLineIndex: opts.completed ? lineIndex + 1 : lineIndex, + stable: opts.completed || lineIndex < lines.length, + }, + ); } turn = null; toolCallsById = new Map(); }; - for (const line of lines) { + for (lineIndex = 0; lineIndex < lines.length; lineIndex++) { + const line = lines[lineIndex]; const ts = Number.isFinite(Date.parse(line.timestamp)) ? Date.parse(line.timestamp) : lastTimestamp; diff --git a/plugins/tracing/src/rollout.ts b/plugins/tracing/src/rollout.ts new file mode 100644 index 0000000..aca9e2a --- /dev/null +++ b/plugins/tracing/src/rollout.ts @@ -0,0 +1,183 @@ +import * as fs from "node:fs/promises"; +import { createHash, randomBytes } from "node:crypto"; + +import type { ParserContext } from "./parse.js"; +import type { RolloutLine } from "./types.js"; +import { loadUploadedTurnIds } from "./sidecar.js"; + +export type HistoryMessage = { + role: "system" | "user" | "assistant" | "tool"; + [key: string]: unknown; +}; + +export type RolloutCheckpoint = { + version: 1; + offset: number; + turnCount: number; + lastTurnId: string; + maxHistoryChars: number; + context: ParserContext; + history: HistoryMessage[]; + omittedMessages: number; + identity: string; + head: string; + anchor: string; +}; + +export type RolloutRecord = { line: RolloutLine; start: number; end: number }; +const CHUNK_SIZE = 64 * 1024; +const GUARD_SIZE = 4096; +const digest = (bytes: Buffer) => createHash("sha256").update(bytes).digest("hex"); + +async function readGuard(handle: fs.FileHandle, start: number, size: number): Promise { + const bytes = Buffer.alloc(size); + const { bytesRead } = await handle.read(bytes, 0, size, start); + return digest(bytes.subarray(0, bytesRead)); +} + +function validCheckpoint(value: unknown): value is RolloutCheckpoint { + if (!value || typeof value !== "object") return false; + const c = value as RolloutCheckpoint; + const context = c.context; + return ( + c.version === 1 && + Number.isSafeInteger(c.offset) && + c.offset >= 0 && + Number.isSafeInteger(c.turnCount) && + c.turnCount >= 0 && + typeof c.lastTurnId === "string" && + c.lastTurnId.length > 0 && + Number.isSafeInteger(c.omittedMessages) && + c.omittedMessages >= 0 && + typeof c.identity === "string" && + typeof c.head === "string" && + typeof c.anchor === "string" && + Array.isArray(c.history) && + c.history.every( + (m) => m && typeof m === "object" && ["system", "user", "assistant", "tool"].includes(m.role), + ) && + !!context && + typeof context.sessionMeta?.sessionId === "string" && + Array.isArray(context.developerMessages) && + context.developerMessages.every((m) => typeof m === "string") && + Array.isArray(context.injectedContext) && + context.injectedContext.every((m) => typeof m === "string") && + Array.isArray(context.toolDefinitions) && + context.toolDefinitions.every((t) => t && typeof t.name === "string") && + Number.isSafeInteger(context.exportedSegmentCount) && + context.exportedSegmentCount >= 0 + ); +} + +/** A checkpoint is an optimization, never a replacement for upload receipts. */ +export async function openIncrementalRollout(file: string, maxHistoryChars: number) { + const statePath = `${file}.langfuse.cursor`; + + let handle: fs.FileHandle | undefined; + try { + handle = await fs.open(file, "r"); + const stat = await handle.stat(); + if (!stat.isFile()) throw new Error("Rollout must be a regular file"); + const identity = `${stat.dev}:${stat.ino}`; + let checkpoint: RolloutCheckpoint | undefined; + if (maxHistoryChars > 0) { + try { + const candidate: unknown = JSON.parse(await fs.readFile(statePath, "utf8")); + if ( + validCheckpoint(candidate) && + candidate.identity === identity && + candidate.maxHistoryChars === maxHistoryChars && + candidate.offset <= stat.size && + (await loadUploadedTurnIds(file)).has(candidate.lastTurnId) && + candidate.head === (await readGuard(handle, 0, Math.min(GUARD_SIZE, candidate.offset))) && + candidate.anchor === + (await readGuard( + handle, + Math.max(0, candidate.offset - GUARD_SIZE), + Math.min(GUARD_SIZE, candidate.offset), + )) + ) { + checkpoint = candidate; + } + } catch { + /* Missing/corrupt checkpoints rebuild from the original transcript. */ + } + } + const records: RolloutRecord[] = []; + const buffer = Buffer.alloc(CHUNK_SIZE); + let position = checkpoint?.offset ?? 0; + let lineStart = position; + let parts: Buffer[] = []; + let bytesReadTotal = 0; + // Snapshot EOF: a concurrently appended half-line belongs to the next hook. + while (position < stat.size) { + const { bytesRead } = await handle.read( + buffer, + 0, + Math.min(buffer.length, stat.size - position), + position, + ); + if (!bytesRead) break; + bytesReadTotal += bytesRead; + let consumed = 0; + for (let i = 0; i < bytesRead; i++) { + if (buffer[i] !== 10) continue; + const fragment = buffer.subarray(consumed, i); + const raw = (parts.length ? Buffer.concat([...parts, fragment]) : fragment) + .toString("utf8") + .trim(); + const end = position + i + 1; + if (raw) { + try { + records.push({ line: JSON.parse(raw) as RolloutLine, start: lineStart, end }); + } catch { + /* Same malformed-complete-line policy as the existing parser. */ + } + } + parts = []; + consumed = i + 1; + lineStart = end; + } + if (consumed < bytesRead) parts.push(Buffer.from(buffer.subarray(consumed, bytesRead))); + position += bytesRead; + } + return { + checkpoint, + records, + bytesRead: bytesReadTotal, + async commit( + state: Omit< + RolloutCheckpoint, + "version" | "identity" | "head" | "anchor" | "maxHistoryChars" + >, + ) { + if (maxHistoryChars <= 0) return; + const value: RolloutCheckpoint = { + ...state, + version: 1, + identity, + maxHistoryChars, + head: await readGuard(handle!, 0, Math.min(GUARD_SIZE, state.offset)), + anchor: await readGuard( + handle!, + Math.max(0, state.offset - GUARD_SIZE), + Math.min(GUARD_SIZE, state.offset), + ), + }; + const temporary = `${statePath}.tmp-${randomBytes(8).toString("hex")}`; + try { + await fs.writeFile(temporary, JSON.stringify(value), { flag: "wx", mode: 0o600 }); + await fs.rename(temporary, statePath); + } finally { + await fs.unlink(temporary).catch(() => undefined); + } + }, + async close() { + await handle!.close(); + }, + }; + } catch (error) { + await handle?.close(); + throw error; + } +} diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index b0de00d..75a60e4 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -15,8 +15,9 @@ import { TraceFlags, type SpanContext } from "@opentelemetry/api"; import type { Config } from "./config.js"; import { currentIdSeed, seedIds } from "./instrumentation.js"; -import { parseArgs, parseSession } from "./parse.js"; +import { parseArgs, parseSession, type ParserContext } from "./parse.js"; import { loadUploadedTurnIds } from "./sidecar.js"; +import { openIncrementalRollout, type HistoryMessage } from "./rollout.js"; import { skillsForToolCall, traceTags } from "./skills.js"; import type { EventMsgPayload, @@ -388,6 +389,7 @@ function boundedConversation( earlier: readonly ChatMlMessage[], current: readonly ChatMlMessage[], maxChars: number, + previouslyOmitted = 0, ): ChatMlMessage[] { const all = [...earlier, ...current]; if (maxChars <= 0) return all; @@ -399,16 +401,32 @@ function boundedConversation( used += chars; start--; } - if (start === 0) return all; + if (start === 0 && previouslyOmitted === 0) return all; return [ { role: "system", - content: `[${start} earlier message(s) omitted: the conversation exceeds max_history_chars (${maxChars})]`, + content: `[${start + previouslyOmitted} earlier message(s) omitted: the conversation exceeds max_history_chars (${maxChars})]`, }, ...all.slice(start), ]; } +function retainHistory( + messages: ChatMlMessage[], + maxChars: number, +): { messages: ChatMlMessage[]; omitted: number } { + if (maxChars <= 0) return { messages, omitted: 0 }; + let used = 0; + let start = messages.length; + while (start > 0) { + const chars = serializedChars(messages[start - 1]); + if (used + chars > maxChars) break; + used += chars; + start--; + } + return { messages: messages.slice(start), omitted: start }; +} + function generationInput( systemMessage: string | undefined, conversation: ChatMlMessage[], @@ -469,6 +487,7 @@ async function emitTurn( unannouncedSubagents?: SubagentRollout[]; inheritableTurnIds?: ReadonlySet; historyPrefix?: ChatMlMessage[]; + omittedHistoryMessages?: number; }, ): Promise { const isSubagent = sessionMeta.isSubagentThread === true || ctx.parentObservation != null; @@ -530,6 +549,7 @@ async function emitTurn( historyPrefix, turnMessages, ctx.config.max_history_chars, + ctx.omittedHistoryMessages, ); const generation = startObservation( isSubagent ? "LLM Subagent" : "LLM", @@ -685,141 +705,213 @@ export async function convertRollout( onTurnExported?: (turnId: string) => Promise; }, ): 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)); - } - } - debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`); - + // Child/ancestor traversal keeps its full-history ownership semantics. Only + // an acknowledged root hook may resume and persist an incremental cursor. + const meta = await readSessionMeta(rolloutFile); const subagentIndex = options.subagentIndex ?? (await buildSubagentIndex(rolloutFile, { - includeEarlierDays: sessionMeta.isSubagentThread === true, + includeEarlierDays: meta?.parentThreadId !== undefined, })); - const seenThreadIds = options.seenThreadIds ?? new Set(); - seenThreadIds.add(sessionMeta.sessionId); + const hasSubagents = + meta?.parentThreadId !== undefined || + (subagentIndex.byParent.get(meta?.threadId ?? "")?.length ?? 0) > 0; + const reader = + !options.parentObservation && + options.onTurnExported && + !hasSubagents && + options.config.max_history_chars > 0 + ? await openIncrementalRollout(rolloutFile, options.config.max_history_chars) + : undefined; + try { + const checkpoints = new Map(); + const records = reader?.records; + const { sessionMeta, turns } = parseSession( + records ? records.map((r) => r.line) : await loadSession(rolloutFile), + { + context: reader?.checkpoint?.context, + onTurn: (turn, context, boundary) => { + if (!records || !boundary.stable) return; + const offset = + boundary.nextLineIndex === 0 + ? (reader!.checkpoint?.offset ?? 0) + : records[boundary.nextLineIndex - 1].end; + checkpoints.set(turn, { context, offset }); + }, + }, + ); + const baseTurnCount = reader?.checkpoint?.turnCount ?? 0; + const historyPrefixes: ChatMlMessage[][] = []; + const omittedPrefixes: number[] = []; + const historiesAfterTurn: Array<{ messages: ChatMlMessage[]; omitted: number }> = []; + let history = (reader?.checkpoint?.history ?? []) as ChatMlMessage[]; + let omitted = reader?.checkpoint?.omittedMessages ?? 0; + for (const turn of turns) { + historyPrefixes.push(history); + omittedPrefixes.push(omitted); + const retained = retainHistory( + [...history, ...turnHistoryMessages(turn)], + options.config.max_history_chars, + ); + history = retained.messages; + omitted += retained.omitted; + historiesAfterTurn.push({ messages: history, omitted }); + } + if (reader) + debugLog( + `read ${reader.bytesRead} rollout byte(s) from offset ${reader.checkpoint?.offset ?? 0}`, + ); + debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`); - const announced = new Set(turns.flatMap((t) => t.subagentThreadIds)); - const unannounced = (subagentIndex.byParent.get(sessionMeta.sessionId) ?? []).filter( - (s) => !announced.has(s.threadId) && !seenThreadIds.has(s.threadId), - ); - const spawnTurnOf = - unannounced.length > 0 ? turnIndexByNickname(turns) : new Map(); - const byTurn = new Map(); - for (const sub of unannounced) { - const viaNickname = sub.nickname !== undefined ? spawnTurnOf.get(sub.nickname) : undefined; - const i = viaNickname ?? turnIndexAt(turns, sub.startTime); - debugLog( - `recovered unannounced subagent ${sub.threadId} for thread ${sessionMeta.sessionId}: ` + - `turn ${i + 1} via ${viaNickname !== undefined ? `nickname ${sub.nickname}` : "start time"}`, + const seenThreadIds = options.seenThreadIds ?? new Set(); + seenThreadIds.add(sessionMeta.sessionId); + + const announced = new Set(turns.flatMap((t) => t.subagentThreadIds)); + const unannounced = (subagentIndex.byParent.get(sessionMeta.sessionId) ?? []).filter( + (s) => !announced.has(s.threadId) && !seenThreadIds.has(s.threadId), ); - byTurn.set(i, [...(byTurn.get(i) ?? []), sub]); - } + const spawnTurnOf = + unannounced.length > 0 ? turnIndexByNickname(turns) : new Map(); + const byTurn = new Map(); + for (const sub of unannounced) { + const viaNickname = sub.nickname !== undefined ? spawnTurnOf.get(sub.nickname) : undefined; + const i = viaNickname ?? turnIndexAt(turns, sub.startTime); + debugLog( + `recovered unannounced subagent ${sub.threadId} for thread ${sessionMeta.sessionId}: ` + + `turn ${i + 1} via ${viaNickname !== undefined ? `nickname ${sub.nickname}` : "start time"}`, + ); + byTurn.set(i, [...(byTurn.get(i) ?? []), sub]); + } - const ancestorTurnIds = - options.ancestorTurnIds ?? - (sessionMeta.isSubagentThread - ? await ancestorTurnIdsOf(sessionMeta, subagentIndex) - : undefined); - - const carriedSubagents: SubagentRollout[] = []; - const subagentsFor = (turnIndex: number): SubagentRollout[] | undefined => { - const own = byTurn.get(turnIndex) ?? []; - const all = carriedSubagents.length > 0 ? [...carriedSubagents, ...own] : own; - carriedSubagents.length = 0; - return all.length > 0 ? all : undefined; - }; + const ancestorTurnIds = + options.ancestorTurnIds ?? + (sessionMeta.isSubagentThread + ? await ancestorTurnIdsOf(sessionMeta, subagentIndex) + : undefined); + + const carriedSubagents: SubagentRollout[] = []; + const subagentsFor = (turnIndex: number): SubagentRollout[] | undefined => { + const own = byTurn.get(turnIndex) ?? []; + const all = carriedSubagents.length > 0 ? [...carriedSubagents, ...own] : own; + carriedSubagents.length = 0; + return all.length > 0 ? all : undefined; + }; - const skipInherited = (turn: Turn, turnIndex: number): boolean => { - if (!turn.turnId || !ancestorTurnIds?.has(turn.turnId)) return false; - debugLog(`skipping turn ${turn.turnId}: inherited from an ancestor thread`); - carriedSubagents.push(...(byTurn.get(turnIndex) ?? [])); - return true; - }; + const skipInherited = (turn: Turn, turnIndex: number): boolean => { + if (!turn.turnId || !ancestorTurnIds?.has(turn.turnId)) return false; + debugLog(`skipping turn ${turn.turnId}: inherited from an ancestor thread`); + carriedSubagents.push(...(byTurn.get(turnIndex) ?? [])); + return true; + }; - const inheritableTurnIds = new Set(ancestorTurnIds); - for (const turn of turns) { - if (turn.turnId) inheritableTurnIds.add(turn.turnId); - } + const inheritableTurnIds = new Set(ancestorTurnIds); + for (const turn of turns) { + if (turn.turnId) inheritableTurnIds.add(turn.turnId); + } - if (options.parentObservation) { - for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { - const turn = turns[turnIndex]; - if (skipInherited(turn, turnIndex)) continue; - await emitTurn(turn, sessionMeta, { - config: options.config, - rolloutFile, - parentObservation: options.parentObservation, - subagentIndex, - seenThreadIds, - unannouncedSubagents: subagentsFor(turnIndex), - inheritableTurnIds, - historyPrefix: historyPrefixes[turnIndex], - }); + if (options.parentObservation) { + for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { + const turn = turns[turnIndex]; + if (skipInherited(turn, turnIndex)) continue; + await emitTurn(turn, sessionMeta, { + config: options.config, + rolloutFile, + parentObservation: options.parentObservation, + subagentIndex, + seenThreadIds, + unannouncedSubagents: subagentsFor(turnIndex), + inheritableTurnIds, + historyPrefix: historyPrefixes[turnIndex], + omittedHistoryMessages: omittedPrefixes[turnIndex], + }); + } + return []; } - return []; - } - const uploaded = await loadUploadedTurnIds(rolloutFile); - const exportedTurnIds: string[] = []; - const attached = options.parentSpanContext != null; + const uploaded = await loadUploadedTurnIds(rolloutFile); + const exportedTurnIds: string[] = []; + const attached = options.parentSpanContext != null; + + const commitTurn = async (turn: Turn, turnIndex: number) => { + const checkpoint = checkpoints.get(turn); + if ( + !reader || + !checkpoint || + sessionMeta.isSubagentThread || + turns.some((t) => t.subagentThreadIds.length > 0) + ) + return; + const after = historiesAfterTurn[turnIndex]; + await reader.commit({ + offset: checkpoint.offset, + turnCount: baseTurnCount + turnIndex + 1, + context: checkpoint.context, + history: after.messages as HistoryMessage[], + omittedMessages: after.omitted, + lastTurnId: turn.turnId!, + }); + }; - for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { - const turn = turns[turnIndex]; + for (let turnIndex = 0; turnIndex < turns.length; turnIndex++) { + const turn = turns[turnIndex]; - if (skipInherited(turn, turnIndex)) continue; + if (skipInherited(turn, turnIndex)) continue; - if (!isFinal(turn, options.stoppedTurnId, turnIndex < turns.length - 1)) { - debugLog(`skipping turn ${turn.turnId ?? "(no turn id)"}: not final`); - continue; - } - if (uploaded.has(turn.turnId)) { - continue; - } + if (!isFinal(turn, options.stoppedTurnId, turnIndex < turns.length - 1)) { + debugLog(`skipping turn ${turn.turnId ?? "(no turn id)"}: not final`); + continue; + } + if (uploaded.has(turn.turnId)) { + await commitTurn(turn, turnIndex); + continue; + } - const parentSpanContext = - options.parentSpanContext ?? - (await seededTraceParent(options.config, sessionMeta, turnIndex + 1)); - - const emit = () => - emitTurn(turn, sessionMeta, { - config: options.config, - rolloutFile, - parentSpanContext, - attached, - subagentIndex, - seenThreadIds, - unannouncedSubagents: subagentsFor(turnIndex), - inheritableTurnIds, - historyPrefix: historyPrefixes[turnIndex], - }); + const parentSpanContext = + options.parentSpanContext ?? + (await seededTraceParent(options.config, sessionMeta, baseTurnCount + turnIndex + 1)); + + const emit = () => + emitTurn(turn, sessionMeta, { + config: options.config, + rolloutFile, + parentSpanContext, + attached, + subagentIndex, + seenThreadIds, + unannouncedSubagents: subagentsFor(turnIndex), + inheritableTurnIds, + historyPrefix: historyPrefixes[turnIndex], + omittedHistoryMessages: omittedPrefixes[turnIndex], + }); + + if (attached) { + await emit(); + } else { + const tags = traceTags(options.config, turn); + await propagateAttributes( + { + sessionId: sessionMeta.sessionId, + traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", + ...(options.config.user_id ? { userId: options.config.user_id } : {}), + ...(tags.length > 0 ? { tags } : {}), + ...(options.config.metadata ? { metadata: options.config.metadata } : {}), + }, + emit, + ); + } - if (attached) { - await emit(); - } else { - const tags = traceTags(options.config, turn); - await propagateAttributes( - { - sessionId: sessionMeta.sessionId, - traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", - ...(options.config.user_id ? { userId: options.config.user_id } : {}), - ...(tags.length > 0 ? { tags } : {}), - ...(options.config.metadata ? { metadata: options.config.metadata } : {}), - }, - emit, - ); + uploaded.add(turn.turnId); + exportedTurnIds.push(turn.turnId); + await options.onTurnExported?.(turn.turnId); + // A callback may observe delivery without persisting a receipt. In that + // case retain the cold-read behavior rather than hiding retryable turns. + if (!options.onTurnExported || !(await loadUploadedTurnIds(rolloutFile)).has(turn.turnId)) + continue; + await commitTurn(turn, turnIndex); } - uploaded.add(turn.turnId); - exportedTurnIds.push(turn.turnId); - await options.onTurnExported?.(turn.turnId); + return exportedTurnIds; + } finally { + await reader?.close(); } - - return exportedTurnIds; } diff --git a/plugins/tracing/test/incremental-rollout.test.ts b/plugins/tracing/test/incremental-rollout.test.ts new file mode 100644 index 0000000..f1186b4 --- /dev/null +++ b/plugins/tracing/test/incremental-rollout.test.ts @@ -0,0 +1,352 @@ +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; + +import { InMemorySpanExporter, SimpleSpanProcessor } from "@opentelemetry/sdk-trace-base"; +import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; + +import type { Config } from "../src/config.js"; +import { openIncrementalRollout } from "../src/rollout.js"; +import { markTurnUploaded } from "../src/sidecar.js"; +import { convertRollout } from "../src/trace.js"; + +const exporter = new InMemorySpanExporter(); +let provider: NodeTracerProvider; +const temporary: string[] = []; +const config: Config = { + enabled: true, + public_key: "test-public", + secret_key: "test-secret", + base_url: "http://127.0.0.1", + skill_tags: true, + max_history_chars: 2_000, + debug: false, + fail_on_error: true, +}; +const row = (type: string, payload: Record) => + JSON.stringify({ + timestamp: "2026-11-10T10:00:00.000Z", + type, + payload, + }) + "\n"; +const event = (type: string, payload: Record = {}) => + row("event_msg", { type, ...payload }); +function turn(n: number, complete = true): string { + return ( + event("task_started", { turn_id: `t${n}` }) + + row("turn_context", { model: "test-model", effort: "medium" }) + + event("user_message", { message: `question ${n} 中文` }) + + row("response_item", { + type: "function_call", + name: "exec_command", + call_id: `call-${n}`, + arguments: '{"cmd":"echo proof"}', + }) + + row("response_item", { + type: "function_call_output", + call_id: `call-${n}`, + output: `proof ${n}`, + }) + + row("response_item", { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: `answer ${n}` }], + }) + + event("token_count", { + info: { last_token_usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 } }, + }) + + (complete ? event("task_complete", { turn_id: `t${n}` }) : "") + ); +} +function stage(): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "codex-incremental-")); + temporary.push(dir); + const day = path.join(dir, "sessions", "2026", "11", "10"); + fs.mkdirSync(day, { recursive: true }); + const file = path.join(day, "rollout.jsonl"); + fs.writeFileSync( + file, + row("session_meta", { + id: "test-session", + base_instructions: { text: "system instructions" }, + }) + + row("response_item", { + type: "message", + role: "developer", + content: [{ type: "input_text", text: "developer instructions" }], + }), + ); + return file; +} +const options = (file: string, overrides: Partial = {}) => ({ + config: { ...config, ...overrides }, + onTurnExported: (id: string) => markTurnUploaded(file, id), +}); +const attrs = (key: string) => exporter.getFinishedSpans().map((s) => s.attributes[key]); +const cursor = (file: string) => JSON.parse(fs.readFileSync(`${file}.langfuse.cursor`, "utf8")); + +beforeAll(() => { + provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + provider.register(); +}); +afterAll(() => provider.shutdown()); +beforeEach(() => exporter.reset()); +afterEach(() => { + for (const dir of temporary.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); +}); + +describe("incremental root Stop hooks", () => { + it("resumes from acknowledged bytes and retains generation history, instructions, tools and usage", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + const offset = fs.statSync(file).size; + expect(cursor(file).offset).toBe(offset); + exporter.reset(); + const addition = turn(2); + fs.appendFileSync(file, addition); + const reader = await openIncrementalRollout(file, config.max_history_chars); + try { + expect(reader.checkpoint?.offset).toBe(offset); + expect(reader.bytesRead).toBe(Buffer.byteLength(addition)); + expect(reader.records[0].line.payload.type).toBe("task_started"); + } finally { + await reader.close(); + } + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + const warm = exporter.getFinishedSpans(); + const generation = warm.find( + (s) => s.attributes["langfuse.observation.type"] === "generation", + )!; + expect(String(generation.attributes["langfuse.observation.input"])).toContain( + "question 1 中文", + ); + expect(String(generation.attributes["langfuse.observation.input"])).toContain( + "system instructions", + ); + expect(String(generation.attributes["langfuse.observation.input"])).toContain( + "developer instructions", + ); + expect( + warm.find((s) => s.name === "exec_command")!.attributes["langfuse.observation.output"], + ).toContain("proof 2"); + expect( + JSON.parse(String(generation.attributes["langfuse.observation.usage_details"])).total_tokens, + ).toBe(15); + fs.unlinkSync(`${file}.langfuse.cursor`); + fs.writeFileSync(`${file}.langfuse`, "t1\n"); + exporter.reset(); + await convertRollout(file, options(file)); + const cold = exporter + .getFinishedSpans() + .find((s) => s.attributes["langfuse.observation.type"] === "generation")!; + expect(cold.attributes["langfuse.observation.input"]).toEqual( + generation.attributes["langfuse.observation.input"], + ); + if (process.platform !== "win32") + expect(fs.statSync(`${file}.langfuse.cursor`).mode & 0o777).toBe(0o600); + }); + + it("keeps omission counts identical to a cold parse after a bounded-history restart", async () => { + const file = stage(); + for (let n = 1; n <= 6; n++) { + fs.appendFileSync(file, turn(n)); + await convertRollout(file, options(file, { max_history_chars: 400 })); + } + const warmGeneration = exporter + .getFinishedSpans() + .filter((s) => s.attributes["langfuse.observation.type"] === "generation") + .at(-1)!; + expect(String(warmGeneration.attributes["langfuse.observation.input"])).toContain( + "earlier message(s) omitted", + ); + fs.unlinkSync(`${file}.langfuse.cursor`); + fs.writeFileSync(`${file}.langfuse`, "t1\nt2\nt3\nt4\nt5\n"); + exporter.reset(); + await convertRollout(file, options(file, { max_history_chars: 400 })); + const coldGeneration = exporter + .getFinishedSpans() + .find((s) => s.attributes["langfuse.observation.type"] === "generation")!; + expect(coldGeneration.attributes["langfuse.observation.input"]).toEqual( + warmGeneration.attributes["langfuse.observation.input"], + ); + }); + + it("does not skip an incomplete UTF-8 tail, and retries after it becomes a complete line", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + const offset = cursor(file).offset; + const next = Buffer.from(turn(2)); + const split = next.indexOf(Buffer.from("中文")) + 1; + fs.appendFileSync(file, next.subarray(0, split)); + exporter.reset(); + expect(await convertRollout(file, options(file))).toEqual([]); + expect(cursor(file).offset).toBe(offset); + fs.appendFileSync(file, next.subarray(split)); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + expect( + attrs("langfuse.observation.input").some((x) => String(x).includes("question 2 中文")), + ).toBe(true); + }); + + it("exports the last stopped turn before task_complete without consuming its late records", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1, false)); + expect(await convertRollout(file, { ...options(file), stoppedTurnId: "t1" })).toEqual(["t1"]); + expect(fs.existsSync(`${file}.langfuse.cursor`)).toBe(false); + fs.appendFileSync(file, event("task_complete", { turn_id: "t1" }) + turn(2)); + exporter.reset(); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + expect(cursor(file).turnCount).toBe(2); + expect(cursor(file).offset).toBe(fs.statSync(file).size); + }); + + it("does not advance the cursor on failed acknowledgement, and retries the same turn", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + const before = cursor(file); + fs.appendFileSync(file, turn(2)); + await expect( + convertRollout(file, { + config, + onTurnExported: async () => { + throw new Error("upload failed"); + }, + }), + ).rejects.toThrow("upload failed"); + expect(cursor(file)).toEqual(before); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + expect(cursor(file).lastTurnId).toBe("t2"); + }); + + it("decodes long UTF-8 lines crossing read chunks without losing the completion boundary", async () => { + const file = stage(); + const text = "中文🌍".repeat(20_000); + const first = turn(1).replace("answer 1", text); + fs.appendFileSync(file, first); + await convertRollout(file, options(file)); + const root = exporter.getFinishedSpans().find((s) => s.name === "Codex Turn")!; + expect(root.attributes["langfuse.observation.output"]).toBe(text); + expect(cursor(file).offset).toBe(fs.statSync(file).size); + fs.appendFileSync(file, turn(2)); + exporter.reset(); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + expect( + exporter.getFinishedSpans().find((s) => s.name === "Codex Turn")!.attributes[ + "langfuse.observation.output" + ], + ).toBe("answer 2"); + }); + + it("rebuilds a corrupt, replaced or truncated transcript checkpoint without hiding new turns", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + fs.writeFileSync(`${file}.langfuse.cursor`, "not json"); + fs.appendFileSync(file, turn(2)); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + const replacement = `${file}.replacement`; + fs.writeFileSync(replacement, row("session_meta", { id: "test-session" }) + turn(3)); + fs.renameSync(replacement, file); + expect(await convertRollout(file, options(file))).toEqual(["t3"]); + fs.writeFileSync(file, row("session_meta", { id: "test-session" }) + turn(4)); + expect(await convertRollout(file, options(file))).toEqual(["t4"]); + }); + + it("does not trust a checkpoint after receipts were removed or history limit changed", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + fs.unlinkSync(`${file}.langfuse`); + expect(await convertRollout(file, options(file))).toEqual(["t1"]); + const reader = await openIncrementalRollout(file, 100); + try { + expect(reader.checkpoint).toBeUndefined(); + } finally { + await reader.close(); + } + }); + + it("retains tool definitions discovered before the acknowledged cursor", async () => { + const file = stage(); + const first = turn(1).replace( + event("task_complete", { turn_id: "t1" }), + row("response_item", { + type: "tool_search_output", + tools: [ + { + name: "remembered_tool", + description: "prior definition", + parameters: { type: "object" }, + }, + ], + }) + event("task_complete", { turn_id: "t1" }), + ); + fs.appendFileSync(file, first); + await convertRollout(file, options(file)); + fs.appendFileSync(file, turn(2)); + exporter.reset(); + await convertRollout(file, options(file)); + const generation = exporter + .getFinishedSpans() + .find((s) => s.attributes["langfuse.observation.type"] === "generation")!; + const input = JSON.parse(String(generation.attributes["langfuse.observation.input"])); + expect(input[0].tools).toEqual([ + { name: "remembered_tool", description: "prior definition", parameters: { type: "object" } }, + ]); + }); + + it("switches to ownership-preserving reads when a subagent appears after a checkpoint", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file)); + const child = path.join(path.dirname(file), "rollout-child.jsonl"); + fs.writeFileSync( + child, + row("session_meta", { id: "child-session", parent_thread_id: "test-session" }) + + turn(1) + + turn(3), + ); + const second = turn(2).replace( + event("task_complete", { turn_id: "t2" }), + event("collab_agent_spawn_end", { new_thread_id: "child-session", call_id: "call-2" }) + + event("task_complete", { turn_id: "t2" }), + ); + fs.appendFileSync(file, second); + exporter.reset(); + expect(await convertRollout(file, options(file))).toEqual(["t2"]); + const roots = exporter + .getFinishedSpans() + .filter((s) => s.name === "Codex Turn" || s.name === "Codex Subagent Turn"); + expect(roots.map((s) => s.attributes["langfuse.observation.metadata.codex.turn_id"])).toEqual([ + "t3", + "t2", + ]); + }); + + it("preserves seeded turn numbering across warm hooks", async () => { + const file = stage(); + fs.appendFileSync(file, turn(1)); + await convertRollout(file, options(file, { trace_seed: "seed" })); + fs.appendFileSync(file, turn(2)); + exporter.reset(); + await convertRollout(file, options(file, { trace_seed: "seed" })); + const warmTrace = exporter + .getFinishedSpans() + .find((s) => s.name === "Codex Turn")! + .spanContext().traceId; + fs.unlinkSync(`${file}.langfuse.cursor`); + fs.writeFileSync(`${file}.langfuse`, "t1\n"); + exporter.reset(); + await convertRollout(file, options(file, { trace_seed: "seed" })); + expect( + exporter + .getFinishedSpans() + .find((s) => s.name === "Codex Turn")! + .spanContext().traceId, + ).toBe(warmTrace); + }); +}); From 8702cc73e303816b7206d7666e22cae90a5e27f9 Mon Sep 17 00:00:00 2001 From: albin-zhu <1317911+albin-zhu@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:20:05 +0800 Subject: [PATCH 2/3] fix: bound subagent discovery on Windows and skip earlier sessions --- README.md | 31 ++++++++++++-- plugins/tracing/src/trace.ts | 37 ++++++++++++++++- .../tracing/test/incremental-rollout.test.ts | 41 ++++++++++++++++++- 3 files changed, 103 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index bf74331..8c26efa 100644 --- a/README.md +++ b/README.md @@ -91,10 +91,33 @@ Missing/corrupt checkpoints, a replaced or truncated file, removed receipts, and changes to `max_history_chars` rebuild from the transcript. Concurrent hooks may replace the checkpoint with an older acknowledged position: this causes extra reads, not lost turns. Subagent/ancestor traversal and unbounded history -(`max_history_chars: 0`) retain the existing full-read path for fidelity. Session -directory metadata discovery is unchanged. This optimization reduces repeated -main-rollout reads; it does not claim constant total I/O or eliminate the cost of -converting a single large turn. +(`max_history_chars: 0`) retain the existing full-read path for fidelity. This +optimization reduces repeated main-rollout reads; it does not claim constant +total I/O or eliminate the cost of converting a single large turn. + +Subagent discovery still runs on every hook, but its scope is now bounded on all +platforms: the day-directory pruning previously compared a `path.relative` +result against a `/`-separated pattern, so on Windows it never matched and every +hook walked the entire session history. Separators are normalized before the +check. Within the remaining scope, timestamped rollouts that started before the +current session are skipped without opening them, since they cannot be its +subagents. Unannounced-subagent recovery still reads session headers in scope, +and child threads still scan earlier days. Windows behavior is verified by +source inspection and path-level tests, not yet on a Windows host. + +## Hook diagnostics + +Set `LANGFUSE_CODEX_DIAGNOSTICS=true` to retain structured hook progress in +`$CODEX_HOME/langfuse-diagnostics/hook.jsonl` (default `~/.codex`). The file rotates +at 1 MiB to `hook.jsonl.previous`. Records include skip reasons, parsing mode, +cursor position, memory counters, conversion/flush progress and safe error +categories, never transcript content, credentials, raw errors or full paths. +They are also written to stderr. Unlike verbose `debug` output, these files can +be used for failure reports. POSIX files are mode `0600`. + +A missing final record can indicate termination or timeout; it does not by +itself prove an OOM. Diagnostics are best-effort and do not change failure-open +or upload behavior. ## Contributing diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 75a60e4..6a9eef5 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -14,6 +14,7 @@ import { import { TraceFlags, type SpanContext } from "@opentelemetry/api"; import type { Config } from "./config.js"; +import { diagnostic } from "./diagnostics.js"; import { currentIdSeed, seedIds } from "./instrumentation.js"; import { parseArgs, parseSession, type ParserContext } from "./parse.js"; import { loadUploadedTurnIds } from "./sidecar.js"; @@ -102,9 +103,18 @@ export async function buildSubagentIndex( options: { includeEarlierDays?: boolean } = {}, ): Promise { const root = path.resolve(path.dirname(rolloutFile), "../../.."); - const fromDay = path.relative(root, path.dirname(rolloutFile)); + // `path.relative` yields backslashes on Windows; normalizing keeps the + // day-directory pruning active there instead of silently walking all history. + const fromDay = path.relative(root, path.dirname(rolloutFile)).split(path.sep).join("/"); const bounded = !options.includeEarlierDays && /^\d{4}\/\d{2}\/\d{2}$/.test(fromDay); const index: SubagentIndex = { byParent: new Map(), byThread: new Map() }; + // Real rollout names are `rollout--.jsonl`, + // so lexical order is chronological. A session started before this one cannot + // be its subagent, and ancestor lookups only run for child threads (which are + // unbounded). Only timestamped names qualify; other layouts keep the old scan. + const ROLLOUT_NAME = /^rollout-\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}-/; + const ownName = path.basename(rolloutFile); + const skipEarlier = bounded && ROLLOUT_NAME.test(ownName); async function walk(dir: string, rel: string): Promise { if (bounded && rel && rel < fromDay.slice(0, rel.length)) return; @@ -121,6 +131,7 @@ export async function buildSubagentIndex( continue; } if (!entry.isFile() || !entry.name.endsWith(".jsonl")) continue; + if (skipEarlier && ROLLOUT_NAME.test(entry.name) && entry.name < ownName) continue; const meta = await readSessionMeta(full); if (!meta) continue; if (index.byThread.has(meta.threadId)) continue; @@ -762,6 +773,15 @@ export async function convertRollout( `read ${reader.bytesRead} rollout byte(s) from offset ${reader.checkpoint?.offset ?? 0}`, ); debugLog(`parsed ${turns.length} turn(s) from ${path.basename(rolloutFile)}`); + diagnostic("rollout_parsed", { + mode: reader ? "incremental" : "full", + cursor_hit: !!reader?.checkpoint, + offset: reader?.checkpoint?.offset ?? 0, + bytes_read: reader?.bytesRead ?? -1, + turns: turns.length, + rss_bytes: process.memoryUsage().rss, + heap_bytes: process.memoryUsage().heapUsed, + }); const seenThreadIds = options.seenThreadIds ?? new Set(); seenThreadIds.add(sessionMeta.sessionId); @@ -831,6 +851,11 @@ export async function convertRollout( const uploaded = await loadUploadedTurnIds(rolloutFile); const exportedTurnIds: string[] = []; const attached = options.parentSpanContext != null; + diagnostic("delivery_plan", { + parsed_turns: turns.length, + receipt_count: uploaded.size, + pending_turns: turns.filter((t) => !!t.turnId && !uploaded.has(t.turnId)).length, + }); const commitTurn = async (turn: Turn, turnIndex: number) => { const checkpoint = checkpoints.get(turn); @@ -866,6 +891,12 @@ export async function convertRollout( continue; } + diagnostic("turn_conversion_started", { + turn_index: baseTurnCount + turnIndex + 1, + steps: turn.steps.length, + tools: turn.steps.reduce((n, s) => n + s.toolCalls.length, 0), + rss_bytes: process.memoryUsage().rss, + }); const parentSpanContext = options.parentSpanContext ?? (await seededTraceParent(options.config, sessionMeta, baseTurnCount + turnIndex + 1)); @@ -899,6 +930,10 @@ export async function convertRollout( emit, ); } + diagnostic("turn_conversion_finished", { + turn_index: baseTurnCount + turnIndex + 1, + rss_bytes: process.memoryUsage().rss, + }); uploaded.add(turn.turnId); exportedTurnIds.push(turn.turnId); diff --git a/plugins/tracing/test/incremental-rollout.test.ts b/plugins/tracing/test/incremental-rollout.test.ts index f1186b4..6c5328c 100644 --- a/plugins/tracing/test/incremental-rollout.test.ts +++ b/plugins/tracing/test/incremental-rollout.test.ts @@ -9,7 +9,7 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from import type { Config } from "../src/config.js"; import { openIncrementalRollout } from "../src/rollout.js"; import { markTurnUploaded } from "../src/sidecar.js"; -import { convertRollout } from "../src/trace.js"; +import { buildSubagentIndex, convertRollout } from "../src/trace.js"; const exporter = new InMemorySpanExporter(); let provider: NodeTracerProvider; @@ -349,4 +349,43 @@ describe("incremental root Stop hooks", () => { .spanContext().traceId, ).toBe(warmTrace); }); + + it("does not open earlier sessions when discovering subagents of a timestamped rollout", async () => { + const file = stage(); + const day = path.dirname(file); + const own = path.join(day, "rollout-2026-11-10T12-00-00-parent.jsonl"); + fs.renameSync(file, own); + fs.appendFileSync(own, turn(1)); + const earlier = path.join(day, "rollout-2026-11-10T09-00-00-unrelated.jsonl"); + fs.writeFileSync(earlier, row("session_meta", { id: "unrelated" }) + turn(1)); + const later = path.join(day, "rollout-2026-11-10T13-00-00-child.jsonl"); + fs.writeFileSync( + later, + row("session_meta", { id: "child-session", parent_thread_id: "test-session" }) + turn(1), + ); + // An unreadable earlier file proves it is never opened during discovery. + fs.writeFileSync(earlier, "not json at all\n"); + const index = await buildSubagentIndex(own); + expect(index.byThread.has("unrelated")).toBe(false); + expect(index.byParent.get("test-session")?.map((s) => s.threadId)).toEqual(["child-session"]); + }); + + it("prunes earlier day directories but still finds them when explicitly unbounded", async () => { + const root = fs.mkdtempSync(path.join(os.tmpdir(), "codex-day-prune-")); + temporary.push(root); + const day = path.join(root, "2026", "11", "10"); + fs.mkdirSync(day, { recursive: true }); + const own = path.join(day, "rollout-2026-11-10T12-00-00-parent.jsonl"); + fs.writeFileSync(own, row("session_meta", { id: "test-session" }) + turn(1)); + const old = path.join(root, "2025", "01", "01"); + fs.mkdirSync(old, { recursive: true }); + fs.writeFileSync( + path.join(old, "rollout-2025-01-01T00-00-00-ancient.jsonl"), + row("session_meta", { id: "ancient-child", parent_thread_id: "test-session" }) + turn(1), + ); + expect((await buildSubagentIndex(own)).byThread.has("ancient-child")).toBe(false); + expect( + (await buildSubagentIndex(own, { includeEarlierDays: true })).byThread.has("ancient-child"), + ).toBe(true); + }); }); From e5ab365c2afe682c8cc32e1f5a1c518657b85a44 Mon Sep 17 00:00:00 2001 From: albin-zhu <1317911+albin-zhu@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:25:11 +0800 Subject: [PATCH 3/3] feat: retain opt-in secret-safe hook diagnostics --- plugins/tracing/src/diagnostics.ts | 64 ++++++++++++++++++++++++++++++ plugins/tracing/src/index.ts | 26 +++++++++++- 2 files changed, 89 insertions(+), 1 deletion(-) create mode 100644 plugins/tracing/src/diagnostics.ts diff --git a/plugins/tracing/src/diagnostics.ts b/plugins/tracing/src/diagnostics.ts new file mode 100644 index 0000000..a06dfd7 --- /dev/null +++ b/plugins/tracing/src/diagnostics.ts @@ -0,0 +1,64 @@ +import { appendFileSync, chmodSync, mkdirSync, renameSync, statSync } from "node:fs"; +import { homedir } from "node:os"; +import { join } from "node:path"; + +const enabled = process.env.LANGFUSE_CODEX_DIAGNOSTICS === "true"; +const run = `${process.pid}-${Date.now()}`; + +/** Deliberately excludes raw error messages, paths, payloads and credentials. */ +export function diagnostic( + event: string, + fields: Record = {}, +): void { + if (!enabled) return; + const record = JSON.stringify({ time: new Date().toISOString(), run, event, ...fields }); + console.error(`[langfuse-diagnostic] ${record}`); + try { + const dir = join(process.env.CODEX_HOME || join(homedir(), ".codex"), "langfuse-diagnostics"); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + const file = join(dir, "hook.jsonl"); + try { + if (statSync(file).size > 1024 * 1024) renameSync(file, `${file}.previous`); + } catch { + /* First run or competing rotation. */ + } + appendFileSync(file, record + "\n", { mode: 0o600 }); + chmodSync(file, 0o600); + } catch { + /* Diagnostics must never block the hook. */ + } +} + +export function diagnosticError(event: string, error: unknown): void { + const e = error as { + code?: unknown; + name?: unknown; + status?: unknown; + statusCode?: unknown; + message?: unknown; + } | null; + const code = + typeof e?.code === "string" && /^[A-Z][A-Z0-9_]{0,40}$/.test(e.code) ? e.code : "UNKNOWN"; + const status = e?.statusCode ?? e?.status; + const message = typeof e?.message === "string" ? e.message : ""; + const httpMatch = message.match(/\b(?:HTTP|status|code)\s*[:=]?\s*(\d{3})\b/i); + diagnostic(event, { + code, + ...(typeof status === "number" + ? { http_status: status } + : httpMatch + ? { http_status: Number(httpMatch[1]) } + : {}), + category: /heap|memory/i.test(message) + ? "memory" + : /timeout|timed out/i.test(message) + ? "timeout" + : /401|403|unauthorized/i.test(message) + ? "authentication" + : /404/i.test(message) + ? "endpoint" + : /network|socket|connect/i.test(message) + ? "network" + : "unknown", + }); +} diff --git a/plugins/tracing/src/index.ts b/plugins/tracing/src/index.ts index ffe7a87..fbcf80f 100644 --- a/plugins/tracing/src/index.ts +++ b/plugins/tracing/src/index.ts @@ -1,4 +1,5 @@ import { getConfig } from "./config.js"; +import { diagnostic, diagnosticError } from "./diagnostics.js"; import { setupInstrumentation } from "./instrumentation.js"; import { readExternalParentSpanContext } from "./parent-context.js"; import { markTurnUploaded } from "./sidecar.js"; @@ -21,10 +22,13 @@ let failOnError = process.env.LANGFUSE_CODEX_FAIL_ON_ERROR === "true"; * hook failures instead. */ export async function runHook(): Promise { + const started = Date.now(); + diagnostic("hook_started"); let hookInput: HookInput; try { hookInput = await readStdin(); } catch (error) { + diagnosticError("stdin_failed", error); return; } @@ -32,16 +36,25 @@ export async function runHook(): Promise { setDebug(config.debug); failOnError = config.fail_on_error; + diagnostic("config_loaded", { + enabled: config.enabled, + has_credentials: !!config.public_key && !!config.secret_key, + has_transcript: !!hookInput.transcript_path, + has_turn_id: !!hookInput.turn_id, + }); if (!config.enabled) { debugLog("tracing disabled (set TRACE_TO_LANGFUSE=true to enable)"); + diagnostic("hook_skipped", { reason: "disabled" }); return; } if (!config.public_key || !config.secret_key) { debugLog("missing LANGFUSE_PUBLIC_KEY / LANGFUSE_SECRET_KEY; skipping"); + diagnostic("hook_skipped", { reason: "missing_credentials" }); return; } if (!hookInput.transcript_path) { debugLog("hook payload missing transcript_path; skipping"); + diagnostic("hook_skipped", { reason: "missing_transcript" }); return; } @@ -65,13 +78,21 @@ export async function runHook(): Promise { // flush: a turn marked despite a failed export is lost silently, while // an unmarked one is retried. onTurnExported: async (turnId) => { - await instrumentation.flush(); + diagnostic("flush_started"); + try { + await instrumentation.flush(); + } catch (error) { + diagnosticError("flush_failed", error); + throw error; + } await markTurnUploaded(transcriptPath, turnId); + diagnostic("turn_acknowledged"); }, }); } catch (error) { failure = error; debugLog("failed to convert rollout:", error); + diagnosticError("conversion_failed", error); } try { @@ -79,8 +100,10 @@ export async function runHook(): Promise { } catch (error) { failure ??= error; debugLog("error during flush/shutdown:", error); + diagnosticError("shutdown_failed", error); } + diagnostic("hook_finished", { success: !failure, duration_ms: Date.now() - started }); if (failure) { if (config.fail_on_error) throw failure; debugLog("export incomplete; leaving the remaining turns unmarked so they stay retryable"); @@ -88,6 +111,7 @@ export async function runHook(): Promise { } runHook().catch((error) => { + diagnosticError("hook_fatal", error); if (process.env.LANGFUSE_CODEX_DEBUG === "true") { console.error("[langfuse-codex] fatal:", error); }