fix(deep-scan): recover checkpoints at discovery deadlines

plugins/codex-security/mcp-app/tests/test_deep_scan_store_integration.ts

@@ -6,7 +6,14 @@ import { mock } from "node:test";
 import { execFile, spawn } from "node:child_process";
 import { randomUUID } from "node:crypto";
 import { once } from "node:events";
-import { mkdir, readFile, rm, writeFile } from "node:fs/promises";
+import {
+  mkdir,
+  readFile,
+  readdir,
+  rm,
+  stat,
+  writeFile,
+} from "node:fs/promises";
 import path from "node:path";
 import { promisify } from "node:util";
 import { importModule } from "./import-module.ts";
@@ -1180,20 +1187,106 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
       scanId: run.scanId,
       threadId,
     });
-    const worker = await createWorkerFixture(run, "saved", "discovery");
-    await store.updateWorker({ ...worker, status: "running" });
-    await saveScanDraftCheckpoint(
-      { root: worker.artifactDir, layout: "worker", repoRoot: targetPath },
-      {
+    const workers: { id: string }[] = [];
+    for (const label of [
+      "valid",
+      "second-valid",
+      "anonymous",
+      "mismatch",
+      "corrupt",
+      "missing",
+    ]) {
+      const worker = await createWorkerFixture(run, label, "discovery");
+      await store.updateWorker({ ...worker, status: "running" });
+      const context = {
+        root: worker.artifactDir,
+        layout: "worker",
+        repoRoot: targetPath,
+      };
+      const saved = {
         ...checkpoint(
           run.scanId,
           "pending-review",
           "Worker did not finish reviewing this source.",
         ),
-        findings: [finding("saved", "fixture.py")],
-      },
-    );
-    await store.updateWorker({ ...worker, status: "canceled" });
+        findings: [finding(label, "fixture.py")],
+      };
+      if (label === "valid" || label === "second-valid") {
+        if (label === "valid") {
+          saved.findings.push(finding("out-of-scope", "../outside.py"));
+          saved.findings.push({
+            title: "Unvalidated, incomplete finding",
+          } as (typeof saved.findings)[number]);
+        } else {
+          saved.findings = [];
+        }
+        await mkdir(path.join(worker.artifactDir, "artifacts"));
+        await writeFile(
+          path.join(worker.artifactDir, "artifacts/review.md"),
+          `Source evidence for ${label}.\n`,
+        );
+        await saveScanDraftCheckpoint(context, {
+          ...saved,
+          coverage: {
+            ...saved.coverage,
+            surfaces: [
+              {
+                id: "api",
+                label: `${label} API`,
+                disposition: "no_issue_found",
+                receiptRefs: ["artifacts/review.md"],
+              },
+            ],
+            deferred: [
+              ...saved.coverage.deferred,
+              {
+                id: "surface-review",
+                reason: "Review remains.",
+                surfaceIds: ["api"],
+              },
+            ],
+          },
+        });
+      } else if (label === "anonymous") {
+        for (const reason of [
+          "Review authentication.",
+          "Review authorization.",
+        ]) {
+          await saveScanDraftCheckpoint(context, {
+            scanId: run.scanId,
+            complete: false,
+            findings: [],
+            threatModel: {},
+            coverage: {
+              surfaces: [],
+              explicitExclusions: [],
+              deferred: [{ reason }],
+            },
+          });
+        }
+      } else if (label === "mismatch") {
+        await saveScanDraftCheckpoint(context, {
+          ...saved,
+          scanId: randomUUID(),
+        });
+      } else {
+        await mkdir(path.join(worker.artifactDir, "checkpoints"), {
+          recursive: true,
+        });
+        const name = "a".repeat(64) + ".json";
+        await writeFile(
+          path.join(worker.artifactDir, "checkpoint-head.json"),
+          JSON.stringify({ checkpoint: name }),
+        );
+        if (label === "corrupt")
+          await writeFile(
+            path.join(worker.artifactDir, "checkpoints", name),
+            '{"scanId":',
+          );
+      }
+      await store.updateWorker({ ...worker, status: "canceled" });
+      workers.push(worker);
+    }
     const context = await createScanArtifactContext(run.scanId, runWorkbench, {
       requireRunning: true,
     });
