Resume Deep Scans after coordinator restarts

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")
 

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":

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,

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.

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<string, unknown>;
+    };
+
+    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<string, unknown>;
+    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<string, unknown>;
+    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");
+  });
 });

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");