Bound diagnostics and job retention

Add timeouts for Codex version probes and bounds for unterminated child output lines.

Make debug bundle log tails opt-in and bounded, add key-aware redaction for production logs, and cap async job retention.

Co-Authored-By: OpenAI Codex <noreply@openai.com>

Author: xuio

dist/index.js

@@ -25,6 +25,8 @@ import { OutputArtifactWriter } from "./artifacts.js";
 import { recordDiagnosticEvent } from "./diagnostics.js";
 
 type JsonObject = Record<string, unknown>;
+
+const maxPendingJsonLineChars = 1_000_000;
 type AppServerRequestMethod =
   | "initialize"
   | "thread/start"
@@ -213,6 +215,7 @@ export class CodexAppServerSession {
   private readonly closedPromise: Promise<void>;
   private resolveClosed: (() => void) | undefined;
   private lineBuffer = "";
+  private lineBufferOverflowReported = false;
   private requestCounter = 0;
   private activeTurn?: ActiveTurnState;
   private acceptingStartNotifications = false;
@@ -719,11 +722,29 @@ export class CodexAppServerSession {
       chunk: summarizeRawTrafficForLog(chunk),
     });
     this.lineBuffer += chunk;
+    if (this.lineBuffer.length > maxPendingJsonLineChars) {
+      const dropped = this.lineBuffer.length - maxPendingJsonLineChars;
+      this.lineBuffer = this.lineBuffer.slice(-maxPendingJsonLineChars);
+      if (!this.lineBufferOverflowReported) {
+        this.lineBufferOverflowReported = true;
+        const error = `Codex app-server stdout JSON line exceeded ${maxPendingJsonLineChars} chars; dropped leading data from an unterminated line.`;
+        logger.warn("codex.app_server.stdout_line_oversized", {
+          ...this.logContext,
+          appServerId: this.id,
+          threadId: this.threadId || undefined,
+          activeTurnId: this.activeTurnId,
+          droppedChars: dropped,
+          maxPendingJsonLineChars,
+        });
+        this.recordBufferedLineError(error);
+      }
+    }
     let newlineIndex = this.lineBuffer.indexOf("\n");
     while (newlineIndex >= 0) {
       const line = this.lineBuffer.slice(0, newlineIndex);
       this.lineBuffer = this.lineBuffer.slice(newlineIndex + 1);
       this.handleLine(line);
+      this.lineBufferOverflowReported = false;
       newlineIndex = this.lineBuffer.indexOf("\n");
     }
   }
@@ -982,6 +1003,19 @@ export class CodexAppServerSession {
     this.activeTurn?.publishSnapshot(true);
   }
 
