fix(deep-scan): keep selected results stable during retries

plugins/codex-security/mcp-app/tests/scan-identity-recovery-fixture.ts

@@ -71,7 +71,12 @@ binding.update(scanId=scan_id,producer={'name':'codex-security-plugin','version'
 warnings=[]
 recovery=merge_saved_results(recovered,scan_id,binding,json.loads(sys.argv[5]),warnings,stopped=True,reason='Synthetic interruption',scan_mode=sys.argv[10]) if sys.argv[9]=='saved' else tuple(json.loads((recovered/name).read_text()) for name in ['scan-manifest.json','findings.json','coverage.json'])
 if json.loads(sys.argv[7]):
- replayed=merge_saved_results(recovered,scan_id,binding,json.loads(sys.argv[5]),[],stopped=True,reason='Synthetic interruption',frozen_source_digests=recovery[0]['scan']['preservedSources'],scan_mode=sys.argv[10])
+ frozen=recovery[0]['scan']
+ replayed=merge_saved_results(
+  recovered,scan_id,binding,json.loads(sys.argv[5]),[],stopped=True,reason='Synthetic interruption',
+  frozen_source_digests=frozen['preservedSources'],checkpoint_heads=frozen['preservedCheckpointHeads'].copy(),
+  scan_mode=sys.argv[10],
+ )
  assert replayed == recovery, 'Frozen recovery changed the retained documents'
 ordinary=(manifest,json.loads((normal/'findings.json').read_text()),coverage)
 results=[]

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

@@ -133,24 +133,26 @@ def capture(*args, **kwargs):
     return captured
 saved._capture_saved_source = capture
 retained._capture_saved_source = capture
-frozen = None
+frozen = heads = None
 if sys.argv[9] == "true":
     db = SimpleNamespace(require_canonical_scan_directory=lambda path: path, artifact_path=lambda *args, **kwargs: None, ARTIFACTS={"manifest": "scan-manifest.json"})
     connection = SimpleNamespace(execute=lambda *args: SimpleNamespace(fetchall=lambda: [worker]))
-    scan = {"scan_dir": str(root), "id": scan_id, "retained_source_digests_json": None, "seal_manifest_digest": None}
-    frozen, _ = saved._recovery_source_digests(db, connection, scan)
-merged = merge_saved_results(*args, stopped=True, reason="interrupted", frozen_source_digests=frozen)
+    scan = {"scan_dir": str(root), "id": scan_id, "retained_source_digests_json": None, "retained_checkpoint_heads_json": None, "seal_manifest_digest": None}
+    frozen, _, heads = saved._recovery_source_digests(db, connection, scan)
+merged = merge_saved_results(*args, stopped=True, reason="interrupted", frozen_source_digests=frozen, checkpoint_heads=heads)
 saved._capture_saved_source = original_capture
 retained._capture_saved_source = original_retained_capture
-replay = merge_saved_results(*args, stopped=True, reason="interrupted", frozen_source_digests=merged[0]["scan"]["preservedSources"])
-assert merged[2] == replay[2]
+snapshot = merged[0]["scan"]
+def replay():
+    return merge_saved_results(*args, stopped=True, reason="interrupted", frozen_source_digests=snapshot["preservedSources"], checkpoint_heads=snapshot["preservedCheckpointHeads"])
+assert merged[2] == replay()[2]
 rejects_changed_checkpoint = None
 if published:
     checkpoint = root / published[-1]
     original = checkpoint.read_bytes()
     checkpoint.write_text("{}")
     try:
-        merge_saved_results(*args, stopped=True, reason="interrupted", frozen_source_digests=merged[0]["scan"]["preservedSources"])
+        replay()
     except saved.ContractError:
         rejects_changed_checkpoint = True
     else:
@@ -161,7 +163,7 @@ refs = merged[2]["surfaces"][0]["receiptRefs"]
 warnings = []
 coverage = merged[2]
 _recover_unsealed_coverage(coverage, plugin / "schemas", root, warnings, [])
-print(json.dumps({"refs": refs, "surfaces": coverage["surfaces"], "warnings": warnings, "published": published, "captured": merged[0]["scan"]["preservedSources"], "rejectsChangedCheckpoint": rejects_changed_checkpoint}))
+print(json.dumps({"refs": refs, "surfaces": coverage["surfaces"], "warnings": warnings, "published": published, "captured": snapshot["preservedSources"], "rejectsChangedCheckpoint": rejects_changed_checkpoint}))
 `,
         path.resolve(import.meta.dirname, "../.."),
         f.root,

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

@@ -168,6 +168,180 @@ test("every released schema upgrades to the same current schema and remains idem
   }
 });
 
+for (const { missingVersions, legacyCheckpoint, mainReader } of [
+  { missingVersions: [57], legacyCheckpoint: 0, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 47, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 47, mainReader: true },
+  { missingVersions: [], legacyCheckpoint: 48, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 50, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 52, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 53, mainReader: false },
+  { missingVersions: [], legacyCheckpoint: 54, mainReader: false },
+  { missingVersions: [43], legacyCheckpoint: 47, mainReader: false },
+  { missingVersions: [44, 45, 46], legacyCheckpoint: 47, mainReader: false },
+]) {
+  test(`database-info retains stopped-scan checkpoints when adding migration ${missingVersions.join(",")} (legacy checkpoint: ${legacyCheckpoint}, main reader: ${mainReader})`, async () => {
+    const directory = await temporary.create("workbench-checkpoint-upgrade-");
+    const databasePath = join(directory, "workbench.sqlite3");
+    const sources = JSON.stringify({ "results.json": "retained-digest" });
+    const heads = JSON.stringify({ worker: "accepted-checkpoint" });
+    const context = JSON.stringify("Original discovery context");
+    let originalDiscoveryMigration: Record<string, unknown> | undefined;
+    let originalCheckpointMigration: Record<string, unknown> | undefined;
+    const old = new DatabaseSync(databasePath);
+    try {
+      applyMigrations(
+        old,
+        migrations
+          .filter(
+            (item) =>
+              !missingVersions.includes(item.version) &&
+              !(legacyCheckpoint && [54, 55, 56].includes(item.version)) &&
+              !(
+                [47, 48, 50, 52, 53].includes(legacyCheckpoint) &&
+                [52, 53].includes(item.version)
+              ) &&
+              !(
+                [47, 48, 50].includes(legacyCheckpoint) &&
+                [50, 51].includes(item.version)
+              ) &&
+              !(
+                [47, 48].includes(legacyCheckpoint) &&
+                [48, 49].includes(item.version)
+              ) &&
+              !(legacyCheckpoint === 47 && item.version === 47),
+          )
+          .map((item) =>
+            legacyCheckpoint && item.version === 57
+              ? { ...item, version: legacyCheckpoint }
+              : item,
+          ),
+      );
+      if (mainReader)
+        old.exec(
+          migrations.find((item) => item.version === 47)!.statements.join("\n"),
+        );
+      insertScan(old);
+      old.exec(`INSERT INTO deep_scan_runs (scan_id, schema_version, workflow_version, status, phase,
+        workers, subagents, stop_after_no_new, max_discovery_runs, created_at, updated_at)
+        VALUES ('scan', 1, 'v1', 'failed', 'terminal', 1, 0, 7, 10, 'created', 'updated')`);
+      if (legacyCheckpoint !== 47 || mainReader) {
+        old
+          .prepare("UPDATE deep_scan_runs SET discovery_user_context_json = ?")
+          .run(context);
+        originalDiscoveryMigration = old
+          .prepare("SELECT * FROM schema_migrations WHERE version = 47")
+          .get();
+      }
+      old
+        .prepare(
+          "UPDATE scans SET status = 'failed', retained_source_digests_json = ?",
+        )
+        .run(sources);
+      if (!missingVersions.includes(57)) {
+        old
+          .prepare("UPDATE scans SET retained_checkpoint_heads_json = ?")
+          .run(heads);
+      }
+      originalCheckpointMigration = old
+        .prepare(
+          "SELECT name, applied_at FROM schema_migrations WHERE version = ?",
+        )
+        .get(legacyCheckpoint || 57);
+    } finally {
+      old.close();
+    }
+    await databaseInfo(directory);
+    const upgraded = new DatabaseSync(databasePath);
+    try {
+      assert.equal(
+        upgraded
+          .prepare("SELECT retained_checkpoint_heads_json FROM scans")
+          .get()?.retained_checkpoint_heads_json,
+        !missingVersions.includes(57) ? heads : null,
+      );
+      upgraded
+        .prepare("UPDATE scans SET retained_checkpoint_heads_json = ?")
+        .run(heads);
+    } finally {
+      upgraded.close();
+    }
+    await databaseInfo(directory);
+    const reopened = new DatabaseSync(databasePath);
+    try {
+      const scan = reopened.prepare("SELECT * FROM scans").get()!;
+      assert.equal(scan.status, "failed");
+      assert.equal(scan.retained_source_digests_json, sources);
+      assert.equal(scan.retained_checkpoint_heads_json, heads);
+      if (originalCheckpointMigration !== undefined) {
+        assert.deepEqual(
+          reopened
+            .prepare(
+              "SELECT name, applied_at FROM schema_migrations WHERE version = 57",
+            )
+            .get(),
+          originalCheckpointMigration,
+        );
+      }
+      assert.equal(
+        reopened
+          .prepare("SELECT COUNT(*) AS count FROM schema_migrations")
+          .get()?.count,
+        migrations.length,
+      );
+      assert.equal(
+        reopened
+          .prepare("SELECT MAX(version) AS version FROM schema_migrations")
+          .get()?.version,
+        57,
+      );
+      for (const { version } of migrations.filter(
+        (item) => item.version >= 43,
+      )) {
+        assert.equal(
+          reopened
+            .prepare(
+              "SELECT COUNT(*) AS count FROM schema_migrations WHERE version = ?",
+            )
+            .get(version)?.count,
+          1,
+        );
+      }
+      assert.deepEqual(reopened.prepare("PRAGMA foreign_key_check").all(), []);
+      assert.equal(
+        reopened
+          .prepare("SELECT discovery_user_context_json FROM deep_scan_runs")
+          .get()?.discovery_user_context_json,
+        legacyCheckpoint !== 47 || mainReader ? context : null,
+      );
+      assertMigrationNames(
+        reopened,
+        47,
+        48,
+        49,
+        50,
+        51,
+        52,
+        53,
+        54,
+        55,
+        56,
+        57,
+      );
+      if (legacyCheckpoint !== 47) {
+        assert.deepEqual(
+          reopened
+            .prepare("SELECT * FROM schema_migrations WHERE version = 47")
+            .get(),
+          originalDiscoveryMigration,
+        );
+      }
+    } finally {
+      reopened.close();
+    }
+  });
+}
+
 test("worker thread history survives retry, replay, and worker retirement", (t) => {
   const database = memory(t, 49);
   insertScan(database);

plugins/codex-security/plugin-files.json

@@ -82,6 +82,7 @@
     "scripts/workbench/__init__.py",
     "scripts/workbench/handoff.py",
     "scripts/workbench/retained_coverage.py",
+    "scripts/workbench/retained_records.py",
     "scripts/workbench/runtime_toml.py",
     "scripts/workbench/saved_coverage.py",
     "scripts/workbench/saved_outputs.py",