@@ -1229,6 +1322,26 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
     const beforeFailure = await Promise.all(
       draftNames.map((name) => readFile(path.join(run.scanDir, name), "utf8")),
     );
+    const readObservationTimes = () =>
+      Promise.all(
+        draftNames.map(
+          async (name) =>
+            (await stat(path.join(run.scanDir, name), { bigint: true }))
+              .mtimeNs,
+        ),
+      );
+    const observationTimesBeforeFailure = await readObservationTimes();
+    const checkpointDirectory = path.join(run.scanDir, "checkpoints");
+    const readParentCheckpoints = async () =>
+      Promise.all(
+        (await readdir(checkpointDirectory))
+          .sort()
+          .map(async (name) => [
+            name,
+            await readFile(path.join(checkpointDirectory, name), "utf8"),
+          ]),
+      );
+    const checkpointsBeforeFailure = await readParentCheckpoints();
     const database = path.join(
       environment.CODEX_SECURITY_STATE_DIR,
       "workbench.sqlite3",
@@ -1265,6 +1378,16 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
       beforeFailure,
       "a failed finish commit must restore the unsealed parent and accepted checkpoint head",
     );
+    assert.deepEqual(
+      await readParentCheckpoints(),
+      checkpointsBeforeFailure,
+      "rollback must not publish an aggregate checkpoint as new parent evidence",
+    );
+    assert.deepEqual(
+      await readObservationTimes(),
+      observationTimesBeforeFailure,
+      "rollback must preserve parent observation order",
+    );
     assert.equal((await store.get(run.scanId, threadId)).status, "running");
     await runSql("DROP TRIGGER reject_deadline_commit;");
     const finished = await store.finish({
@@ -1283,11 +1406,27 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
     const recovered = await readJson(run.scanDir, "findings.json");
     assert.deepEqual(
       recovered.findings.map((item: { title: string }) => item.title),
-      ["Unsafe request output saved"],
+      ["Unsafe request output valid"],
     );
-    assert.equal(recovered.findings[0].provenance.workerId, worker.id);
+    assert.equal(recovered.findings[0].provenance.workerId, workers[0].id);
     const coverage = await readJson(run.scanDir, "coverage.json");
-
+    assert.equal(coverage.surfaces.length, 2);
+    for (const [index, label] of ["valid", "second-valid"].entries()) {
+      const workerId: string = workers[index].id;
+      const surface = coverage.surfaces.find(
+        (row: { workerId: string }) => row.workerId === workerId,
+      );
+      const task = coverage.deferred.find(
+        (row: { id: string }) => row.id === `${workerId}/surface-review`,
+      );
+      assert.equal(surface.id, `${workerId}/api`);
+      assert.equal(surface.disposition, "no_issue_found");
+      assert.deepEqual(task.surfaceIds, [surface.id]);
+      assert.equal(
+        await readFile(path.join(run.scanDir, surface.receiptRefs[0]), "utf8"),
+        `Source evidence for ${label}.\n`,
+      );
+    }
     assert.equal(
       coverage.deferred.some((item: { reason: string }) =>
         /time limit elapsed/.test(item.reason),
@@ -1301,12 +1440,25 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
       ),
       true,
     );
+    const anonymousTasks = coverage.deferred.filter(
+      (row: { workerId?: string }) => row.workerId === workers[2].id,
+    );
+    assert.deepEqual(
+      new Set(anonymousTasks.map((row: { reason: string }) => row.reason)),
+      new Set(["Review authentication.", "Review authorization."]),
+    );
     const manifest = await readJson(run.scanDir, "scan-manifest.json");
+    assert.equal(manifest.scan.threatModel, undefined);
+    const savedScan = await runWorkbench(["get-scan", "--scan-id", run.scanId]);
+    assert.match(
+      JSON.stringify(savedScan),
+      /Skipped malformed saved threat model.*missing required schema property/,
+    );
     assert.equal(manifest.scan.sealedAt, undefined);
     assert.equal(manifest.scan.complete, undefined);
     assert.equal(
       Object.keys(manifest.scan.preservedSources).some((source) =>
-        source.includes("saved/output/checkpoints/"),
+        source.includes("valid/output/checkpoints/"),
       ),
       true,
     );
@@ -1325,6 +1477,14 @@ async function testDeadlinePreservesValidWorkerCheckpoints() {
       beforeReplay,
     );
     await runWorkbench(["complete-scan", "--scan-id", run.scanId]);
+    assert.deepEqual(
+      (await readJson(run.scanDir, "coverage.json")).surfaces,
+      coverage.surfaces,
+    );
+    assert.deepEqual(
+      (await readJson(run.scanDir, "coverage.json")).deferred,
+      coverage.deferred,
+    );
     const sealed = await readJson(run.scanDir, "scan-manifest.json");
     assert.equal(typeof sealed.scan.sealedAt, "string");
     assert.deepEqual(

plugins/codex-security/scripts/deep_scan_workbench.py

@@ -10,7 +10,6 @@
 import sqlite3
 import sys
 import uuid
-from contextlib import nullcontext
 from datetime import datetime, timedelta
 from pathlib import Path
 from typing import Any, Callable
@@ -1676,6 +1675,9 @@ def finish_deep_scan_locked(
     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
+    deadline_publication: (
+        tuple[Path, dict[str, tuple[bytes, os.stat_result] | None], set[str]] | None
+    ) = None
     connection.execute("BEGIN IMMEDIATE")
     try:
         run = require_deep_scan_run(connection, scan_id)
@@ -1799,25 +1801,33 @@ def finish_deep_scan_locked(
                 kind="file",
             )
             promotion = promote_staged_file(staged_manifest_path, manifest_path)
-        with (
-            saved_results.preserve_deep_deadline_draft(dependencies(), connection, scan)
-            if zero_discovery_deadline and standard_scan_manifest
-            else nullcontext()
-        ):
-            timestamp = dependencies().now()
-            connection.execute(
-                """
-                UPDATE deep_scan_runs
-                SET status = 'succeeded', phase = 'terminal', terminal_reason = ?,
-                    manifest_path = ?, completed_at = ?, updated_at = ?
-                WHERE scan_id = ?
-                """,
-                (args.terminal_reason, manifest_path, timestamp, timestamp, scan_id),
+        if zero_discovery_deadline:
+            deadline_publication = saved_results.preserve_deep_deadline_draft(
+                dependencies(), connection, scan, allow_empty=not standard_scan_manifest
             )
-            cancel_active_workers(connection, scan_id, timestamp)
-            connection.commit()
+        timestamp = dependencies().now()
+        connection.execute(
+            """
+            UPDATE deep_scan_runs
+            SET status = 'succeeded', phase = 'terminal', terminal_reason = ?,
+                manifest_path = ?, completed_at = ?, updated_at = ?
+            WHERE scan_id = ?
+            """,
+            (args.terminal_reason, manifest_path, timestamp, timestamp, scan_id),
+        )
+        cancel_active_workers(connection, scan_id, timestamp)
+        connection.commit()
     except BaseException:
         connection.rollback()
+        if deadline_publication is not None or promotion is not None:
+            # A signal can arrive after SQLite commits but before commit returns.
+            # The completion lock makes persisted success authoritative here.
+            if require_deep_scan_run(connection, scan_id)["status"] == "succeeded":
+                if promotion is not None:
+                    finish_staged_file(promotion)
+                raise
+        if deadline_publication is not None:
+            saved_results._restore_published_outputs(*deadline_publication)
         if promotion is not None:
             rollback_staged_file(promotion)
         raise

plugins/codex-security/scripts/finalize_scan_contract.py

@@ -1075,6 +1075,8 @@ def _recover_unsealed_findings(
     schema_dir: Path,
     scan_dir: Path,
     warnings: list[str],
+    *,
+    current_contract: bool = False,
 ) -> list[str]:
     schema = _read_json(schema_dir / "findings.schema.json")
     properties = _require_dict(schema, "properties", "findings.schema")
@@ -1190,6 +1192,8 @@ def _recover_unsealed_findings(
                         f"Skipped malformed {auxiliary} for finding {index + 1}: {exc}."
                     )
             _validate_schema_node(finding, finding_schema, context, schema)
+            if current_contract:
+                _validate_current_findings({"findings": [finding]}, schema_dir)
         except ContractError as exc:
             warning = f"Skipped malformed finding {index + 1}: {exc}."
             warnings.append(warning)