From 98c3581a942dc5fdc006f10b7a81a8c547c28b46 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Sun, 25 Oct 2026 19:03:02 +0000 Subject: [PATCH] fix(sdk): preserve interrupted Deep Scans for resume Keep an eligible CLI scan and its coordinator running when the parent returns before a manifest exists. Verify the original checkout and session on explicit resume, reuse saved workers, and report only measured usage. Keep cancellations, terminal failures, and workflows with additional actions on their existing paths. Refs #1967. --- .../mcp-app/tests/test_workbench_database.ts | 2 +- .../scripts/finalize_scan_contract.py | 2 + .../codex-security/scripts/workbench_cli.py | 2 + .../codex-security/scripts/workbench_db.py | 4 +- .../scripts/workbench_saved_results.py | 126 ++++- .../scripts/workbench_scan_history.py | 6 +- .../scripts/workbench_scan_usage.py | 24 +- .../shared/workbench-migrations.json | 5 + .../tests/test_checkpoint_schema.py | 2 +- .../tests/test_workbench_deep_scan.py | 1 + .../tests/test_workbench_scan_history.py | 95 +++- .../test_workbench_setup_and_migrations.py | 1 + .../test_workbench_standard_deep_results.py | 23 + sdk/typescript/docs/cli.md | 27 +- sdk/typescript/src/api.ts | 245 ++++++++- sdk/typescript/src/cli.ts | 58 +- sdk/typescript/src/component-scan.ts | 1 + sdk/typescript/src/config.ts | 16 + sdk/typescript/src/cost.ts | 52 +- sdk/typescript/src/multiscan.ts | 1 + sdk/typescript/src/saved-scan-bootstrap.ts | 57 +- sdk/typescript/src/scan-logs.ts | 21 +- .../tests-ts/api-preflight-config.test.ts | 26 + .../tests-ts/cli-scan-failures.test.ts | 34 +- sdk/typescript/tests-ts/cost.test.ts | 12 + .../tests-ts/saved-scan-bootstrap.test.ts | 126 ++++- sdk/typescript/tests-ts/scan-resume.test.ts | 498 +++++++++++++++++- 27 files changed, 1359 insertions(+), 108 deletions(-) diff --git a/plugins/codex-security/mcp-app/tests/test_workbench_database.ts b/plugins/codex-security/mcp-app/tests/test_workbench_database.ts index 6104edc133..0353ca7151 100644 --- a/plugins/codex-security/mcp-app/tests/test_workbench_database.ts +++ b/plugins/codex-security/mcp-app/tests/test_workbench_database.ts @@ -293,7 +293,7 @@ for (const { missingVersions, legacyCheckpoint, mainReader } of [ reopened .prepare("SELECT MAX(version) AS version FROM schema_migrations") .get()?.version, - 57, + migrations.at(-1)!.version, ); for (const { version } of migrations.filter( (item) => item.version >= 43, diff --git a/plugins/codex-security/scripts/finalize_scan_contract.py b/plugins/codex-security/scripts/finalize_scan_contract.py index 0917daf8c1..23de4f94d9 100644 --- a/plugins/codex-security/scripts/finalize_scan_contract.py +++ b/plugins/codex-security/scripts/finalize_scan_contract.py @@ -731,6 +731,8 @@ def _read_scan_local_json_with_metadata( if not isinstance(payload, dict): raise ContractError(f"{context}: expected a JSON object") _require_safe_json_value(payload, context, validate_strings=False) + except RecursionError as exc: + raise ContractError(f"{context}: JSON nesting is too deep: {exc}") from exc finally: if descriptor >= 0: os.close(descriptor) diff --git a/plugins/codex-security/scripts/workbench_cli.py b/plugins/codex-security/scripts/workbench_cli.py index 37df392c01..a06bf10765 100644 --- a/plugins/codex-security/scripts/workbench_cli.py +++ b/plugins/codex-security/scripts/workbench_cli.py @@ -228,6 +228,8 @@ def add_command(name: str, *required: str) -> argparse.ArgumentParser: fail_scan = add_command("fail-scan", "--scan-id", "--message") fail_scan.add_argument("--claim-token") fail_scan.add_argument("--cost-json") + fail_scan.add_argument("--preserve-deep-owner") + fail_scan.add_argument("--handoff-cost-json") preserve_scan = add_command("preserve-scan-results", "--scan-id") preserve_scan.add_argument("--thread-id") diff --git a/plugins/codex-security/scripts/workbench_db.py b/plugins/codex-security/scripts/workbench_db.py index 71aaaa35bb..96e0fdb781 100644 --- a/plugins/codex-security/scripts/workbench_db.py +++ b/plugins/codex-security/scripts/workbench_db.py @@ -1513,6 +1513,8 @@ def complete_scan_locked( turn_id: str | None = None, ) -> dict[str, Any]: scan = require_scan(connection, scan_id) + if not prepare_only: + cost_json = scan_usage.terminal_scan_cost(scan, cost_json) if scan["status"] == "complete": return saved_results.refresh_completed_scan( _WORKBENCH_DB_CONTEXT, connection, scan, cost_json @@ -2912,7 +2914,7 @@ def scan_result( "artifacts": artifacts, **model_fields, "canceledAt": scan["canceled_at"], - **scan_usage.stored_scan_cost_fields(scan["cost_json"]), + **scan_usage.saved_scan_cost_fields(scan), "contract": scan_contract(scan), "continuationThreadId": scan["continuation_thread_id"], "threadIds": scan_usage._scan_root_thread_ids(connection, scan, None), diff --git a/plugins/codex-security/scripts/workbench_saved_results.py b/plugins/codex-security/scripts/workbench_saved_results.py index 51f40f2c81..7470a476f1 100644 --- a/plugins/codex-security/scripts/workbench_saved_results.py +++ b/plugins/codex-security/scripts/workbench_saved_results.py @@ -17,6 +17,7 @@ from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parent)) +from filesystem_identity import stored_filesystem_identity_matches from finalize_scan_contract import ( ContractError, ScanLocalIOError, @@ -1296,12 +1297,124 @@ def _scan_draft_contents_digest(documents: dict[str, bytes | None]) -> str: return digest.hexdigest() +def _cli_owned_deep_run( + connection: sqlite3.Connection, scan_id: str, args: Any +) -> sqlite3.Row | None: + owner = getattr(args, "preserve_deep_owner", None) + if not owner or args.claim_token is not None: + return None + return connection.execute( + """ + SELECT run.status, run.manifest_path FROM deep_scan_runs AS run + JOIN scans ON scans.id = run.scan_id JOIN workspaces ON workspaces.id = scans.workspace_id + WHERE scans.id = :id AND scans.mode = 'deep' AND scans.recipe_json IS NOT NULL + AND scans.status = 'running' AND scans.canceled_at IS NULL AND run.cancel_requested = 0 + AND scans.handoff_status = 'delivered' AND scans.handoff_claim_token IS NULL + AND COALESCE(scans.deep_scan_owner_thread_id, workspaces.thread_id) = :owner + AND (scans.continuation_thread_id IS NULL OR scans.continuation_thread_id = :owner) + """, + {"id": scan_id, "owner": owner}, + ).fetchone() + + +def _verified_cli_publication( + db: Any, connection: sqlite3.Connection, scan_id: str, args: Any +) -> str | None: + # The per-scan completion lock fences native publication. Validate its files + # before taking SQLite's write lock, then recheck ownership and selection in it. + run = _cli_owned_deep_run(connection, scan_id, args) + if run is None or run["status"] != "succeeded": + return None + scan = db.require_scan(connection, scan_id) + scan_dir = Path(scan["scan_dir"]) + if run["manifest_path"] != str(scan_dir / "scan-manifest.json"): + return None + try: + db.deep_scan.require_deep_scan_ready_for_parent_completion(connection, scan) + manifest, findings, coverage = ( + _read_scan_local_json(scan_dir, name, name) + for name in ("scan-manifest.json", "findings.json", "coverage.json") + ) + published = manifest.get("scan") + if not isinstance(published, dict) or published.get("complete") is False: + return None + binding = db.workbench_completion_binding(scan, published.get("completedAt"), manifest) + _validate_completion_binding(manifest, findings, coverage, binding) + _prepare_scan_finalization( + scan_dir, + completion_binding=binding, + allow_missing_writeups=True, + finding_scope=db.requested_finding_scope(scan), + ) + except (ContractError, OSError, RecursionError, SystemExit): + return None + return run["manifest_path"] + + +def preserve_deep_owner( + db: Any, + connection: sqlite3.Connection, + scan: sqlite3.Row, + args: Any, + cost_json: str | None, + *, + verified_manifest: str | None = None, +) -> bool: + """Keep this CLI attempt's coordinator under the caller's write transaction.""" + run = _cli_owned_deep_run(connection, scan["id"], args) + if run is None or not ( + run["status"] == "running" + or run["status"] == "succeeded" + and verified_manifest is not None + and run["manifest_path"] == verified_manifest + ): + return False + try: + identity = db.require_remediation_target(scan["target_path"]).stat() + if not ( + stored_filesystem_identity_matches(scan["target_device"], identity.st_dev) + and stored_filesystem_identity_matches(scan["target_inode"], identity.st_ino) + ): + return False + db.require_canonical_scan_directory(Path(scan["scan_dir"])) + except (OSError, RuntimeError, SystemExit): + return False + # Only fresh attempts send this guard. A concurrent estimate can fill a + # missing original measurement, but cannot replace one already recorded. + subtotal = scan["cli_handoff_cost_json"] + if subtotal in {None, "null"}: + cost = db.scan_usage.stored_scan_cost_fields(cost_json).get("cost") + if cost is None and args.handoff_cost_json is not None: + cost = db.scan_usage.stored_scan_cost_fields( + db.parse_scan_cost(args.handoff_cost_json) + ).get("cost") + subtotal = json.dumps(cost) + connection.execute( + "UPDATE scans SET continuation_thread_id = ?, cli_handoff_cost_json = ?, cost_json = ? WHERE id = ?", + ( + args.preserve_deep_owner, + subtotal, + db.scan_usage.usage_without_handoff_total(scan, cost_json), + scan["id"], + ), + ) + return True + + def fail_scan(db: Any, connection: Any, args: Any) -> dict[str, Any]: - with db.scan_completion_lock(db.require_uuid(args.scan_id, "scan-id")): - return fail_scan_locked(db, connection, args) + scan_id = db.require_uuid(args.scan_id, "scan-id") + with db.scan_completion_lock(scan_id): + return fail_scan_locked( + db, + connection, + args, + verified_manifest=_verified_cli_publication(db, connection, scan_id, args), + ) -def fail_scan_locked(db: Any, connection: Any, args: Any) -> dict[str, Any]: +def fail_scan_locked( + db: Any, connection: Any, args: Any, *, verified_manifest: str | None = None +) -> dict[str, Any]: scan_id = db.require_uuid(args.scan_id, "scan-id") cost_json = db.parse_scan_cost(args.cost_json) connection.execute("BEGIN IMMEDIATE") @@ -1318,6 +1431,11 @@ def fail_scan_locked(db: Any, connection: Any, args: Any) -> dict[str, Any]: args.claim_token, error_message="Scan failure is owned by another continuation.", ) + if preserve_deep_owner( + db, connection, scan, args, cost_json, verified_manifest=verified_manifest + ): + return {"preserved": True} + cost_json = db.scan_usage.terminal_scan_cost(scan, cost_json) message = db.optional_text(args.message) updated = connection.execute( """ @@ -1386,7 +1504,7 @@ def cancel_scan_locked(db: Any, connection: Any, args: Any) -> dict[str, Any]: def preserve_stopped_results_after_transition(db: Any, connection: Any, scan_id: str) -> None: try: published = preserve_scan_results_locked(db, connection, scan_id) - except (ContractError, OSError, SystemExit, ValueError) as exc: + except (ContractError, OSError, RecursionError, SystemExit, ValueError) as exc: scan = db.require_scan(connection, scan_id) warnings = json.loads(scan["completion_warnings_json"]) warning = f"Saved scan evidence remains on disk; result publication needs follow-up: {exc}" diff --git a/plugins/codex-security/scripts/workbench_scan_history.py b/plugins/codex-security/scripts/workbench_scan_history.py index 69aa02015b..1d618ea59e 100644 --- a/plugins/codex-security/scripts/workbench_scan_history.py +++ b/plugins/codex-security/scripts/workbench_scan_history.py @@ -25,7 +25,7 @@ from report_projection import SEVERITY_ORDER from workbench_constants import ARTIFACTS, FINDINGS_PAGE_MAX from workbench_scan_start import scan_target_identity -from workbench_scan_usage import stored_scan_cost_fields +from workbench_scan_usage import saved_scan_cost_fields from workbench_source_scopes import public_scan_recipe from workbench_target import git_output, require_scan_target_identity from workbench_validation import timestamp_key @@ -122,6 +122,8 @@ def cli_scan_resume( "threadId": thread_id, "userContext": scan["user_context"], } + if scan["cli_handoff_cost_json"] is not None: + result["handoffCost"] = json.loads(scan["cli_handoff_cost_json"]) # Active coordinators may still be writing drafts. Validate sealed results # before attaching to a coordinator that has finished. if run is not None and run["status"] == "succeeded": @@ -287,7 +289,7 @@ def scan_summary(row: sqlite3.Row) -> dict[str, Any]: return { "completedAt": row["completed_at"], "continuationThreadId": row["continuation_thread_id"], - **stored_scan_cost_fields(row["cost_json"]), + **saved_scan_cost_fields(row), "findingCount": row["finding_count"], "handoffStatus": row["handoff_status"], "mode": row["mode"], diff --git a/plugins/codex-security/scripts/workbench_scan_usage.py b/plugins/codex-security/scripts/workbench_scan_usage.py index 301ff9cbef..66fdd850a7 100644 --- a/plugins/codex-security/scripts/workbench_scan_usage.py +++ b/plugins/codex-security/scripts/workbench_scan_usage.py @@ -59,6 +59,28 @@ def stored_scan_cost_fields(value: str | None) -> dict[str, Any]: } +def saved_scan_cost_fields(scan: sqlite3.Row) -> dict[str, Any]: + fields = stored_scan_cost_fields(scan["cost_json"]) + if scan["cli_handoff_cost_json"] is not None: + fields["handoffCost"] = json.loads(scan["cli_handoff_cost_json"]) + return fields + + +def usage_without_handoff_total(scan: sqlite3.Row, value: str | None) -> str | None: + # A later turn can supply independent measured usage, but cannot establish + # a total cost across the interruption. Retain the original subtotal separately. + usage = stored_scan_cost_fields(value).get("usage") or stored_scan_cost_fields( + scan["cost_json"] + ).get("usage") + return json.dumps({"usage": usage}, separators=(",", ":")) if usage is not None else None + + +def terminal_scan_cost(scan: sqlite3.Row, value: str | None) -> str | None: + return ( + value if scan["cli_handoff_cost_json"] is None else usage_without_handoff_total(scan, value) + ) + + def reconcile_completed_scan_cost( connection: sqlite3.Connection, scan: sqlite3.Row, @@ -67,7 +89,7 @@ def reconcile_completed_scan_cost( """Persist authoritative SDK cost without discarding measured worker usage.""" existing = json.loads(scan["cost_json"]) if scan["cost_json"] is not None else {} - if isinstance(existing, dict) and "usage" in existing: + if scan["cli_handoff_cost_json"] is None and isinstance(existing, dict) and "usage" in existing: cost_json = json.dumps( {**existing, "cost": json.loads(cost_json)}, separators=(",", ":"), diff --git a/plugins/codex-security/shared/workbench-migrations.json b/plugins/codex-security/shared/workbench-migrations.json index ac72416389..2919b100f2 100644 --- a/plugins/codex-security/shared/workbench-migrations.json +++ b/plugins/codex-security/shared/workbench-migrations.json @@ -469,5 +469,10 @@ "statements": [ "ALTER TABLE scans ADD COLUMN retained_checkpoint_heads_json TEXT;" ] + }, + { + "version": 58, + "name": "record CLI Deep Scan handoff cost", + "statements": ["ALTER TABLE scans ADD COLUMN cli_handoff_cost_json TEXT;"] } ] diff --git a/plugins/codex-security/tests/test_checkpoint_schema.py b/plugins/codex-security/tests/test_checkpoint_schema.py index 459ddefadc..29ebaf4000 100644 --- a/plugins/codex-security/tests/test_checkpoint_schema.py +++ b/plugins/codex-security/tests/test_checkpoint_schema.py @@ -157,7 +157,7 @@ def migrate(connection, selected): ) assert connection.execute( "SELECT COUNT(*), MAX(version) FROM schema_migrations" - ).fetchone()[:] == (len(migrations), 57) + ).fetchone()[:] == (len(migrations), migrations[-1][0]) for version in (43, 44, 45, 46, 47): assert ( connection.execute( diff --git a/plugins/codex-security/tests/test_workbench_deep_scan.py b/plugins/codex-security/tests/test_workbench_deep_scan.py index 83d4695c02..a11112d3bd 100644 --- a/plugins/codex-security/tests/test_workbench_deep_scan.py +++ b/plugins/codex-security/tests/test_workbench_deep_scan.py @@ -81,6 +81,7 @@ def begin_target_scan( def restore_v53_schema(state_dir: Path) -> None: with sqlite3.connect(state_dir / "workbench.sqlite3") as connection: + connection.execute("ALTER TABLE scans DROP COLUMN cli_handoff_cost_json") connection.execute("ALTER TABLE scans DROP COLUMN retained_checkpoint_heads_json") connection.execute("ALTER TABLE workspaces DROP COLUMN deep_scan_config_json") connection.execute("ALTER TABLE deep_scan_runs DROP COLUMN pending_target_project_path") diff --git a/plugins/codex-security/tests/test_workbench_scan_history.py b/plugins/codex-security/tests/test_workbench_scan_history.py index 688fe09755..162dd8bc68 100644 --- a/plugins/codex-security/tests/test_workbench_scan_history.py +++ b/plugins/codex-security/tests/test_workbench_scan_history.py @@ -18,6 +18,7 @@ from test_workbench_deep_scan import begin_target_scan from test_workbench_prompt_only_scan import start_headless_standard_scan, start_prompt_only_scan from workbench_test_support import ( + BUDGET_COST, fail_scan, get_scan, initialize_git_repository, @@ -687,32 +688,86 @@ def test_cli_scan_completion_persists_authoritative_cost_after_plugin_completion assert get_scan(state_dir, scan["scanId"])["scan"]["cost"] == cost -def test_failed_cli_scan_history_persists_measured_cost(tmp_path: Path) -> None: +@pytest.mark.parametrize( + ("full", "partial"), + [(True, None), (True, "{"), (False, json.dumps(BUDGET_COST)), (False, "{")], + ids=["ordinary", "unrelated-owner", "partial-only", "invalid-partial-only"], +) +def test_failed_cli_scan_history_persists_measured_cost(tmp_path: Path, full, partial) -> None: state_dir = tmp_path / "state" repository = tmp_path / "repository" repository.mkdir() - cost = { - "model": "gpt-5.6-terra", - "inputTokens": 1250, - "cachedInputTokens": 200, - "cacheWriteInputTokens": 0, - "outputTokens": 30, - "estimatedUsd": 0.003125, - } + cost = {**BUDGET_COST, "model": "gpt-5.6-terra", "estimatedUsd": 0.003125} scan = create_cli_scan(state_dir, tmp_path / "results", repository, complete=False) + args = ["--cost-json", json.dumps(cost)] if full else [] + if partial is not None: + args.extend(("--preserve-deep-owner", "unrelated", "--handoff-cost-json", partial)) + failed = fail_scan(state_dir, scan["scanId"], "Scan stopped: cost limit exceeded.", *args) - failed = fail_scan( - state_dir, - scan["scanId"], - "Scan stopped: cost limit exceeded.", - "--cost-json", - json.dumps(cost), - ) - - assert failed["scan"]["cost"] == cost assert failed["scan"]["progress"]["status"] == "failed" - assert run_workbench(state_dir, "list-scans")["scans"][0]["cost"] == cost - assert get_scan(state_dir, scan["scanId"])["scan"]["cost"] == cost + expected = {"cost": cost} if full else {} + for saved in ( + failed["scan"], + run_workbench(state_dir, "list-scans")["scans"][0], + get_scan(state_dir, scan["scanId"])["scan"], + ): + assert {key: saved[key] for key in ("cost", "handoffCost") if key in saved} == expected + + +@pytest.mark.parametrize( + ("terminal", "receipt_flag", "receipts"), + [ + ("complete", "--handoff-cost-json", (BUDGET_COST,)), + ("failed", "--cost-json", (BUDGET_COST, {**BUDGET_COST, "inputTokens": 2500})), + ("canceled", "--cost-json", (None, BUDGET_COST)), + ("complete", "--handoff-cost-json", (None,)), + ], +) +def test_preserved_owner_keeps_cost_subtotal(tmp_path, terminal, receipt_flag, receipts): + state, repository = tmp_path / "state", tmp_path / "repository" + repository.mkdir() + scan = create_cli_scan(state, tmp_path, repository, mode="deep", complete=False) + scan_id, scan_dir = scan["scanId"], Path(scan["scanDir"]) + resume_deep_scan(state, scan_id, "owner", environment={"CODEX_HOME": str(tmp_path / "home")}) + prepared = scan_command(state, "prepare-scan-completion", scan_id, check=False) + assert prepared["returncode"] != 0 + assert "must finish and persist its manifest" in prepared["stderr"] + guard = ("--preserve-deep-owner", "owner") + for receipt in receipts: + args = (receipt_flag, json.dumps(receipt)) if receipt else () + result = fail_scan(state, scan_id, "Parent interrupted.", *guard, *args) + assert result == {"preserved": True} + resume = scan_command(state, "get-cli-scan-resume", scan_id) + assert resume["threadId"] == "owner" + measured = BUDGET_COST in receipts + assert resume["handoffCost"] == (BUDGET_COST if measured else None) + + later = {**BUDGET_COST, "inputTokens": 2500, "outputTokens": 60, "estimatedUsd": 0.0125} + usage = { + **{key: value for key, value in BUDGET_COST.items() if key.endswith("Tokens")}, + "outputTokens": 90, + "reasoningOutputTokens": 0, + "totalTokens": 1340, + "source": "codex_rollout", + "coverage": "complete", + "threadCount": 1, + } + if terminal == "complete": + mark_deep_coordinator_succeeded(state, scan_id, scan_dir) + write_completed_contract(scan_dir, scan_id, repository, coverage_mode="deep_repository") + complete = ("complete-scan", scan_id, "--cost-json") + scan_command(state, *complete, json.dumps({"usage": usage})) + scan_command(state, *complete, json.dumps(later)) + elif terminal == "failed": + fail_scan(state, scan_id, "Could not resume.", "--cost-json", json.dumps(later)) + else: + scan_command(state, "cancel-scan", scan_id, "--thread-id", "owner") + saved = get_scan(state, scan_id)["scan"] + assert saved["progress"]["status"] == terminal + assert saved["handoffCost"] == (BUDGET_COST if measured else None) + assert "cost" not in saved + if terminal == "complete": + assert saved["usage"] == usage def test_scan_failure_rejects_invalid_cost_without_stopping_the_scan(tmp_path: Path) -> None: diff --git a/plugins/codex-security/tests/test_workbench_setup_and_migrations.py b/plugins/codex-security/tests/test_workbench_setup_and_migrations.py index 4113e8df5c..eff5c2ebd7 100644 --- a/plugins/codex-security/tests/test_workbench_setup_and_migrations.py +++ b/plugins/codex-security/tests/test_workbench_setup_and_migrations.py @@ -99,6 +99,7 @@ (55, "retain undelivered headless deep scan results"), (56, "recover unfinished legacy headless deep scans"), (57, "freeze stopped scan checkpoint selections"), + (58, "record CLI Deep Scan handoff cost"), ] diff --git a/plugins/codex-security/tests/test_workbench_standard_deep_results.py b/plugins/codex-security/tests/test_workbench_standard_deep_results.py index 0988f540c2..ef85fdff7e 100644 --- a/plugins/codex-security/tests/test_workbench_standard_deep_results.py +++ b/plugins/codex-security/tests/test_workbench_standard_deep_results.py @@ -2012,6 +2012,29 @@ def test_standard_worker_results_finish_with_only_canonical_parent_manifest( assert not (scan_dir / "artifacts" / "02_discovery").exists() +@pytest.mark.parametrize( + ("filename", "depth"), + [("findings.json", sys.getrecursionlimit() + 10), ("scan-manifest.json", 600)], +) +def test_guarded_cli_failure_rejects_deeply_nested_publication( + tmp_path: Path, filename: str, depth: int +) -> None: + state, home, _, scan_dir, scan_id = standard_parent_results_fixture(tmp_path, budget=True) + env = {"CODEX_HOME": str(home)} + manifest = scan_dir / "scan-manifest.json" + finish_deep_scan(state, scan_id, "capped", str(manifest), environment=env) + artifact = scan_dir / filename + nested = "[" * depth + "0" + "]" * depth + artifact.write_text(artifact.read_text().replace("{", '{"nested":' + nested + ",", 1)) + + owner = "standard-worker-thread" + result = fail_scan(state, scan_id, "Parent stopped.", "--preserve-deep-owner", owner) + assert result["scan"]["progress"]["status"] == "failed" + assert result["scan"]["failureMessage"] == "Parent stopped." + coordinator = get_deep_scan(state, scan_id, owner, environment=env)["deepScan"] + assert coordinator["status"] == "succeeded" + + @pytest.mark.parametrize( "incidental_artifacts", ("inventory", "ledger", "both", "inventory_symlink", "ledger_symlink"), diff --git a/sdk/typescript/docs/cli.md b/sdk/typescript/docs/cli.md index d8d4803808..001161f958 100644 --- a/sdk/typescript/docs/cli.md +++ b/sdk/typescript/docs/cli.md @@ -2141,12 +2141,31 @@ The scan must still be `running`, with its original checkout, output directory, and Codex session in the same state directory. Checkout identity, revision, and contents must match. Completed, failed, and canceled scans need a rerun instead. -Resume retains scan ID, completed workers, artifacts, accumulated cost, and saved -settings/instructions. Knowledge-bearing scans save their extracted documents in +If a standalone Deep Scan parent stops while discovery is still working, use +`codex-security scans show SCAN_ID` to inspect it. When the CLI reports that it +preserved the scan, resume from the original environment to continue with the +same workers. Scans with a spending limit, a post-scan prompt, a knowledge base +outside the checkout, or private launch settings that cannot be restored keep +their usual stopping behavior. + +If a scan started with `--validate` or `--patch` fails because its parent +returned early, retry the original `scan` command with the same options and a +fresh output directory. This keeps validation or patching with its scan settings. + +If the original scan used `--python`, set `PYTHON` to the same interpreter's +absolute path before running `scans show` or `scans resume`. + +When the parent leaves a live scan for resume, `scans show SCAN_ID --json` +reports the subtotal measured before interruption as `handoffCost`, or `null` +when no estimate was available. A resumed +turn can report usage for that turn alone, so a total cost is unavailable. +Resume retains the scan ID, completed workers, artifacts, and saved settings. +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 -document text. Old individual scans without a saved knowledge snapshot require a +deleted. If the snapshot is missing, restore the original `.scan-knowledge.json` +to the scan directory. Canonical reports do not embed its document text. +Old individual scans without a saved knowledge snapshot require a new scan when they used a knowledge base. Old scans without a knowledge base and bulk campaigns with an already-bound snapshot retain their existing resume path. New records save the authentication mode, explicit safety diff --git a/sdk/typescript/src/api.ts b/sdk/typescript/src/api.ts index 1a983ee7ea..6086c04235 100644 --- a/sdk/typescript/src/api.ts +++ b/sdk/typescript/src/api.ts @@ -77,6 +77,7 @@ import { import { DEFAULT_CODEX_CONFIG, EXTERNAL_CODEX_PROVIDERS, + containsCodexOverrides, inlineToml, isExternalModelProvider, hasCommandAuth, @@ -309,6 +310,7 @@ interface PreparedSession { runtimeHome: string; effectiveConfig: JsonObject; preflightConfig: JsonObject; + canReplayOverrides: boolean; sessionConfig: JsonObject; modelProvider: unknown; externalProvider: @@ -329,6 +331,8 @@ export interface ScanOptions extends ScanSettings { knowledgeBaseSnapshot?: KnowledgeBaseSnapshot; /** @internal Resume a CLI Deep Scan with its saved launch recipe. */ resumeScanId?: string; + /** @internal The caller owns retries or follow-up work outside the saved recipe. */ + managedAttempt?: boolean; /** Save synthetic Standard scan results without calling Codex or a model. */ mock?: boolean; /** Opt into a durable scan -> custom publication -> dedupe workflow. */ @@ -1966,6 +1970,17 @@ export class CodexSecurity { let artifactRestorationFailure: OutputDirectoryError | null = null; let customValidationComplete = false; let completionCost: ScanCost | null = null; + let handoffResume = false; + let deferResumeTotals = false; + const pendingTotals: (() => void)[] = []; + const reportTotal = (notify: () => void) => { + if (handoffResume) return; + if (deferResumeTotals) pendingTotals.push(notify); + else notify(); + }; + let finalizationStarted = false; + let prepareFailed = false; + let recoverableOwner: (() => Promise) | undefined; let scanTurnStatus: "in_progress" | "completed" | "failed" | null = null; let budgetRecovery: { expectation: ScanExpectation; @@ -2121,8 +2136,10 @@ export class CodexSecurity { modelProvider, ), }; + const checkoutRoot = + (await gitMarkerRoot(repo, signal, "outermost")) ?? repo; for (const root of [ - (await gitMarkerRoot(repo, signal, "outermost")) ?? repo, + checkoutRoot, ...(knowledgeBase?.protectedRoots ?? []), ]) { git = await inspectTrustedExecutable("git", git.environment, root); @@ -2371,11 +2388,20 @@ export class CodexSecurity { onUsage: options.onUsage === undefined ? undefined - : (usage) => notifyObserver(options, "onUsage")(usage), + : (usage) => + reportTotal(() => notifyObserver(options, "onUsage")(usage)), onUsageObservation: options.onUsageObservation === undefined ? undefined - : notifyObserver(options, "onUsageObservation"), + : (observation) => + notifyObserver( + options, + "onUsageObservation", + )( + handoffResume + ? { ...observation, tokensUpdatedAt: null } + : observation, + ), onActivity: options.onActivity === undefined ? undefined @@ -2404,7 +2430,9 @@ export class CodexSecurity { latestCost = cost; latestCostComplete = complete; if (complete) - notifyObserver(options, "onCost")(cost, maxCostUsd); + reportTotal(() => + notifyObserver(options, "onCost")(cost, maxCostUsd), + ); if ( maxCostUsd !== undefined && cost.estimatedUsd > maxCostUsd @@ -2456,7 +2484,7 @@ export class CodexSecurity { ); if (budgetSignal.aborted) return; maxCostUsd = next; - if (latestCostComplete) + if (latestCostComplete && !handoffResume) notifyObserver(options, "onCost")( latestCost!, maxCostUsd, @@ -2478,13 +2506,17 @@ export class CodexSecurity { costTracker = tracker; const scanKnowledge = knowledgeBase === null ? undefined : options.knowledgeBaseSnapshot; + const recipeConfig: JsonObject = { + ...preflightConfig, + approval_policy: approvalPolicy, + }; const recipe = scanRecipe({ repository: repo, target: normalized, mode, repositoryRevision: expectation.repositoryRevision, pluginVersion: runtime.plugin.version, - config: { ...preflightConfig, approval_policy: approvalPolicy }, + config: recipeConfig, failOnSeverity: options.failureSeverity, knowledgeBasePaths: knowledgeBase?.sources, maxCostUsd: options.maxCostUsd, @@ -2501,6 +2533,13 @@ export class CodexSecurity { ); if (options.safetyIdentifier !== undefined) recipe["safetyIdentifier"] = options.safetyIdentifier; + if ( + mode === "deep" && + !options.managedAttempt && + options.workflowId === undefined && + checkoutRoot !== repo + ) + recipe["checkoutRoot"] = checkoutRoot; if (options.postScanPrompt !== undefined) recipe["postScanPrompt"] = options.postScanPrompt; if (options.validationPrompt !== undefined) @@ -2545,6 +2584,8 @@ export class CodexSecurity { ? undefined : registration["threadId"]; if (options.resumeScanId !== undefined) { + // This turn may report only its own usage. The Workbench measures scan history separately. + handoffResume = Object.hasOwn(registration, "handoffCost"); const savedRecipe = registration["recipe"]; if ( scanId !== options.resumeScanId || @@ -2923,6 +2964,22 @@ export class CodexSecurity { signal, }) : (await thread.runStreamed(prompt, turnOptions)).events; + const canPreserve = + mode === "deep" && + !directDeepScan && + options.resumeScanId === undefined && + options.maxCostUsd === undefined && + options.workflowId === undefined && + !options.managedAttempt && + options.validationPrompt === undefined && + // A worker that fails after this caller exits cannot run its follow-up. + !postScanPrompt?.trim() && + // Bare resume protects the target's enclosing checkout before restoring knowledge. + (knowledgeBase?.protectedRoots ?? []).every((root) => + isWithin(protectedRoot, root), + ) && + this.config.pluginPath === undefined && + session.canReplayOverrides; scanTurnStatus = "in_progress"; checkOpen(); @@ -2940,8 +2997,10 @@ export class CodexSecurity { modelProvider, workbenchValidated: true, model, - onTurnStatus: (status) => { + onTurnStatus: (status, usage) => { scanTurnStatus = status; + if (status === "completed" && recoverableOwner) + tracker.recordUsage(usage); }, onThreadStarted: async (threadId) => { if (resumeThreadId !== undefined && threadId !== resumeThreadId) { @@ -2967,6 +3026,17 @@ export class CodexSecurity { if (directDeepScan) tracker.recordUsage(DIRECT_DEEP_SCAN_CONTROL_USAGE, threadId); tracker.start(threadId); + if (canPreserve && recoverableOwner === undefined) { + recoverableOwner = async () => { + const saved = await findScanSession( + runtime.codexHome, + threadId, + ).catch(() => null); + return !signal.aborted && saved?.workingDirectory === scanDir + ? threadId + : undefined; + }; + } try { await workbench( workbenchOptions, @@ -2980,6 +3050,8 @@ export class CodexSecurity { } }, onFinalize: async (usage) => { + // Codex can yield turn.completed before reporting a process exit. + finalizationStarted = true; if (options.validationPrompt !== undefined) { tracker.recordUsage(usage); await tracker.refresh().catch(reportTrackingError); @@ -3039,6 +3111,8 @@ export class CodexSecurity { customValidationComplete = true; } budgetAbortController.abort(); + // A concurrent parent can hand off before this resumed turn's fallback receipt. + deferResumeTotals = options.resumeScanId !== undefined; const snapshot = await tracker.stop(usage).catch((error: unknown) => { if (options.maxCostUsd !== undefined) throw error; reportTrackingError(error); @@ -3053,7 +3127,7 @@ export class CodexSecurity { "Scan completed, but its cost limit could not be verified because model pricing or token usage is unavailable.", ); } - completionCost = snapshot.cost; + completionCost = handoffResume ? null : snapshot.cost; let preparation: JsonObject; try { preparation = await workbench( @@ -3061,6 +3135,7 @@ export class CodexSecurity { workbenchArgs("prepare-scan-completion", { scanId }), ); } catch (error) { + prepareFailed = true; const saved = await workbench( workbenchOptions, workbenchArgs("get-scan", { scanId }), @@ -3087,7 +3162,7 @@ export class CodexSecurity { (warning): warning is string => typeof warning === "string", ) : []; - return snapshot.usage; + return handoffResume ? null : snapshot.usage; }, onScanStarted: options.onScanStarted, onTrustedAccessStatus: options.onTrustedAccessStatus, @@ -3126,6 +3201,15 @@ export class CodexSecurity { completedScan, "completed", ); + handoffResume ||= + isRecord(completedScan) && Object.hasOwn(completedScan, "handoffCost"); + if (handoffResume && result.turnResult.usage != null) { + result = new ScanResult({ + ...result, + turnResult: { ...result.turnResult, usage: null }, + }); + } + if (!handoffResume) for (const notify of pendingTotals) notify(); if (directDeepScan) { result = new ScanResult({ ...result, @@ -3481,15 +3565,75 @@ export class CodexSecurity { }).catch(() => undefined); } try { - await workbench({ ...activeScan.options, signal: undefined }, [ - "fail-scan", - "--scan-id", - activeScan.id, - `--message=${errorMessage(failure)}`, - ...(snapshot?.cost - ? ["--cost-json", JSON.stringify(snapshot.cost)] - : []), - ]); + const disconnected = + (scanTurnStatus === "in_progress" && + failure instanceof IncompleteScanError) || + (!finalizationStarted && lostParentTransport(failure)); + const owner = + (prepareFailed || disconnected) && !signal.aborted + ? await recoverableOwner?.() + : undefined; + const subtotal = + owner && !snapshot?.cost + ? (costTracker?.lastMeasuredCost ?? completionCost) + : null; + const saved = await workbench( + { ...activeScan.options, signal: undefined }, + [ + "fail-scan", + "--scan-id", + activeScan.id, + `--message=${errorMessage(failure)}`, + ...(owner === undefined ? [] : ["--preserve-deep-owner", owner]), + ...(snapshot?.cost + ? ["--cost-json", JSON.stringify(snapshot.cost)] + : []), + ...(subtotal + ? ["--handoff-cost-json", JSON.stringify(subtotal)] + : []), + ], + ); + if (owner && saved["preserved"] === true) { + const scanId = activeScan.id; + const current = await workbench( + { ...activeScan.options, signal: undefined }, + workbenchArgs("get-scan", { scanId }), + ).catch(() => null); + const row = current?.["scan"]; + const status = + isRecord(row) && isRecord(row["progress"]) + ? row["progress"]["status"] + : undefined; + let action = `Check progress with codex-security scans show ${scanId}.`; + if (this.config.pythonPath !== undefined) + action = `Set PYTHON to the interpreter used for the original scan. ${action}`; + let reason = errorMessage(failure); + if ( + isRecord(row) && + status === "failed" && + typeof row["failureMessage"] === "string" + ) + reason = row["failureMessage"]; + if ( + signal.aborted && + status !== "complete" && + status !== "failed" && + status !== "canceled" + ) { + // Cancellation can arrive after the workbench preserved the parent. + await workbench( + { ...activeScan.options, signal: undefined }, + workbenchArgs("fail-scan", { + scanId, + message: scanInterruptionMessage(scanDir), + }), + ); + } else if (status === "running") + action += ` With the original checkout, output directory, Codex session, Python and knowledge settings restored, run codex-security scans resume ${scanId}.`; + failure = new IncompleteScanError(`${reason}\n${action}`, { + cause: failure, + }); + } } catch {} } if (runPostScan !== null && !signal.aborted) { @@ -4000,6 +4144,7 @@ export class CodexSecurity { runtimeHome, effectiveConfig, preflightConfig, + canReplayOverrides: false, sessionConfig: preflightConfig, modelProvider: "host", externalProvider: null, @@ -4078,8 +4223,12 @@ export class CodexSecurity { } const runtime = this.#runtime; if (deepScan) await requireCurrentDeepScanPlugin(runtime); + const configured = scanCyberAccessConfig( + requestedConfig, + options.cyberAccessProgram, + ); const effectiveConfig = await scanSkillConfig( - scanCyberAccessConfig(requestedConfig, options.cyberAccessProgram), + configured, runtime.plugin, runtime.codexHome, ); @@ -4199,6 +4348,20 @@ export class CodexSecurity { runtimeHome, effectiveConfig, preflightConfig, + // Resume reapplies the bundled skill exclusions. Compare explicit settings + // before that overlay, using the same safe projection as the saved recipe. + canReplayOverrides: + deepScan && + (this.config.codexOverrides === undefined || + containsCodexOverrides( + { + ...(effectiveConfig === configured + ? preflightConfig + : scanPreflightCodexConfig(configured)), + approval_policy: approvalPolicy, + }, + this.config.codexOverrides, + )), sessionConfig, modelProvider, externalProvider, @@ -4901,7 +5064,7 @@ interface ScanEventRunOptions extends Pick { workbenchValidated?: boolean; model?: string; onFinalize?: (usage: unknown) => Promise; - onTurnStatus?: (status: "completed" | "failed") => void; + onTurnStatus?: (status: "completed" | "failed", usage?: unknown) => void; onThreadStarted?: (threadId: string) => Promise | void; onScanStarted?: () => void; onTrustedAccessStatus?: (status: ScanTrustedAccessStatus) => void; @@ -4925,7 +5088,7 @@ export async function runScanEvents( events: options.events, onEvent: async (event) => { if (event.type === "turn.completed") - options.onTurnStatus?.("completed"); + options.onTurnStatus?.("completed", event["usage"]); if (event.type === "turn.failed") options.onTurnStatus?.("failed"); if ( !tacStatusReported && @@ -5751,6 +5914,33 @@ function turnFailureMessage(error: unknown): string { return "The Codex Security scan turn failed without a readable error message."; } +function lostParentTransport(error: unknown): boolean { + const message = errorMessage(error); + const clientStatus = + /\b(?:HTTP(?:\/[\d.]+)?|status(?: code)?)\s*[:=]?\s*(4\d{2})\b/iu.exec( + message, + )?.[1]; + if (clientStatus !== undefined && clientStatus !== "408") return false; + const classification = classifyConnectionFailure(error); + if (classification === "timeout") return true; + // Native exec prefixes provider response.failed messages with this text too. + const reason = message.replace( + /^stream disconnected before completion:\s*/iu, + "", + ); + if (classification === "network_error") { + return /\b(?:ENOTFOUND|ECONNRESET|ECONNREFUSED|EHOSTUNREACH|ETIMEDOUT|error sending request|(?:network|connection) (?:error|failed|lost|refused|reset|unreachable)|(?:DNS|TLS) (?:error|failure|failed)|timed? out)\b/iu.test( + reason, + ); + } + return ( + classification === "unknown" && + /\b(?:(?:stream|transport) disconnected|(?:stream|websocket) closed (?:before|by server)|websocket closed$|Codex Exec exited with (?:code|signal))\b/iu.test( + reason, + ) + ); +} + export function classifyConnectionFailure( error: unknown, ): @@ -6223,13 +6413,18 @@ async function pluginSupportsIsolatedDeepScanConfig( ); } +function scanInterruptionMessage(scanDir: string): string { + return scanDir + ? `Codex Security scan was interrupted; partial output remains at ${scanDir}.` + : "Codex Security scan was interrupted during preparation."; +} + function throwIfAborted(signal?: AbortSignal, scanDir = ""): void { if (!signal?.aborted) return; if (signal.reason instanceof ScanCostLimitExceededError) throw signal.reason; - const message = scanDir - ? `Codex Security scan was interrupted; partial output remains at ${scanDir}.` - : "Codex Security scan was interrupted during preparation."; - throw new ScanInterruptedError(message, scanDir, { cause: signal.reason }); + throw new ScanInterruptedError(scanInterruptionMessage(scanDir), scanDir, { + cause: signal.reason, + }); } function throwIfPatchAborted(signal?: AbortSignal, repository = ""): void { diff --git a/sdk/typescript/src/cli.ts b/sdk/typescript/src/cli.ts index 5ac6869d99..696c5935d0 100644 --- a/sdk/typescript/src/cli.ts +++ b/sdk/typescript/src/cli.ts @@ -2573,7 +2573,23 @@ export async function main( async run({ args, error: incurError, format, options }) { let scanArguments: ScanArguments; try { - const saved = await dependencies.runWorkbench( + const workbench = + dependencies.runWorkbench === DEFAULT_DEPENDENCIES.runWorkbench + ? await savedScanWorkbench( + { resumeId: args.scanId }, + { + environment: { + ...exportEnvironment(dependencies.environment), + CODEX_SECURITY_STATE_DIR: codexSecurityStateDirectory( + dependencies.environment, + ), + }, + pluginRoot: await bundledPluginRoot(), + currentDirectory: dependencies.currentDirectory(), + }, + ) + : dependencies.runWorkbench; + const saved = await workbench( workbenchArgs("get-cli-scan-resume", { scanId: args.scanId }), ); if ( @@ -9266,6 +9282,7 @@ async function executeScan( let repository = ""; let failed = false; let failure: unknown; + let incompleteBeforeCleanup: IncompleteScanError | undefined; try { deepScanOptions(arguments_); const directory = dependencies.currentDirectory(); @@ -9465,6 +9482,9 @@ async function executeScan( const options: ScanOptions = { ...pickScanSettings(arguments_), knowledgeBaseSnapshot: arguments_.knowledgeBaseSnapshot, + ...(arguments_.validate || arguments_.patch + ? { managedAttempt: true } + : {}), ...(arguments_.resumeScanId === undefined ? {} : { resumeScanId: arguments_.resumeScanId }), @@ -9929,6 +9949,12 @@ async function executeScan( } catch (error) { failed = true; failure = error; + if ( + requestedSignal === null && + arguments_.mode === "deep" && + error instanceof IncompleteScanError + ) + incompleteBeforeCleanup = error; } finally { stopPresentation(); if (security !== null) { @@ -9993,16 +10019,14 @@ async function executeScan( }, }; } - const exitCode = interruptedExit(requestedSignal, scanDir, errorOutput); + const outcome = interruptedOutcome( + requestedSignal, + scanDir, + errorOutput, + incompleteBeforeCleanup, + ); printFailedScanReference(registeredScan, phase, arguments_, errorOutput); - return { - exitCode, - data: failureData, - error: - requestedSignal === "SIGINT" - ? "Scan canceled by Ctrl-C." - : "Scan terminated by SIGTERM.", - }; + return { ...outcome, data: failureData }; } if (failed) { const costLimitFailure = @@ -11148,21 +11172,23 @@ export class Progress { } } -function interruptedExit( +function interruptedOutcome( signal: SignalName, scanDir: string | null, errorOutput: Writable, -): number { + incompleteBeforeCleanup: IncompleteScanError | undefined, +): Pick { const ctrlC = signal === "SIGINT"; - errorOutput.write( - `codex-security: Scan ${ctrlC ? "canceled by Ctrl-C" : "terminated by SIGTERM"}.\n`, - ); + const error = incompleteBeforeCleanup + ? `Scan cleanup interrupted by ${signal}.\n${diagnosticLines(incompleteBeforeCleanup)}` + : `Scan ${ctrlC ? "canceled by Ctrl-C" : "terminated by SIGTERM"}.`; + errorOutput.write(`codex-security: ${error}\n`); errorOutput.write( scanDir === null ? "codex-security: No partial output was kept.\n" : `codex-security: Partial output was kept at ${errorMessage(scanDir)}.\n`, ); - return ctrlC ? 130 : 143; + return { exitCode: ctrlC ? 130 : 143, error }; } function invokedAsMain(): boolean { diff --git a/sdk/typescript/src/component-scan.ts b/sdk/typescript/src/component-scan.ts index f6076d5787..4e6901455b 100644 --- a/sdk/typescript/src/component-scan.ts +++ b/sdk/typescript/src/component-scan.ts @@ -243,6 +243,7 @@ export async function runComponentScans( mode: options.scanOptions?.mode ?? "standard", target: receipt.paths.map((path) => resolve(repository, path)), outputDir: receipt.outputDir, + managedAttempt: true, signal: options.signal, }); if (result.manifest.scan.status !== "completed") diff --git a/sdk/typescript/src/config.ts b/sdk/typescript/src/config.ts index 167fb54ade..2047f699cd 100644 --- a/sdk/typescript/src/config.ts +++ b/sdk/typescript/src/config.ts @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { mkdir, open, rename, unlink } from "node:fs/promises"; import { dirname, join, resolve } from "node:path"; +import { isDeepStrictEqual } from "node:util"; import type { CyberAccessProgram } from "@openai/codex-sdk"; import { parse, stringify } from "smol-toml"; import { ConfigurationError } from "./errors.js"; @@ -443,6 +444,21 @@ export function mergeCodexOverrides( return deepMerge(structuredClone(base), overrides); } +/** @internal Saved tables may include defaults alongside explicit overrides. */ +export function containsCodexOverrides( + saved: JsonValue | undefined, + requested: JsonValue, +): boolean { + if (!isObject(requested)) return isDeepStrictEqual(saved, requested); + return ( + isObject(saved) && + Object.entries(requested).every( + ([key, value]) => + Object.hasOwn(saved, key) && containsCodexOverrides(saved[key], value), + ) + ); +} + /** @internal */ export function deepMerge(base: JsonObject, overrides: JsonObject): JsonObject { for (const [key, value] of Object.entries(overrides)) { diff --git a/sdk/typescript/src/cost.ts b/sdk/typescript/src/cost.ts index 23300da9f0..6905ad0cf3 100644 --- a/sdk/typescript/src/cost.ts +++ b/sdk/typescript/src/cost.ts @@ -159,6 +159,7 @@ export class ScanCostTracker { #pending: Promise = Promise.resolve(); #snapshot: ScanCostSnapshot = { usage: null, cost: null }; #lastCost: string | null = null; + #lastMeasuredUsage = new Map(); #lastUsage: string | null = null; #lastObservation: ScanUsageObservation | null = null; #highestFilesCompleted = 0; @@ -169,6 +170,20 @@ export class ScanCostTracker { this.#expectedFilesTotal = options.expectedFilesTotal; } + /** A measured subtotal can survive an unmeasured worker; it is not a scan total. */ + public get lastMeasuredCost(): Readonly | null { + // A final parent receipt can arrive after the last readable session poll. + const measured = new Map(this.#lastMeasuredUsage); + for (const [id, receipt] of this.#receipts) { + measured.set(id, threadUsage(measured.get(id), receipt)); + } + let usage: ScanTokenUsage | null = null; + for (const value of measured.values()) { + if (value !== null) usage = addTokenUsage(usage, value); + } + return estimateScanCost(this.#options.model, usage); + } + public setExpectedFilesTotal(filesTotal: number): void { this.#expectedFilesTotal = filesTotal; } @@ -358,19 +373,8 @@ export class ScanCostTracker { } this.#reportWorkerProgress(session); } - const receipt = usages.get(threadId); // A discovered thread without a receipt has unknown usage, not zero. - // Retain it so a parent-only subtotal cannot appear to be complete. - if (receipt === undefined) usages.set(threadId, null); - if ( - session.usage !== null && - (session.usage.total_tokens > (receipt?.total_tokens ?? -1) || - // The SDK inserts zero when the final receipt omits cache writes. - (session.usage.total_tokens === receipt?.total_tokens && - receipt.cache_write_input_tokens === 0)) - ) { - usages.set(threadId, session.usage); - } + usages.set(threadId, threadUsage(session.usage, usages.get(threadId))); } let usage: ScanTokenUsage | null = null; let complete = true; @@ -384,6 +388,7 @@ export class ScanCostTracker { if (observeUsage) this.#reportUsageObservation(usages, complete); if (!complete) this.#snapshot = { usage: null, cost: null }; if (usage === null) return; + this.#lastMeasuredUsage = usages; const cost = estimateScanCost(this.#options.model, usage); if (complete) { this.#snapshot = { usage, cost }; @@ -499,18 +504,23 @@ export class ScanCostTracker { export async function* sessionFiles( directory: string, compressed = false, + onReadError?: (error: unknown) => void, ): AsyncGenerator { let entries; try { entries = await readdir(directory, { withFileTypes: true }); } catch (error) { if (isMissingFile(error)) return; + if (onReadError) { + onReadError(error); + return; + } throw error; } for (const entry of entries) { const path = join(directory, entry.name); if (entry.isDirectory()) { - yield* sessionFiles(path, compressed); + yield* sessionFiles(path, compressed, onReadError); } else if ( entry.isFile() && (entry.name.endsWith(".jsonl") || @@ -904,6 +914,22 @@ function sessionProgressUpdates( return scanProgressUpdatesFromText(output); } +function threadUsage( + native: ScanTokenUsage | null | undefined, + receipt: ScanTokenUsage | null | undefined, +): ScanTokenUsage | null { + if ( + native && + (!receipt || + native.total_tokens > receipt.total_tokens || + // The SDK inserts zero when its receipt omits cache writes. + (native.total_tokens === receipt.total_tokens && + receipt.cache_write_input_tokens === 0)) + ) + return native; + return receipt ?? null; +} + function addTokenUsage( previous: ScanTokenUsage | null, next: ScanTokenUsage, diff --git a/sdk/typescript/src/multiscan.ts b/sdk/typescript/src/multiscan.ts index a9f8610b59..e3fde9d382 100644 --- a/sdk/typescript/src/multiscan.ts +++ b/sdk/typescript/src/multiscan.ts @@ -561,6 +561,7 @@ async function runCampaign( : {}), mode: task.mode, outputDir: scanDir, + managedAttempt: true, ...(scanPrompt ? { scanPrompt } : {}), ...(options.validationPrompt === undefined ? {} diff --git a/sdk/typescript/src/saved-scan-bootstrap.ts b/sdk/typescript/src/saved-scan-bootstrap.ts index d013cf8a87..98a2762f98 100644 --- a/sdk/typescript/src/saved-scan-bootstrap.ts +++ b/sdk/typescript/src/saved-scan-bootstrap.ts @@ -1,7 +1,7 @@ import { isRecord } from "./record.js"; import { compareTimestamps } from "./timestamps.js"; import { environmentEntry } from "./auth.js"; -import { inspectTrustedExecutable } from "./trusted-executable.js"; +import { inspectTrustedExecutable, isWithin } from "./trusted-executable.js"; import { realpath } from "node:fs/promises"; import { createRequire } from "node:module"; import { isAbsolute, join, resolve } from "node:path"; @@ -14,6 +14,7 @@ import { CodexSecurityError, errorMessage } from "./errors.js"; import type { SavedScanDependencies } from "./saved-scan.js"; import { codexSecurityStateDirectory, + environmentWithGit, resolvePluginPython, runWorkbench, workbenchEnvironment, @@ -24,6 +25,7 @@ interface ScanTarget { id: string; target_path: string; target_id?: string | null; + checkout_root?: string | null; } interface BootstrapDatabase { @@ -37,7 +39,7 @@ interface BootstrapDatabase { /** Read only enough history to protect Python discovery; the workbench still validates the scan. */ export async function savedScanWorkbench( - requestedId: string | { workflowId: string }, + selection: string | { workflowId: string } | { resumeId: string }, options: { environment: ProcessEnvironment; pluginRoot: string; @@ -51,7 +53,9 @@ export async function savedScanWorkbench( } > { options.signal?.throwIfAborted(); - const environment: ProcessEnvironment = workbenchEnvironment( + const resuming = typeof selection !== "string" && "resumeId" in selection; + const requestedId = resuming ? selection.resumeId : selection; + let environment: ProcessEnvironment = workbenchEnvironment( options.environment, ); const targets = await readTargets( @@ -59,8 +63,10 @@ export async function savedScanWorkbench( environment, options.currentDirectory, options.signal, + resuming, ); - const latest = requestedId === "latest" || typeof requestedId !== "string"; + const latest = + !resuming && (requestedId === "latest" || typeof requestedId !== "string"); if (targets.length === 0) throw new CodexSecurityError( typeof requestedId !== "string" @@ -99,6 +105,29 @@ export async function savedScanWorkbench( ) ).flat(), ]; + if (resuming) { + protectedRoots.unshift(options.currentDirectory); + // The original enclosing checkout can outlive the Git marker that identified it. + const originalRoot = targets[0]!.checkout_root; + if (originalRoot != null) { + if ( + typeof originalRoot !== "string" || + !isAbsolute(originalRoot) || + !isWithin(originalRoot, targets[0]!.target_path) + ) + throw new CodexSecurityError( + "Saved scan history has an invalid checkout root.", + ); + protectedRoots.push(originalRoot); + } + // Git starts configured helpers from PATH while verifying the original checkout. + const git = await inspectTrustedExecutable( + "git", + environment, + protectedRoots, + ); + environment = environmentWithGit(environment, git); + } let python: string | undefined; const workbench: SavedScanDependencies["runWorkbench"] = async ( args, @@ -106,6 +135,8 @@ export async function savedScanWorkbench( signal = options.signal, ) => { signal?.throwIfAborted(); + if (resuming && args[0] === "get-cli-scan-resume") + args = [args[0], "--scan-id", targets[0]!.id]; // Return the selected ID directly: probing Python must not revisit unrelated history. if (requestedId === "latest" && args[0] === "list-scans") return { scans: targets.map((target) => ({ scanId: target.id })) }; @@ -163,7 +194,9 @@ export async function savedScanWorkbench( }; return Object.assign(workbench, { environment, - protectedRoots: [options.currentDirectory, ...protectedRoots], + protectedRoots: resuming + ? protectedRoots + : [options.currentDirectory, ...protectedRoots], }); } @@ -172,6 +205,7 @@ async function readTargets( environment: ProcessEnvironment, currentDirectory: string, signal?: AbortSignal, + resuming = false, ): Promise { const require = createRequire(import.meta.url); const bun = process.versions["bun"] !== undefined; @@ -217,13 +251,20 @@ async function readTargets( requestedId.workflowId, ); } - if (requestedId === "latest") + if (!resuming && requestedId === "latest") return await latestTargets( database, currentDirectory, environment, signal, ); + const fields = + resuming && + readRows<{ name: string }>(database, "PRAGMA table_info(scans)").some( + (column) => column.name === "recipe_json", + ) + ? "id, target_path, json_extract(recipe_json, '$.checkoutRoot') AS checkout_root" + : "id, target_path"; // Match uuid.UUID's accepted full-ID spellings before considering a prefix. const compact = trimBoundary( requestedId.replaceAll("urn:", "").replaceAll("uuid:", ""), @@ -236,7 +277,7 @@ async function readTargets( .replace(/^(.{8})(.{4})(.{4})(.{4})(.{12})$/, "$1-$2-$3-$4-$5"); return readRows( database, - "SELECT id, target_path FROM scans WHERE id = ?", + `SELECT ${fields} FROM scans WHERE id = ?`, id, ); } @@ -246,7 +287,7 @@ async function readTargets( ); return readRows( database, - "SELECT id, target_path FROM scans WHERE substr(id, 1, ?) = ? LIMIT 2", + `SELECT ${fields} FROM scans WHERE substr(id, 1, ?) = ? LIMIT 2`, requestedId.length, requestedId.toLowerCase(), ); diff --git a/sdk/typescript/src/scan-logs.ts b/sdk/typescript/src/scan-logs.ts index 33a284760a..5f5ebd1608 100644 --- a/sdk/typescript/src/scan-logs.ts +++ b/sdk/typescript/src/scan-logs.ts @@ -72,12 +72,17 @@ const readsCompressedSessions = typeof zlib.createZstdDecompress === "function"; type SessionLog = NonNullable>>; -async function* scanSessions(codexHome: string, directory = "sessions") { +async function* scanSessions( + codexHome: string, + directory = "sessions", + onReadError?: (error: unknown) => void, +) { for await (const path of sessionFiles( join(codexHome, directory), readsCompressedSessions, + onReadError, )) { - for await (const first of sessionEvents(path)) { + for await (const first of sessionEvents(path, onReadError)) { if (first["type"] !== "session_meta" || !isRecord(first["payload"])) { break; } @@ -98,9 +103,13 @@ async function* scanSessions(codexHome: string, directory = "sessions") { } export async function findScanSession(codexHome: string, threadId: string) { - for await (const session of scanSessions(codexHome)) { + let readError: unknown; + for await (const session of scanSessions(codexHome, "sessions", (error) => { + readError ??= error; + })) { if (session.threadId === threadId) return session; } + if (readError !== undefined) throw readError; return null; } @@ -261,6 +270,7 @@ async function extendsSessionLog( async function* sessionEvents( path: string, + onReadError?: (error: unknown) => void, ): AsyncGenerator> { const file = createReadStream(path); const stream = path.endsWith(".zst") @@ -278,7 +288,10 @@ async function* sessionEvents( } } } catch (error) { - if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + if (!onReadError) throw error; + onReadError(error); + } } finally { lines.close(); stream.destroy(); diff --git a/sdk/typescript/tests-ts/api-preflight-config.test.ts b/sdk/typescript/tests-ts/api-preflight-config.test.ts index 7d1d19392b..14562746b2 100644 --- a/sdk/typescript/tests-ts/api-preflight-config.test.ts +++ b/sdk/typescript/tests-ts/api-preflight-config.test.ts @@ -10,6 +10,7 @@ import { CHEAPERINFERENCE_CODEX_PROVIDER, FIREWORKS_CODEX_PROVIDER, OPENROUTER_CODEX_PROVIDER, + containsCodexOverrides, scanModelProvider, writeCodexConfig, type JsonObject, @@ -504,6 +505,31 @@ describe("CodexSecurity preflight configuration", () => { ).resolves.toBeUndefined(); }); + test("resume rejects private settings and requires whole arrays to match", () => { + const skills = { config: [{ path: "review", enabled: true }] }; + const saved = scanPreflightCodexConfig({ + model_provider: "amazon-bedrock", + model_providers: { "amazon-bedrock": { aws: { profile: "review" } } }, + skills, + }); + expect(containsCodexOverrides(saved, { skills })).toBe(true); + const omitted: JsonObject[] = [ + { skills: { config: [{ path: "review" }] } }, + { + skills: { + config: [{ path: "review", enabled: true, command: "helper" }], + }, + }, + { openai_base_url: "fixture" }, + { + model_providers: { "amazon-bedrock": { auth: { command: "helper" } } }, + }, + { sandbox_workspace_write: { network_access: true } }, + ]; + for (const requested of omitted) + expect(containsCodexOverrides(saved, requested)).toBe(false); + }); + test("keeps every valid profile, project, and root marker", () => { const activeProject = "/workspace/active"; const profiles = Object.fromEntries([ diff --git a/sdk/typescript/tests-ts/cli-scan-failures.test.ts b/sdk/typescript/tests-ts/cli-scan-failures.test.ts index ba66831e88..f03bf7bfba 100644 --- a/sdk/typescript/tests-ts/cli-scan-failures.test.ts +++ b/sdk/typescript/tests-ts/cli-scan-failures.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test"; import { main } from "../src/cli.js"; -import { CodexSecurityError } from "../src/index.js"; +import { CodexSecurityError, IncompleteScanError } from "../src/index.js"; import { FakeSignals, capture, @@ -250,6 +250,38 @@ test.each(commands)( }, ); +test.each([ + ["SIGINT", 130], + ["SIGTERM", 143], +] as const)( + "keeps the Deep Scan handoff when %s arrives during cleanup", + async (signal, code) => { + const stdout = capture(); + const stderr = capture(); + const signals = new FakeSignals(); + const handoff = + "Parent disconnected.\nRun codex-security scans resume saved-scan."; + const deps = dependencies({ + signals, + onRun: () => { + throw new IncompleteScanError(handoff); + }, + onClose: () => signals.emit(signal), + }); + expect( + await main( + ["scan", ".", "--mode", "deep", "--json"], + stdout.stream, + stderr.stream, + deps, + ), + ).toBe(code); + const message = `Scan cleanup interrupted by ${signal}.\n${handoff}`; + expect(JSON.parse(stdout.text()).message).toBe(message); + expect(stderr.text()).toContain(message); + }, +); + test("retains diagnostics when the latest-scan working directory is unavailable", async () => { for (const flags of [[], ["--json"], ["--json", "--full-output"]]) { const stdout = capture(); diff --git a/sdk/typescript/tests-ts/cost.test.ts b/sdk/typescript/tests-ts/cost.test.ts index 12efd2c7d3..cb0cf39a1c 100644 --- a/sdk/typescript/tests-ts/cost.test.ts +++ b/sdk/typescript/tests-ts/cost.test.ts @@ -639,6 +639,18 @@ describe("live scan cost tracking", () => { }); }); + test("retains the latest parent receipt alongside measured worker usage", async () => { + const { home } = await workerSessionFixture(); + const tracker = costTracker(home); + tracker.start("scan-thread"); + await tracker.stop(); + tracker.recordUsage({ input_tokens: 120, output_tokens: 15 }); + expect(tracker.lastMeasuredCost).toMatchObject({ + inputTokens: 220, + outputTokens: 25, + }); + }); + test("keeps freshness with the last reportable token total", async () => { const home = await codexHome(); const parent = await writeSession(home, scanThreadId, {}); diff --git a/sdk/typescript/tests-ts/saved-scan-bootstrap.test.ts b/sdk/typescript/tests-ts/saved-scan-bootstrap.test.ts index 67e9582663..42a1559802 100644 --- a/sdk/typescript/tests-ts/saved-scan-bootstrap.test.ts +++ b/sdk/typescript/tests-ts/saved-scan-bootstrap.test.ts @@ -7,10 +7,11 @@ import { mkdir, readFile, realpath, + rename, symlink, writeFile, } from "node:fs/promises"; -import { delimiter, dirname, join } from "node:path"; +import { basename, delimiter, dirname, join } from "node:path"; import { afterEach, expect, mock, test } from "bun:test"; import { deduplicateScanInternal } from "../src/deduplication/scan.js"; import { savedScanWorkbench } from "../src/saved-scan-bootstrap.js"; @@ -33,7 +34,7 @@ const { temporaryDirectory, cleanup } = createApiTestFixtures( ); afterEach(cleanup); -async function fixture(git = false, separateSetupCommands = false) { +async function repositoryFixture(git: boolean) { const root = await temporaryDirectory(); const repository = join(root, "repository"); await mkdir(join(repository, "src"), { recursive: true }); @@ -71,6 +72,12 @@ async function fixture(git = false, separateSetupCommands = false) { CODEX_SECURITY_STATE_DIR: join(root, "state"), }; const python = await resolvePluginPython({ environment }); + return { root, repository, environment, python }; +} + +async function fixture(git = false, separateSetupCommands = false) { + const { root, repository, environment, python } = + await repositoryFixture(git); const options = { environment, python, pluginRoot: PLUGIN_ROOT }; const registrationArguments = (scanDir: string) => [ "register-cli-scan", @@ -901,6 +908,121 @@ test.skipIf(process.platform === "win32")( }, ); +async function registeredNestedDeepScan() { + const f = await repositoryFixture(true); + const target = join(f.repository, "src"); + const scanDir = join(f.root, "deep"); + await mkdir(scanDir, { mode: 0o700 }); + const workbench = (args: string[]) => + runWorkbench({ ...f, pluginRoot: PLUGIN_ROOT }, args); + const registered = await workbench([ + "register-cli-scan", + "--repository", + target, + "--scan-dir", + scanDir, + "--recipe-json", + JSON.stringify({ + repository: target, + checkoutRoot: f.repository, + mode: "deep", + config: {}, + target: { kind: "repository", paths: [] }, + }), + ]); + const scanId = registered["scanId"] as string; + await workbench([ + "set-scan-thread", + "--scan-id", + scanId, + "--thread-id", + "owner", + ]); + return { ...f, target, scanId }; +} + +test.skipIf(process.platform === "win32")( + "nested Deep resume protects the recorded checkout after its Git marker disappears", + async () => { + const f = await registeredNestedDeepScan(); + await rename(join(f.repository, ".git"), join(f.root, "original-git")); + const bin = join(f.repository, "bin"); + const marker = join(f.root, "checkout-tool-ran"); + const python = basename(f.python); + await mkdir(bin); + for (const tool of ["git", python]) + await writeFile( + join(bin, tool), + '#!/bin/sh\necho "$0" >> "$TEST_PROBE"\nexit 1\n', + { mode: 0o700 }, + ); + const workbench = await savedScanWorkbench( + { resumeId: f.scanId }, + { + currentDirectory: await temporaryDirectory(), + pluginRoot: PLUGIN_ROOT, + environment: { + ...f.environment, + PATH: [bin, dirname(f.python), f.environment.PATH].join(delimiter), + PYTHON: python, + TEST_PROBE: marker, + }, + }, + ); + const result = await workbench([ + "get-cli-scan-resume", + "--scan-id", + f.scanId, + ]).catch((error: unknown) => error); + expect(existsSync(marker)).toBe(false); + expect(workbench.environment["PATH"]?.split(delimiter)).not.toContain(bin); + expect(result).toMatchObject({ + message: expect.stringContaining( + "original checkout revision or contents changed", + ), + }); + }, +); + +test.skipIf(process.platform === "win32")( + "scans resume rejects a relative Python interpreter in the caller", + async () => { + const f = await registeredNestedDeepScan(); + const caller = await temporaryDirectory(); + const python = ".venv/bin/python"; + const interpreter = join(caller, python); + await mkdir(dirname(interpreter), { recursive: true }); + await symlink(f.python, join(dirname(interpreter), "host-python")); + await writeFile( + interpreter, + '#!/bin/sh\necho "$@" >> "$0.calls"\nexec "${0%/*}/host-python" "$@"\n', + { mode: 0o700 }, + ); + await writeFile( + join(f.target, "extract.py"), + "# Changed since registration\n", + ); + const result = spawnSync( + process.execPath, + [join(import.meta.dir, "../src/cli.ts"), "scans", "resume", f.scanId], + { + cwd: caller, + encoding: "utf8", + env: { + ...f.environment, + PYTHON: python, + CODEX_SECURITY_NO_UPDATE_CHECK: "1", + }, + }, + ); + expect(existsSync(interpreter + ".calls")).toBe(false); + expect(result.status).toBe(2); + expect(result.stderr).toContain( + "PYTHON interpreter is unavailable or unusable", + ); + }, +); + for (const selector of ["latest", "workflow"] as const) { test(`bootstrap ${selector} ignores unrelated history enclosing trusted Python`, async () => { const f = await fixture(true); diff --git a/sdk/typescript/tests-ts/scan-resume.test.ts b/sdk/typescript/tests-ts/scan-resume.test.ts index 020357bfbe..e1fc7cd8cf 100644 --- a/sdk/typescript/tests-ts/scan-resume.test.ts +++ b/sdk/typescript/tests-ts/scan-resume.test.ts @@ -1,7 +1,8 @@ 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 { gitText, runCommand } from "./support/shell.js"; +import { git as fixtureGit } from "./git-fixture.js"; import { readJsonLines } from "./support/json.js"; import { randomUUID } from "node:crypto"; import { stripVTControlCharacters } from "node:util"; @@ -15,13 +16,15 @@ import { rm, writeFile, } from "node:fs/promises"; -import { join } from "node:path"; -import { afterEach, expect, test } from "bun:test"; +import { delimiter, join } from "node:path"; +import { afterEach, expect, spyOn, test } from "bun:test"; import { parse as parseToml } from "smol-toml"; import { main } from "../src/cli.js"; import { scanPreflightCodexConfig, type ScanOptions } from "../src/api.js"; +import { ScanCostTracker } from "../src/cost.js"; import type { JsonObject } from "../src/config.js"; import { runWorkbench } from "../src/runtime.js"; +import { workbenchArgs } from "../src/workbench-arguments.js"; import { readKnowledgeBaseSnapshot } from "../src/knowledge-base.js"; import { saveScanKnowledge, scanInputIdentity } from "../src/scan-inputs.js"; import { workflowDigest } from "../src/finding-workflow.js"; @@ -313,6 +316,68 @@ test("resume resolves an interrupted scan without changing its ID, recipe, or co expect(await readFile(f.checkpoint, "utf8")).toBe('{"completed":"setup"}\n'); }); +test("bare resume checks changed files with the host Git clean filter", async () => { + const f = await interruptedScan("deep", true); + const filter = "codex-test-clean"; + const checkoutBin = join(f.repository, "bin"); + const hostBin = join(f.root, "host-bin"); + const caller = join(f.root, "caller"); + for (const [label, bin] of [ + ["checkout", checkoutBin], + ["caller", caller], + ["host", hostBin], + ] as const) { + await mkdir(bin); + await writeFile( + join(bin, filter), + `#!/bin/sh\necho ${label} >> .git/clean-calls\ncat\n`, + { mode: 0o755 }, + ); + } + gitText([ + "-C", + f.repository, + "config", + "--local", + "filter.fixture.clean", + filter, + ]); + await writeFile( + join(f.repository, ".gitattributes"), + "source.py filter=fixture\n", + ); + await appendFile(join(f.repository, "source.py"), "# changed\n"); + const result = await runCommand( + process.execPath, + [ + join(import.meta.dir, "../src/cli.ts"), + "scans", + "resume", + f.scanId.slice(0, 8), + ], + { + cwd: caller, + env: { + ...f.environment, + PATH: [caller, checkoutBin, hostBin, f.environment.PATH].join( + delimiter, + ), + PYTHON: f.python, + CODEX_SECURITY_NO_UPDATE_CHECK: "1", + }, + }, + ); + expect(result.status).toBe(2); + expect(result.stderr).toContain( + "original checkout revision or contents changed", + ); + const calls = await readFile( + join(f.repository, ".git", "clean-calls"), + "utf8", + ); + expect(new Set(calls.trim().split(/\s+/u))).toEqual(new Set(["host"])); +}); + test.each([ "failed", "canceled", @@ -571,6 +636,7 @@ function resumeDependencies( ConstructorParameters[1]["createCodex"] >, runtimeOverrides: Partial> = {}, + workbench = runWorkbench, ) { return { ...dependencies({ environment: f.environment, currentDirectory: f.root }), @@ -598,7 +664,7 @@ function resumeDependencies( return runtime; }, resolvePluginPython: async () => f.python, - runWorkbench, + runWorkbench: workbench, createCodex, }), }; @@ -648,6 +714,427 @@ async function finishDiscovery(f: Awaited>) { ]); } +test("resumes the original Deep scan after its parent returns during consolidation", () => + parentReturnsDuringConsolidation()); +test("resumes the original Deep scan when the process exits after turn.completed", () => + parentReturnsDuringConsolidation("signal")); +test("resumes the original Deep scan when Codex reports a closed WebSocket", () => + parentReturnsDuringConsolidation("websocket-close")); +test.each([ + [ + "a provider policy rejection", + "stream disconnected before completion: Network access is disabled by policy", + ], + [ + "an invalid network policy", + '{"error":{"code":"invalid_request_error","message":"Unsupported network policy"}}', + ], + [ + "HTTP 422 with timeout wording", + "unexpected status 422 Unprocessable Entity: Network request timed out", + ], +])("keeps a parent terminal after %s", (_kind, message) => + parentReturnsDuringConsolidation(undefined, message), +); +test("resumes a nested Git target with knowledge in the scanned directory", () => + parentReturnsDuringConsolidation("nested-knowledge")); +test("does not offer bare resume with knowledge from a separate checkout", () => + parentReturnsDuringConsolidation("external-knowledge")); +test.each(["fail-scan", "get-scan"] as const)( + "honors an explicit abort after the handoff's %s returns", + (transition) => parentReturnsDuringConsolidation(transition), +); + +test("a peer resume can seal a preserved scan", () => + parentReturnsDuringConsolidation("peer-resume")); +test("runs the requested post-scan prompt once after a failed Deep scan", () => + parentReturnsDuringConsolidation("post-scan")); +test.each(["--validate", "--patch"] as const)( + "retains the terminal Deep Scan outcome when the CLI owns %s", + (action) => parentReturnsDuringConsolidation(action), +); + +async function parentReturnsDuringConsolidation( + transition?: + | "signal" + | "websocket-close" + | "fail-scan" + | "get-scan" + | "peer-resume" + | "nested-knowledge" + | "external-knowledge" + | "--validate" + | "--patch" + | "post-scan", + rejection?: string, +) { + const cancellation = new AbortController(); + const f = await interruptedScan("standard"); + const { root, codexHome, python, environment, command } = f; + const nested = transition === "nested-knowledge"; + const repository = nested ? join(f.repository, "service") : f.repository; + const userSkill = { path: join(root, "custom", "SKILL.md"), enabled: false }; + const installedRoot = join(codexHome, "plugins", "cache", "codex-security"); + const extraArgs: string[] = []; + const cliFollowup = transition === "--validate" || transition === "--patch"; + if (cliFollowup) extraArgs.push(transition); + if (nested || transition === "external-knowledge") { + const checkout = nested ? f.repository : await temporaryDirectory(); + fixtureGit(checkout, "init", "-q"); + if (nested) { + await mkdir(repository); + await writeFile(join(repository, "source.py"), "# synthetic service\n"); + } + const document = join(nested ? repository : checkout, "architecture.md"); + await writeFile(document, "Synthetic architecture.\n"); + if (nested) { + fixtureGit(checkout, "add", "."); + fixtureGit(checkout, "commit", "--no-gpg-sign", "-qm", "Add service"); + } + extraArgs.push("--knowledge-base", document); + } + const scanDir = join(root, "fresh-deep-scan"); + const threadId = randomUUID(); + const workers = Array.from({ length: 3 }, () => randomUUID()); + const reducer = randomUUID(); + const promptPath = join(scanDir, "worker.md"); + const resultManifestPath = join(scanDir, "worker-result.json"); + const postScanPrompt = "Report only when the scan is terminal."; + if (transition === "post-scan") { + const promptFile = join(root, "post-scan.md"); + await writeFile(promptFile, postScanPrompt); + extraArgs.push("--post-scan-prompt-file", promptFile); + } + let scanId = ""; + let parentTurns = 0; + let resumeTurns = 0; + const followupStates: string[] = []; + let registration: JsonObject = {}; + async function* startDiscovery() { + // Only session identity is known; this fixture has no cumulative token receipt. + await writeFile( + join(codexHome, "sessions", "rollout.jsonl"), + JSON.stringify({ + type: "session_meta", + payload: { id: threadId, cwd: scanDir }, + }) + "\n", + ); + await writeFile(promptPath, "Review the synthetic source.\n"); + await writeFile(resultManifestPath, "{}\n"); + const paths = { scanId, promptPath, artifactDir: scanDir }; + const worker = (workerId: string, status: string, kind = "discovery") => + workbenchArgs("upsert-deep-scan-worker", { + ...paths, + workerId, + kind, + status, + resultManifestPath: + status === "succeeded" ? resultManifestPath : undefined, + attempt: kind === "dedup" ? 1 : undefined, + }); + const run = { scanId, threadId, availableParallelism: 4 }; + const claim = { ...paths, workerId: reducer, inputWorkerId: workers }; + await runWorkbenchFixtureCommands(python, environment, [ + workbenchArgs("begin-deep-scan", run), + ...workers.flatMap((id) => [ + worker(id, "running"), + worker(id, "succeeded"), + ]), + workbenchArgs("claim-deep-scan-dedup", claim), + worker(reducer, "running", "dedup"), + ]); + const message = + transition === "websocket-close" + ? "stream disconnected before completion: websocket closed" + : rejection; + if (message) yield { type: "turn.failed" as const, error: { message } }; + } + async function* parentEvents() { + yield* completedEvents(threadId, startDiscovery()); + if (transition === "signal") + throw new Error("Codex Exec exited with signal SIGKILL"); + } + const createCodex: Parameters[1] = (options) => ({ + startThread: () => ({ + id: threadId, + async runStreamed(prompt) { + scanId = + options.env?.["CODEX_SECURITY_SCAN_ID"] ?? fail("Expected scan ID"); + parentTurns++; + if (prompt === postScanPrompt) { + const { scan } = await command(workbenchArgs("get-scan", { scanId })); + followupStates.push( + (scan as { progress: { status: string } }).progress.status, + ); + return { events: completedEvents(threadId) }; + } + return { events: parentEvents() }; + }, + }), + resumeThread(id) { + expect(id).toBe(threadId); + expect(options.config?.["features"]).toMatchObject({ + multi_agent_v2: { + enabled: true, + max_concurrent_threads_per_session: 4, + }, + }); + expect(options.config?.["skills"]).toMatchObject({ + config: expect.arrayContaining([ + userSkill, + { + path: join(installedRoot, "skills", "security-scan", "SKILL.md"), + enabled: false, + }, + ]), + }); + return { + id, + async runStreamed() { + resumeTurns++; + if (nested) { + const staged = + options.env?.["CODEX_SECURITY_KNOWLEDGE_BASE"] ?? + fail("Expected saved knowledge"); + expect( + await readFile(join(staged, "0-architecture.md.txt"), "utf8"), + ).toBe("Synthetic architecture.\n"); + } + await command( + workbenchArgs("commit-deep-scan-dedup", { + scanId, + workerId: reducer, + resultManifestPath, + newFindingsCount: 0, + }), + ); + await finishDiscovery({ ...f, scanId, scanDir, registration }); + return { events: completedEvents(id) }; + }, + }; + }, + }); + let preserved = false; + const deps = resumeDependencies( + f, + createCodex, + { plugin: { ...preparedRuntime(codexHome).plugin, installedRoot } }, + async (options, args, input) => { + if ( + preserved && + transition === "peer-resume" && + args[0] === "get-scan" && + options.signal === undefined + ) { + preserved = false; + await resume(); + } + const result = await runWorkbench(options, args, input); + if (result["preserved"] === true) preserved = true; + if (preserved && args[0] === transition) cancellation.abort(); + return result; + }, + ); + if (transition === "fail-scan" || transition === "get-scan") { + await using client = deps.createSecurity({}); + await expect( + client.run(repository, { + mode: "deep", + auth: "chatgpt", + outputDir: scanDir, + signal: cancellation.signal, + }), + ).rejects.toThrow("interrupted"); + expect(cancellation.signal.aborted).toBe(true); + expect(await command(workbenchArgs("get-scan", { scanId }))).toMatchObject({ + scan: { + progress: { status: "failed" }, + handoffCost: { inputTokens: 10, outputTokens: 3 }, + }, + }); + expect( + await command(workbenchArgs("get-deep-scan", { scanId, threadId })), + ).toMatchObject({ + deepScan: { status: "failed" }, + }); + const retry = createCliTest(main); + expect( + await retry.runCli(["scans", "resume", scanId, "--json"], deps), + ).toBe(2); + expect(parentTurns).toBe(1); + expect(resumeTurns).toBe(0); + return; + } + const first = createCliTest(main); + const scan = ["scan", repository, "--mode", "deep", "--auth", "chatgpt"]; + using unreadableArchive = + transition === undefined && rejection === undefined + ? spyOn(ScanCostTracker.prototype, "refresh").mockRejectedValue( + new Error("Session archive unavailable"), + ) + : undefined; + expect( + await first.runCli( + [ + ...scan, + ...extraArgs, + "--codex", + "features.multi_agent_v2.max_concurrent_threads_per_session=4", + "--codex", + `skills.config=[{path=${JSON.stringify(userSkill.path)},enabled=false}]`, + "--output-dir", + scanDir, + "--json", + ], + deps, + ), + first.stderr.text(), + ).toBe(2); + unreadableArchive?.mockRestore(); + const message = JSON.parse(first.stdout.text()).message; + const saved = await command(workbenchArgs("get-scan", { scanId })); + if (rejection) { + expect(message).not.toContain("scans resume"); + expect(saved["scan"]).toMatchObject({ + progress: { status: "failed" }, + failureMessage: rejection, + }); + expect(resumeTurns).toBe(0); + return; + } + if (transition === "post-scan") { + expect(followupStates).toEqual(["failed"]); + expect(parentTurns).toBe(2); + expect(resumeTurns).toBe(0); + } + if ( + cliFollowup || + transition === "external-knowledge" || + transition === "post-scan" + ) { + expect(message).toContain("Deep Scan discovery orchestration must finish"); + expect(message).not.toContain("scans resume"); + expect(saved["scan"]).toMatchObject({ + progress: { status: "failed" }, + failureMessage: message, + }); + if (cliFollowup) { + expect(first.stderr.text()).toContain(`scans show ${scanId}`); + expect(first.stderr.text()).not.toMatch( + /codex-security (?:validate |patch --scan)/u, + ); + const retry = createCliTest(main); + expect(await retry.runCli(["scans", "resume", scanId], deps)).toBe(2); + expect(parentTurns).toBe(1); + expect(resumeTurns).toBe(0); + } + return; + } + if (transition === "peer-resume") { + expect(message).not.toContain("scans resume"); + expect(saved["scan"]).toMatchObject({ progress: { status: "complete" } }); + expect(saved["scan"]).not.toHaveProperty("cost"); + expect(parentTurns).toBe(1); + expect(resumeTurns).toBe(1); + return; + } + expect({ + message, + parentTurns, + scan: saved["scan"], + }).toMatchObject({ + message: expect.stringContaining("codex-security scans resume " + scanId), + parentTurns: 1, + scan: { + progress: { + status: "running", + independentReviews: { completed: 3, consolidating: true }, + }, + }, + }); + await resume(); + expect(resumeTurns).toBe(1); + expect(parentTurns).toBe(1); + + async function resume() { + registration = await command( + workbenchArgs("get-cli-scan-resume", { scanId }), + ); + expect(registration).toMatchObject({ + scanId, + threadId, + recipe: { repository }, + }); + if (nested) + expect(registration["recipe"]).toMatchObject({ + checkoutRoot: f.repository, + }); + if (transition === undefined || transition === "signal") + expect(registration).toMatchObject({ + handoffCost: { inputTokens: 10, outputTokens: 3 }, + }); + const { stdout, stderr, runCli } = createCliTest(main); + expect( + await runCli( + ["scans", "resume", scanId, "--json"], + resumeDependencies(f, createCodex), + ), + stderr.text(), + ).toBe(0); + expect(JSON.parse(stdout.text())).toMatchObject({ + manifest: { scan: { id: scanId } }, + cost: null, + }); + } +} + +test("a resume cannot return a scan total when the parent hands off after attachment", () => + verifyConcurrentHandoff("gpt-5.6-sol")); +test("an unpriced model withholds turn-only tokens after a concurrent handoff", () => + verifyConcurrentHandoff("unpriced-test-model")); + +async function verifyConcurrentHandoff(model: string) { + const f = await interruptedScan("deep", false, { config: { model } }); + const cli = createCliTest(main); + const deps = resumeDependencies(f, () => ({ + startThread: () => fail("Expected the original session"), + resumeThread(id) { + return { + id, + async runStreamed() { + // The SDK has read the running scan, but its original owner now detaches. + const handoff = await f.command([ + "fail-scan", + "--scan-id", + f.scanId, + "--message", + "Parent disconnected", + "--preserve-deep-owner", + id, + ]); + expect(handoff).toHaveProperty("preserved", true); + await finishDiscovery(f); + return { events: completedEvents(id) }; + }, + }; + }, + })); + const exit = await cli.runCli( + ["scans", "resume", f.scanId, "--json", "--show-cost"], + deps, + ); + expect(exit, cli.stderr.text()).toBe(0); + expect(cli.stderr.text()).not.toContain("13 total"); + expect(cli.stderr.text()).not.toMatch(/\$[0-9]/); + expect( + (await f.command(workbenchArgs("get-scan", { scanId: f.scanId })))["scan"], + ).not.toHaveProperty("cost"); + expect(JSON.parse(cli.stdout.text())).toMatchObject({ + cost: null, + turn: { usage: null }, + }); +} + test.each([false, true])( "resumed CLI starts with terminal Deep progress (interactive=%p)", async (interactive) => { @@ -723,7 +1210,7 @@ test.each([ const code = await runCli( bulk ? ["bulk-scan", f.input, "--output-dir", f.root, "--recover", "--json"] - : ["scans", "resume", f.scanId, "--json"], + : ["scans", "resume", f.scanId, "--json", "--show-cost"], resumeDependencies(f, () => ({ startThread: () => fail("Unexpected new session"), resumeThread(threadId) { @@ -795,6 +1282,7 @@ test.each([ failed: 0, }); } else { + expect(stderr.text()).toContain("Estimated cost: $"); expect(result.coverage).not.toHaveProperty("completeness"); expect(result.manifest.scan.id).toBe(f.scanId); expect(result.manifest.scan.sealedAt).toBeString();