+  private recordBufferedLineError(error: string): void {
+    this.lastError = error;
+    if (this.activeTurn) {
+      this.activeTurn.summary.errors.push(error);
+      this.activeTurn.publishSnapshot(true);
+    } else if (this.acceptingStartNotifications) {
+      this.queuePendingStartNotification({
+        method: "internal/unparseableLine",
+        params: { error, line: "" },
+      });
+    }
+  }
+
   private async probeThreadRead(timeoutMs: number): Promise<void> {
     try {
       await this.request("thread/read", {

src/app-server.ts

@@ -1,4 +1,4 @@
-import { accessSync, constants } from "node:fs";
+import { accessSync, constants, statSync } from "node:fs";
 import os from "node:os";
 import path from "node:path";
 
@@ -31,6 +31,7 @@ export function cleanOption(value: string | undefined): string | undefined {
 
 export function isExecutable(candidate: string): boolean {
   try {
+    if (!statSync(candidate).isFile()) return false;
     accessSync(candidate, constants.X_OK);
     return true;
   } catch {

src/binary.ts

@@ -1,4 +1,4 @@
-import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises";
+import { mkdir, mkdtemp, open, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { loggingDiagnostics } from "./logging.js";
@@ -62,11 +62,18 @@ export function diagnosticStats(): Record<string, unknown> {
 }
 
 async function tailFile(file: string, maxBytes = 200_000): Promise<string | undefined> {
+  let handle: Awaited<ReturnType<typeof open>> | undefined;
   try {
-    const text = await readFile(file, "utf8");
-    return text.length <= maxBytes ? text : text.slice(text.length - maxBytes);
+    handle = await open(file, "r");
+    const stat = await handle.stat();
+    const length = Math.min(stat.size, maxBytes);
+    const buffer = Buffer.alloc(length);
+    await handle.read(buffer, 0, length, stat.size - length);
+    return buffer.toString("utf8");
   } catch {
     return undefined;
+  } finally {
+    await handle?.close().catch(() => {});
   }
 }
 
@@ -76,6 +83,7 @@ export async function createDebugBundle(input: {
   status?: unknown;
   notes?: string[];
   env?: NodeJS.ProcessEnv;
+  includeLogTail?: boolean;
 } = {}): Promise<{ bundleDir: string; diagnosticsPath: string }> {
   const env = input.env ?? process.env;
   const base = path.resolve(env.CODEX_SUBAGENTS_DEBUG_BUNDLE_DIR?.trim() || os.tmpdir());
@@ -101,7 +109,7 @@ export async function createDebugBundle(input: {
     session: input.session,
     job: input.job,
     notes: input.notes,
-    logTail: logFile ? await tailFile(logFile) : undefined,
+    logTail: input.includeLogTail && logFile ? await tailFile(logFile) : undefined,
   });
   const diagnosticsPath = path.join(bundleDir, "diagnostics.json");
   await writeFile(diagnosticsPath, JSON.stringify(payload, null, 2), "utf8");

src/diagnostics.ts

@@ -11,6 +11,7 @@ import {
   probeCodexVersion,
   reasoningEfforts,
   reasoningSummaries,
+  resolveWorkingDirectory,
   sandboxModes,
   serviceTiers,
 } from "./runner.js";
@@ -2020,6 +2021,10 @@ server.registerTool(
       session_id: sessionIdSchema.optional(),
       job_id: jobIdSchema.optional(),
       include_all_sessions: z.boolean().default(false),
+      include_log_tail: z
+        .boolean()
+        .default(false)
+        .describe("Include a bounded tail of CODEX_SUBAGENTS_LOG_FILE in the bundle. This may contain raw MCP traffic."),
     },
   },
   async (args, extra) =>
@@ -2048,8 +2053,11 @@ server.registerTool(
         },
         notes: [
           "The bundle intentionally records environment key names, not environment values.",
-          "If CODEX_SUBAGENTS_LOG_FILE is configured, diagnostics.json includes a bounded tail of that log file.",
+          args.include_log_tail
+            ? "A bounded CODEX_SUBAGENTS_LOG_FILE tail was included because include_log_tail was true."
+            : "The configured log file tail was not included; rerun with include_log_tail=true when raw MCP traffic is needed.",
         ],
+        includeLogTail: args.include_log_tail,
       });
       await progress.flush();
       return jsonResult({
@@ -2158,10 +2166,10 @@ server.registerTool(
       }
 
       try {
-        const projectDir = args.project_dir ?? process.env.CLAUDE_PROJECT_DIR ?? process.cwd();
+        const projectDir = await resolveWorkingDirectory(args.project_dir);
         checks.push({
           name: "project_dir",
-          ok: Boolean(projectDir),
+          ok: true,
           detail: { projectDir: cleanOption(projectDir) },
         });
       } catch (error) {

src/index.ts

@@ -349,6 +349,11 @@ export async function runQueuedAgents(
 export class CodexJobManager {
   private readonly jobs = new Map<string, JobRecord>();
   private readonly ttlMs = readPositiveInt(process.env.CODEX_SUBAGENTS_JOB_TTL_SECONDS, 3600, 86_400) * 1000;
+  private readonly maxJobs: number;
+
+  constructor(maxJobs = readPositiveInt(process.env.CODEX_SUBAGENTS_MAX_JOBS, 200, 10_000)) {
+    this.maxJobs = maxJobs;
+  }
 
   startAgent(options: AgentRunOptions): JobSnapshot {
     return this.start("agent", async (job) => {
@@ -446,11 +451,12 @@ export class CodexJobManager {
     return snapshot(job);
   }
 
-  stats(): QueueStats & { jobs: number; waiters: number } {
+  stats(): QueueStats & { jobs: number; maxJobs: number; waiters: number } {
     this.prune();
     return {
       ...agentRunQueue.stats(),
       jobs: this.jobs.size,
+      maxJobs: this.maxJobs,
       waiters: [...this.jobs.values()].reduce((count, job) => count + job.waiters.size, 0),
     };
   }
@@ -472,6 +478,13 @@ export class CodexJobManager {
 
   private start(kind: JobKind, run: (job: JobRecord) => Promise<unknown>): JobSnapshot {
     this.prune();
+    this.pruneOverflow();
+    if (this.jobs.size >= this.maxJobs) {
+      logger.warn("job.start_rejected_backpressure", { kind, jobs: this.jobs.size, maxJobs: this.maxJobs });
+      throw new BackpressureError(
+        `Codex async job table is full (${this.jobs.size}/${this.maxJobs}). Wait for or cancel existing jobs before starting another asynchronous run.`,
+      );
+    }
     const now = new Date().toISOString();
     const job: JobRecord = {
       id: `job-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 10)}`,
@@ -542,6 +555,17 @@ export class CodexJobManager {
       if (Date.parse(job.completedAt) < cutoff) this.jobs.delete(id);
     }
   }
+
+  private pruneOverflow(): void {
+    if (this.jobs.size < this.maxJobs) return;
+    const completed = [...this.jobs.entries()]
+      .filter(([, job]) => Boolean(job.completedAt))
+      .sort(([, left], [, right]) => Date.parse(left.completedAt ?? "") - Date.parse(right.completedAt ?? ""));
+    for (const [id] of completed) {
+      if (this.jobs.size < this.maxJobs) return;
+      this.jobs.delete(id);
+    }
+  }
 }
 
 export const jobManager = new CodexJobManager();

src/jobs.ts

@@ -1,5 +1,5 @@
 import { createHash, randomUUID } from "node:crypto";
-import { appendFileSync, mkdirSync, renameSync, statSync } from "node:fs";
+import { appendFileSync, chmodSync, mkdirSync, renameSync, statSync } from "node:fs";
 import path from "node:path";
 import { redactJsonValue, redactSensitiveText } from "./redaction.js";
 
@@ -20,6 +20,7 @@ const sensitiveKeyRe = /(api[_-]?key|token|secret|password|private[_-]?key|cooki
 let logWriter: (line: string) => void = (line) => {
   writeDefaultLog(line);
 };
+let lastLogFileError: string | undefined;
 
 export function configuredLogProfile(env: NodeJS.ProcessEnv = process.env): LogProfile {
   const raw = env.CODEX_SUBAGENTS_LOG_PROFILE?.trim().toLowerCase();
@@ -184,6 +185,7 @@ export function setLogWriterForTest(writer: (line: string) => void): void {
 }
 
 export function resetLogWriterForTest(): void {
+  lastLogFileError = undefined;
   logWriter = (line) => {
     writeDefaultLog(line);
   };
@@ -198,6 +200,7 @@ export function loggingDiagnostics(env: NodeJS.ProcessEnv = process.env): Record
     maxStringChars: maxStringChars(env),
     logFile: logFile || undefined,
     logFileMaxBytes: logFile ? logFileMaxBytes(env) : undefined,
+    logFileLastError: lastLogFileError,
   };
 }
 
@@ -209,11 +212,17 @@ function writeDefaultLog(line: string): void {
     mkdirSync(path.dirname(logFile), { recursive: true });
     try {
       if (statSync(logFile).size > logFileMaxBytes()) renameSync(logFile, `${logFile}.1`);
-    } catch {
+    } catch (error) {
       // Missing files or rotation races are harmless.
+      if ((error as NodeJS.ErrnoException | undefined)?.code !== "ENOENT") {
+        lastLogFileError = error instanceof Error ? error.message : String(error);
+      }
     }
-    appendFileSync(logFile, `${line}\n`, "utf8");
-  } catch {
+    appendFileSync(logFile, `${line}\n`, { encoding: "utf8", mode: 0o600 });
+    chmodSync(logFile, 0o600);
+    lastLogFileError = undefined;
+  } catch (error) {
+    lastLogFileError = error instanceof Error ? error.message : String(error);
     // Logging must never break MCP traffic or Codex execution.
   }
 }

src/logging.ts

@@ -48,15 +48,20 @@ export function redactSensitiveText(text: string): string {
   return redacted;
 }
 
-export function redactJsonValue<T>(value: T): T {
+export function isSensitiveKey(key: string): boolean {
+  return SENSITIVE_ENV_KEY.test(key);
+}
+
+export function redactJsonValue<T>(value: T, key = ""): T {
+  if (key && isSensitiveKey(key)) return "[REDACTED]" as T;
   if (typeof value === "string") return redactSensitiveText(value) as T;
-  if (Array.isArray(value)) return value.map((item) => redactJsonValue(item)) as T;
+  if (Array.isArray(value)) return value.map((item) => redactJsonValue(item, key)) as T;
   if (!value || typeof value !== "object") return value;
 
   return Object.fromEntries(
     Object.entries(value as Record<string, unknown>).map(([key, child]) => [
       key,
-      redactJsonValue(child),
+      redactJsonValue(child, key),
     ]),
   ) as T;
 }

src/redaction.ts

@@ -1,4 +1,5 @@
 import type { AgentRunPartial, AgentRunResult, CodexEventSummary } from "./runner.js";
+import { redactSensitiveText } from "./redaction.js";
 
 interface TruncatedString {
   text: string;
@@ -104,7 +105,9 @@ export function compactAgentResultForMcp(
     stdoutTail: stdoutTail.text,
     eventSummary: compactSummary(result.eventSummary, limits),
     structuredOutput: compactUnknown(result.structuredOutput, limits.structuredStringChars),
-    commandPreview: result.commandPreview.slice(0, 40).map((arg) => truncateString(arg, 1_000).text),
+    commandPreview: result.commandPreview
+      .slice(0, 40)
+      .map((arg) => truncateString(redactSensitiveText(arg), 1_000).text),
     mcpResponse: {
       compacted,
       finalMessageOmittedChars: finalMessage.omittedChars,