diff --git a/plugins/codex-security/mcp-app/src/artifact-scan-draft.ts b/plugins/codex-security/mcp-app/src/artifact-scan-draft.ts index 125a02ce9..32132b51e 100644 --- a/plugins/codex-security/mcp-app/src/artifact-scan-draft.ts +++ b/plugins/codex-security/mcp-app/src/artifact-scan-draft.ts @@ -1,6 +1,7 @@ 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"; import scanDraftDocument from "../../schemas/tools/scan-draft.schema.json"; @@ -187,14 +188,16 @@ 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, - ...(publication === undefined ? {} : { deepScanPublication: publication }), - }), + replaceArtifactJson(draftPath, stagedDraft), ]); const arguments_ = [ "write-scan-draft", @@ -214,6 +217,18 @@ export async function recordCodexSecurityScanDraftViaWorkbench( try { await runWorkbench(arguments_); } 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(); + await runWorkbench(arguments_); + return; + } + } if (!workbenchScanDraftConflict(error)) throw error; throw Object.assign( new Error( @@ -226,6 +241,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 }), ]); } }, diff --git a/plugins/codex-security/mcp-app/tests/clock_workbench.py b/plugins/codex-security/mcp-app/tests/clock_workbench.py index 22d2c52b6..4bf982389 100644 --- a/plugins/codex-security/mcp-app/tests/clock_workbench.py +++ b/plugins/codex-security/mcp-app/tests/clock_workbench.py @@ -1,5 +1,6 @@ """Control the workbench clock while retaining its real commands and database writes.""" +import errno import os import runpy import sys @@ -11,4 +12,22 @@ api = runpy.run_path(str(source), run_name="test_workbench") instant = os.environ["TEST_WORKBENCH_NOW"] api["main"].__globals__["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"]() diff --git a/plugins/codex-security/mcp-app/tests/test_deep_scan_publication_integration.mjs b/plugins/codex-security/mcp-app/tests/test_deep_scan_publication_integration.mjs index 93bf5498d..09e8c7198 100644 --- a/plugins/codex-security/mcp-app/tests/test_deep_scan_publication_integration.mjs +++ b/plugins/codex-security/mcp-app/tests/test_deep_scan_publication_integration.mjs @@ -202,7 +202,7 @@ async function createFixture(t, paths = {}) { transport.stderr?.resume(); await client.connect(transport); return { - run, store, runWorkbench, instant, + run, store, runWorkbench, instant, environment, setTime(offset) { now = new Date(Date.parse(instant) + offset * 1_000).toISOString(); }, call(name, args) { return client.callTool({ name, arguments: args, _meta: { "openai/threadId": owner } }); @@ -219,7 +219,8 @@ async function commitReducers(fixture, offsets, ids) { const id = `10000000-0000-4000-8000-${String(index).padStart(12, "0")}`; const label = `discovery-${String(index).padStart(4, "0")}`; const artifact = await workerArtifact(run, "workers", label); - const update = { id, scanId: run.scanId, kind: "discovery", status: "running", attempt: 1, ...artifact }; + const update = { id, scanId: run.scanId, kind: "discovery", status: "running", attempt: 1, + threadId: `session-${label}`, ...artifact }; await store.updateWorker(update); const resultManifestPath = path.join(artifact.artifactDir, "result.json"); await writeFile(resultManifestPath, JSON.stringify({ scanId: run.scanId, findings: [], coverage })); @@ -230,7 +231,8 @@ async function commitReducers(fixture, offsets, ids) { const artifact = await workerArtifact(run, "dedup", label); const id = ids[batch]; await store.claimDedup({ id, scanId: run.scanId, workerIds, ...artifact }); - await store.updateWorker({ id, scanId: run.scanId, kind: "dedup", status: "running", attempt: 1, ...artifact }); + await store.updateWorker({ id, scanId: run.scanId, kind: "dedup", status: "running", attempt: 1, + threadId: `session-${label}`, ...artifact }); const resultManifestPath = path.join(artifact.artifactDir, "result.json"); await writeFile(resultManifestPath, JSON.stringify({ scanId: run.scanId, findings: [], threatModel: { summary: label } })); fixture.setTime(offsets[batch]); @@ -283,3 +285,226 @@ function assertToolError(result, pattern) { assert.equal(result.isError, true, JSON.stringify(result)); if (pattern) assert.match(JSON.stringify(result), pattern); } + +async function publicationFixture(t, workflow = "deep-scan-mcp/v1") { + const fixture = await createFixture(t); + const { run, store, runWorkbench, environment } = fixture; + const database = path.join(environment.CODEX_SECURITY_STATE_DIR, "workbench.sqlite3"); + const sql = async (script, ...args) => (await execFileAsync(python, + ["-c", script, database, run.scanId, ...args], { env: environment })).stdout; + await sql(`import sqlite3, sys +with sqlite3.connect(sys.argv[1]) as c: + c.execute("UPDATE deep_scan_runs SET workflow_version = ? WHERE scan_id = ?", (sys.argv[3], sys.argv[2])) +`, workflow); + const claim = await store.claimCoordinator({ scanId: run.scanId, threadId: owner }); + assert.equal(claim.run.coordinatorGeneration, 2); + const results = await commitReducers(fixture, [1, 2], [highId, lowId]); + await mkdir(path.join(run.scanDir, "checkpoints"), { recursive: true }); + const draft = { ...JSON.parse(await readFile(results[1], "utf8")), coverage }; + const context = await createScanArtifactContext(run.scanId, runWorkbench, { requireRunning: true }); + return { + ...fixture, sql, results, draft, + publish: (runner = runWorkbench) => recordCodexSecurityScanDraftViaWorkbench( + context, draft, runner, undefined, { coordinatorGeneration: 2, resultPath: results[1] }, + ), + workers: async () => JSON.parse(await sql(`import json, sqlite3, sys +c = sqlite3.connect(sys.argv[1]); c.row_factory = sqlite3.Row +print(json.dumps({ + "workflow": c.execute("SELECT workflow_version FROM deep_scan_runs WHERE scan_id = ?", (sys.argv[2],)).fetchone()[0], + "owner": dict(c.execute("SELECT workspace_id, deep_scan_owner_thread_id, continuation_thread_id, handoff_claim_token FROM scans WHERE id = ?", (sys.argv[2],)).fetchone()), + "workers": [dict(r) for r in c.execute("SELECT * FROM deep_scan_workers WHERE scan_id = ? ORDER BY id", (sys.argv[2],))], + "inputs": [dict(r) for r in c.execute("SELECT * FROM deep_scan_dedup_inputs WHERE dedup_worker_id IN (SELECT id FROM deep_scan_workers WHERE scan_id = ?) ORDER BY dedup_worker_id, discovery_worker_id", (sys.argv[2],))], +}))`)), + }; +} + +function publicationRunner(fixture, { before, after } = {}) { + const requests = []; + let writes = 0; + const run = async (args, input) => { + if (args[0] !== "write-scan-draft") return fixture.runWorkbench(args, input); + const draftPath = args[args.indexOf("--draft-path") + 1]; + const checkpointPath = args[args.indexOf("--checkpoint-path") + 1]; + requests.push({ args: [...args], input, + draft: await readFile(draftPath, "utf8"), checkpoint: await readFile(checkpointPath, "utf8") }); + await before?.(requests.length, { draftPath, checkpointPath }); + const result = await fixture.runWorkbench(args, input); + writes++; + await after?.(requests.length, { draftPath, checkpointPath }); + return result; + }; + return { run, requests, writes: () => writes }; +} + +for (const workflow of ["deep-scan-mcp/v1", "deep-security-scan/v1"]) { + for (const [loseResponse, receiptIoFailure] of [[false, false], [true, false], [false, true]]) { + test(`${workflow} finishes accepted publication (lost=${loseResponse}, receipt I/O failure=${receiptIoFailure})`, async (t) => { + const f = await publicationFixture(t, workflow); + if (receiptIoFailure) f.environment.TEST_WORKBENCH_RECEIPT_IO_FAILURE = "1"; + const before = await f.workers(); + const sources = await Promise.all(f.results.map((file) => readFile(file))); + let accepted; + const runner = publicationRunner(f, { after: async (attempt, { draftPath }) => { + if (attempt === 1) accepted = await snapshot(f.run); + else assert.deepEqual(await snapshot(f.run), accepted); + if (receiptIoFailure) { + await assert.rejects(readFile(draftPath.replace(/\.json$/, ".accepted.json")), { code: "ENOENT" }); + } + if (loseResponse && attempt === 1) throw new Error("Synthetic accepted publication response loss"); + } }); + let publications = 0; + const coordinator = new DeepScanCoordinator({ + run: await f.store.get(f.run.scanId, owner), store: f.store, pluginRoot, + executor: { run() { assert.fail("Publication recovery cannot dispatch workers"); } }, + clock: { now: () => Date.parse(f.instant), sleep: async () => {} }, + onComplete: async (draft, signal, publication) => { + publications++; + assert.deepEqual(draft, f.draft); + assert.deepEqual(publication, { coordinatorGeneration: 2, resultPath: f.results[1] }); + const context = await createScanArtifactContext(f.run.scanId, runner.run, { requireRunning: true }); + await recordCodexSecurityScanDraftViaWorkbench(context, draft, runner.run, signal, publication); + }, + }); + coordinator.start(); + const terminal = await coordinator.settled(); + assert.equal(terminal.status, "succeeded", terminal.error); + assert.equal(terminal.terminalReason, "saturated"); + assert.equal((await f.workers()).workflow, workflow); + assert.equal(terminal.coordinatorGeneration, 2); + assert.equal(publications, 1); + assert.equal(runner.requests.length, loseResponse ? 2 : 1); + assert.equal(runner.writes(), loseResponse ? 2 : 1); + if (loseResponse) assert.deepEqual(runner.requests[1], runner.requests[0]); + assert.deepEqual(await f.workers(), before); + assert.deepEqual(await Promise.all(f.results.map((file) => readFile(file))), sources); + const parent = await f.runWorkbench(["complete-scan", "--scan-id", f.run.scanId]); + assert.equal(parent.scan.progress.status, "complete"); + assert.deepEqual(await f.workers(), before); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); + }); + } +} + +test("receipt I/O failure cannot authorize replay after response loss", async (t) => { + const f = await publicationFixture(t); + f.environment.TEST_WORKBENCH_RECEIPT_IO_FAILURE = "1"; + const workers = await f.workers(); + const lost = new Error("Synthetic accepted publication response loss"); + let accepted; + const runner = publicationRunner(f, { after: async (_attempt, { draftPath }) => { + accepted = await snapshot(f.run); + await assert.rejects(readFile(draftPath.replace(/\.json$/, ".accepted.json")), { code: "ENOENT" }); + throw lost; + } }); + await assert.rejects(f.publish(runner.run), (error) => error === lost); + assert.equal(runner.requests.length, 1); + assert.equal(runner.writes(), 1); + assert.deepEqual(await snapshot(f.run), accepted); + assert.deepEqual(await f.workers(), workers); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); +}); + +for (const code of ["EACCES", "ECONNRESET"]) { + for (const matching of [false, true]) { + test(`publication ${code} before acceptance is one attempt (matching=${matching})`, async (t) => { + const f = await publicationFixture(t); + if (matching) await f.publish(); + const before = await snapshot(f.run); + const workers = await f.workers(); + const original = Object.assign(new Error(`Synthetic precommit ${code}`), { code }); + const runner = publicationRunner(f, { before() { throw original; } }); + await assert.rejects(f.publish(runner.run), (error) => error === original); + assert.equal(runner.requests.length, 1); + assert.equal(runner.writes(), 0); + assert.deepEqual(await snapshot(f.run), before); + assert.deepEqual(await f.workers(), workers); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); + }); + } +} + +for (const takeover of ["coordinator", "continuation"]) { + test(`accepted publication replay rejects a replacement ${takeover}`, async (t) => { + const f = await publicationFixture(t); + const workers = await f.workers(); + let accepted; + const runner = publicationRunner(f, { after: async (attempt) => { + assert.equal(attempt, 1, "A stale replay must not commit"); + accepted = await snapshot(f.run); + if (takeover === "coordinator") { + f.setTime(120); + const replacement = new WorkbenchDeepScanStore(f.runWorkbench); + const claim = await replacement.claimCoordinator({ scanId: f.run.scanId, threadId: owner }); + assert.equal(claim.acquired, true); + assert.equal(claim.run.coordinatorGeneration, 3); + } else { + await f.sql(`import sqlite3, sys +with sqlite3.connect(sys.argv[1]) as c: + c.execute("UPDATE scans SET handoff_claim_token = ?, continuation_thread_id = ? WHERE id = ?", ("00000000-0000-4000-8000-000000000099", "new-owner", sys.argv[2])) +`); + workers.owner.handoff_claim_token = "00000000-0000-4000-8000-000000000099"; + workers.owner.continuation_thread_id = "new-owner"; + } + throw new Error("Synthetic response loss after takeover"); + } }); + await assert.rejects(f.publish(runner.run), takeover === "coordinator" ? /newer generation/ : /another continuation/); + assert.equal(runner.requests.length, 2); + assert.equal(runner.writes(), 1); + assert.deepEqual(runner.requests[1], runner.requests[0]); + assert.deepEqual(await snapshot(f.run), accepted); + assert.deepEqual(await f.workers(), workers); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); + }); +} + +test("a failed accepted publication replay has no third attempt", async (t) => { + const f = await publicationFixture(t); + let accepted; + const failure = new Error("Synthetic replay failure"); + const runner = publicationRunner(f, { + before(attempt) { if (attempt === 2) throw failure; }, + async after() { accepted = await snapshot(f.run); throw new Error("Synthetic accepted response loss"); }, + }); + await assert.rejects(f.publish(runner.run), (error) => error === failure); + assert.equal(runner.requests.length, 2); + assert.equal(runner.writes(), 1); + assert.deepEqual(runner.requests[1], runner.requests[0]); + assert.deepEqual(await snapshot(f.run), accepted); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); +}); + +test("raw publication repeats the same staged paths without changing their bytes", async (t) => { + const f = await publicationFixture(t); + const workers = await f.workers(); + await f.publish(async (args, input) => { + const stagedPaths = ["--draft-path", "--checkpoint-path"].map((flag) => args[args.indexOf(flag) + 1]); + const staged = await Promise.all(stagedPaths.map((file) => readFile(file))); + const first = await f.runWorkbench(args, input); + const accepted = await snapshot(f.run); + assert.deepEqual(await Promise.all(stagedPaths.map((file) => readFile(file))), staged); + assert.deepEqual(await f.runWorkbench(args, input), first); + assert.deepEqual(await Promise.all(stagedPaths.map((file) => readFile(file))), staged); + assert.deepEqual(await snapshot(f.run), accepted); + }); + assert.deepEqual(await f.workers(), workers); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); +}); + +for (const changed of ["draftPath", "checkpointPath"]) { + test(`publication acknowledgment must match the original ${changed}`, async (t) => { + const f = await publicationFixture(t); + const lost = new Error("Synthetic response loss after different staged input"); + const runner = publicationRunner(f, { + async before(attempt, paths) { + const document = JSON.parse(await readFile(paths[changed], "utf8")); + document.threatModel = { summary: "Different staged input" }; + await writeFile(paths[changed], JSON.stringify(document)); + }, + after() { throw lost; }, + }); + await assert.rejects(f.publish(runner.run), (error) => error === lost); + assert.equal(runner.requests.length, 1); + assert.equal(runner.writes(), 1); + assert.deepEqual(await readdir(path.join(f.run.scanDir, "drafts")), []); + }); +} diff --git a/plugins/codex-security/scripts/finalize_scan_contract.py b/plugins/codex-security/scripts/finalize_scan_contract.py index 58cac8f1f..fa6557a81 100644 --- a/plugins/codex-security/scripts/finalize_scan_contract.py +++ b/plugins/codex-security/scripts/finalize_scan_contract.py @@ -83,6 +83,10 @@ class ContractError(ValueError): """Raised when a completed scan does not satisfy the additive contract.""" +class ScanLocalIOError(ContractError): + """Raised when scan-local storage I/O fails, distinct from path validation.""" + + class RecoverableContractError(ContractError): """Raised when report projection can safely be retried before publication.""" @@ -536,15 +540,18 @@ def write_scan_local_bytes( if not _descriptor_relative_writes_available(): if not _is_windows(): raise ContractError("scan-local output requires descriptor-relative file operations") + backend = _windows_scan_local_files() try: - _windows_scan_local_files().atomic_write( + backend.atomic_write( scan_dir, relative_path, payload, expected_root_identity=expected_root_identity, ) - except OSError as exc: + except backend.WindowsScanLocalPathError as exc: raise ContractError(f"{relative_path}: {exc}") from exc + except OSError as exc: + raise ScanLocalIOError(f"{relative_path}: {exc}") from exc return root_fd: int | None = None parent_fd: int | None = None diff --git a/plugins/codex-security/scripts/windows_scan_local_files.py b/plugins/codex-security/scripts/windows_scan_local_files.py index c67c86eb6..3997fd2cb 100644 --- a/plugins/codex-security/scripts/windows_scan_local_files.py +++ b/plugins/codex-security/scripts/windows_scan_local_files.py @@ -37,6 +37,10 @@ class WindowsScanLocalFileError(OSError): """Raised when a scan-local operation cannot be completed securely.""" +class WindowsScanLocalPathError(WindowsScanLocalFileError): + """Raised when a scan-local path fails an integrity check.""" + + # CreateFile access and sharing flags. _DELETE = 0x00010000 _FILE_READ_ATTRIBUTES = 0x00000080 @@ -224,8 +228,8 @@ def _raise_last_error(operation: str, path: Path | None = None) -> None: raise WindowsScanLocalFileError(error, f"{operation}{target}: {detail}", str(path or "")) -def _invalid_path(path: Path | str, reason: str) -> WindowsScanLocalFileError: - return WindowsScanLocalFileError(errno.EINVAL, reason, str(path)) +def _invalid_path(path: Path | str, reason: str) -> WindowsScanLocalPathError: + return WindowsScanLocalPathError(errno.EINVAL, reason, str(path)) def _validated_parts(relative_path: str) -> tuple[str, ...]: diff --git a/plugins/codex-security/scripts/workbench_saved_results.py b/plugins/codex-security/scripts/workbench_saved_results.py index abb428519..6322236d4 100644 --- a/plugins/codex-security/scripts/workbench_saved_results.py +++ b/plugins/codex-security/scripts/workbench_saved_results.py @@ -21,6 +21,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from finalize_scan_contract import ( ContractError, + ScanLocalIOError, _finding_strength, _populate_unsealed_artifact_envelope, _populate_unsealed_manifest_envelope, @@ -1501,6 +1502,32 @@ def write_scan_draft(db: Any, connection: Any, args: Any) -> dict[str, Any]: filename, (json.dumps(document, allow_nan=False, indent=2) + "\n").encode(), ) + if ( + scan["mode"] == "deep" + and draft.get("deepScanPublication") is not None + and db.deep_scan.require_deep_scan_run(connection, scan_id)["workflow_version"] + in {"deep-scan-mcp/v1", "deep-security-scan/v1"} + ): + # Acknowledge this staged operation only after validation and all canonical writes. + acceptance = { + "status": "draft_written", + "input": { + **draft, + "checkpoint": checkpoint if args.checkpoint_path is not None else None, + }, + } + try: + write_scan_local_bytes( + scan_dir, + Path(args.draft_path) + .with_suffix(".accepted.json") + .relative_to(scan_dir) + .as_posix(), + (json.dumps(acceptance, allow_nan=False, indent=2) + "\n").encode(), + ) + except (OSError, ScanLocalIOError): + # Publication succeeded; a missing receipt still prevents lost-response replay. + pass # Accepted Standard drafts are evidence of review or report assembly, # even when the parent omitted its explicit progress call. if scan["mode"] == "standard": diff --git a/plugins/codex-security/tests/test_deep_scan_publication_authority.py b/plugins/codex-security/tests/test_deep_scan_publication_authority.py index 1d17ca8be..0e176f78d 100644 --- a/plugins/codex-security/tests/test_deep_scan_publication_authority.py +++ b/plugins/codex-security/tests/test_deep_scan_publication_authority.py @@ -1,9 +1,15 @@ from __future__ import annotations import copy +import errno import json +import os +import sys import uuid from argparse import Namespace +from contextlib import contextmanager +from pathlib import Path +from types import SimpleNamespace import pytest from test_deep_scan_successful_publication import add_worker @@ -141,3 +147,232 @@ def test_current_publication_replays_without_changing_checkpoint_or_worker_state assert dict(workbench_db.execute("SELECT * FROM deep_scan_workers").fetchone()) == worker_before findings = json.loads((scan.scan_dir / "findings.json").read_text())["findings"] assert findings[0]["title"] == "Accepted aggregate" + + +@pytest.mark.parametrize( + "failure", + ["validation", "findings.json", "coverage.json", "scan-manifest.json"] + + [ + f"windows-emulated:{name}" + for name in ("findings.json", "coverage.json", "scan-manifest.json") + ], +) +def test_failed_publication_does_not_acknowledge_staged_input( + workbench_api, workbench_db, publication_scan, monkeypatch, failure +): + scan = publication_scan() + result = add_worker(workbench_db, scan) + with workbench_db: + workbench_db.execute( + "UPDATE deep_scan_runs SET workflow_version = 'deep-scan-mcp/v1', " + "coordinator_generation = 2 WHERE scan_id = ?", + (scan.scan_id,), + ) + workbench_db.execute( + "UPDATE deep_scan_workers SET kind = 'dedup', merge_state = 'none' WHERE scan_id = ?", + (scan.scan_id,), + ) + args = stage_publication(scan, generation=2, result_path=result, title="Accepted aggregate") + staged = {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} + saved_results = workbench_api["saved_results"] + original_write = saved_results.write_scan_local_bytes + + def fail_validation(*args): + raise OSError("Synthetic validation failure") + + def fail_write(scan_dir, filename, contents): + if filename == failure: + raise OSError("Synthetic canonical write failure") + if failure == f"windows-emulated:{filename}": + with monkeypatch.context() as patch: + finalizer = sys.modules[original_write.__module__] + backend = emulate_windows_atomic_write(patch, finalizer) + patch.setattr(backend, "_rename_handle", fail_validation) + return original_write(scan_dir, filename, contents) + original_write(scan_dir, filename, contents) + + if failure == "validation": + monkeypatch.setattr(saved_results, "_validate_completion_binding", fail_validation) + else: + monkeypatch.setattr(saved_results, "write_scan_local_bytes", fail_write) + + expected_error = ( + saved_results.ContractError if failure.startswith("windows-emulated:") else OSError + ) + with pytest.raises(expected_error, match="Synthetic"): + workbench_api["write_scan_draft"](workbench_db, args) + + assert {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} == staged + + +def emulate_windows_atomic_write(monkeypatch, finalizer, *, reparse_point=False): + """Run the real Windows atomic writer with emulated handles, not native Win32 I/O.""" + backend = finalizer._windows_scan_local_files() + paths = {} + pending_deletions = set() + + @contextmanager + def locked_parent(scan_dir, relative_path, **kwargs): + parts = backend._validated_parts(relative_path) + yield scan_dir.joinpath(*parts[:-1]), parts[-1] + + def create_file(path, **kwargs): + descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + paths[descriptor] = path + return backend._OwnedHandle(descriptor) + + def rename_handle(handle, destination): + os.replace(paths[handle], destination) + paths[handle] = destination + + def close_handle(handle): + os.close(handle) + if handle in pending_deletions: + paths[handle].unlink() + + monkeypatch.setattr(finalizer, "_descriptor_relative_writes_available", lambda: False) + monkeypatch.setattr(finalizer, "_is_windows", lambda: True) + monkeypatch.setattr(backend, "_locked_parent", locked_parent) + monkeypatch.setattr(backend, "_validate_existing_output", lambda path: None) + monkeypatch.setattr(backend, "_create_file", create_file) + monkeypatch.setattr(backend, "_close_handle", close_handle) + monkeypatch.setattr( + backend, + "_attributes", + lambda handle: SimpleNamespace( + FileAttributes=backend._FILE_ATTRIBUTE_REPARSE_POINT if reparse_point else 0 + ), + ) + monkeypatch.setattr( + backend, "_GetFileType", lambda handle: backend._FILE_TYPE_DISK, raising=False + ) + monkeypatch.setattr(backend, "_verify_handle_path", lambda *args: None) + monkeypatch.setattr(backend, "_write_all", os.write) + monkeypatch.setattr(backend, "_rename_handle", rename_handle) + monkeypatch.setattr(backend, "_mark_handle_for_deletion", pending_deletions.add) + return backend + + +@pytest.mark.parametrize("workflow", ["deep-scan-mcp/v1", "deep-security-scan/v1"]) +@pytest.mark.parametrize( + ("backend_kind", "failure"), + [("host", "write"), ("host", "rename")] + + [("windows-emulated", failure) for failure in ("create", "write", "rename")], +) +def test_receipt_io_failure_preserves_successful_publication( + workbench_api, workbench_db, publication_scan, monkeypatch, workflow, backend_kind, failure +): + scan = publication_scan() + result = add_worker(workbench_db, scan) + with workbench_db: + workbench_db.execute( + "UPDATE deep_scan_runs SET workflow_version = ?, coordinator_generation = 2 " + "WHERE scan_id = ?", + (workflow, scan.scan_id), + ) + workbench_db.execute( + "UPDATE deep_scan_workers SET kind = 'dedup', merge_state = 'none' WHERE scan_id = ?", + (scan.scan_id,), + ) + args = stage_publication(scan, generation=2, result_path=result, title="Accepted aggregate") + draft = json.loads(Path(args.draft_path).read_text()) + staged = {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} + saved_results = workbench_api["saved_results"] + original_write = saved_results.write_scan_local_bytes + original_replace = os.replace + failures = [] + + def fail_receipt(filename): + failures.append(filename) + # The fault occurs only after all three canonical files have been published. + for name, document in ( + ("findings.json", draft["findings"]), + ("coverage.json", draft["coverage"]), + ("scan-manifest.json", draft["manifest"]), + ): + assert json.loads((scan.scan_dir / name).read_text()) == document + raise OSError(errno.ENOSPC, "Synthetic receipt I/O failure", filename) + + def write_file(scan_dir, filename, contents): + if not filename.endswith(".accepted.json"): + return original_write(scan_dir, filename, contents) + with monkeypatch.context() as patch: + finalizer = sys.modules[original_write.__module__] + if backend_kind == "windows-emulated": + backend = emulate_windows_atomic_write(patch, finalizer) + elif os.name == "nt": + backend = finalizer._windows_scan_local_files() + else: + if failure == "write": + fail_receipt(filename) + return original_write(scan_dir, filename, contents) + + def fail_backend(*args, **kwargs): + fail_receipt(filename) + + operation = { + "create": "_create_file", + "write": "_write_all", + "rename": "_rename_handle", + } + patch.setattr(backend, operation[failure], fail_backend) + return original_write(scan_dir, filename, contents) + + def replace_file(source, destination, *args, **kwargs): + if failure == "rename" and str(destination).endswith(".accepted.json"): + fail_receipt(destination) + return original_replace(source, destination, *args, **kwargs) + + monkeypatch.setattr(saved_results, "write_scan_local_bytes", write_file) + monkeypatch.setattr(os, "replace", replace_file) + + assert workbench_api["write_scan_draft"](workbench_db, args) == { + "scanId": scan.scan_id, + "status": "draft_written", + } + assert len(failures) == 1 + assert {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} == staged + assert len(list((scan.scan_dir / "checkpoints").glob("*.json"))) == 1 + + +@pytest.mark.parametrize("failure", ["validation", "unsafe-path", "windows-reparse-emulated"]) +def test_receipt_contract_errors_still_reject_publication( + workbench_api, workbench_db, publication_scan, monkeypatch, failure +): + scan = publication_scan() + result = add_worker(workbench_db, scan) + with workbench_db: + workbench_db.execute( + "UPDATE deep_scan_runs SET workflow_version = 'deep-scan-mcp/v1', " + "coordinator_generation = 2 WHERE scan_id = ?", + (scan.scan_id,), + ) + workbench_db.execute( + "UPDATE deep_scan_workers SET kind = 'dedup', merge_state = 'none' WHERE scan_id = ?", + (scan.scan_id,), + ) + args = stage_publication(scan, generation=2, result_path=result, title="Accepted aggregate") + staged = {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} + saved_results = workbench_api["saved_results"] + original_write = saved_results.write_scan_local_bytes + finalizer = sys.modules[original_write.__module__] + attempts = [] + + def write_file(scan_dir, filename, contents): + if not filename.endswith(".accepted.json"): + return original_write(scan_dir, filename, contents) + attempts.append(filename) + if failure == "validation": + raise finalizer.ContractError("Synthetic receipt validation failure") + if failure == "unsafe-path": + return original_write(scan_dir, "../outside.accepted.json", contents) + with monkeypatch.context() as patch: + emulate_windows_atomic_write(patch, finalizer, reparse_point=True) + return original_write(scan_dir, filename, contents) + + monkeypatch.setattr(saved_results, "write_scan_local_bytes", write_file) + with pytest.raises(finalizer.ContractError, match="validation|safe|reparse"): + workbench_api["write_scan_draft"](workbench_db, args) + assert len(attempts) == 1 + assert {path: path.read_bytes() for path in (scan.scan_dir / "drafts").iterdir()} == staged + assert not (scan.scan_dir.parent / "outside.accepted.json").exists()