fix(deep-scan): recover publication after a lost reply

plugins/codex-security/mcp-app/server.ts

@@ -1185,7 +1185,7 @@ export function createCodexSecurityServer(): McpServer {
               log: logDeepScanEvent,
               handoffClaimToken,
               threadId,
-              onComplete: async (draft, signal) => {
+              onComplete: async (draft, signal, publication) => {
                 const context = await createScanArtifactContext(
                   begun.scanId,
                   runWorkbench,
@@ -1206,6 +1206,7 @@ export function createCodexSecurityServer(): McpServer {
                   },
                   runWorkbench,
                   signal,
+                  publication,
                 );
               },
               onStopped: async (run) => {

plugins/codex-security/mcp-app/src/artifact-scan-draft.ts

@@ -2,7 +2,7 @@ import type { JsonObject } from "./types.js";
 import { isRecord as isObject } from "./record.js";
 import { createHash, randomUUID } from "node:crypto";
 import { promises as fs } from "node:fs";
-import { dirname, join, sep } from "node:path";
+import { basename, dirname, join, sep } from "node:path";
 import { isDeepStrictEqual } from "node:util";
 import type * as z from "zod/v4";
 import commonSchema from "../../schemas/definitions/artifact-common.schema.json";
@@ -57,6 +57,12 @@ interface PreparedScanDraft {
   coverage: JsonObject;
 }
 
+/** Host-selected Deep aggregate, separate from model-authored draft fields. */
+export interface DeepScanPublication {
+  coordinatorGeneration?: number;
+  resultPath: string | null;
+}
+
 interface SavedScanDraft {
   input: ScanDraftInput;
   modifiedMs: number;
@@ -207,6 +213,7 @@ export async function recordCodexSecurityScanDraftViaWorkbench(
   input: ScanDraftInput,
   runWorkbench: RunArtifactWorkbench,
   signal?: AbortSignal,
+  publication?: DeepScanPublication,
 ): Promise<ScanDraftResult> {
   return recordCodexSecurityScanDraft(
     context,
@@ -222,11 +229,18 @@ export async function recordCodexSecurityScanDraftViaWorkbench(
         ["drafts", `${randomUUID()}.json`],
         "staged scan draft",
       );
+      const acceptanceName = `${basename(draftPath, ".json")}.accepted.json`;
       try {
         const { handoffClaimToken: _claim, ...snapshot } = checkpoint;
+        const stagedDraft = {
+          ...draft,
+          ...(publication === undefined
+            ? {}
+            : { deepScanPublication: publication }),
+        };
         await Promise.all([
           replaceArtifactJson(checkpointPath, snapshot),
-          replaceArtifactJson(draftPath, draft),
+          replaceArtifactJson(draftPath, stagedDraft),
         ]);
         const arguments_ = [
           "write-scan-draft",
@@ -243,14 +257,38 @@ export async function recordCodexSecurityScanDraftViaWorkbench(
         if (context.handoffClaimToken) {
           arguments_.push("--claim-token", context.handoffClaimToken);
         }
-        try {
+        const publish = async () => {
           const result = await runWorkbench(arguments_);
           return Array.isArray(result?.warnings)
             ? result.warnings.filter(
                 (warning): warning is string => typeof warning === "string",
               )
             : undefined;
+        };
+        try {
+          return await publish();
         } catch (error) {
+          if (context.mode === "deep" && publication !== undefined) {
+            const accepted = await readArtifactJsonObject(
+              context,
+              ["drafts", acceptanceName],
+              "accepted scan draft",
+            ).catch(() => undefined);
+            if (
+              isDeepStrictEqual(
+                accepted,
+                JSON.parse(
+                  JSON.stringify({
+                    status: "draft_written",
+                    input: { ...stagedDraft, checkpoint: snapshot },
+                  }),
+                ),
+              )
+            ) {
+              signal?.throwIfAborted();
+              return await publish();
+            }
+          }
           if (!workbenchScanDraftConflict(error)) throw error;
           throw Object.assign(
             new Error(
@@ -263,6 +301,7 @@ export async function recordCodexSecurityScanDraftViaWorkbench(
         await Promise.all([
           fs.rm(checkpointPath, { force: true }),
           fs.rm(draftPath, { force: true }),
+          fs.rm(join(dirname(draftPath), acceptanceName), { force: true }),
         ]);
       }
     },

plugins/codex-security/mcp-app/src/deep-scan/coordinator.ts

@@ -1,7 +1,7 @@
 /// <reference lib="es2023.array" />
 import { randomUUID } from "node:crypto";
 import { promises as fs } from "node:fs";
-import { join } from "node:path";
+import { basename, dirname, join } from "node:path";
 import { setTimeout as delay } from "node:timers/promises";
 import {
   createDeepScanArtifacts,
@@ -14,6 +14,7 @@ import {
 } from "./artifact-validation.js";
 import {
   scanDraftInputSchema,
+  type DeepScanPublication,
   type ScanDraftInput,
 } from "../artifact-scan-draft.js";
 import type { DeepScanArtifacts } from "./artifacts.js";
@@ -56,6 +57,7 @@ interface SchedulerResult {
   reason: DeepScanTerminalReason;
   omittedWorkerIds: string[];
   result?: DeepReductionInput;
+  resultPath?: string;
 }
 
 export interface CoordinatorOptions {
@@ -72,7 +74,11 @@ export interface CoordinatorOptions {
   threadId?: string;
   heartbeatIntervalMs?: number;
   observeReplacement?: (run: DeepScanRunState) => Promise<DeepScanRunState>;
-  onComplete?: (draft: ScanDraftInput, signal: AbortSignal) => Promise<void>;
+  onComplete?: (
+    draft: ScanDraftInput,
+    signal: AbortSignal,
+    publication: DeepScanPublication,
+  ) => Promise<void>;
   onStopped?: (run: DeepScanRunState) => Promise<void>;
 }
 
@@ -318,6 +324,10 @@ export class DeepScanCoordinator {
       await this.options.onComplete?.(
         draft,
         this.publicationAbortController.signal,
+        {
+          coordinatorGeneration: this.state.coordinatorGeneration,
+          resultPath: schedulerResult.resultPath ?? null,
+        },
       );
       if (this.canceled || this.externallyFailed) return;
       this.state = await this.finishWithReplay(schedulerResult);
@@ -924,6 +934,7 @@ export class DeepScanCoordinator {
       reason: stopReason,
       omittedWorkerIds: [...new Set(omittedWorkerIds)],
       result: latestResult,
+      resultPath: previousReducerResultPath,
     };
   }
 
@@ -1111,7 +1122,9 @@ function workerLabelSequence(
   worker: PersistedDeepScanWorker,
   kind: "discovery" | "dedup",
 ): number {
-  const match = worker.promptPath.match(new RegExp(`${kind}-(\\d+)`));
+  const match = basename(dirname(worker.promptPath)).match(
+    new RegExp(`${kind}-(\\d+)`),
+  );
   return match ? Number(match[1]) : 0;
 }
 

plugins/codex-security/mcp-app/tests/clock_workbench.py

@@ -0,0 +1,34 @@
+"""Control the workbench clock while retaining its real commands and database writes."""
+
+import errno
+import os
+import runpy
+import sys
+from pathlib import Path
+
+source = Path(sys.argv[1])
+sys.argv = [str(source), *sys.argv[2:]]
+sys.path.insert(0, str(source.parent))
+api = runpy.run_path(str(source), run_name="test_workbench")
+instant = os.environ["TEST_WORKBENCH_NOW"]
+api["main"].__globals__["now"] = lambda: instant
+api["_WORKBENCH_DB_CONTEXT"].now = lambda: instant
+if os.environ.get("TEST_WORKBENCH_RECEIPT_IO_FAILURE"):
+    if os.name == "nt":
+        from finalize_scan_contract import _windows_scan_local_files
+
+        backend = _windows_scan_local_files()
+        original_replace = backend._rename_handle
+    else:
+        original_replace = os.replace
+
+    def replace(source, destination, *args, **kwargs):
+        if str(destination).endswith(".accepted.json"):
+            raise OSError(errno.ENOSPC, "Synthetic receipt I/O failure", destination)
+        return original_replace(source, destination, *args, **kwargs)
+
+    if os.name == "nt":
+        backend._rename_handle = replace
+    else:
+        os.replace = replace
+api["main"]()

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

@@ -838,6 +838,10 @@ try {
     "obsolete.json",
   );
   await writeFile(obsoleteCheckpointPath, "{malformed obsolete checkpoint\n");
+  const deepPublication = {
+    coordinatorGeneration: 3,
+    resultPath: path.join(deepParentRoot, "workers", "reducer", "result.json"),
+  };
   const deepWorkbenchWrites = mock.fn(async (arguments_: string[]) => {
     assert.deepEqual(arguments_.slice(0, 3), [
       "write-scan-draft",
@@ -851,6 +855,8 @@ try {
       arguments_[arguments_.indexOf("--checkpoint-path") + 1];
     const staged = await readJson(draftPath);
     const stagedCheckpoint = await readJson(checkpointPath);
+    assert.deepEqual(staged.deepScanPublication, deepPublication);
+    assert.equal(stagedCheckpoint.deepScanPublication, undefined);
     assert.deepEqual(staged.findings, acceptedDeepFindings);
     assert.deepEqual(staged.coverage, acceptedDeepCoverage);
     assert.deepEqual(stagedCheckpoint.findings, acceptedDeepDraft.findings);
@@ -860,6 +866,8 @@ try {
     deepParentContext,
     acceptedDeepDraft,
     deepWorkbenchWrites,
+    undefined,
+    deepPublication,
   );
   assert.equal(
     deepWorkbenchWrites.mock.callCount(),

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

@@ -3446,6 +3446,8 @@ async function testSaturationIgnoresDiscoveryCancellationWriteFailure() {
 
 async function testPublicationUsesAcceptedReducerSnapshot() {
   const { fixture, store } = await coordinatorFixture();
+  fixture.run.coordinatorGeneration = 3;
+  store.run.coordinatorGeneration = 3;
   const commitDedup = store.commitDedup.bind(store);
   store.commitDedup = async (commit) => {
     const accepted = await commitDedup(commit);
@@ -3462,12 +3464,19 @@ async function testPublicationUsesAcceptedReducerSnapshot() {
     return structuredClone(store.run);
   };
   const completed: ScanDraftInput[] = [];
+  const published: {
+    coordinatorGeneration?: number;
+    resultPath: string | null;
+  }[] = [];
   const terminal = await runCoordinator(
     fixture,
     store,
     new FakeExecutor({ discoveryCandidateId: "accepted-finding" }),
     {
-      onComplete: async (draft) => void completed.push(structuredClone(draft)),
+      onComplete: async (draft, _signal, publication) => {
+        completed.push(structuredClone(draft));
+        published.push(publication);
+      },
     },
   );
   assert.equal(terminal?.status, "succeeded", terminal?.error);
@@ -3477,8 +3486,13 @@ async function testPublicationUsesAcceptedReducerSnapshot() {
     "accepted-finding",
   );
   assert.equal(completed[0].coverage.completeness, "complete");
+  const reducer = [...store.workers.values()].find(
+    (worker) => worker.kind === "dedup",
+  );
+  assert.deepEqual(published, [
+    { coordinatorGeneration: 3, resultPath: reducer!.resultManifestPath },
+  ]);
 }
-
 async function testResumeRequiresHistoricalWorkerPrompt(
   status: "failed" | "canceled",
 ) {