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_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..2d241c33c8 100644 --- a/plugins/codex-security/scripts/workbench_db.py +++ b/plugins/codex-security/scripts/workbench_db.py @@ -1580,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 @@ -1793,6 +1788,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, @@ -1805,8 +1805,11 @@ def set_scan_thread(connection: sqlite3.Connection, args: argparse.Namespace) -> 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"]), + """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_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/plugins/codex-security/scripts/workbench_scan_usage.py b/plugins/codex-security/scripts/workbench_scan_usage.py index 0972ca6912..ae533a663c 100644 --- a/plugins/codex-security/scripts/workbench_scan_usage.py +++ b/plugins/codex-security/scripts/workbench_scan_usage.py @@ -336,6 +336,12 @@ 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 "recipe_json" in scan.keys() + 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 +932,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 +954,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 +1085,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 +1153,20 @@ 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 + and (open_turns - {current_turn_id} or failed_turns) + ): + # 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/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) 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/docs/cli.md b/sdk/typescript/docs/cli.md index 570ec08d9b..e2be4b0150 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,15 +2251,22 @@ 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 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 +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. 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/src/api.ts b/sdk/typescript/src/api.ts index d40069edea..2374f85753 100644 --- a/sdk/typescript/src/api.ts +++ b/sdk/typescript/src/api.ts @@ -174,6 +174,7 @@ import { } from "./custom-validation-prompt.js"; import { AuthenticationRequiredError, + CodexConnectionError, CodexSecurityError, ConfigurationError, InvalidTargetError, @@ -254,6 +255,7 @@ import { bundledCodexSdkEnvironment } from "./codex-sdk-environment.js"; import { canonicalConfigPath, acquireCodexSecurityCredentialHomeLock, + acquireScanExecutionLock, withCredentialHomeLock, writeCredentialHomeConfig, bootstrapPlugin, @@ -2236,6 +2238,7 @@ export class CodexSecurity { let latestMeasurement: Readonly | null = null; let notifiedLimit: number | undefined; let scanDir = ""; + let releaseStandardScan: (() => void) | undefined; let targetPathsFile: string | null = null; let stagedPluginRoot: string | undefined; let stagedPluginDirectory: string | undefined; @@ -2462,6 +2465,16 @@ export class CodexSecurity { ); requireOutputOutsideRepository(protectedRoot, scanDir); requireModelSafeOutputDir(scanDir); + if ( + !hosted && + mode === "standard" && + (normalized.kind === "repository" || normalized.kind === "paths") + ) { + releaseStandardScan = await acquireScanExecutionLock( + stateDirectory, + scanDir, + ); + } if (!options.archiveExisting && options.resumeScanId === undefined) { notifyObserver(options, "onOutputDirReady")(scanDir); } @@ -2951,6 +2964,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 @@ -2970,6 +2984,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, @@ -3460,7 +3478,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." + : "\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 && @@ -3637,13 +3657,14 @@ export class CodexSecurity { runtime.plugin.pluginRoot, ); checkOpen(); - prompt += `\n\n${renderScanTimeBudgetPolicy( - timePolicy, - scanTimeBudget( - options.maxTimeMinutes, - (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); const events = directDeepScan @@ -3711,7 +3732,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( @@ -4256,9 +4283,38 @@ export class CodexSecurity { ); } } - // A failed attachment must not turn a resumable coordinator into a terminal failure. - // Deep Scan orchestration persists its own terminal failures and cancellations. - if (activeScan !== null && options.resumeScanId === undefined) { + if ( + activeScan !== null && + releaseStandardScan !== undefined && + !signal.aborted && + options.validationPrompt === undefined && + scanTurnStatus !== null && + terminalTurnStatus !== "completed" && + failure instanceof CodexConnectionError + ) { + 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; + } + } + // A resumed Deep Scan persists its own terminal status. + if ( + activeScan !== null && + (options.resumeScanId === undefined || + releaseStandardScan !== undefined) + ) { if ( options.validationPrompt !== undefined && customValidation.status !== "complete" @@ -4340,6 +4396,7 @@ export class CodexSecurity { warnCleanupFailed(options, error); } observeSettledScan(options, activeScan); + releaseStandardScan?.(); } } @@ -5639,7 +5696,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 +5704,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 +6150,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 +6261,13 @@ export async function runScanEvents( } } +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 +6285,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"]; @@ -6235,8 +6313,7 @@ async function readCodexTurn(options: { } const reconnect = reconnectAttempt(message); if (reconnect !== null) options.onReconnect?.(message, reconnect); - else if (!waitingForNetwork(message)) - throw new CodexSecurityError(message); + else if (!waitingForNetwork(message)) throw codexTurnError(message); } }, }); diff --git a/sdk/typescript/src/cli.ts b/sdk/typescript/src/cli.ts index 7535ffd731..65094c2faf 100644 --- a/sdk/typescript/src/cli.ts +++ b/sdk/typescript/src/cli.ts @@ -3109,11 +3109,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/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/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-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/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/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); 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 new file mode 100644 index 0000000000..77e4126573 --- /dev/null +++ b/sdk/typescript/tests-ts/scan-execution-lock.test.ts @@ -0,0 +1,52 @@ +import { join } from "node:path"; +import { mkdir } from "node:fs/promises"; +import { afterEach, expect, test } from "bun:test"; +import { acquireScanExecutionLock } from "../src/runtime.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/runtime.ts", import.meta.url).href; + const child = Bun.spawn( + [ + process.execPath, + "--eval", + ` + const { acquireScanExecutionLock } = await import(${JSON.stringify(module)}); + const release = await acquireScanExecutionLock(${JSON.stringify(state)}, ${JSON.stringify(scan)}); + console.log("locked"); + process.on("exit", release); + setInterval(() => {}, 60_000); + `, + ], + { 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("SIGKILL"); + 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..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"; @@ -59,6 +63,7 @@ async function interruptedScan( ScanOptions, | "safetyIdentifier" | "maxCostUsd" + | "maxTimeMinutes" | "ownerSuggestions" | "postScanPrompt" | "auth" @@ -331,11 +336,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 +366,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"], @@ -379,18 +390,18 @@ test.skipIf(process.platform !== "win32")( ); test.each([ - "failed", - "canceled", - "standard", - "changed", - "replaced", - "wrong-owner", -])( - "resume refuses %s scans without altering their saved state", - async (scenario) => { - const f = await interruptedScan( - scenario === "standard" ? "standard" : "deep", - ); + ["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", @@ -434,9 +445,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 +783,260 @@ 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, + }); + 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, + ]); + } + 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", + "--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 readDraft()).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) + return { + events: failedEvents("Connection failed: error sending request"), + }; + await saveDraft(true); + 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(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"); +}); + +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 = ""; + 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( + 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 }; + 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); + 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, + recipe: { + mode: "standard", + target: { kind: "paths", paths: ["source.py"] }, + }, + }); +}); + +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", + ); + return { events: failedEvents("403 forbidden: access denied") }; + }, + }), + })), + ); + 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], 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) {