fix(cost): quarantine unreadable session logs instead of failing every poll

sdk/typescript/src/cost.ts

@@ -30,6 +30,7 @@ interface SessionUsage {
   offset: number;
   pendingLine: Buffer[];
   pendingLineBytes: number;
+  openFailures: number;
   unreadable: boolean;
   threadId: string | null;
   parentThreadId: string | null;
@@ -56,6 +57,22 @@ const MODEL_PRICING_NANODOLLARS: Readonly<Record<string, ModelPricing>> = {
   "gpt-5.6-luna": [1_000, 100, 1_250, 6_000],
 };
 
+// Open failures that cannot succeed again until the file itself changes, so
+// retrying them only wastes a syscall per poll. Every other code (EMFILE,
+// ENFILE, EBUSY, EIO and friends) may clear on its own and is retried.
+const PERMANENT_ACCESS_ERROR_CODES: ReadonlySet<string> = new Set([
+  "EACCES",
+  "EPERM",
+  "EISDIR",
+  "ELOOP",
+  "ENAMETOOLONG",
+  "ENOTDIR",
+]);
+
+// A retryable code that never clears must still stop somewhere, otherwise the
+// "reported once, then skipped" guarantee holds only for the codes listed above.
+const MAX_SESSION_OPEN_ATTEMPTS = 5;
+
 const COST_POLL_INTERVAL_MS = 100;
 const SESSION_READ_SIZE = 64 * 1_024;
 const MAX_SESSION_EVENT_BYTES = 1 * 1_024 * 1_024;
@@ -116,9 +133,23 @@ export class ScanCostTracker {
       clearInterval(this.#timer);
       this.#timer = null;
     }
-    await this.refresh();
-    if (this.#snapshot.usage !== null) return this.#snapshot;
+    let refreshed = true;
+    try {
+      await this.refresh();
+    } catch {
+      refreshed = false;
+    }
+    if (refreshed && this.#snapshot.usage !== null) return this.#snapshot;
+    // This refresh failed, so the snapshot predates the completed turn. The
+    // caller's own usage is authoritative for the scan thread, so it has to be
+    // able to win; keeping the stale snapshot would hide spend from `onCost`.
     const cost = estimateScanCost(this.#options.model, fallbackUsage);
+    if (
+      this.#snapshot.usage !== null &&
+      !chargesMore(cost, this.#snapshot.cost)
+    ) {
+      return this.#snapshot;
+    }
     this.#snapshot = { usage: fallbackUsage ?? null, cost };
     this.#reportCost(cost);
     return this.#snapshot;
@@ -136,6 +167,7 @@ export class ScanCostTracker {
           offset: 0,
           pendingLine: [],
           pendingLineBytes: 0,
+          openFailures: 0,
           unreadable: false,
           threadId: null,
           parentThreadId: null,
@@ -222,8 +254,20 @@ async function readSessionUsage(
     file = await open(path, "r");
   } catch (error) {
     if (isMissingFile(error)) return;
+    // A process-wide shortage such as EMFILE clears on its own, so quarantining
+    // on the first failure would stop observing this session's usage for the
+    // rest of the scan. A file-specific access failure cannot clear on its own,
+    // and anything else is retried a few times before it is retired.
+    session.openFailures += 1;
+    if (
+      isPermanentAccessError(error) ||
+      session.openFailures >= MAX_SESSION_OPEN_ATTEMPTS
+    ) {
+      quarantineSession(session);
+    }
     throw error;
   }
+  session.openFailures = 0;
   try {
     const buffer = Buffer.alloc(SESSION_READ_SIZE);
     while (true) {
@@ -238,9 +282,7 @@ async function readSessionUsage(
       try {
         readSessionChunk(buffer.subarray(0, bytesRead), session);
       } catch (error) {
-        session.unreadable = true;
-        session.pendingLine = [];
-        session.pendingLineBytes = 0;
+        quarantineSession(session);
         throw error;
       }
     }
@@ -249,6 +291,12 @@ async function readSessionUsage(
   }
 }
 
+function quarantineSession(session: SessionUsage): void {
+  session.unreadable = true;
+  session.pendingLine = [];
+  session.pendingLineBytes = 0;
+}
+
 function readSessionChunk(contents: Buffer, session: SessionUsage): void {
   let lineStart = 0;
   while (lineStart < contents.length) {
@@ -372,6 +420,20 @@ function isMissingFile(error: unknown): boolean {
   return isRecord(error) && error["code"] === "ENOENT";
 }
 
+function isPermanentAccessError(error: unknown): boolean {
+  if (!isRecord(error)) return false;
+  const code = error["code"];
+  return typeof code === "string" && PERMANENT_ACCESS_ERROR_CODES.has(code);
+}
+
+function chargesMore(
+  cost: ScanCost | null,
+  previous: ScanCost | null,
+): boolean {
+  if (cost === null) return false;
+  return previous === null || cost.estimatedUsd > previous.estimatedUsd;
+}
+
 export function estimateScanCost(
   model: string | undefined,
   usage: unknown,

sdk/typescript/tests-ts/api.test.ts

@@ -1,4 +1,5 @@
 import {
+  appendFile,
   copyFile,
   cp,
   mkdir,
@@ -2176,6 +2177,103 @@ describe("CodexSecurity orchestration", () => {
     await client.close();
   });
 
+  test("enforces the cost limit on a completed turn the final refresh cannot read", async () => {
+    const root = await temporaryDirectory();
+    const repository = join(root, "repository");
+    const codexHome = join(root, "codex-home");
+    const scanDir = join(root, "scan");
+    await mkdir(repository);
+    await mkdir(codexHome);
+    await mkdir(scanDir, { mode: 0o700 });
+    const sessionPath = join(
+      codexHome,
+      "sessions",
+      "2026",
+      "07",
+      "26",
+      "rollout-scan-thread.jsonl",
+    );
+    const commands: Array<readonly string[]> = [];
+    const costs: number[] = [];
+    const client = new TestClient(
+      {},
+      {
+        environment: {},
+        prepareRuntime: async () => preparedRuntime(codexHome),
+        resolvePluginPython: async () => "/managed/python",
+        prepareOutputDir: async () => scanDir,
+        repositoryRevision: async () => "deadbeef",
+        runWorkbench: async (_options: unknown, args: readonly string[]) => {
+          commands.push(args);
+          if (args[0] === "register-cli-scan") {
+            return mockScanRegistration(args);
+          }
+          if (args[0] === "get-scan-feedback") {
+            return {
+              scanId: "scan_example_001",
+              targetId: "target_sha256_example",
+              falsePositives: [],
+            };
+          }
+          return {};
+        },
+        createCodex: () => ({
+          startThread: () => ({
+            id: null,
+            async runStreamed() {
+              await copyCompletedScan(root);
+              async function* events(): AsyncGenerator<ThreadEvent> {
+                yield { type: "thread.started", thread_id: "scan-thread" };
+                // Half the completed turn's spend, so the recorded snapshot
+                // stays below the limit while the turn itself passes it.
+                await writeUsageSession(codexHome, "scan-thread", {
+                  input_tokens: 500,
+                  cached_input_tokens: 100,
+                  output_tokens: 10,
+                });
+                for (let attempt = 0; costs.length === 0; attempt += 1) {
+                  if (attempt >= 200) throw new Error("No cost was polled.");
+                  await new Promise<void>((resolve) => setTimeout(resolve, 5));
+                }
+                // The rollout becomes unreadable exactly as the turn completes.
+                await appendFile(
+                  sessionPath,
+                  "x".repeat(1 * 1_024 * 1_024 + 1),
+                );
+                yield {
+                  type: "turn.completed",
+                  usage: {
+                    input_tokens: 1_250,
+                    cached_input_tokens: 200,
+                    output_tokens: 30,
+                    reasoning_output_tokens: 5,
+                  },
+                };
+              }
+              return { events: events() };
+            },
+          }),
+        }),
+      },
+    );
+
+    await expect(
+      client.run(repository, {
+        maxCostUsd: 0.005,
+        onCost: (cost) => costs.push(cost.estimatedUsd),
+        signal: AbortSignal.timeout(10_000),
+      }),
+    ).rejects.toMatchObject({
+      name: ScanCostLimitExceededError.name,
+      maxCostUsd: 0.005,
+      cost: { estimatedUsd: 0.00625, inputTokens: 1_250, outputTokens: 30 },
+    });
+    expect(costs).toContain(0.00625);
+    expect(commands.some((args) => args[0] === "complete-scan")).toBe(false);
+    expect(commands.some((args) => args[0] === "fail-scan")).toBe(true);
+    await client.close();
+  });
+
   test("saves a budgeted scan with a warning when token usage is unavailable", async () => {
     const root = await temporaryDirectory();
     const repository = join(root, "repository");