From 5fdc91c041cc202756f2c5611b729fc34e689f20 Mon Sep 17 00:00:00 2001 From: ianw-oai Date: Wed, 11 Nov 2026 10:43:37 -0800 Subject: [PATCH 1/5] feat(scans): resume interrupted Standard scans --- docs/agent-workflows.md | 4 +- .../scripts/workbench_scan_history.py | 4 +- .../scripts/workbench_scan_start.py | 2 +- sdk/typescript/docs/cli.md | 27 +- sdk/typescript/scripts/package-dist-files.mjs | 1 + sdk/typescript/src/api.ts | 90 +++++- sdk/typescript/src/cli.ts | 5 +- sdk/typescript/src/scan-execution-lock.ts | 39 +++ sdk/typescript/src/scan-history-renderer.ts | 2 +- .../tests-ts/scan-execution-lock.test.ts | 51 +++ sdk/typescript/tests-ts/scan-resume.test.ts | 304 ++++++++++++++++-- 11 files changed, 484 insertions(+), 45 deletions(-) create mode 100644 sdk/typescript/src/scan-execution-lock.ts create mode 100644 sdk/typescript/tests-ts/scan-execution-lock.test.ts diff --git a/docs/agent-workflows.md b/docs/agent-workflows.md index fcf8512e40..e4a5bc00b1 100644 --- a/docs/agent-workflows.md +++ b/docs/agent-workflows.md @@ -161,10 +161,10 @@ cs scans logs SCAN_ID ``` Without an ID, `scans show` selects the latest completed scan, while `scans logs` -selects the latest scan of any status. An interrupted Deep Scan still marked +selects the latest scan of any status. An interrupted Standard or Deep Scan still marked `running` can use `cs scans resume SCAN_ID` with its original checkout, output, session, and state directory. Completed, failed, and canceled scans need a new attempt with `cs scans rerun SCAN_ID`. A rerun uses the current checkout and may need the original custom prompt files supplied again. See -[resume requirements](../sdk/typescript/docs/cli.md#resuming-an-interrupted-deep-scan) +[resume requirements](../sdk/typescript/docs/cli.md#resuming-an-interrupted-scan) and [runtime troubleshooting](../sdk/typescript/docs/cli.md#troubleshooting). diff --git a/plugins/codex-security/scripts/workbench_scan_history.py b/plugins/codex-security/scripts/workbench_scan_history.py index 54dd0409dc..247341d357 100644 --- a/plugins/codex-security/scripts/workbench_scan_history.py +++ b/plugins/codex-security/scripts/workbench_scan_history.py @@ -140,8 +140,8 @@ def cli_scan_resume( ) -> dict[str, Any]: scan = scan_with_progress(connection, scan["id"]) - if scan["mode"] != "deep" or scan["recipe_json"] is None: - raise SystemExit("Resume requires a Deep Scan with a saved CLI launch recipe.") + if scan["mode"] not in {"standard", "deep"} or scan["recipe_json"] is None: + raise SystemExit("Resume requires a Standard or Deep Scan with a saved CLI launch recipe.") if scan["status"] != "running" or scan["canceled_at"] is not None: raise SystemExit( "Resume requires a running scan; completed, failed, and canceled scans cannot resume." diff --git a/plugins/codex-security/scripts/workbench_scan_start.py b/plugins/codex-security/scripts/workbench_scan_start.py index 641c52f702..2742507593 100644 --- a/plugins/codex-security/scripts/workbench_scan_start.py +++ b/plugins/codex-security/scripts/workbench_scan_start.py @@ -100,7 +100,7 @@ class WindowsResumeTarget: def prepare_cli_scan_target( target: Path, diff_target: dict[str, str] | None, mode: str, launch_kind: str ) -> tuple[tuple[str, str | None, int | str, int | str], WindowsResumeTarget | None]: - if os.name == "nt" and mode == "deep" and launch_kind == "scan": + if os.name == "nt" and mode in {"standard", "deep"} and launch_kind == "scan": from windows_scan_local_files import pinned_resume_checkout with pinned_resume_checkout(target) as native: diff --git a/sdk/typescript/docs/cli.md b/sdk/typescript/docs/cli.md index 5c07381427..61b3ea3d7f 100644 --- a/sdk/typescript/docs/cli.md +++ b/sdk/typescript/docs/cli.md @@ -724,8 +724,8 @@ Choose the next step from the recorded reason: start a new scan with the intended scope, including for plugin scans without a recipe. Retained findings can still be reviewed, but absence of findings from a stopped run is inconclusive. -- **A Deep Scan was interrupted and is still recorded as `running`:** it may - support [resume](#resuming-an-interrupted-deep-scan) with the original checkout, +- **A Standard or Deep Scan was interrupted and is still recorded as `running`:** it may + support [resume](#resuming-an-interrupted-scan) with the original checkout, artifacts, and session. Completed, failed, and canceled scans cannot resume. For example, after identifying an omitted API directory: @@ -956,7 +956,7 @@ behavior are unchanged. The SDK setting is `maxTimeMinutes`; project files use `scan.max_time_minutes`. Each launched scan or rerun gets its own allowance after setup, and reconnects -do not reset it. Standard scans still do not support resume. For batch and +do not reset it. [Resumed Standard scans](#resuming-an-interrupted-scan) keep the original deadline. For batch and component scans, the project setting applies separately to each scan attempt, not to planning, result matching, or the total campaign. @@ -2135,7 +2135,7 @@ in the workbench, without changing sealed report timestamps. | `scans list [REPOSITORY]` | List scans; `--scan-root DIR` filters artifact roots. | | `scans show [SCAN_ID]` | Show a scan; defaults to latest completed. | | `scans logs [SCAN_ID]` | Show session events; defaults to latest, including active. | -| `scans resume SCAN_ID` | Continue an interrupted Deep Scan. | +| `scans resume SCAN_ID` | Continue an interrupted Standard or Deep Scan. | | `scans rerun [SCAN_ID]` | Rerun a saved recipe in its checkout. | | `scans match --all` | Match completed scans in this repository generation, including linked worktrees. | | `scans compare [BEFORE] [AFTER]` | Compare scans; defaults to latest two completed. | @@ -2231,7 +2231,7 @@ work on all supported Node.js versions, including logs, feedback attachments, usage accounting, and saved-session discovery for resume. A bundled decoder reads concatenated frames and reports truncated streams as collection errors. -### Resuming an interrupted Deep Scan +### Resuming an interrupted scan ```bash codex-security scans list --scan-root /path/to/security-scans @@ -2239,7 +2239,7 @@ codex-security scans resume SCAN_ID ``` The terminal list shows the full scan ID and labels the last activity of a -`running` scan after 24 hours without an update. For a Deep Scan with a saved +`running` scan after 24 hours without an update. For a Standard or Deep Scan with a saved launch recipe and its original session, it also shows the resume command to use after interruption. The scan must still be `running`, with its original checkout, output directory, @@ -2251,7 +2251,7 @@ output. It lets the workbench verify the original output even before the first checkpoint. If an older scan stopped with an empty output, start a new scan with `codex-security scans rerun SCAN_ID`. -New Windows Deep Scans also record the checkout's native volume and file ID +New Windows Standard and Deep Scans also record the checkout's native volume and file ID when the filesystem supplies them. Validation and remediation verify the same checkout identity. An older scan can resume across Python versions when its saved checkout identity can be verified. If the workbench @@ -2259,7 +2259,18 @@ cannot verify it, resume with the Python runtime that started the scan or use `codex-security scans rerun SCAN_ID` to start a new scan. Resume retains scan ID, completed workers, artifacts, accumulated cost, and saved -settings/instructions. Knowledge-bearing scans save their extracted documents in +settings/instructions. Standard scans continue the original Codex session from +saved analysis, findings, and review checkpoints; they do not start a new scan. +Only one CLI or SDK call can run a Standard scan at a time. A network failure +after the session is saved leaves the scan resumable and prints the recovery +command. The original `--max-cost` limit and `--max-time-minutes` deadline still +apply. Work not saved before the interruption may need to be repeated. +Explicit cancellation, authentication failures, and other terminal failures +still require a rerun. This does not reopen scans marked failed by older versions. +Custom-validation scans still require their original validation instructions and +are not supported by `scans resume`. + +Knowledge-bearing scans save their extracted documents in the private `.scan-knowledge.json` continuation file beside the scan artifacts. Resume verifies and uses that snapshot even if the original files changed or were deleted. Preserve it with the scan directory; canonical reports do not embed its diff --git a/sdk/typescript/scripts/package-dist-files.mjs b/sdk/typescript/scripts/package-dist-files.mjs index 8d9fb29d2f..3dfc7507d4 100644 --- a/sdk/typescript/scripts/package-dist-files.mjs +++ b/sdk/typescript/scripts/package-dist-files.mjs @@ -2,6 +2,7 @@ const distModules = [ "api", "batch-cost", "scan-executor", + "scan-execution-lock", "host-protocol", "hosted-remediation", "hosted-remediation-protocol", diff --git a/sdk/typescript/src/api.ts b/sdk/typescript/src/api.ts index 9536d3cec5..d808b301e6 100644 --- a/sdk/typescript/src/api.ts +++ b/sdk/typescript/src/api.ts @@ -117,6 +117,7 @@ import type { ScanPricingContext, ScanTokenUsage } from "./cost-model.js"; import { measuredScanCost, combineMeasurements } from "./scan-measurement.js"; import { ScanProgressTracker, type DeepScanProgress } from "./scan-progress.js"; import { findScanSession } from "./scan-logs.js"; +import { acquireScanExecutionLock } from "./scan-execution-lock.js"; import { DIRECT_DEEP_SCAN_CONTROL_USAGE, deepScanParentToolOverrides, @@ -2236,6 +2237,8 @@ export class CodexSecurity { let latestMeasurement: Readonly | null = null; let notifiedLimit: number | undefined; let scanDir = ""; + let releaseScanExecution: (() => void) | undefined; + let standardScan = false; let targetPathsFile: string | null = null; let stagedPluginRoot: string | undefined; let stagedPluginDirectory: string | undefined; @@ -2414,6 +2417,10 @@ export class CodexSecurity { ), }; const hosted = this.#dependencies.hostedScan !== undefined; + standardScan = + mode === "standard" && + !hosted && + (normalized.kind === "repository" || normalized.kind === "paths"); for (const root of [ (await gitMarkerRoot(repo, signal, "outermost")) ?? repo, ...(knowledgeBase?.protectedRoots ?? []), @@ -2462,6 +2469,12 @@ export class CodexSecurity { ); requireOutputOutsideRepository(protectedRoot, scanDir); requireModelSafeOutputDir(scanDir); + if (standardScan) { + releaseScanExecution = await acquireScanExecutionLock( + stateDirectory, + scanDir, + ); + } if (!options.archiveExisting && options.resumeScanId === undefined) { notifyObserver(options, "onOutputDirReady")(scanDir); } @@ -3460,7 +3473,9 @@ export class CodexSecurity { } if (options.resumeScanId !== undefined) { prompt += - "\nResume the existing Deep Scan through its coordinator. Preserve completed workers and saved artifacts; do not recreate the scan directory or restart completed analysis. If the coordinator already finished, continue with completion of this same scan."; + mode === "deep" + ? "\nResume the existing Deep Scan through its coordinator. Preserve completed workers and saved artifacts; do not recreate the scan directory or restart completed analysis. If the coordinator already finished, continue with completion of this same scan." + : "\nResume this same Standard scan in its original session and output directory. Read the saved scan-manifest.json, findings.json, and scan.reviewState checkpoints when present. Preserve completed analysis, validation decisions, evidence, and stable finding/surface IDs. Continue unfinished validation and deferred work; do not restart completed analysis or recreate the scan directory. If the review is finished, complete this same scan using its saved results."; } if ( falsePositiveExamples.length > 0 && @@ -3641,11 +3656,15 @@ export class CodexSecurity { timePolicy, scanTimeBudget( options.maxTimeMinutes, - (this.#dependencies.now ?? Date.now)(), + options.resumeScanId !== undefined && + typeof registration["startedAt"] === "string" + ? Date.parse(registration["startedAt"]) + : (this.#dependencies.now ?? Date.now)(), ), )}`; } tracker.beginParentTurn(resumeThreadId !== undefined); + scanTurnStatus = "in_progress"; const events = directDeepScan ? (this.#dependencies.runDeepScan ?? runDeepScan)({ onDiagnostic: this.#onDeepScanDiagnostic, @@ -3662,7 +3681,6 @@ export class CodexSecurity { signal, }) : (await thread.runStreamed(prompt, turnOptions)).events; - scanTurnStatus = "in_progress"; checkOpen(); let result = await runScanEvents({ @@ -4256,9 +4274,39 @@ export class CodexSecurity { ); } } - // A failed attachment must not turn a resumable coordinator into a terminal failure. + if ( + activeScan !== null && + standardScan && + !signal.aborted && + options.validationPrompt === undefined && + scanTurnStatus !== null && + terminalTurnStatus !== "completed" && + failure instanceof CodexConnectionError + ) { + // Keep the existing running scan only when its saved session and inputs can resume. + const saved = await workbench( + { ...activeScan.options, signal: undefined }, + workbenchArgs("get-cli-scan-resume", { + scanId: activeScan.id, + allowUnavailable: true, + }), + ).catch(() => null); + if (saved?.["scanId"] === activeScan.id) { + failure = new ScanInterruptedError( + `${errorMessage(failure)}\nResume when connected: codex-security scans resume ${activeScan.id}`, + scanDir, + { cause: failure }, + ); + activeScan = null; + runPostScan = null; + } + } // Deep Scan orchestration persists its own terminal failures and cancellations. - if (activeScan !== null && options.resumeScanId === undefined) { + // Standard scans have no coordinator, so fatal errors and cancellation still stop them. + if ( + activeScan !== null && + (options.resumeScanId === undefined || standardScan) + ) { if ( options.validationPrompt !== undefined && customValidation.status !== "complete" @@ -4340,6 +4388,7 @@ export class CodexSecurity { warnCleanupFailed(options, error); } observeSettledScan(options, activeScan); + releaseScanExecution?.(); } } @@ -5639,7 +5688,7 @@ export class CodexSecurity { ): Promise { if ( options.resumeScanId !== undefined && - (options.mode !== "deep" || + ((options.mode !== "deep" && options.mode !== "standard") || !options.outputDir || options.archiveExisting || options.parentScanId !== undefined || @@ -5647,7 +5696,7 @@ export class CodexSecurity { options.mock) ) { throw new CodexSecurityError( - "Resume requires the original Deep Scan output directory without archive, rerun, or workflow options.", + "Resume requires the original Standard or Deep Scan output directory without archive, rerun, or workflow options.", ); } if ( @@ -6093,6 +6142,10 @@ export async function runScanEvents( ); } if (status !== "completed") { + if (lastStreamError !== null) { + const error = codexTurnError(lastStreamError); + if (error instanceof CodexConnectionError) throw error; + } throw new IncompleteScanError( lastStreamError ?? "Codex Security event stream ended before the turn completed.", @@ -6200,6 +6253,15 @@ export async function runScanEvents( } } +class CodexConnectionError extends CodexSecurityError {} + +function codexTurnError(error: unknown): CodexSecurityError { + const classification = classifyConnectionFailure(error); + return classification === "network_error" || classification === "timeout" + ? new CodexConnectionError(errorMessage(error), { cause: error }) + : new CodexSecurityError(errorMessage(error), { cause: error }); +} + async function readCodexTurn(options: { thread: Pick; events: AsyncGenerator; @@ -6217,12 +6279,22 @@ async function readCodexTurn(options: { join(await bundledPluginRoot(), "scripts", "codex_session.mjs"), ).href ); + async function* transportEvents() { + try { + yield* options.events; + } catch (error) { + const failure = codexTurnError(error); + if (failure instanceof CodexConnectionError) throw failure; + throw error; + } + } return readCodexSessionTurn({ ...options, + events: transportEvents(), onEvent: async (event: ScanEvent) => { await options.onEvent?.(event); if (event.type === "turn.failed") { - throw new CodexSecurityError(turnFailureMessage(event["error"])); + throw codexTurnError(turnFailureMessage(event["error"])); } if (event.type === "error" && typeof event["message"] === "string") { const message = event["message"]; @@ -6234,7 +6306,7 @@ async function readCodexTurn(options: { throw new CodexSecurityError(message); } const reconnect = reconnectAttempt(message); - if (reconnect === null) throw new CodexSecurityError(message); + if (reconnect === null) throw codexTurnError(message); options.onReconnect?.(message, reconnect); } }, diff --git a/sdk/typescript/src/cli.ts b/sdk/typescript/src/cli.ts index fc11e3275a..195cf3f33c 100644 --- a/sdk/typescript/src/cli.ts +++ b/sdk/typescript/src/cli.ts @@ -3110,11 +3110,12 @@ export async function main( }, }) .command("resume", { - description: "Resume an interrupted Deep Scan in its original session.", + description: + "Resume an interrupted Standard or Deep Scan in its original session.", examples: [{ args: { scanId: "SCAN_ID" } }], hint: "Execution failures keep available scan results under data in failed full-output JSON, JSONL, YAML, or TOON envelopes.", args: z.object({ - scanId: z.string().min(1).describe("Interrupted Deep Scan identifier."), + scanId: z.string().min(1).describe("Interrupted scan identifier."), }), options: z.object({ showCost: SHOW_COST_OPTION, diff --git a/sdk/typescript/src/scan-execution-lock.ts b/sdk/typescript/src/scan-execution-lock.ts new file mode 100644 index 0000000000..1e20925d96 --- /dev/null +++ b/sdk/typescript/src/scan-execution-lock.ts @@ -0,0 +1,39 @@ +import { createHash } from "node:crypto"; +import { chmod, mkdir, realpath, stat } from "node:fs/promises"; +import { join } from "node:path"; +import { CodexSecurityError } from "./errors.js"; +import { isRecord } from "./record.js"; +import { requirePrivateDirectory, sqliteDatabaseOpener } from "./runtime.js"; + +/** Hold one Standard scan's execution across processes, including resume. */ +export async function acquireScanExecutionLock( + stateDirectory: string, + scanDirectory: string, +): Promise<() => void> { + const directory = join(stateDirectory, "scan-locks"); + await mkdir(directory, { recursive: true, mode: 0o700 }); + await requirePrivateDirectory(await stat(directory), directory, "scan locks"); + const identity = createHash("sha256") + .update(await realpath(scanDirectory)) + .digest("hex"); + const path = join(directory, `${identity}.sqlite3`); + const database = sqliteDatabaseOpener()(path); + try { + if (process.platform !== "win32") await chmod(path, 0o600); + database.exec("PRAGMA busy_timeout = 0"); + database.exec("BEGIN EXCLUSIVE"); + } catch (error) { + database.close(); + if ( + isRecord(error) && + (error["errcode"] === 5 || error["code"] === "SQLITE_BUSY") + ) { + throw new CodexSecurityError( + "This scan is already running in another CLI or SDK call.", + ); + } + throw error; + } + // The OS also releases the lock if the scan process exits unexpectedly. + return () => database.close(); +} diff --git a/sdk/typescript/src/scan-history-renderer.ts b/sdk/typescript/src/scan-history-renderer.ts index c7abf6c1a2..da2b505d59 100644 --- a/sdk/typescript/src/scan-history-renderer.ts +++ b/sdk/typescript/src/scan-history-renderer.ts @@ -260,7 +260,7 @@ export function renderScanHistory( ) { lines.push(` ${dim(`Last activity: ${clean(updatedAt)}`)}`); if ( - scan["mode"] === "deep" && + (scan["mode"] === "deep" || scan["mode"] === "standard") && scan["recipeAvailable"] === true && scan["handoffStatus"] === "delivered" && scan["continuationThreadId"] diff --git a/sdk/typescript/tests-ts/scan-execution-lock.test.ts b/sdk/typescript/tests-ts/scan-execution-lock.test.ts new file mode 100644 index 0000000000..71ada709e5 --- /dev/null +++ b/sdk/typescript/tests-ts/scan-execution-lock.test.ts @@ -0,0 +1,51 @@ +import { join } from "node:path"; +import { mkdir } from "node:fs/promises"; +import { afterEach, expect, test } from "bun:test"; +import { acquireScanExecutionLock } from "../src/scan-execution-lock.js"; +import { createApiTestFixtures } from "./support/temporary-directories.js"; + +const { temporaryDirectory, cleanup } = createApiTestFixtures(); +afterEach(cleanup); + +test("scan execution excludes a second process and recovers after the owner exits", async () => { + const root = await temporaryDirectory(); + const state = join(root, "state"); + const scan = join(root, "scan"); + await mkdir(scan); + const module = new URL("../src/scan-execution-lock.ts", import.meta.url).href; + const child = Bun.spawn( + [ + process.execPath, + "--eval", + ` + const { acquireScanExecutionLock } = await import(${JSON.stringify(module)}); + await acquireScanExecutionLock(${JSON.stringify(state)}, ${JSON.stringify(scan)}); + console.log("locked"); + process.stdin.resume(); + `, + ], + { stdin: "pipe", stdout: "pipe", stderr: "pipe" }, + ); + try { + const reader = child.stdout.getReader(); + expect(new TextDecoder().decode((await reader.read()).value)).toBe( + "locked\n", + ); + reader.releaseLock(); + await expect(acquireScanExecutionLock(state, scan)).rejects.toThrow( + "already running", + ); + } finally { + child.kill(); + await child.exited; + } + const release = await acquireScanExecutionLock(state, scan); + try { + await expect(acquireScanExecutionLock(state, scan)).rejects.toThrow( + "already running", + ); + } finally { + release(); + } + (await acquireScanExecutionLock(state, scan))(); +}); diff --git a/sdk/typescript/tests-ts/scan-resume.test.ts b/sdk/typescript/tests-ts/scan-resume.test.ts index 0c49923841..785a095cc0 100644 --- a/sdk/typescript/tests-ts/scan-resume.test.ts +++ b/sdk/typescript/tests-ts/scan-resume.test.ts @@ -59,6 +59,7 @@ async function interruptedScan( ScanOptions, | "safetyIdentifier" | "maxCostUsd" + | "maxTimeMinutes" | "ownerSuggestions" | "postScanPrompt" | "auth" @@ -331,11 +332,11 @@ test("resume resolves an interrupted scan without changing its ID, recipe, or co expect(await readFile(f.checkpoint, "utf8")).toBe('{"completed":"setup"}\n'); }); -test.skipIf(process.platform !== "win32")( - "CLI resumes the original Windows session before its first output checkpoint", - async () => { +test.skipIf(process.platform !== "win32").each(["deep", "standard"] as const)( + "CLI resumes the original Windows %s session before its first output checkpoint", + async (mode) => { const f = await interruptedScan( - "deep", + mode, false, {}, false, @@ -361,13 +362,19 @@ test.skipIf(process.platform !== "win32")( return { id: threadId, async runStreamed() { - throw new Error("Synthetic continuation reached"); + async function* disconnected() { + yield { + type: "error" as const, + message: "Connection failed: error sending request", + }; + } + return { events: disconnected() }; }, }; }, })), ); - expect(stderr.text()).toContain("Synthetic continuation reached"); + expect(stderr.text()).toContain("Connection failed"); expect(resumedThread).toBe(f.threadId); expect( (await f.command(["get-scan", "--scan-id", f.scanId]))["scan"], @@ -378,19 +385,10 @@ test.skipIf(process.platform !== "win32")( }, ); -test.each([ - "failed", - "canceled", - "standard", - "changed", - "replaced", - "wrong-owner", -])( +test.each(["failed", "canceled", "changed", "replaced", "wrong-owner"])( "resume refuses %s scans without altering their saved state", async (scenario) => { - const f = await interruptedScan( - scenario === "standard" ? "standard" : "deep", - ); + const f = await interruptedScan(); if (scenario === "failed") await f.command([ "fail-scan", @@ -434,9 +432,7 @@ test.each([ ? "checkout is missing or was replaced" : scenario === "wrong-owner" ? "original owning CLI session" - : scenario === "standard" - ? "Deep Scan with a saved CLI launch recipe" - : "running scan; completed, failed, and canceled", + : "running scan; completed, failed, and canceled", ); expect(await f.command(["get-scan", "--scan-id", f.scanId])).toEqual( before, @@ -774,6 +770,274 @@ test.each([false, true])( }, ); +test("Standard resume preserves checkpoints through another disconnect and completes the same scan", async () => { + const f = await interruptedScan("standard", false, { + maxTimeMinutes: 5, + ownerSuggestions: false, + }); + await copyCompletedScanFixture(f.scanDir); + const manifestPath = join(f.scanDir, "scan-manifest.json"); + const manifest = JSON.parse(await readFile(manifestPath, "utf8")); + manifest.scan.id = f.scanId; + manifest.scan.target = { kind: "directory_snapshot" }; + manifest.scan.complete = false; + delete manifest.scan.sealedAt; + delete manifest.scan.artifacts; + await writeFile(manifestPath, JSON.stringify(manifest)); + for (const name of ["coverage.json", "findings.json"]) { + const path = join(f.scanDir, name); + const document = JSON.parse(await readFile(path, "utf8")); + document.scanId = f.scanId; + await writeFile(path, JSON.stringify(document)); + } + const original = await Promise.all( + ["scan-manifest.json", "findings.json", "coverage.json"].map((name) => + readFile(join(f.scanDir, name), "utf8"), + ), + ); + let attempts = 0; + const initial = await f.command([ + "get-cli-scan-resume", + "--scan-id", + f.scanId, + ]); + const deps = resumeDependencies(f, () => ({ + startThread: () => fail("Resume must not start a new session"), + resumeThread(id, options) { + expect(id).toBe(f.threadId); + expect(options.workingDirectory).toBe(f.scanDir); + return { + id, + async runStreamed(prompt) { + expect(prompt).toContain( + new Date(initial["startedAt"] as string).toISOString(), + ); + expect( + await Promise.all( + ["scan-manifest.json", "findings.json", "coverage.json"].map( + (name) => readFile(join(f.scanDir, name), "utf8"), + ), + ), + ).toEqual(original); + attempts++; + await appendFile( + f.sessionPath, + JSON.stringify({ + type: "event_msg", + payload: { + type: "token_count", + info: { + total_token_usage: { + input_tokens: 10000 + attempts * 100, + output_tokens: 2000 + attempts * 20, + }, + }, + }, + }) + "\n", + ); + if (attempts === 1) { + async function* disconnected() { + yield { type: "thread.started" as const, thread_id: id }; + yield { + type: "turn.failed" as const, + error: { message: "Connection failed: error sending request" }, + }; + } + return { events: disconnected() }; + } + delete manifest.scan.complete; + await writeFile(manifestPath, JSON.stringify(manifest)); + return { events: completedEvents(id) }; + }, + }; + }, + })); + const interrupted = createCliTest(main); + expect( + await interrupted.runCli(["scans", "resume", f.scanId, "--json"], deps), + ).toBe(2); + expect(interrupted.stderr.text()).toContain( + `codex-security scans resume ${f.scanId}`, + ); + expect( + await f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), + ).toEqual(initial); + const completed = createCliTest(main); + expect( + await completed.runCli(["scans", "resume", f.scanId, "--json"], deps), + completed.stderr.text(), + ).toBe(0); + expect(attempts).toBe(2); + expect( + (await f.command(["get-scan", "--scan-id", f.scanId]))["scan"], + ).toMatchObject({ + scanId: f.scanId, + continuationThreadId: f.threadId, + progress: { status: "complete" }, + usage: { totalTokens: 12240 }, + }); + expect(await readFile(f.checkpoint, "utf8")).toBe('{"completed":"setup"}\n'); + await expect( + f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), + ).rejects.toThrow("running scan"); +}); + +test("a new default Standard scan remains resumable after a network failure", async () => { + const f = await interruptedScan("standard"); + const scanDir = join(f.root, "new-scan"); + let scanId = ""; + const deps = resumeDependencies(f, ({ env }) => ({ + startThread: () => ({ + id: f.threadId, + async runStreamed() { + scanId = env!["CODEX_SECURITY_SCAN_ID"]!; + await writeFile( + f.sessionPath, + JSON.stringify({ + type: "session_meta", + payload: { id: f.threadId, cwd: scanDir }, + }) + "\n", + ); + async function* disconnected() { + yield { type: "thread.started" as const, thread_id: f.threadId }; + await writeFile( + join(scanDir, "checkpoint.json"), + '{"validated":"saved"}\n', + ); + yield { + type: "error" as const, + message: + "Reconnecting... waiting for network (Connection failed: error sending request)", + }; + } + return { events: disconnected() }; + }, + }), + })); + await using client = deps.createSecurity({}); + await expect( + client.run(f.repository, { + outputDir: scanDir, + target: ["source.py"], + ownerSuggestions: false, + }), + ).rejects.toThrow("codex-security scans resume"); + expect(scanId).not.toBe(f.scanId); + expect( + await f.command(["get-cli-scan-resume", "--scan-id", scanId]), + ).toMatchObject({ + scanId, + scanDir, + threadId: f.threadId, + recipe: { + mode: "standard", + target: { kind: "paths", paths: ["source.py"] }, + }, + }); + expect(await readFile(join(scanDir, "checkpoint.json"), "utf8")).toBe( + '{"validated":"saved"}\n', + ); +}); + +test.each(["failed", "canceled", "changed"])( + "Standard resume refuses %s scans", + async (scenario) => { + const f = await interruptedScan("standard"); + if (scenario === "failed") + await f.command([ + "fail-scan", + "--scan-id", + f.scanId, + "--message", + "Synthetic failure", + ]); + if (scenario === "canceled") + await f.command(["cancel-scan", "--scan-id", f.scanId]); + if (scenario === "changed") + await writeFile(join(f.repository, "source.py"), "# changed\n"); + const before = await f.command(["get-scan", "--scan-id", f.scanId]); + await expect( + f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), + ).rejects.toThrow( + scenario === "changed" ? "revision or contents changed" : "running scan", + ); + expect(await f.command(["get-scan", "--scan-id", f.scanId])).toEqual( + before, + ); + }, +); + +test.each(["access", "local"])( + "a Standard resume %s failure stops the scan", + async (kind) => { + const f = await interruptedScan("standard"); + const cli = createCliTest(main); + const code = await cli.runCli( + ["scans", "resume", f.scanId, "--json"], + resumeDependencies(f, () => ({ + startThread: () => fail("Unexpected new session"), + resumeThread: (id) => ({ + id, + async runStreamed() { + if (kind === "local") + throw new Error( + "Invalid network configuration in saved settings", + ); + async function* failed() { + yield { + type: "turn.failed" as const, + error: { message: "403 forbidden: access denied" }, + }; + } + return { events: failed() }; + }, + }), + })), + ); + expect(code).toBe(2); + expect( + (await f.command(["get-scan", "--scan-id", f.scanId]))["scan"], + ).toMatchObject({ progress: { status: "failed" } }); + expect(cli.stderr.text()).not.toContain( + `codex-security scans resume ${f.scanId}`, + ); + }, +); + +test("canceling Standard resume keeps the scan canceled", async () => { + const f = await interruptedScan("standard"); + const abort = new AbortController(); + const deps = resumeDependencies(f, () => ({ + startThread: () => fail("Unexpected new session"), + resumeThread: (id) => ({ + id, + async runStreamed() { + async function* canceled() { + yield { type: "thread.started" as const, thread_id: id }; + abort.abort(); + abort.signal.throwIfAborted(); + } + return { events: canceled() }; + }, + }), + })); + await using client = deps.createSecurity({}); + await expect( + client.run(f.repository, { + mode: "standard", + resumeScanId: f.scanId, + outputDir: f.scanDir, + signal: abort.signal, + }), + ).rejects.toThrow("interrupted"); + expect( + (await f.command(["get-scan", "--scan-id", f.scanId]))["scan"], + ).toMatchObject({ progress: { status: "canceled" } }); + await expect( + f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), + ).rejects.toThrow("running scan"); +}); + test.each([ [undefined, 2], [1, 1], From b0179da0375adee196fcbdac15ae355267093ad0 Mon Sep 17 00:00:00 2001 From: ianw-oai Date: Wed, 11 Nov 2026 10:54:13 -0800 Subject: [PATCH 2/5] test(scans): preserve Standard resume data during archive rollback --- .../tests/test_windows_workbench_output_permissions.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/plugins/codex-security/tests/test_windows_workbench_output_permissions.py b/plugins/codex-security/tests/test_windows_workbench_output_permissions.py index a358413069..1f213f92e9 100644 --- a/plugins/codex-security/tests/test_windows_workbench_output_permissions.py +++ b/plugins/codex-security/tests/test_windows_workbench_output_permissions.py @@ -532,8 +532,10 @@ def test_resume_verification_failure_rolls_back_the_created_handle( with closing(workbench_db.connect()) as connection: old_id = None + previous_identity = None if archive: old_id = register("standard", connection=connection)["scanId"] + previous_identity = windows_scan_local_files.deep_resume_output_identity(output) connection.execute("UPDATE scans SET status = 'failed' WHERE id = ?", (old_id,)) connection.commit() previous_file.write_bytes(b"original result") @@ -560,13 +562,14 @@ def competing_reader(handle, path, **options): with pytest.raises(OSError) as failure: register(archive=archive, connection=connection) assert (getattr(failure.value, "winerror", None) or failure.value.errno) == 32 - assert not retained.exists() assert not list(output.parent.glob(f"{output.name}.previous-*")) rows = connection.execute("SELECT id, scan_dir FROM scans").fetchall() if archive: + assert windows_scan_local_files.deep_resume_output_identity(output) == previous_identity assert [tuple(row) for row in rows] == [(old_id, str(output))] assert previous_file.read_bytes() == b"original result" else: + assert not retained.exists() assert rows == [] assert list(output.iterdir()) == [] assert register(archive=archive, connection=connection)["scanDir"] == str(output) From bd127c45485b1945d9f48720686c2a306cf60e98 Mon Sep 17 00:00:00 2001 From: ianw-oai <269041055+ianw-oai@users.noreply.github.com> Date: Wed, 11 Nov 2026 19:22:27 +0000 Subject: [PATCH 3/5] fix(scans): reconcile usage and preserve timing on Standard resume --- .../codex-security/scripts/workbench_cli.py | 1 + .../codex-security/scripts/workbench_db.py | 12 ++-- .../scripts/workbench_scan_start.py | 26 +++++++ .../scripts/workbench_scan_usage.py | 26 +++++++ .../tests/test_workbench_scan_usage_resume.py | 52 ++++++++++++++ sdk/typescript/src/api.ts | 34 +++++---- sdk/typescript/src/errors.ts | 1 + sdk/typescript/src/scan-executor.ts | 5 +- sdk/typescript/tests-ts/scan-resume.test.ts | 72 +++++++++++-------- 9 files changed, 182 insertions(+), 47 deletions(-) create mode 100644 plugins/codex-security/tests/test_workbench_scan_usage_resume.py diff --git a/plugins/codex-security/scripts/workbench_cli.py b/plugins/codex-security/scripts/workbench_cli.py index 7dad25c9af..17bc1e77fd 100644 --- a/plugins/codex-security/scripts/workbench_cli.py +++ b/plugins/codex-security/scripts/workbench_cli.py @@ -256,6 +256,7 @@ def add_command(name: str, *required: str) -> argparse.ArgumentParser: set_scan_thread = subparsers.add_parser("set-scan-thread") set_scan_thread.add_argument("--scan-id", required=True) set_scan_thread.add_argument("--thread-id", required=True) + set_scan_thread.add_argument("--time-budget-started-at") set_scan_cost_limit = add_command("set-scan-cost-limit", "--scan-id") set_scan_cost_limit.add_argument("--max-cost-usd", required=True, type=float) diff --git a/plugins/codex-security/scripts/workbench_db.py b/plugins/codex-security/scripts/workbench_db.py index ee4ff62596..195c982214 100644 --- a/plugins/codex-security/scripts/workbench_db.py +++ b/plugins/codex-security/scripts/workbench_db.py @@ -84,6 +84,7 @@ from workbench_finding_workflows import finding_workflow, register_workflow_scan from workbench_scan_start import ( archive_scan, + bind_scan_thread, insert_running_scan, loaded_plugin_version, prepare_cli_scan_target, @@ -1793,6 +1794,11 @@ def acknowledge_output_change() -> None: "contract": scan_contract(scan), "lifecycle": json.loads(scan["lifecycle_json"]), **({"phase": scan["phase"]} if mode == "standard" else {}), + **( + {"canSaveTimeBudgetStart": True} + if "maxTimeMinutes" in recipe and mode in {"standard", "diff"} + else {} + ), "scanDir": str(scan_dir), "scanId": scan_id, "scopeFileCount": scope_file_count, @@ -1803,11 +1809,7 @@ def acknowledge_output_change() -> None: def set_scan_thread(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: scan = require_scan(connection, args.scan_id) - with connection: - connection.execute( - "UPDATE scans SET continuation_thread_id = ?, updated_at = ? WHERE id = ?", - (args.thread_id, now(), scan["id"]), - ) + bind_scan_thread(connection, scan["id"], args.thread_id, args.time_budget_started_at, now()) return {"scanId": scan["id"], "threadId": args.thread_id} diff --git a/plugins/codex-security/scripts/workbench_scan_start.py b/plugins/codex-security/scripts/workbench_scan_start.py index 2742507593..f2df758d4e 100644 --- a/plugins/codex-security/scripts/workbench_scan_start.py +++ b/plugins/codex-security/scripts/workbench_scan_start.py @@ -501,5 +501,31 @@ def insert_running_scan( return scan_id +def bind_scan_thread( + connection: sqlite3.Connection, + scan_id: str, + thread_id: str, + time_budget_started_at: str | None, + timestamp: str, +) -> None: + """Bind the owning session and its advisory clock in the same transaction.""" + with connection: + connection.execute( + """ + UPDATE scans SET continuation_thread_id = ?, updated_at = ?, + recipe_json = CASE WHEN ? IS NULL THEN recipe_json + ELSE json_insert(recipe_json, '$.timeBudgetStartedAt', ?) END + WHERE id = ? + """, + ( + thread_id, + timestamp, + time_budget_started_at, + time_budget_started_at, + scan_id, + ), + ) + + if __name__ == "__main__": argparse.ArgumentParser(description=__doc__).parse_args() diff --git a/plugins/codex-security/scripts/workbench_scan_usage.py b/plugins/codex-security/scripts/workbench_scan_usage.py index 0972ca6912..ad592b74ea 100644 --- a/plugins/codex-security/scripts/workbench_scan_usage.py +++ b/plugins/codex-security/scripts/workbench_scan_usage.py @@ -336,6 +336,11 @@ def collect_scan_usage( started_at=started_at, completed_at=stopped_at, completed_turn_id=completed_turn_id if session.thread_id == thread_id else None, + resume_owner=( + scan["mode"] == "standard" + and scan["recipe_json"] is not None + and session.thread_id == scan["continuation_thread_id"] + ), is_scan_owner=( scan["mode"] == "deep" and session.thread_id == scan["deep_scan_owner_thread_id"] @@ -926,6 +931,7 @@ def _read_rollout_usage( completed_at: datetime | None, completed_turn_id: str | None = None, is_scan_owner: bool = False, + resume_owner: bool = False, ) -> RolloutUsage: result = RolloutUsage(_empty_token_usage(), set(), None, {}) total = result.tokens @@ -947,6 +953,7 @@ def _read_rollout_usage( started_turns: set[str] = set() open_turns: set[str] = set() closed_turns: set[str] = set() + failed_turns: set[str] = set() measured_turns: set[str] = set() def past_scan_end(timestamp: datetime, turn_id: object) -> bool: @@ -1077,6 +1084,8 @@ def past_scan_end(timestamp: datetime, turn_id: object) -> bool: if timestamp >= started_at: if terminal_turn_id in started_turns: closed_turns.add(terminal_turn_id) + if payload.get("type") == "task_complete" and payload.get("error"): + failed_turns.add(terminal_turn_id) if payload.get("type") == "turn_aborted": # Later successful work does not recover interrupted request usage. warnings.add("turn_interrupted") @@ -1143,6 +1152,23 @@ def past_scan_end(timestamp: datetime, turn_id: object) -> bool: warnings.add("token_record_incomplete") result.tokens_updated_at = max(result.tokens_updated_at or timestamp, timestamp) _add_token_usage(response_usage, usage) + if ( + resume_owner + and receipt_turn_id == current_turn_id + and current_turn_id in started_turns + ): + earlier_open = open_turns - {current_turn_id} + if earlier_open or failed_turns: + # A new owner turn supersedes the crashed turn. Reconcile its + # receipts (and any failed reconnects) against native thread + # totals before retiring their missing end or usage markers. + cumulative = _usage_counts(payload.get("thread_token_usage")) + if cumulative is not None and all( + cumulative[key] == total[key] + response_usage[key] + for key in TOKEN_FIELDS.values() + ): + open_turns.difference_update(earlier_open) + closed_turns.difference_update(failed_turns - {current_turn_id}) continue if event.get("type") != "event_msg" or payload.get("type") != "token_count": continue diff --git a/plugins/codex-security/tests/test_workbench_scan_usage_resume.py b/plugins/codex-security/tests/test_workbench_scan_usage_resume.py new file mode 100644 index 0000000000..01b2daf35d --- /dev/null +++ b/plugins/codex-security/tests/test_workbench_scan_usage_resume.py @@ -0,0 +1,52 @@ +"""Account for interrupted and resumed turns of a Standard scan.""" + +from datetime import datetime +from pathlib import Path + +import pytest +from test_workbench_scan_usage import ( + _event, + _measure_rollout, + _response_event, + _rollout, + _token_event, +) +from workbench_test_support import usage_counts as _counts + + +@pytest.mark.parametrize("cumulative_input", [30, 31, None]) +def test_resumed_owner_reconciles_crash_and_failed_reconnect_only_with_matching_thread_usage( + workbench_api, tmp_path: Path, cumulative_input: int | None +) -> None: + at = datetime.fromisoformat("2026-01-01T00:00:00+00:00") + thread_id = "scan-parent" + resumed = _response_event(at, thread_id, "second", 10, 2, turn_id="resumed") + if cumulative_input is None: + del resumed["payload"]["thread_token_usage"] + else: + resumed["payload"]["thread_token_usage"] = _token_event(at, cumulative_input, 5)["payload"][ + "info" + ]["total_token_usage"] + events = [ + _event(at, "event_msg", {"type": "task_started", "turn_id": "crashed"}), + _response_event(at, thread_id, "first", 20, 3, turn_id="crashed"), + _event(at, "event_msg", {"type": "task_started", "turn_id": "reconnect"}), + _event( + at, + "event_msg", + { + "type": "task_complete", + "turn_id": "reconnect", + "error": {"message": "Connection failed"}, + }, + ), + _event(at, "event_msg", {"type": "task_started", "turn_id": "resumed"}), + resumed, + _event(at, "event_msg", {"type": "task_complete", "turn_id": "resumed"}), + ] + rollout = _rollout(tmp_path, thread_id, events) + usage = _measure_rollout(workbench_api["scan_usage"], rollout, at, at, resume_owner=True) + assert usage.tokens == _counts(30, 0, 5) + assert usage.warnings == ( + set() if cumulative_input == 30 else {"token_record_missing", "turn_in_progress"} + ) diff --git a/sdk/typescript/src/api.ts b/sdk/typescript/src/api.ts index c7fb1d5482..ab97fc4751 100644 --- a/sdk/typescript/src/api.ts +++ b/sdk/typescript/src/api.ts @@ -175,6 +175,7 @@ import { } from "./custom-validation-prompt.js"; import { AuthenticationRequiredError, + CodexConnectionError, CodexSecurityError, ConfigurationError, InvalidTargetError, @@ -2964,6 +2965,7 @@ export class CodexSecurity { // The registered recipe now owns the private snapshot, including after completion. savedProviderProfile = undefined; const registeredScanId = registration["scanId"]; + let timeBudgetStartedAt: string | undefined; const resumeThreadId = options.resumeScanId === undefined ? undefined @@ -2983,6 +2985,10 @@ export class CodexSecurity { "The workbench returned mismatched scan resume context.", ); } + const originalStart = + savedRecipe["timeBudgetStartedAt"] ?? registration["startedAt"]; + if (typeof originalStart === "string") + timeBudgetStartedAt = originalStart; const savedSession = await findScanSession( runtime.codexHome, resumeThreadId, @@ -3652,16 +3658,14 @@ export class CodexSecurity { runtime.plugin.pluginRoot, ); checkOpen(); - prompt += `\n\n${renderScanTimeBudgetPolicy( - timePolicy, - scanTimeBudget( - options.maxTimeMinutes, - options.resumeScanId !== undefined && - typeof registration["startedAt"] === "string" - ? Date.parse(registration["startedAt"]) - : (this.#dependencies.now ?? Date.now)(), - ), - )}`; + const timing = scanTimeBudget( + options.maxTimeMinutes, + timeBudgetStartedAt === undefined + ? (this.#dependencies.now ?? Date.now)() + : Date.parse(timeBudgetStartedAt), + ); + timeBudgetStartedAt = timing.startedAt; + prompt += `\n\n${renderScanTimeBudgetPolicy(timePolicy, timing)}`; } tracker.beginParentTurn(resumeThreadId !== undefined); scanTurnStatus = "in_progress"; @@ -3729,7 +3733,13 @@ export class CodexSecurity { try { await workbench( workbenchOptions, - workbenchArgs("set-scan-thread", { scanId, threadId }), + workbenchArgs("set-scan-thread", { + scanId, + threadId, + ...(registration["canSaveTimeBudgetStart"] === true && { + timeBudgetStartedAt, + }), + }), ); } catch (error) { notifyObserver( @@ -6253,8 +6263,6 @@ export async function runScanEvents( } } -class CodexConnectionError extends CodexSecurityError {} - function codexTurnError(error: unknown): CodexSecurityError { const classification = classifyConnectionFailure(error); return classification === "network_error" || classification === "timeout" diff --git a/sdk/typescript/src/errors.ts b/sdk/typescript/src/errors.ts index cb9627739e..8f2b1475b0 100644 --- a/sdk/typescript/src/errors.ts +++ b/sdk/typescript/src/errors.ts @@ -43,6 +43,7 @@ export class DeduplicationReviewError extends CodexSecurityError { } } +export class CodexConnectionError extends CodexSecurityError {} export class ConfigurationError extends CodexSecurityError {} export class AuthenticationRequiredError extends CodexSecurityError {} export class PluginBootstrapError extends CodexSecurityError {} diff --git a/sdk/typescript/src/scan-executor.ts b/sdk/typescript/src/scan-executor.ts index 84a2140118..6fc5a73a48 100644 --- a/sdk/typescript/src/scan-executor.ts +++ b/sdk/typescript/src/scan-executor.ts @@ -1,6 +1,7 @@ import { z } from "zod"; import type { severityRubricSnapshotSchema } from "./classify-severity-records.js"; import { + CodexConnectionError, CodexSecurityError, ScanCostLimitExceededError, ScanInterruptedError, @@ -144,7 +145,9 @@ export function hostedScanFailureStatus( if (status === "acceptance_unknown") return status; if (signal?.aborted) return explicitlyCanceled(signal) ? "canceled" : "failed"; - return status === "canceled" || error instanceof ScanInterruptedError + return status === "canceled" || + (error instanceof ScanInterruptedError && + !(error.cause instanceof CodexConnectionError)) ? "canceled" : "failed"; } diff --git a/sdk/typescript/tests-ts/scan-resume.test.ts b/sdk/typescript/tests-ts/scan-resume.test.ts index 785a095cc0..891df9f422 100644 --- a/sdk/typescript/tests-ts/scan-resume.test.ts +++ b/sdk/typescript/tests-ts/scan-resume.test.ts @@ -886,46 +886,62 @@ test("a new default Standard scan remains resumable after a network failure", as const f = await interruptedScan("standard"); const scanDir = join(f.root, "new-scan"); let scanId = ""; - const deps = resumeDependencies(f, ({ env }) => ({ - startThread: () => ({ - id: f.threadId, - async runStreamed() { - scanId = env!["CODEX_SECURITY_SCAN_ID"]!; - await writeFile( - f.sessionPath, - JSON.stringify({ - type: "session_meta", - payload: { id: f.threadId, cwd: scanDir }, - }) + "\n", - ); - async function* disconnected() { - yield { type: "thread.started" as const, thread_id: f.threadId }; + let initialPrompt = ""; + const outcomes: string[] = []; + const deps = resumeDependencies( + f, + ({ env }) => ({ + startThread: () => ({ + id: f.threadId, + async runStreamed(prompt) { + initialPrompt = prompt; + scanId = env!["CODEX_SECURITY_SCAN_ID"]!; await writeFile( - join(scanDir, "checkpoint.json"), - '{"validated":"saved"}\n', + f.sessionPath, + JSON.stringify({ + type: "session_meta", + payload: { id: f.threadId, cwd: scanDir }, + }) + "\n", ); - yield { - type: "error" as const, - message: - "Reconnecting... waiting for network (Connection failed: error sending request)", - }; - } - return { events: disconnected() }; - }, + async function* disconnected() { + yield { type: "thread.started" as const, thread_id: f.threadId }; + await writeFile( + join(scanDir, "checkpoint.json"), + '{"validated":"saved"}\n', + ); + yield { + type: "error" as const, + message: + "Reconnecting... waiting for network (Connection failed: error sending request)", + }; + } + return { events: disconnected() }; + }, + }), }), - })); + {}, + (options) => new OperationTelemetry(options), + ); await using client = deps.createSecurity({}); await expect( client.run(f.repository, { + maxTimeMinutes: 5, + onTelemetry: (event) => { + if (event.outcome) outcomes.push(event.outcome); + }, outputDir: scanDir, target: ["source.py"], ownerSuggestions: false, }), ).rejects.toThrow("codex-security scans resume"); + expect(outcomes).toEqual(["failed"]); expect(scanId).not.toBe(f.scanId); - expect( - await f.command(["get-cli-scan-resume", "--scan-id", scanId]), - ).toMatchObject({ + const saved = await f.command(["get-cli-scan-resume", "--scan-id", scanId]); + const start = (saved["recipe"] as { timeBudgetStartedAt: string }) + .timeBudgetStartedAt; + expect(typeof start).toBe("string"); + expect(initialPrompt).toContain(start); + expect(saved).toMatchObject({ scanId, scanDir, threadId: f.threadId, From d487d0d4652a84a53793d45ca74843f6c5f8dcde Mon Sep 17 00:00:00 2001 From: Ian Webster Date: Wed, 11 Nov 2026 12:21:02 -0800 Subject: [PATCH 4/5] refactor(scans): simplify Standard scan recovery --- .../codex-security/scripts/workbench_db.py | 17 +- .../scripts/workbench_scan_start.py | 26 --- .../scripts/workbench_scan_usage.py | 22 ++- sdk/typescript/docs/cli.md | 18 +- sdk/typescript/scripts/package-dist-files.mjs | 1 - sdk/typescript/src/api.ts | 32 ++-- sdk/typescript/src/runtime.ts | 38 ++++- sdk/typescript/src/scan-execution-lock.ts | 39 ----- sdk/typescript/tests-ts/api.test.ts | 21 +-- sdk/typescript/tests-ts/runtime.test.ts | 3 +- .../tests-ts/scan-execution-lock.test.ts | 13 +- sdk/typescript/tests-ts/scan-resume.test.ts | 155 ++++++++---------- sdk/typescript/tests-ts/support/api-events.ts | 9 +- 13 files changed, 168 insertions(+), 226 deletions(-) delete mode 100644 sdk/typescript/src/scan-execution-lock.ts diff --git a/plugins/codex-security/scripts/workbench_db.py b/plugins/codex-security/scripts/workbench_db.py index 195c982214..2d241c33c8 100644 --- a/plugins/codex-security/scripts/workbench_db.py +++ b/plugins/codex-security/scripts/workbench_db.py @@ -84,7 +84,6 @@ from workbench_finding_workflows import finding_workflow, register_workflow_scan from workbench_scan_start import ( archive_scan, - bind_scan_thread, insert_running_scan, loaded_plugin_version, prepare_cli_scan_target, @@ -1581,12 +1580,7 @@ def register_cli_scan( *, before_archive: Callable[[], None] | None = None, ) -> dict[str, Any]: - """Register a CLI scan and its selected target before execution begins. - - Record the launch recipe and target identity with the new scan. When reusing - an output directory, coordinate archiving with registration so previous results - can still be located. Return the scan ID, contract, and target metadata. - """ + """Register the scan recipe and target identity, archiving any previous output atomically.""" # A selected plugin may replace __main__ while delegating to this workbench. before_output_change = before_archive or getattr( sys, "_codex_security_before_output_change", None @@ -1809,7 +1803,14 @@ def acknowledge_output_change() -> None: def set_scan_thread(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: scan = require_scan(connection, args.scan_id) - bind_scan_thread(connection, scan["id"], args.thread_id, args.time_budget_started_at, now()) + with connection: + connection.execute( + """UPDATE scans SET continuation_thread_id = ?1, updated_at = ?2, + recipe_json = CASE WHEN ?3 IS NULL THEN recipe_json + ELSE json_insert(recipe_json, '$.timeBudgetStartedAt', ?3) END + WHERE id = ?4""", + (args.thread_id, now(), getattr(args, "time_budget_started_at", None), scan["id"]), + ) return {"scanId": scan["id"], "threadId": args.thread_id} diff --git a/plugins/codex-security/scripts/workbench_scan_start.py b/plugins/codex-security/scripts/workbench_scan_start.py index f2df758d4e..2742507593 100644 --- a/plugins/codex-security/scripts/workbench_scan_start.py +++ b/plugins/codex-security/scripts/workbench_scan_start.py @@ -501,31 +501,5 @@ def insert_running_scan( return scan_id -def bind_scan_thread( - connection: sqlite3.Connection, - scan_id: str, - thread_id: str, - time_budget_started_at: str | None, - timestamp: str, -) -> None: - """Bind the owning session and its advisory clock in the same transaction.""" - with connection: - connection.execute( - """ - UPDATE scans SET continuation_thread_id = ?, updated_at = ?, - recipe_json = CASE WHEN ? IS NULL THEN recipe_json - ELSE json_insert(recipe_json, '$.timeBudgetStartedAt', ?) END - WHERE id = ? - """, - ( - thread_id, - timestamp, - time_budget_started_at, - time_budget_started_at, - scan_id, - ), - ) - - if __name__ == "__main__": argparse.ArgumentParser(description=__doc__).parse_args() diff --git a/plugins/codex-security/scripts/workbench_scan_usage.py b/plugins/codex-security/scripts/workbench_scan_usage.py index ad592b74ea..ae533a663c 100644 --- a/plugins/codex-security/scripts/workbench_scan_usage.py +++ b/plugins/codex-security/scripts/workbench_scan_usage.py @@ -338,6 +338,7 @@ def collect_scan_usage( completed_turn_id=completed_turn_id if session.thread_id == thread_id else None, resume_owner=( scan["mode"] == "standard" + and "recipe_json" in scan.keys() and scan["recipe_json"] is not None and session.thread_id == scan["continuation_thread_id"] ), @@ -1156,19 +1157,16 @@ def past_scan_end(timestamp: datetime, turn_id: object) -> bool: resume_owner and receipt_turn_id == current_turn_id and current_turn_id in started_turns + and (open_turns - {current_turn_id} or failed_turns) ): - earlier_open = open_turns - {current_turn_id} - if earlier_open or failed_turns: - # A new owner turn supersedes the crashed turn. Reconcile its - # receipts (and any failed reconnects) against native thread - # totals before retiring their missing end or usage markers. - cumulative = _usage_counts(payload.get("thread_token_usage")) - if cumulative is not None and all( - cumulative[key] == total[key] + response_usage[key] - for key in TOKEN_FIELDS.values() - ): - open_turns.difference_update(earlier_open) - closed_turns.difference_update(failed_turns - {current_turn_id}) + # Matching thread totals account for turns left open or failed before resume. + cumulative = _usage_counts(payload.get("thread_token_usage")) + if cumulative is not None and all( + cumulative[key] == total[key] + response_usage[key] + for key in TOKEN_FIELDS.values() + ): + open_turns.intersection_update({current_turn_id}) + closed_turns.difference_update(failed_turns - {current_turn_id}) continue if event.get("type") != "event_msg" or payload.get("type") != "token_count": continue diff --git a/sdk/typescript/docs/cli.md b/sdk/typescript/docs/cli.md index 8586a5fd6f..e2be4b0150 100644 --- a/sdk/typescript/docs/cli.md +++ b/sdk/typescript/docs/cli.md @@ -2258,17 +2258,13 @@ versions when its saved checkout identity can be verified. If the workbench cannot verify it, resume with the Python runtime that started the scan or use `codex-security scans rerun SCAN_ID` to start a new scan. -Resume retains scan ID, completed workers, artifacts, accumulated cost, and saved -settings/instructions. Standard scans continue the original Codex session from -saved analysis, findings, and review checkpoints; they do not start a new scan. -Only one CLI or SDK call can run a Standard scan at a time. A network failure -after the session is saved leaves the scan resumable and prints the recovery -command. The original `--max-cost` limit and `--max-time-minutes` deadline still -apply. Work not saved before the interruption may need to be repeated. -Explicit cancellation, authentication failures, and other terminal failures -still require a rerun. This does not reopen scans marked failed by older versions. -Custom-validation scans still require their original validation instructions and -are not supported by `scans resume`. +Resume keeps the scan ID, saved settings and instructions, completed workers, artifacts, +and accumulated cost. A Standard scan continues in its original Codex session with +the same `--max-cost` limit and `--max-time` deadline. Only one CLI or SDK call +can run it at a time. If the connection fails after the session is saved, the CLI +prints the resume command. Work since the last checkpoint may need to be repeated. +Use `scans rerun` for canceled or failed scans, including authentication failures. +For custom validation, rerun with the original validation instructions. Knowledge-bearing scans save their extracted documents in the private `.scan-knowledge.json` continuation file beside the scan artifacts. diff --git a/sdk/typescript/scripts/package-dist-files.mjs b/sdk/typescript/scripts/package-dist-files.mjs index efcf30fb73..9890db929d 100644 --- a/sdk/typescript/scripts/package-dist-files.mjs +++ b/sdk/typescript/scripts/package-dist-files.mjs @@ -2,7 +2,6 @@ const distModules = [ "api", "batch-cost", "scan-executor", - "scan-execution-lock", "host-protocol", "hosted-remediation", "hosted-remediation-protocol", diff --git a/sdk/typescript/src/api.ts b/sdk/typescript/src/api.ts index ab97fc4751..2374f85753 100644 --- a/sdk/typescript/src/api.ts +++ b/sdk/typescript/src/api.ts @@ -117,7 +117,6 @@ import type { ScanPricingContext, ScanTokenUsage } from "./cost-model.js"; import { measuredScanCost, combineMeasurements } from "./scan-measurement.js"; import { ScanProgressTracker, type DeepScanProgress } from "./scan-progress.js"; import { findScanSession } from "./scan-logs.js"; -import { acquireScanExecutionLock } from "./scan-execution-lock.js"; import { DIRECT_DEEP_SCAN_CONTROL_USAGE, deepScanParentToolOverrides, @@ -256,6 +255,7 @@ import { bundledCodexSdkEnvironment } from "./codex-sdk-environment.js"; import { canonicalConfigPath, acquireCodexSecurityCredentialHomeLock, + acquireScanExecutionLock, withCredentialHomeLock, writeCredentialHomeConfig, bootstrapPlugin, @@ -2238,8 +2238,7 @@ export class CodexSecurity { let latestMeasurement: Readonly | null = null; let notifiedLimit: number | undefined; let scanDir = ""; - let releaseScanExecution: (() => void) | undefined; - let standardScan = false; + let releaseStandardScan: (() => void) | undefined; let targetPathsFile: string | null = null; let stagedPluginRoot: string | undefined; let stagedPluginDirectory: string | undefined; @@ -2418,10 +2417,6 @@ export class CodexSecurity { ), }; const hosted = this.#dependencies.hostedScan !== undefined; - standardScan = - mode === "standard" && - !hosted && - (normalized.kind === "repository" || normalized.kind === "paths"); for (const root of [ (await gitMarkerRoot(repo, signal, "outermost")) ?? repo, ...(knowledgeBase?.protectedRoots ?? []), @@ -2470,8 +2465,12 @@ export class CodexSecurity { ); requireOutputOutsideRepository(protectedRoot, scanDir); requireModelSafeOutputDir(scanDir); - if (standardScan) { - releaseScanExecution = await acquireScanExecutionLock( + if ( + !hosted && + mode === "standard" && + (normalized.kind === "repository" || normalized.kind === "paths") + ) { + releaseStandardScan = await acquireScanExecutionLock( stateDirectory, scanDir, ); @@ -3481,7 +3480,7 @@ export class CodexSecurity { prompt += mode === "deep" ? "\nResume the existing Deep Scan through its coordinator. Preserve completed workers and saved artifacts; do not recreate the scan directory or restart completed analysis. If the coordinator already finished, continue with completion of this same scan." - : "\nResume this same Standard scan in its original session and output directory. Read the saved scan-manifest.json, findings.json, and scan.reviewState checkpoints when present. Preserve completed analysis, validation decisions, evidence, and stable finding/surface IDs. Continue unfinished validation and deferred work; do not restart completed analysis or recreate the scan directory. If the review is finished, complete this same scan using its saved results."; + : "\nContinue this Standard scan from the saved scan-manifest.json, findings.json, and review checkpoints in its output directory. Keep existing evidence, validation decisions, finding IDs, and surface IDs. Finish outstanding work and finalize the same scan."; } if ( falsePositiveExamples.length > 0 && @@ -3668,7 +3667,6 @@ export class CodexSecurity { prompt += `\n\n${renderScanTimeBudgetPolicy(timePolicy, timing)}`; } tracker.beginParentTurn(resumeThreadId !== undefined); - scanTurnStatus = "in_progress"; const events = directDeepScan ? (this.#dependencies.runDeepScan ?? runDeepScan)({ onDiagnostic: this.#onDeepScanDiagnostic, @@ -3685,6 +3683,7 @@ export class CodexSecurity { signal, }) : (await thread.runStreamed(prompt, turnOptions)).events; + scanTurnStatus = "in_progress"; checkOpen(); let result = await runScanEvents({ @@ -4286,14 +4285,13 @@ export class CodexSecurity { } if ( activeScan !== null && - standardScan && + releaseStandardScan !== undefined && !signal.aborted && options.validationPrompt === undefined && scanTurnStatus !== null && terminalTurnStatus !== "completed" && failure instanceof CodexConnectionError ) { - // Keep the existing running scan only when its saved session and inputs can resume. const saved = await workbench( { ...activeScan.options, signal: undefined }, workbenchArgs("get-cli-scan-resume", { @@ -4311,11 +4309,11 @@ export class CodexSecurity { runPostScan = null; } } - // Deep Scan orchestration persists its own terminal failures and cancellations. - // Standard scans have no coordinator, so fatal errors and cancellation still stop them. + // A resumed Deep Scan persists its own terminal status. if ( activeScan !== null && - (options.resumeScanId === undefined || standardScan) + (options.resumeScanId === undefined || + releaseStandardScan !== undefined) ) { if ( options.validationPrompt !== undefined && @@ -4398,7 +4396,7 @@ export class CodexSecurity { warnCleanupFailed(options, error); } observeSettledScan(options, activeScan); - releaseScanExecution?.(); + releaseStandardScan?.(); } } diff --git a/sdk/typescript/src/runtime.ts b/sdk/typescript/src/runtime.ts index 93d17c0e07..a0c17cd03e 100644 --- a/sdk/typescript/src/runtime.ts +++ b/sdk/typescript/src/runtime.ts @@ -1346,6 +1346,39 @@ export function sqliteDatabaseOpener( }; } +function sqliteBusy(error: unknown): boolean { + return ( + isRecord(error) && + (error["errcode"] === 5 || error["code"] === "SQLITE_BUSY") + ); +} + +export async function acquireScanExecutionLock( + stateDirectory: string, + scanDirectory: string, +): Promise<() => void> { + const directory = join(stateDirectory, "scan-locks"); + await mkdir(directory, { recursive: true, mode: 0o700 }); + await requirePrivateDirectory(await stat(directory), directory, "scan locks"); + const identity = createHash("sha256").update(scanDirectory).digest("hex"); + const path = join(directory, `${identity}.sqlite3`); + const database = sqliteDatabaseOpener()(path); + try { + if (process.platform !== "win32") await chmod(path, 0o600); + database.exec("PRAGMA busy_timeout = 0"); + database.exec("BEGIN EXCLUSIVE"); + } catch (error) { + database.close(); + if (sqliteBusy(error)) { + throw new CodexSecurityError( + "This scan is already running in another CLI or SDK call.", + ); + } + throw error; + } + return () => database.close(); +} + export async function withCredentialHomeLock( codexHome: string, operation: () => Promise, @@ -1423,10 +1456,7 @@ export async function acquireCodexSecurityCredentialHomeLock( try { database.exec("BEGIN EXCLUSIVE"); } catch (error) { - if ( - !isRecord(error) || - (error["errcode"] !== 5 && error["code"] !== "SQLITE_BUSY") - ) { + if (!sqliteBusy(error)) { throw error; } await delay(CREDENTIAL_LOCK_POLL_MILLISECONDS, undefined, { signal }); diff --git a/sdk/typescript/src/scan-execution-lock.ts b/sdk/typescript/src/scan-execution-lock.ts deleted file mode 100644 index 1e20925d96..0000000000 --- a/sdk/typescript/src/scan-execution-lock.ts +++ /dev/null @@ -1,39 +0,0 @@ -import { createHash } from "node:crypto"; -import { chmod, mkdir, realpath, stat } from "node:fs/promises"; -import { join } from "node:path"; -import { CodexSecurityError } from "./errors.js"; -import { isRecord } from "./record.js"; -import { requirePrivateDirectory, sqliteDatabaseOpener } from "./runtime.js"; - -/** Hold one Standard scan's execution across processes, including resume. */ -export async function acquireScanExecutionLock( - stateDirectory: string, - scanDirectory: string, -): Promise<() => void> { - const directory = join(stateDirectory, "scan-locks"); - await mkdir(directory, { recursive: true, mode: 0o700 }); - await requirePrivateDirectory(await stat(directory), directory, "scan locks"); - const identity = createHash("sha256") - .update(await realpath(scanDirectory)) - .digest("hex"); - const path = join(directory, `${identity}.sqlite3`); - const database = sqliteDatabaseOpener()(path); - try { - if (process.platform !== "win32") await chmod(path, 0o600); - database.exec("PRAGMA busy_timeout = 0"); - database.exec("BEGIN EXCLUSIVE"); - } catch (error) { - database.close(); - if ( - isRecord(error) && - (error["errcode"] === 5 || error["code"] === "SQLITE_BUSY") - ) { - throw new CodexSecurityError( - "This scan is already running in another CLI or SDK call.", - ); - } - throw error; - } - // The OS also releases the lock if the scan process exits unexpectedly. - return () => database.close(); -} diff --git a/sdk/typescript/tests-ts/api.test.ts b/sdk/typescript/tests-ts/api.test.ts index 13f1971e68..76840ace69 100644 --- a/sdk/typescript/tests-ts/api.test.ts +++ b/sdk/typescript/tests-ts/api.test.ts @@ -18329,13 +18329,7 @@ if __name__ == "__main__": }, })), resolveCodexCommand: () => ({ command: executable }), - resolvePluginPython: async () => { - Object.defineProperty(process, "platform", { - value: "win32", - configurable: true, - }); - return "/managed/python"; - }, + resolvePluginPython: async () => "/managed/python", prepareOutputDir: async () => scanDir, repositoryRevision: async () => "deadbeef", createCodex: (options) => ({ @@ -18353,9 +18347,16 @@ if __name__ == "__main__": }), }); try { - await expect(client.run(repository)).rejects.toThrow( - "child environment captured", - ); + await expect( + client.run(repository, { + onOutputDirReady() { + Object.defineProperty(process, "platform", { + value: "win32", + configurable: true, + }); + }, + }), + ).rejects.toThrow("child environment captured"); expect(childPath?.split(delimiter)).toEqual([ bundledTools, inheritedTools, diff --git a/sdk/typescript/tests-ts/runtime.test.ts b/sdk/typescript/tests-ts/runtime.test.ts index 08ba802abc..a077f8f2c8 100644 --- a/sdk/typescript/tests-ts/runtime.test.ts +++ b/sdk/typescript/tests-ts/runtime.test.ts @@ -6380,6 +6380,7 @@ describe("runtime directories and plugin Python boundary", () => { ), ), ); + const beforeCompletion = (await readdir(scanDir)).sort(); const missing = requiredDrafts.filter( (filename) => !present.some((candidate) => candidate === filename), ); @@ -6395,7 +6396,7 @@ describe("runtime directories and plugin Python boundary", () => { ", ", )}. Check that the scan agent can run shell commands and write to the scan directory before retrying.`, ); - expect((await readdir(scanDir)).sort()).toEqual([...present].sort()); + expect((await readdir(scanDir)).sort()).toEqual(beforeCompletion); const stored = await runWorkbench(workbenchOptions, [ "get-scan", "--scan-id", diff --git a/sdk/typescript/tests-ts/scan-execution-lock.test.ts b/sdk/typescript/tests-ts/scan-execution-lock.test.ts index 71ada709e5..77e4126573 100644 --- a/sdk/typescript/tests-ts/scan-execution-lock.test.ts +++ b/sdk/typescript/tests-ts/scan-execution-lock.test.ts @@ -1,7 +1,7 @@ import { join } from "node:path"; import { mkdir } from "node:fs/promises"; import { afterEach, expect, test } from "bun:test"; -import { acquireScanExecutionLock } from "../src/scan-execution-lock.js"; +import { acquireScanExecutionLock } from "../src/runtime.js"; import { createApiTestFixtures } from "./support/temporary-directories.js"; const { temporaryDirectory, cleanup } = createApiTestFixtures(); @@ -12,19 +12,20 @@ test("scan execution excludes a second process and recovers after the owner exit const state = join(root, "state"); const scan = join(root, "scan"); await mkdir(scan); - const module = new URL("../src/scan-execution-lock.ts", import.meta.url).href; + const module = new URL("../src/runtime.ts", import.meta.url).href; const child = Bun.spawn( [ process.execPath, "--eval", ` const { acquireScanExecutionLock } = await import(${JSON.stringify(module)}); - await acquireScanExecutionLock(${JSON.stringify(state)}, ${JSON.stringify(scan)}); + const release = await acquireScanExecutionLock(${JSON.stringify(state)}, ${JSON.stringify(scan)}); console.log("locked"); - process.stdin.resume(); + process.on("exit", release); + setInterval(() => {}, 60_000); `, ], - { stdin: "pipe", stdout: "pipe", stderr: "pipe" }, + { stdout: "pipe", stderr: "pipe" }, ); try { const reader = child.stdout.getReader(); @@ -36,7 +37,7 @@ test("scan execution excludes a second process and recovers after the owner exit "already running", ); } finally { - child.kill(); + child.kill("SIGKILL"); await child.exited; } const release = await acquireScanExecutionLock(state, scan); diff --git a/sdk/typescript/tests-ts/scan-resume.test.ts b/sdk/typescript/tests-ts/scan-resume.test.ts index 891df9f422..46b60b4c61 100644 --- a/sdk/typescript/tests-ts/scan-resume.test.ts +++ b/sdk/typescript/tests-ts/scan-resume.test.ts @@ -2,7 +2,7 @@ import { createCliTest } from "./support/cli-run.js"; import { workbenchCommand } from "./support/workbench-command.js"; import { runWorkbenchFixtureCommands } from "./support/workbench-fixtures.js"; import { gitText } from "./support/shell.js"; -import { readJsonLines } from "./support/json.js"; +import { readJson, readJsonLines, writeJson } from "./support/json.js"; import { hash, randomUUID } from "node:crypto"; import { stripVTControlCharacters } from "node:util"; import { @@ -41,7 +41,11 @@ import { useTestPython } from "./support/python.js"; import { runPython } from "./support/python-probe.js"; import { copyCompletedScanFixture, PLUGIN_ROOT } from "./plugin-root.js"; import { TestClient } from "./support/api-client.js"; -import { completedEvents, preparedRuntime } from "./support/api-events.js"; +import { + completedEvents, + failedEvents, + preparedRuntime, +} from "./support/api-events.js"; import { createApiTestFixtures } from "./support/temporary-directories.js"; import { fail } from "./support/errors.js"; @@ -385,10 +389,19 @@ test.skipIf(process.platform !== "win32").each(["deep", "standard"] as const)( }, ); -test.each(["failed", "canceled", "changed", "replaced", "wrong-owner"])( - "resume refuses %s scans without altering their saved state", - async (scenario) => { - const f = await interruptedScan(); +test.each([ + ["standard", "failed"], + ["standard", "canceled"], + ["standard", "changed"], + ["deep", "failed"], + ["deep", "canceled"], + ["deep", "changed"], + ["deep", "replaced"], + ["deep", "wrong-owner"], +] as const)( + "%s resume refuses %s scans without altering their saved state", + async (mode, scenario) => { + const f = await interruptedScan(mode); if (scenario === "failed") await f.command([ "fail-scan", @@ -775,26 +788,48 @@ test("Standard resume preserves checkpoints through another disconnect and compl maxTimeMinutes: 5, ownerSuggestions: false, }); - await copyCompletedScanFixture(f.scanDir); - const manifestPath = join(f.scanDir, "scan-manifest.json"); - const manifest = JSON.parse(await readFile(manifestPath, "utf8")); - manifest.scan.id = f.scanId; - manifest.scan.target = { kind: "directory_snapshot" }; - manifest.scan.complete = false; - delete manifest.scan.sealedAt; - delete manifest.scan.artifacts; - await writeFile(manifestPath, JSON.stringify(manifest)); - for (const name of ["coverage.json", "findings.json"]) { - const path = join(f.scanDir, name); - const document = JSON.parse(await readFile(path, "utf8")); - document.scanId = f.scanId; - await writeFile(path, JSON.stringify(document)); + const draftPath = join(f.scanDir, "drafts", `${f.scanId}.json`); + await mkdir(dirname(draftPath), { recursive: true }); + async function saveDraft(complete: boolean) { + await writeJson(draftPath, { + manifest: { + scan: { complete, scope: { includePaths: ["."], excludePaths: [] } }, + }, + findings: { findings: [] }, + reviewState: { + inventoryStrategy: "directory", + surfaces: [ + { + id: "source", + label: "Source review", + disposition: "no_issue_found", + receiptRefs: [], + }, + ], + deferred: [], + explicitExclusions: [], + }, + }); + await f.command([ + "write-scan-draft", + "--scan-id", + f.scanId, + "--draft-path", + draftPath, + ]); } - const original = await Promise.all( - ["scan-manifest.json", "findings.json", "coverage.json"].map((name) => - readFile(join(f.scanDir, name), "utf8"), - ), + await saveDraft(false); + const head = await readJson<{ checkpoint: string }>( + join(f.scanDir, "checkpoint-head.json"), ); + const checkpoint = join(f.scanDir, "checkpoints", head.checkpoint); + const readDraft = () => + Promise.all( + ["scan-manifest.json", "findings.json", "checkpoint-head.json"].map( + (name) => readFile(join(f.scanDir, name), "utf8"), + ), + ); + const original = await readDraft(); let attempts = 0; const initial = await f.command([ "get-cli-scan-resume", @@ -812,13 +847,7 @@ test("Standard resume preserves checkpoints through another disconnect and compl expect(prompt).toContain( new Date(initial["startedAt"] as string).toISOString(), ); - expect( - await Promise.all( - ["scan-manifest.json", "findings.json", "coverage.json"].map( - (name) => readFile(join(f.scanDir, name), "utf8"), - ), - ), - ).toEqual(original); + expect(await readDraft()).toEqual(original); attempts++; await appendFile( f.sessionPath, @@ -835,18 +864,11 @@ test("Standard resume preserves checkpoints through another disconnect and compl }, }) + "\n", ); - if (attempts === 1) { - async function* disconnected() { - yield { type: "thread.started" as const, thread_id: id }; - yield { - type: "turn.failed" as const, - error: { message: "Connection failed: error sending request" }, - }; - } - return { events: disconnected() }; - } - delete manifest.scan.complete; - await writeFile(manifestPath, JSON.stringify(manifest)); + if (attempts === 1) + return { + events: failedEvents("Connection failed: error sending request"), + }; + await saveDraft(true); return { events: completedEvents(id) }; }, }; @@ -876,7 +898,9 @@ test("Standard resume preserves checkpoints through another disconnect and compl progress: { status: "complete" }, usage: { totalTokens: 12240 }, }); - expect(await readFile(f.checkpoint, "utf8")).toBe('{"completed":"setup"}\n'); + expect(hash("sha256", await readFile(checkpoint)) + ".json").toBe( + head.checkpoint, + ); await expect( f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), ).rejects.toThrow("running scan"); @@ -905,10 +929,6 @@ test("a new default Standard scan remains resumable after a network failure", as ); async function* disconnected() { yield { type: "thread.started" as const, thread_id: f.threadId }; - await writeFile( - join(scanDir, "checkpoint.json"), - '{"validated":"saved"}\n', - ); yield { type: "error" as const, message: @@ -950,39 +970,8 @@ test("a new default Standard scan remains resumable after a network failure", as target: { kind: "paths", paths: ["source.py"] }, }, }); - expect(await readFile(join(scanDir, "checkpoint.json"), "utf8")).toBe( - '{"validated":"saved"}\n', - ); }); -test.each(["failed", "canceled", "changed"])( - "Standard resume refuses %s scans", - async (scenario) => { - const f = await interruptedScan("standard"); - if (scenario === "failed") - await f.command([ - "fail-scan", - "--scan-id", - f.scanId, - "--message", - "Synthetic failure", - ]); - if (scenario === "canceled") - await f.command(["cancel-scan", "--scan-id", f.scanId]); - if (scenario === "changed") - await writeFile(join(f.repository, "source.py"), "# changed\n"); - const before = await f.command(["get-scan", "--scan-id", f.scanId]); - await expect( - f.command(["get-cli-scan-resume", "--scan-id", f.scanId]), - ).rejects.toThrow( - scenario === "changed" ? "revision or contents changed" : "running scan", - ); - expect(await f.command(["get-scan", "--scan-id", f.scanId])).toEqual( - before, - ); - }, -); - test.each(["access", "local"])( "a Standard resume %s failure stops the scan", async (kind) => { @@ -999,13 +988,7 @@ test.each(["access", "local"])( throw new Error( "Invalid network configuration in saved settings", ); - async function* failed() { - yield { - type: "turn.failed" as const, - error: { message: "403 forbidden: access denied" }, - }; - } - return { events: failed() }; + return { events: failedEvents("403 forbidden: access denied") }; }, }), })), diff --git a/sdk/typescript/tests-ts/support/api-events.ts b/sdk/typescript/tests-ts/support/api-events.ts index 6f53e5b398..4a6bce558a 100644 --- a/sdk/typescript/tests-ts/support/api-events.ts +++ b/sdk/typescript/tests-ts/support/api-events.ts @@ -108,11 +108,10 @@ export function runEvents( }); } -export async function* failedEvents(): AsyncGenerator { - yield { - type: "turn.failed", - error: { message: "Could not draft fixes." }, - }; +export async function* failedEvents( + message = "Could not draft fixes.", +): AsyncGenerator { + yield { type: "turn.failed", error: { message } }; } export function completedCodex(root: string, threadId: string | null = null) { From d6c15ce5cd5e9436039746bbf637aef65f4fd548 Mon Sep 17 00:00:00 2001 From: Ian Webster Date: Wed, 11 Nov 2026 12:25:15 -0800 Subject: [PATCH 5/5] test(scans): release paused registration after cancellation --- sdk/typescript/tests-ts/api-file-profile-replay.test.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/sdk/typescript/tests-ts/api-file-profile-replay.test.ts b/sdk/typescript/tests-ts/api-file-profile-replay.test.ts index 168fb6db23..1aa07217a3 100644 --- a/sdk/typescript/tests-ts/api-file-profile-replay.test.ts +++ b/sdk/typescript/tests-ts/api-file-profile-replay.test.ts @@ -149,6 +149,7 @@ test.serial.each([ ` with socket.create_connection(('127.0.0.1', ${address.port})) as connection:`, " connection.sendall(b'ready')", " connection.recv(1)", + " raise SystemExit('Synthetic registration reply lost')", "def controlled(*args, **kwargs):", ...(boundary === "precommit" ? [" pause()"] : []), " result = original(*args, **kwargs)", @@ -168,6 +169,7 @@ test.serial.each([ async ([socket]) => { const [message] = await once(socket, "data", readinessOptions); expect(String(message)).toBe("ready"); + return socket; }, ) : undefined; @@ -216,13 +218,15 @@ test.serial.each([ .run(repository, { signal: controller.signal }) .catch((error: unknown) => error); if (ready !== undefined) { - await Promise.race([ + const socket = await Promise.race([ ready, operation.then((error) => { throw error; }), ]); controller.abort(); + // Windows lets registration finish once it starts changing the output. + socket.end(); } const failure = await operation; expect(failure).toBeInstanceOf(Error);