diff --git a/sdk/typescript/_bundled_plugin/mcp/mcp-app.html.br b/sdk/typescript/_bundled_plugin/mcp/mcp-app.html.br index 81469f4a..ac39fbbb 100644 Binary files a/sdk/typescript/_bundled_plugin/mcp/mcp-app.html.br and b/sdk/typescript/_bundled_plugin/mcp/mcp-app.html.br differ diff --git a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 index 2b67e55d..2e775f2a 100644 Binary files a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 and b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-000 differ diff --git a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 index 15ef6e8b..f50bd401 100644 Binary files a/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 and b/sdk/typescript/_bundled_plugin/mcp/server.mjs.br.part-001 differ diff --git a/sdk/typescript/_bundled_plugin/scripts/deep_scan_workbench.py b/sdk/typescript/_bundled_plugin/scripts/deep_scan_workbench.py index 779a28b4..41d52228 100644 --- a/sdk/typescript/_bundled_plugin/scripts/deep_scan_workbench.py +++ b/sdk/typescript/_bundled_plugin/scripts/deep_scan_workbench.py @@ -3,12 +3,14 @@ from __future__ import annotations import argparse +import json import os import sqlite3 import sys import tempfile import uuid from dataclasses import dataclass +from datetime import datetime, timedelta from pathlib import Path from typing import Any, Callable @@ -33,6 +35,8 @@ ) DEEP_SCAN_TERMINAL_REASONS = ("saturated", "capped") DEEP_SCAN_WORKFLOW_VERSION = "deep-security-scan/v1" +DEEP_SCAN_COORDINATOR_LEASE_SECONDS = 30 +DEEP_SCAN_LEGACY_COORDINATOR_GRACE_SECONDS = 120 def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) -> None: @@ -54,6 +58,12 @@ def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) -> get_deep_scan.add_argument("--scan-id", required=True) get_deep_scan.add_argument("--thread-id", required=True) + claim_coordinator = subparsers.add_parser("claim-deep-scan-coordinator") + claim_coordinator.add_argument("--scan-id", required=True) + claim_coordinator.add_argument("--thread-id", required=True) + claim_coordinator.add_argument("--claim-token") + claim_coordinator.add_argument("--coordinator-generation", type=positive_int) + upsert_deep_worker = subparsers.add_parser("upsert-deep-scan-worker") upsert_deep_worker.add_argument("--scan-id", required=True) upsert_deep_worker.add_argument("--worker-id", required=True) @@ -68,6 +78,7 @@ def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) -> upsert_deep_worker.add_argument( "--replaceable-failure-kind", choices=DEEP_SCAN_REPLACEABLE_FAILURE_KINDS ) + upsert_deep_worker.add_argument("--coordinator-generation", type=positive_int) claim_deep_dedup = subparsers.add_parser("claim-deep-scan-dedup") claim_deep_dedup.add_argument("--scan-id", required=True) @@ -75,12 +86,15 @@ def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) -> claim_deep_dedup.add_argument("--prompt-path", required=True) claim_deep_dedup.add_argument("--artifact-dir", required=True) claim_deep_dedup.add_argument("--input-worker-id", action="append", required=True) + claim_deep_dedup.add_argument("--coordinator-generation", type=positive_int) commit_deep_dedup = subparsers.add_parser("commit-deep-scan-dedup") commit_deep_dedup.add_argument("--scan-id", required=True) commit_deep_dedup.add_argument("--worker-id", required=True) commit_deep_dedup.add_argument("--result-manifest-path", required=True) + commit_deep_dedup.add_argument("--candidate-ledger-path") commit_deep_dedup.add_argument("--new-findings-count", type=non_negative_int, required=True) + commit_deep_dedup.add_argument("--coordinator-generation", type=positive_int) finish_deep_scan = subparsers.add_parser("finish-deep-scan") finish_deep_scan.add_argument("--scan-id", required=True) @@ -88,15 +102,19 @@ def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) -> "--terminal-reason", choices=DEEP_SCAN_TERMINAL_REASONS, required=True ) finish_deep_scan.add_argument("--manifest-path", required=True) + finish_deep_scan.add_argument("--staged-manifest-path") finish_deep_scan.add_argument("--omitted-worker-id", action="append", default=[]) + finish_deep_scan.add_argument("--coordinator-generation", type=positive_int) fail_deep_scan = subparsers.add_parser("fail-deep-scan") fail_deep_scan.add_argument("--scan-id", required=True) fail_deep_scan.add_argument("--message", required=True) fail_deep_scan.add_argument("--manifest-path") + fail_deep_scan.add_argument("--staged-manifest-path") fail_deep_scan.add_argument( "--deep-status", choices=("failed", "interrupted"), default="failed" ) + fail_deep_scan.add_argument("--coordinator-generation", type=positive_int) def non_negative_int(value: str) -> int: @@ -120,6 +138,7 @@ class DeepScanDependencies: require_canonical_scan_directory: Callable[[Path], Path] safe_segment: Callable[[str], str] compact_timestamp: Callable[[], str] + scan_completion_lock: Callable[[str], Any] _dependencies: DeepScanDependencies | None = None @@ -184,6 +203,10 @@ def compact_timestamp() -> str: return dependencies().compact_timestamp() +def scan_completion_lock(scan_id: str) -> Any: + return dependencies().scan_completion_lock(scan_id) + + def require_deep_scan_run(connection: sqlite3.Connection, scan_id: str) -> sqlite3.Row: scan_id = require_uuid(scan_id, "scan-id") row = connection.execute( @@ -249,6 +272,50 @@ def deep_scan_path( return str(resolved) +def deep_scan_output_path(scan: sqlite3.Row, value: str, label: str) -> str: + supplied = Path(value).expanduser() + if not supplied.is_absolute(): + raise SystemExit(f"{label} must be an absolute path inside the scan directory.") + if supplied.exists(): + return deep_scan_path(scan, str(supplied), label, kind="file") + parent = Path(deep_scan_path(scan, str(supplied.parent), label, kind="directory")) + output = parent / supplied.name + if os.path.normcase(output) != os.path.normcase(supplied.absolute()): + raise SystemExit(f"{label} must be a canonical non-symlink path.") + return str(output) + + +def promote_staged_file(staged_path: str, output_path: str) -> tuple[Path, Path, Path | None]: + staged = Path(staged_path) + output = Path(output_path) + if staged == output: + raise SystemExit("A staged Deep Scan artifact must not be its published output path.") + backup = output.with_name(f".{output.name}.{uuid.uuid4()}.backup") if output.exists() else None + if backup is not None: + os.replace(output, backup) + try: + os.replace(staged, output) + except BaseException: + if backup is not None: + os.replace(backup, output) + raise + return staged, output, backup + + +def rollback_staged_file(promotion: tuple[Path, Path, Path | None]) -> None: + staged, output, backup = promotion + if output.exists(): + os.replace(output, staged) + if backup is not None: + os.replace(backup, output) + + +def finish_staged_file(promotion: tuple[Path, Path, Path | None]) -> None: + backup = promotion[2] + if backup is not None: + backup.unlink(missing_ok=True) + + def canonical_discovery_artifacts(scan: sqlite3.Row) -> dict[str, str]: discovery_dir = Path(scan["scan_dir"]) / "artifacts" / "02_discovery" artifacts = { @@ -817,6 +884,224 @@ def get_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> d return deep_scan_result(connection, scan["id"]) +def coordinator_lease_is_live( + connection: sqlite3.Connection, + run: sqlite3.Row, + scan: sqlite3.Row, + timestamp: str, +) -> bool: + if run["coordinator_generation"] == 1: + active_worker = connection.execute( + """ + SELECT 1 FROM deep_scan_workers + WHERE scan_id = ? AND status IN ('queued', 'running') + LIMIT 1 + """, + (run["scan_id"],), + ).fetchone() + return active_worker is not None and datetime.fromisoformat( + str(run["updated_at"]) + ) > datetime.fromisoformat(timestamp) - timedelta( + seconds=DEEP_SCAN_LEGACY_COORDINATOR_GRACE_SECONDS + ) + heartbeat_time = datetime.fromisoformat(str(run["updated_at"])) + heartbeat_path = ( + Path(scan["scan_dir"]) + / "artifacts" + / "deep_discovery" + / f"coordinator-heartbeat-{run['coordinator_generation']}.json" + ) + try: + heartbeat = json.loads(heartbeat_path.read_text(encoding="utf-8")) + if heartbeat["coordinatorGeneration"] == run["coordinator_generation"]: + heartbeat_time = max(heartbeat_time, datetime.fromisoformat(heartbeat["updatedAt"])) + except (OSError, KeyError, TypeError, ValueError): + pass + current_time = datetime.fromisoformat(timestamp) + return heartbeat_time > current_time - timedelta(seconds=DEEP_SCAN_COORDINATOR_LEASE_SECONDS) + + +def require_current_coordinator(run: sqlite3.Row, args: argparse.Namespace) -> None: + generation = getattr(args, "coordinator_generation", None) + if run["coordinator_generation"] == 1: + if generation is not None: + raise SystemExit("Deep Scan coordinator lease has not been claimed.") + return + if generation is None: + raise SystemExit("Deep Scan mutation requires the current coordinator lease.") + if generation != run["coordinator_generation"]: + raise SystemExit("Deep Scan coordinator lease belongs to a newer generation.") + + +def claim_deep_scan_coordinator( + connection: sqlite3.Connection, args: argparse.Namespace +) -> dict[str, Any]: + scan_id = require_uuid(args.scan_id, "scan-id") + with scan_completion_lock(scan_id): + return claim_deep_scan_coordinator_locked(connection, args, scan_id) + + +def claim_deep_scan_coordinator_locked( + connection: sqlite3.Connection, args: argparse.Namespace, scan_id: str +) -> dict[str, Any]: + connection.execute("BEGIN IMMEDIATE") + try: + scan, _ = require_owned_scan(connection, scan_id, args.thread_id) + require_current_continuation( + scan, + args.claim_token, + error_message="Deep Scan orchestration is owned by another continuation.", + ) + run, _ = require_running_deep_scan(connection, scan_id) + timestamp = now() + if args.coordinator_generation is not None: + require_current_coordinator(run, args) + disposition = "claimed" + elif coordinator_lease_is_live(connection, run, scan, timestamp): + connection.commit() + return { + **deep_scan_result(connection, scan_id), + "coordinatorDisposition": "observing", + } + else: + adopted = run["coordinator_generation"] > 1 or run["phase"] != "setup" + if adopted: + recover_expired_coordinator(connection, run, timestamp) + disposition = "adopted" if adopted else "claimed" + + connection.execute( + """ + UPDATE deep_scan_runs + SET coordinator_generation = coordinator_generation + ?, updated_at = ? + WHERE scan_id = ? AND status = 'running' + """, + (int(args.coordinator_generation != run["coordinator_generation"]), timestamp, scan_id), + ) + connection.commit() + except BaseException: + connection.rollback() + raise + return { + **deep_scan_result(connection, scan_id), + "coordinatorDisposition": disposition, + } + + +def recover_expired_coordinator( + connection: sqlite3.Connection, run: sqlite3.Row, timestamp: str +) -> None: + scan_id = run["scan_id"] + recover_candidate_ledger_publication(connection, scan_id) + legacy_generation = int(run["coordinator_generation"] == 1) + interrupted_discoveries = int( + connection.execute( + """ + SELECT COUNT(*) + FROM deep_scan_workers + WHERE scan_id = ? AND kind = 'discovery' + AND ( + status IN ('queued', 'running') + OR ( + status = 'canceled' + AND ( + error_message LIKE 'coordinator_shutdown:%' + OR (? = 1 AND error_message IS NULL) + ) + ) + ) + """, + (scan_id, legacy_generation), + ).fetchone()[0] + ) + connection.execute( + """ + UPDATE deep_scan_workers + SET merge_state = 'buffered', updated_at = ? + WHERE scan_id = ? AND merge_state = 'merging' + AND id IN ( + SELECT inputs.discovery_worker_id + FROM deep_scan_dedup_inputs AS inputs + JOIN deep_scan_workers AS reducers ON reducers.id = inputs.dedup_worker_id + WHERE reducers.scan_id = ? + AND reducers.kind = 'dedup' + AND ( + reducers.status IN ('queued', 'running', 'failed') + OR ( + reducers.status = 'canceled' + AND ( + reducers.error_message LIKE 'coordinator_shutdown:%' + OR (? = 1 AND reducers.error_message IS NULL) + ) + ) + ) + ) + """, + (timestamp, scan_id, scan_id, legacy_generation), + ) + cancel_active_workers(connection, scan_id, timestamp) + connection.execute( + """ + UPDATE deep_scan_workers + SET error_message = 'coordinator_shutdown_recovered: replacement attempt required', + updated_at = ? + WHERE scan_id = ? AND status = 'canceled' + AND ( + error_message LIKE 'coordinator_shutdown:%' + OR (? = 1 AND error_message IS NULL) + ) + """, + (timestamp, scan_id, legacy_generation), + ) + connection.execute( + """ + UPDATE deep_scan_runs + SET discovery_runs_dispatched = discovery_runs_dispatched - ?, + phase = CASE WHEN phase = 'setup' THEN 'setup' ELSE 'discovery' END, + updated_at = ? + WHERE scan_id = ? + """, + (interrupted_discoveries, timestamp, scan_id), + ) + + +def recover_candidate_ledger_publication(connection: sqlite3.Connection, scan_id: str) -> None: + scan = require_scan(connection, scan_id) + ledger = Path(scan["scan_dir"]) / "artifacts" / "02_discovery" / "candidate_ledger.jsonl" + backups = sorted( + ledger.parent.glob(f".{ledger.name}.*.backup"), + key=lambda backup: backup.stat().st_mtime_ns, + reverse=True, + ) + if not ledger.exists() and not backups: + return + reducers = connection.execute( + """ + SELECT status, artifact_dir + FROM deep_scan_workers + WHERE scan_id = ? AND kind = 'dedup' + AND status IN ('queued', 'running', 'succeeded') + ORDER BY updated_at DESC + """, + (scan_id,), + ) + for reducer in reducers: + snapshot = Path(reducer["artifact_dir"]) / "canonical" / ledger.name + if not snapshot.exists(): + continue + published = ledger.exists() and ledger.samefile(snapshot) + interrupted = reducer["status"] != "succeeded" + if not published and not (interrupted and backups and not ledger.exists()): + continue + if interrupted: + if backups: + os.replace(backups.pop(0), ledger) + else: + ledger.unlink(missing_ok=True) + for backup in backups: + backup.unlink(missing_ok=True) + return + + def require_deep_scan_worker(connection: sqlite3.Connection, worker_id: str) -> sqlite3.Row: worker_id = require_uuid(worker_id, "worker-id") row = connection.execute( @@ -859,6 +1144,7 @@ def upsert_deep_scan_worker( connection.execute("BEGIN IMMEDIATE") try: run = require_deep_scan_run(connection, scan_id) + require_current_coordinator(run, args) scan = require_scan(connection, scan_id) existing = connection.execute( "SELECT * FROM deep_scan_workers WHERE id = ?", (worker_id,) @@ -1073,6 +1359,7 @@ def claim_deep_scan_dedup( connection.execute("BEGIN IMMEDIATE") try: run, scan = require_running_deep_scan(connection, scan_id) + require_current_coordinator(run, args) prompt_path = deep_scan_path(scan, args.prompt_path, "Dedup prompt path", kind="file") artifact_dir = deep_scan_path( scan, args.artifact_dir, "Dedup artifact directory", kind="directory" @@ -1207,10 +1494,20 @@ def commit_deep_scan_dedup( connection: sqlite3.Connection, args: argparse.Namespace ) -> dict[str, Any]: scan_id = require_uuid(args.scan_id, "scan-id") + with scan_completion_lock(scan_id): + return commit_deep_scan_dedup_locked(connection, args, scan_id) + + +def commit_deep_scan_dedup_locked( + connection: sqlite3.Connection, args: argparse.Namespace, scan_id: str +) -> dict[str, Any]: worker_id = require_uuid(args.worker_id, "worker-id") + promotion: tuple[Path, Path, Path | None] | None = None + publication_copy: Path | None = None connection.execute("BEGIN IMMEDIATE") try: run = require_deep_scan_run(connection, scan_id) + require_current_coordinator(run, args) scan = require_scan(connection, scan_id) worker = require_deep_scan_worker(connection, worker_id) if worker["scan_id"] != scan_id or worker["kind"] != "dedup": @@ -1221,7 +1518,29 @@ def commit_deep_scan_dedup( require_running_deep_scan(connection, scan_id) if worker["status"] not in {"queued", "running"}: raise SystemExit("Only an active dedup worker can commit a result.") - canonical_discovery_artifacts(scan) + if args.candidate_ledger_path: + candidate_ledger_path = deep_scan_path( + scan, + args.candidate_ledger_path, + "Staged candidate ledger path", + kind="file", + ) + discovery_dir = Path(scan["scan_dir"]) / "artifacts" / "02_discovery" + deep_scan_path( + scan, + str(discovery_dir / "in_scope_files.txt"), + "Canonical in-scope inventory path", + kind="file", + ) + canonical_candidate_ledger_path = deep_scan_output_path( + scan, + str(discovery_dir / "candidate_ledger.jsonl"), + "Canonical candidate ledger path", + ) + else: + canonical_discovery_artifacts(scan) + candidate_ledger_path = None + canonical_candidate_ledger_path = None result_manifest_path = deep_scan_path( scan, args.result_manifest_path, @@ -1242,6 +1561,16 @@ def commit_deep_scan_dedup( ) if not inputs or any(row["merge_state"] != "merging" for row in inputs): raise SystemExit("Dedup inputs are not in the claimed merging state.") + if candidate_ledger_path and canonical_candidate_ledger_path: + canonical_path = Path(canonical_candidate_ledger_path) + publication_copy = canonical_path.with_name( + f".{canonical_path.name}.{uuid.uuid4()}.publish" + ) + os.link(candidate_ledger_path, publication_copy) + promotion = promote_staged_file( + str(publication_copy), + canonical_candidate_ledger_path, + ) timestamp = now() connection.execute( """ @@ -1278,23 +1607,44 @@ def commit_deep_scan_dedup( connection.commit() except BaseException: connection.rollback() + if promotion is not None: + rollback_staged_file(promotion) + if publication_copy is not None: + publication_copy.unlink(missing_ok=True) raise + if promotion is not None: + finish_staged_file(promotion) + if publication_copy is not None: + publication_copy.unlink(missing_ok=True) return deep_scan_result(connection, scan_id) def finish_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: scan_id = require_uuid(args.scan_id, "scan-id") + with scan_completion_lock(scan_id): + return finish_deep_scan_locked(connection, args, scan_id) + + +def finish_deep_scan_locked( + connection: sqlite3.Connection, args: argparse.Namespace, scan_id: str +) -> dict[str, Any]: omitted_worker_ids = [ require_uuid(value, "omitted-worker-id") for value in args.omitted_worker_id ] if len(set(omitted_worker_ids)) != len(omitted_worker_ids): raise SystemExit("Omitted Deep Scan worker IDs must be unique.") + promotion: tuple[Path, Path, Path | None] | None = None connection.execute("BEGIN IMMEDIATE") try: run = require_deep_scan_run(connection, scan_id) + require_current_coordinator(run, args) scan = require_scan(connection, scan_id) - manifest_path = deep_scan_path( - scan, args.manifest_path, "Deep Scan coordinator manifest path", kind="file" + manifest_path = ( + deep_scan_output_path(scan, args.manifest_path, "Deep Scan coordinator manifest path") + if args.staged_manifest_path + else deep_scan_path( + scan, args.manifest_path, "Deep Scan coordinator manifest path", kind="file" + ) ) buffered_worker_ids = [ row["id"] @@ -1363,8 +1713,31 @@ def finish_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) - raise SystemExit("Deep Scan cannot finish without a successful dedup worker.") failed_worker = connection.execute( """ - SELECT 1 FROM deep_scan_workers - WHERE scan_id = ? AND status = 'failed' + SELECT 1 FROM deep_scan_workers AS failed + WHERE failed.scan_id = ? AND failed.status = 'failed' + AND ( + failed.kind != 'dedup' + OR NOT EXISTS ( + SELECT 1 FROM deep_scan_dedup_inputs AS failed_inputs + WHERE failed_inputs.dedup_worker_id = failed.id + ) + OR EXISTS ( + SELECT 1 FROM deep_scan_dedup_inputs AS failed_inputs + WHERE failed_inputs.dedup_worker_id = failed.id + AND NOT EXISTS ( + SELECT 1 + FROM deep_scan_dedup_inputs AS replacement_inputs + JOIN deep_scan_workers AS replacement + ON replacement.scan_id = replacement_inputs.scan_id + AND replacement.id = replacement_inputs.dedup_worker_id + WHERE replacement_inputs.scan_id = failed.scan_id + AND replacement_inputs.discovery_worker_id = + failed_inputs.discovery_worker_id + AND replacement.kind = 'dedup' + AND replacement.status = 'succeeded' + ) + ) + ) LIMIT 1 """, (scan_id,), @@ -1402,6 +1775,14 @@ def finish_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) - "Deep Scan saturated completion must exactly identify all buffered discovery " "workers with --omitted-worker-id." ) + if args.staged_manifest_path: + staged_manifest_path = deep_scan_path( + scan, + args.staged_manifest_path, + "Staged Deep Scan coordinator manifest path", + kind="file", + ) + promotion = promote_staged_file(staged_manifest_path, manifest_path) timestamp = now() connection.execute( """ @@ -1416,29 +1797,44 @@ def finish_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) - connection.commit() except BaseException: connection.rollback() + if promotion is not None: + rollback_staged_file(promotion) raise + if promotion is not None: + finish_staged_file(promotion) return deep_scan_result(connection, scan_id) def fail_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: scan_id = require_uuid(args.scan_id, "scan-id") + with scan_completion_lock(scan_id): + return fail_deep_scan_locked(connection, args, scan_id) + + +def fail_deep_scan_locked( + connection: sqlite3.Connection, args: argparse.Namespace, scan_id: str +) -> dict[str, Any]: message = optional_text(args.message, maximum=2400) if message is None: raise SystemExit("message is required.") + promotion: tuple[Path, Path, Path | None] | None = None connection.execute("BEGIN IMMEDIATE") try: run = require_deep_scan_run(connection, scan_id) + require_current_coordinator(run, args) scan = require_scan(connection, scan_id) - manifest_path = ( - deep_scan_path( - scan, - args.manifest_path, - "Deep Scan failure manifest path", - kind="file", + manifest_path = None + if args.manifest_path: + manifest_path = ( + deep_scan_output_path(scan, args.manifest_path, "Deep Scan failure manifest path") + if args.staged_manifest_path + else deep_scan_path( + scan, + args.manifest_path, + "Deep Scan failure manifest path", + kind="file", + ) ) - if args.manifest_path - else None - ) if run["status"] in {"failed", "interrupted"} or scan["status"] == "failed": if ( run["status"] == args.deep_status @@ -1465,6 +1861,14 @@ def fail_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> raise SystemExit("Only a running Deep Scan can be failed or interrupted.") if run["manifest_path"] not in {None, manifest_path}: raise SystemExit("Deep Scan coordinator manifest path is immutable.") + if args.staged_manifest_path and manifest_path: + staged_manifest_path = deep_scan_path( + scan, + args.staged_manifest_path, + "Staged Deep Scan failure manifest path", + kind="file", + ) + promotion = promote_staged_file(staged_manifest_path, manifest_path) timestamp = now() connection.execute( """ @@ -1493,7 +1897,11 @@ def fail_deep_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> connection.commit() except BaseException: connection.rollback() + if promotion is not None: + rollback_staged_file(promotion) raise + if promotion is not None: + finish_staged_file(promotion) return deep_scan_result(connection, scan_id) diff --git a/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py b/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py index 31d56c35..f1db7c73 100644 --- a/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py +++ b/sdk/typescript/_bundled_plugin/scripts/workbench_cli.py @@ -238,6 +238,7 @@ def parse_args(description: str) -> argparse.Namespace: update_progress.add_argument("--reportable-findings-count", type=non_negative_int) update_progress.add_argument("--deep-review-pass", type=positive_int) update_progress.add_argument("--claim-token") + update_progress.add_argument("--coordinator-generation", type=positive_int) update_progress.add_argument("--model") update_progress.add_argument("--reasoning-effort") diff --git a/sdk/typescript/_bundled_plugin/scripts/workbench_db.py b/sdk/typescript/_bundled_plugin/scripts/workbench_db.py index ac669cc8..3405b565 100644 --- a/sdk/typescript/_bundled_plugin/scripts/workbench_db.py +++ b/sdk/typescript/_bundled_plugin/scripts/workbench_db.py @@ -3574,6 +3574,7 @@ def main() -> None: require_canonical_scan_directory=require_canonical_scan_directory, safe_segment=safe_segment, compact_timestamp=compact_timestamp, + scan_completion_lock=scan_completion_lock, ) ) if args.command == "inspect-target": @@ -3619,6 +3620,8 @@ def main() -> None: result = deep_scan.begin_deep_scan(connection, args) elif args.command == "get-deep-scan": result = deep_scan.get_deep_scan(connection, args) + elif args.command == "claim-deep-scan-coordinator": + result = deep_scan.claim_deep_scan_coordinator(connection, args) elif args.command == "upsert-deep-scan-worker": result = deep_scan.upsert_deep_scan_worker(connection, args) elif args.command == "claim-deep-scan-dedup": diff --git a/sdk/typescript/_bundled_plugin/scripts/workbench_progress.py b/sdk/typescript/_bundled_plugin/scripts/workbench_progress.py index e9d675fd..6282da2a 100644 --- a/sdk/typescript/_bundled_plugin/scripts/workbench_progress.py +++ b/sdk/typescript/_bundled_plugin/scripts/workbench_progress.py @@ -8,6 +8,7 @@ from typing import Any, Callable sys.path.insert(0, str(Path(__file__).resolve().parent)) +from deep_scan_workbench import require_current_coordinator from workbench.handoff import require_current_continuation from workbench_constants import PHASES from workbench_validation import optional_text, require_uuid, user_text @@ -175,6 +176,16 @@ def update_progress( scan = require_scan(connection, scan_id) if scan["status"] != "running": raise SystemExit("Only a running scan can update progress.") + if scan["mode"] == "deep": + coordinator = connection.execute( + "SELECT * FROM deep_scan_runs WHERE scan_id = ?", (scan_id,) + ).fetchone() + if coordinator is not None and ( + coordinator["status"] == "running" or args.coordinator_generation is not None + ): + require_current_coordinator(coordinator, args) + elif args.coordinator_generation is not None: + raise SystemExit("Coordinator leases apply only to Deep Scan progress.") require_current_continuation( scan, args.claim_token, diff --git a/sdk/typescript/_bundled_plugin/skills/deep-security-scan/SKILL.md b/sdk/typescript/_bundled_plugin/skills/deep-security-scan/SKILL.md index 95d0a311..a8f91bc7 100644 --- a/sdk/typescript/_bundled_plugin/skills/deep-security-scan/SKILL.md +++ b/sdk/typescript/_bundled_plugin/skills/deep-security-scan/SKILL.md @@ -97,11 +97,13 @@ The top-level goal completes only after: Use the same discovery tool in every host: ```text -Desktop: start_codex_security_deep_scan({ scanId }) +Desktop: start_codex_security_deep_scan({ scanId, handoffClaimToken? }) CLI/headless first call: start_codex_security_deep_scan({ targetPath, scope: ".", userContext? }) -Later calls in any host: start_codex_security_deep_scan({ scanId }) +Later calls in any host: start_codex_security_deep_scan({ scanId, handoffClaimToken? }) ``` +When the existing scan has a `handoffClaimToken`, preserve and pass that same token on every discovery start or resume, including after a paused waiter, app update, or MCP server restart. Do not drop the token merely because the scan ID and owning thread are unchanged. + For a scoped-path scan, pass the resolved scoped directory as `targetPath` with `scope: "."`; never silently widen it to the repository root. Make one call and wait for it. The call blocks for up to 24 hours and returns only after discovery completes, fails, or is canceled. The tool owns the transition into the discovery phase, so leave the public scan phase at preflight before calling it. Do not publish discovery progress yourself while the call is pending. @@ -112,7 +114,7 @@ Handle the terminal result as follows: - `status: "canceled"`: stop without starting validation or finalization. - Tool error: report the exact stable MCP error, including its failure-manifest path when present, and stop the current response. This is a terminal failure of that logical scan: do not call `start_codex_security_deep_scan` again in this response; do not call `get_codex_security_scan_context` in this response; do not call `complete_codex_security_scan` in this response; do not call the target form again to create a replacement scan, do not cancel an already terminal failed scan, do not return a final answer, satisfy a structured output schema, do not synthesize no-findings coverage, or emit benchmark JSON. -If the host represents the pending tool call as a running execution cell, keep waiting on that same cell instead of starting another tool call. Stopping the current Codex response or reaching the host's 24-hour timeout detaches only the caller; it does not cancel the scan. Only while the scan is still active may a later desktop turn rejoin with `{ scanId }`, or a CLI/headless turn repeat the identical target form to rejoin the owning thread's active scan. A terminal tool failure is not a detached waiter and must not be replaced. When the user explicitly asks to stop an active scan, call `cancel_codex_security_scan({ scanId })`. +If the host represents the pending tool call as a running execution cell, keep waiting on that same cell instead of starting another tool call. Stopping the current Codex response or reaching the host's 24-hour timeout detaches only the caller; it does not cancel the scan. Only while the scan is still active may a later desktop turn rejoin with `{ scanId, handoffClaimToken? }`, or a CLI/headless turn repeat the identical target form to rejoin the owning thread's active scan. After an MCP process restart, the new coordinator safely adopts the expired lease and preserves completed discovery receipts. A terminal tool failure is not a detached waiter and must not be replaced. When the user explicitly asks to stop an active scan, call `cancel_codex_security_scan({ scanId })`. Do not call `open_codex_security_workspace` again to refresh progress. The Security workspace continues to show discovery progress. @@ -164,6 +166,6 @@ Do not bypass validation because a candidate recurred across workers. Recurrence - Do not edit repository files during scanning. - Do not widen or reinterpret the resolved target. - Do not call `fail_codex_security_scan` because a wait was detached, a turn ended, discovery remains active, or partial artifacts exist. -- If the tool reports that its process ended during discovery, treat the scan as failed; this version cannot resume that run. +- If a waiter detaches or the MCP process ends while discovery is still running, preserve the scan and its handoff claim. A later same-scan call can adopt the expired coordinator lease and resume unfinished discovery without repeating completed reviews. - After any terminal discovery failure, stop the current response and surface the stable MCP failure and preserved failure-manifest path instead. Do not call `start_codex_security_deep_scan` again in that response; do not call `get_codex_security_scan_context` in that response; do not call `complete_codex_security_scan` in that response; do not start a second scan, call cancel for that failed scan, return a final answer, satisfy a structured output schema, or return synthetic no-findings or benchmark output. - On explicit cancellation, call `cancel_codex_security_scan`; after it returns, do not accept late progress or artifacts. diff --git a/sdk/typescript/tests-ts/deep-scan-workbench.test.ts b/sdk/typescript/tests-ts/deep-scan-workbench.test.ts index 1e49734e..bf94c471 100644 --- a/sdk/typescript/tests-ts/deep-scan-workbench.test.ts +++ b/sdk/typescript/tests-ts/deep-scan-workbench.test.ts @@ -1,9 +1,20 @@ +import { mkdir, mkdtemp, realpath, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; import { join } from "node:path"; -import { describe, expect, test } from "bun:test"; +import { afterEach, describe, expect, test } from "bun:test"; import { PLUGIN_ROOT } from "./plugin-root.js"; const originalClaimToken = "22222222-2222-4222-8222-222222222222"; const replacementClaimToken = "33333333-3333-4333-8333-333333333333"; +const temporaryDirectories: string[] = []; + +afterEach(async () => { + await Promise.all( + temporaryDirectories + .splice(0) + .map((path) => rm(path, { recursive: true, force: true })), + ); +}); const deepScanOwnershipProbe = [ "import argparse, json, sqlite3, sys", @@ -138,4 +149,200 @@ describe("deep scan workbench ownership", () => { handoffStatus: "delivered", }); }); + + test("adopts an expired coordinator without repeating completed discovery", async () => { + const root = await realpath( + await mkdtemp(join(tmpdir(), "codex-security-deep-resume-")), + ); + temporaryDirectories.push(root); + const repository = join(root, "repository"); + const stateDir = join(root, "state"); + const codexHome = join(root, "codex-home"); + await mkdir(repository); + await writeFile(join(repository, "source.py"), "# source fixture\n"); + + const python = Bun.which("python3") ?? Bun.which("python"); + expect(python).not.toBeNull(); + const command = (args: string[], allowFailure = false) => { + const result = Bun.spawnSync( + [ + python!, + "-I", + "-B", + join(PLUGIN_ROOT, "scripts", "workbench_db.py"), + ...args, + ], + { + env: { + ...process.env, + CODEX_SECURITY_STATE_DIR: stateDir, + CODEX_HOME: codexHome, + }, + stdout: "pipe", + stderr: "pipe", + }, + ); + const stdout = new TextDecoder().decode(result.stdout); + const stderr = new TextDecoder().decode(result.stderr); + if (allowFailure) return { status: result.exitCode, stderr }; + expect(result.exitCode, stderr).toBe(0); + return JSON.parse(stdout) as Record; + }; + + const started = command([ + "begin-deep-scan", + "--thread-id", + "thread-deep-scan", + "--target-path", + repository, + "--scope", + ".", + "--scan-root", + join(root, "scans"), + "--available-parallelism", + "4", + ]); + const initial = started["deepScan"] as Record; + const scanId = initial["scanId"] as string; + const scanDir = initial["scanDir"] as string; + expect(initial["coordinatorGeneration"]).toBe(1); + + const updateDatabase = (statement: string, ...values: string[]) => { + const result = Bun.spawnSync( + [ + python!, + "-I", + "-B", + "-c", + "import sqlite3,sys; connection=sqlite3.connect(sys.argv[1]); connection.execute(sys.argv[2],sys.argv[3:]); connection.commit()", + join(stateDir, "workbench.sqlite3"), + statement, + ...values, + ], + { stdout: "pipe", stderr: "pipe" }, + ); + expect(result.exitCode, new TextDecoder().decode(result.stderr)).toBe(0); + }; + updateDatabase( + "UPDATE scans SET handoff_claim_token = ? WHERE id = ?", + originalClaimToken, + scanId, + ); + command([ + "update-progress", + "--scan-id", + scanId, + "--phase", + "discovery", + "--claim-token", + originalClaimToken, + ]); + + const completedWorkerId = "44444444-4444-4444-8444-444444444444"; + const interruptedWorkerId = "55555555-5555-4555-8555-555555555555"; + for (const workerId of [completedWorkerId, interruptedWorkerId]) { + const artifactDir = join( + scanDir, + "artifacts", + "deep_discovery", + workerId, + ); + const promptPath = join(artifactDir, "prompt.md"); + await mkdir(artifactDir, { recursive: true }); + await writeFile(promptPath, "Review the source.\n"); + const workerArgs = [ + "upsert-deep-scan-worker", + "--scan-id", + scanId, + "--worker-id", + workerId, + "--kind", + "discovery", + "--prompt-path", + promptPath, + "--artifact-dir", + artifactDir, + "--attempt", + "1", + ]; + command([...workerArgs, "--status", "running"]); + if (workerId === completedWorkerId) { + const resultPath = join(artifactDir, "result.json"); + await writeFile(resultPath, "{}\n"); + command([ + ...workerArgs, + "--status", + "succeeded", + "--result-manifest-path", + resultPath, + ]); + } + } + + updateDatabase( + "UPDATE deep_scan_runs SET updated_at = ? WHERE scan_id = ?", + "2000-01-01T00:00:00+00:00", + scanId, + ); + const claimArgs = [ + "claim-deep-scan-coordinator", + "--scan-id", + scanId, + "--thread-id", + "thread-deep-scan", + ]; + const missingClaim = command(claimArgs, true); + expect(missingClaim["status"]).not.toBe(0); + expect(missingClaim["stderr"]).toContain("another continuation"); + + const resumed = command([ + ...claimArgs, + "--claim-token", + originalClaimToken, + ]); + const recovered = resumed["deepScan"] as Record; + expect(resumed["coordinatorDisposition"]).toBe("adopted"); + expect(recovered).toMatchObject({ + status: "running", + phase: "discovery", + coordinatorGeneration: 2, + dispatchedCount: 1, + }); + expect(recovered["workers"]).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + id: completedWorkerId, + status: "succeeded", + }), + expect.objectContaining({ + id: interruptedWorkerId, + status: "canceled", + }), + ]), + ); + + const observing = command([ + ...claimArgs, + "--claim-token", + originalClaimToken, + ]); + expect(observing["coordinatorDisposition"]).toBe("observing"); + + const staleProgress = command( + [ + "update-progress", + "--scan-id", + scanId, + "--phase", + "discovery", + "--claim-token", + originalClaimToken, + "--coordinator-generation", + "1", + ], + true, + ); + expect(staleProgress["status"]).not.toBe(0); + expect(staleProgress["stderr"]).toContain("newer generation"); + }); }); diff --git a/sdk/typescript/tests-ts/runtime.test.ts b/sdk/typescript/tests-ts/runtime.test.ts index 31cbbef2..d16d9f18 100644 --- a/sdk/typescript/tests-ts/runtime.test.ts +++ b/sdk/typescript/tests-ts/runtime.test.ts @@ -413,6 +413,47 @@ describe("plugin runtime preparation", () => { } }); + test("claims persisted Deep Scans after a coordinator restart", async () => { + const parts = await Promise.all( + ["000", "001"].map((part) => + readFile(join(PLUGIN_ROOT, "mcp", `server.mjs.br.part-${part}`)), + ), + ); + const runtime = brotliDecompressSync(Buffer.concat(parts)).toString("utf8"); + const source = + /async function startOrJoinDeepScanCoordinator\(input\) \{[\s\S]*?\n\}/u.exec( + runtime, + )?.[0]; + expect(source).toBeDefined(); + const startOrJoin = new Function( + `${source}\nreturn startOrJoinDeepScanCoordinator;`, + )() as ( + input: unknown, + ) => Promise<{ coordinator: unknown; joined: boolean }>; + const scan = { scanId: "persisted-scan" }; + const coordinator = {}; + const claimCoordinator = mock(async () => ({ run: scan, acquired: true })); + const start = mock(() => coordinator); + + expect( + await startOrJoin({ + begin: { run: scan, shouldStart: false }, + registry: { get: () => undefined, start }, + options: { + threadId: "scan-thread", + handoffClaimToken: "continuation-claim", + store: { claimCoordinator }, + }, + }), + ).toEqual({ coordinator, joined: false }); + expect(claimCoordinator).toHaveBeenCalledWith({ + scanId: "persisted-scan", + threadId: "scan-thread", + handoffClaimToken: "continuation-claim", + }); + expect(start).toHaveBeenCalledTimes(1); + }); + test("projects only the unchanged external payload from the source checkout", async () => { const root = await temporaryDirectory(); const workspace = join(root, "workspace");