perf: checkpoint acknowledged main rollout reads between Stop hooks

README.md

@@ -71,6 +71,54 @@ 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
+`<rollout>.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. 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
 
 See the [contributing guide](./CONTRIBUTING.md).

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<string, string | number | boolean> = {},
+): 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",
+  });
+}

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,27 +22,39 @@ let failOnError = process.env.LANGFUSE_CODEX_FAIL_ON_ERROR === "true";
  * hook failures instead.
  */
 export async function runHook(): Promise<void> {
+  const started = Date.now();
+  diagnostic("hook_started");
   let hookInput: HookInput;
   try {
     hookInput = await readStdin<HookInput>();
   } catch (error) {
+    diagnosticError("stdin_failed", error);
     return;
   }
 
   const config = await getConfig();
   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,29 +78,40 @@ export async function runHook(): Promise<void> {
       // 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 {
     await instrumentation.shutdown();
   } 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");
   }
 }
 
 runHook().catch((error) => {
+  diagnosticError("hook_fatal", error);
   if (process.env.LANGFUSE_CODEX_DEBUG === "true") {
     console.error("[langfuse-codex] fatal:", error);
   }

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<string, ToolCall>();
   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;