From fb024951883fbf841979b020b116d9ee9bdb74fb Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Fri, 30 Oct 2026 08:26:52 -0700 Subject: [PATCH] fix: share bundled SkillSpector capacity across two roots --- .../lib/runWorkerCommand.cancellation.test.ts | 44 ++++ .../runWorkerCommand.cleanup-timers.test.ts | 24 +++ scripts/lib/runWorkerCommand.cleanup.test.ts | 14 ++ scripts/lib/runWorkerCommand.ts | 57 +++++- .../bundled-skillspector-diagnostic.ts | 1 + .../run-codex-scan-worker-clawscan.test.ts | 5 +- ...run-codex-scan-worker-skillspector.test.ts | 188 +++++++++++++++++- scripts/security/run-codex-scan-worker.ts | 92 +++++++-- specs/security-moderation.md | 13 ++ 9 files changed, 409 insertions(+), 29 deletions(-) create mode 100644 scripts/lib/runWorkerCommand.cancellation.test.ts diff --git a/scripts/lib/runWorkerCommand.cancellation.test.ts b/scripts/lib/runWorkerCommand.cancellation.test.ts new file mode 100644 index 0000000000..8a300401c1 --- /dev/null +++ b/scripts/lib/runWorkerCommand.cancellation.test.ts @@ -0,0 +1,44 @@ +/* @vitest-environment node */ +import { getEventListeners } from "node:events"; +import { mkdtemp, rm, stat } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it, vi } from "vitest"; +import { runWorkerCommand } from "./runWorkerCommand"; + +it("does not launch work that was cancelled before admission", async () => { + const root = await mkdtemp(join(tmpdir(), "worker-preabort-")); + try { + const marker = join(root, "spawned"); + const controller = new AbortController(); + controller.abort(); + const onTermination = vi.fn(); + await expect( + runWorkerCommand( + process.execPath, + ["-e", "require('node:fs').writeFileSync(process.argv[1], 'ran')", marker], + { cwd: root, env: {}, timeoutMs: 1000, signal: controller.signal, onTermination }, + ), + ).rejects.toMatchObject({ message: "Worker command cancelled", failureKind: "cancelled" }); + await expect(stat(marker)).rejects.toMatchObject({ code: "ENOENT" }); + expect(onTermination).not.toHaveBeenCalled(); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + +it("removes the cancellation listener when ordinary work finishes", async () => { + const controller = new AbortController(); + await expect( + runWorkerCommand(process.execPath, ["-e", "process.stdout.write('finished')"], { + cwd: tmpdir(), + env: {}, + timeoutMs: 1000, + signal: controller.signal, + cleanupProcessGroup: true, + }), + ).resolves.toEqual({ stdout: "finished", stderr: "" }); + expect(getEventListeners(controller.signal, "abort")).toHaveLength(0); + controller.abort(); +}); diff --git a/scripts/lib/runWorkerCommand.cleanup-timers.test.ts b/scripts/lib/runWorkerCommand.cleanup-timers.test.ts index 502204ba8b..d1f5cbe3f3 100644 --- a/scripts/lib/runWorkerCommand.cleanup-timers.test.ts +++ b/scripts/lib/runWorkerCommand.cleanup-timers.test.ts @@ -235,6 +235,30 @@ it.skipIf(process.platform === "win32").each([-1_000_000, 1_000_000])( }, ); +it.skipIf(process.platform === "win32")( + "classifies unconfirmed cleanup separately from a findings exit", + async () => { + vi.useFakeTimers(); + const child = simulatedChild(); + vi.spyOn(process, "kill").mockImplementation(() => true); + const pending = runWorkerCommand("generated", [], { + cwd: tmpdir(), + env: {}, + timeoutMs: 100, + cleanupProcessGroup: true, + }).catch((error: unknown) => error); + child.emit("close", 1); + await vi.advanceTimersByTimeAsync(60_000); + expect(await pending).toMatchObject({ + exitCode: 1, + timedOut: false, + failureKind: "cleanup", + message: "Worker cleanup could not be confirmed", + }); + expect(vi.getTimerCount()).toBe(0); + }, +); + it("does not claim Windows process-group absence from direct child termination", async () => { vi.useFakeTimers(); Object.defineProperty(process, "platform", { value: "win32" }); diff --git a/scripts/lib/runWorkerCommand.cleanup.test.ts b/scripts/lib/runWorkerCommand.cleanup.test.ts index caab30db9b..d2db422692 100644 --- a/scripts/lib/runWorkerCommand.cleanup.test.ts +++ b/scripts/lib/runWorkerCommand.cleanup.test.ts @@ -16,8 +16,10 @@ it "nonzero_exit", "output_limit", "observer_failure", + "abort", ])("confirms descendant group absence after %s", async (trigger) => { const shouldTimeout = trigger === "timeout" || trigger === "inherited_pipes"; + const controller = new AbortController(); let groupPid: number | undefined; let termination: WorkerCommandTermination | undefined; try { @@ -48,9 +50,11 @@ it captureOutput: false, maxStdoutBytes: 128, cleanupProcessGroup: true, + signal: controller.signal, onStderrChunk: (chunk) => { groupPid = Number(chunk.toString()); if (trigger === "observer_failure") throw new Error("PRIVATE observer failure"); + if (trigger === "abort") controller.abort(); }, onTermination: (value) => { termination = value; @@ -61,6 +65,16 @@ it else { expect(failure).toBeInstanceOf(CommandFailure); expect(failure).toMatchObject({ timedOut: shouldTimeout }); + const failureKind = shouldTimeout + ? "timeout" + : trigger === "nonzero_exit" + ? "exit" + : trigger === "observer_failure" + ? "observer" + : trigger === "abort" + ? "cancelled" + : "output_limit"; + expect(failure).toMatchObject({ failureKind }); if (trigger === "nonzero_exit") expect(failure).toMatchObject({ exitCode: 7 }); if (trigger === "output_limit") expect(failure).toMatchObject({ outputLimitExceeded: "stdout" }); diff --git a/scripts/lib/runWorkerCommand.ts b/scripts/lib/runWorkerCommand.ts index 21c39ed26b..8683f3130e 100644 --- a/scripts/lib/runWorkerCommand.ts +++ b/scripts/lib/runWorkerCommand.ts @@ -14,6 +14,7 @@ export type WorkerCommandTermination = { export class CommandFailure extends Error { exitCode: number | null; + failureKind: "exit" | "timeout" | "output_limit" | "observer" | "cleanup" | "cancelled"; outputLimitExceeded?: "stdout" | "stderr"; stderr: string; stdout: string; @@ -26,10 +27,16 @@ export class CommandFailure extends Error { stderr: string, timedOut: boolean, outputLimitExceeded?: "stdout" | "stderr", + failureKind: CommandFailure["failureKind"] = timedOut + ? "timeout" + : outputLimitExceeded + ? "output_limit" + : "exit", ) { super(message); this.name = "CommandFailure"; this.exitCode = exitCode; + this.failureKind = failureKind; this.stdout = stdout; this.stderr = stderr; this.timedOut = timedOut; @@ -53,6 +60,7 @@ export async function runWorkerCommand( cleanupProcessGroup?: boolean; onStderrChunk?: (chunk: Buffer) => void; onTermination?: (termination: WorkerCommandTermination) => void; + signal?: AbortSignal; timeoutMs: number; }, ) { @@ -69,6 +77,17 @@ export async function runWorkerCommand( throw new RangeError(`${stream} output limit must be a non-negative safe integer`); } } + if (options.signal?.aborted) { + throw new CommandFailure( + "Worker command cancelled", + null, + "", + "", + false, + undefined, + "cancelled", + ); + } return await new Promise<{ stdout: string; stderr: string }>((resolvePromise, reject) => { const startedAt = performance.now(); const child = spawn(command, args, { @@ -82,6 +101,7 @@ export async function runWorkerCommand( let outputLimitExceeded: "stdout" | "stderr" | undefined; let observerFailed = false; let timedOut = false; + let cancelled = false; let settled = false; let forceKillTimeout: NodeJS.Timeout | undefined; let cleanupTimeout: NodeJS.Timeout | undefined; @@ -141,6 +161,7 @@ export async function runWorkerCommand( output.stderr, timedOut, outputLimitExceeded, + "cleanup", ), ); }, 60_000); @@ -174,14 +195,23 @@ export async function runWorkerCommand( clearTimeout(timeout); if (forceKillTimeout) clearTimeout(forceKillTimeout); if (cleanupTimeout) clearTimeout(cleanupTimeout); + options.signal?.removeEventListener("abort", handleAbort); }; + function handleAbort() { + if (settled) return; + cancelled = true; + clearTimeout(timeout); + terminateProcessTree(); + } + options.signal?.addEventListener("abort", handleAbort, { once: true }); + if (options.signal?.aborted) handleAbort(); child.stdout.on("data", (chunk: Buffer) => appendOutput("stdout", chunk)); child.stderr.on("data", (chunk: Buffer) => appendOutput("stderr", chunk)); child.on("error", (error) => { if (settled) return; settled = true; clearTimers(); - if (timedOut) killProcessTree("SIGKILL"); + if (timedOut || cancelled) killProcessTree("SIGKILL"); reportTermination(null); reject(error); }); @@ -190,7 +220,11 @@ export async function runWorkerCommand( settled = true; clearTimers(); const cleanupRequired = - timedOut || outputLimitExceeded || observerFailed || options.cleanupProcessGroup; + timedOut || + cancelled || + outputLimitExceeded || + observerFailed || + options.cleanupProcessGroup; if (cleanupRequired) { cleanupStartedAt ??= performance.now(); killProcessTree("SIGKILL"); @@ -219,6 +253,8 @@ export async function runWorkerCommand( output.stdout, output.stderr, timedOut, + undefined, + "observer", ), ); return; @@ -233,6 +269,7 @@ export async function runWorkerCommand( output.stderr, timedOut, outputLimitExceeded, + "output_limit", ), ); return; @@ -245,6 +282,22 @@ export async function runWorkerCommand( output.stdout, output.stderr, timedOut, + undefined, + "cleanup", + ), + ); + return; + } + if (cancelled && !timedOut) { + reject( + new CommandFailure( + "Worker command cancelled", + code, + output.stdout, + output.stderr, + false, + undefined, + "cancelled", ), ); return; diff --git a/scripts/security/bundled-skillspector-diagnostic.ts b/scripts/security/bundled-skillspector-diagnostic.ts index 4372e7962a..3fd7692d85 100644 --- a/scripts/security/bundled-skillspector-diagnostic.ts +++ b/scripts/security/bundled-skillspector-diagnostic.ts @@ -21,6 +21,7 @@ export type BundledSkillSpectorFailureContext = { /** Index in the worker's sorted, deduplicated declared-root list. */ rootIndex: number; rootCount: number; + /** Accepted terminal reports; concurrent roots need not finish in index order. */ completedRoots: number; childBudgetMs: number; remainingStageBudgetMs: number; diff --git a/scripts/security/run-codex-scan-worker-clawscan.test.ts b/scripts/security/run-codex-scan-worker-clawscan.test.ts index 227416ee93..fb75868c1c 100644 --- a/scripts/security/run-codex-scan-worker-clawscan.test.ts +++ b/scripts/security/run-codex-scan-worker-clawscan.test.ts @@ -1081,7 +1081,7 @@ JSON`, false, ); - expect((await readFile(skillSpectorTargets, "utf8")).trim().split("\n")).toEqual([ + expect((await readFile(skillSpectorTargets, "utf8")).trim().split("\n").sort()).toEqual([ "artifact/package/skills/alpha", "artifact/package/skills/beta", ]); @@ -1909,6 +1909,9 @@ exit 7`, const fakeClawScan = join(workspace, "fake-clawscan"); const callLog = join(workspace, "clawscan-called"); const firstRootCompleted = join(workspace, "first-root-completed"); + // Keep these exact serial-provider budget and redaction contracts; + // parallel out-of-order accounting is covered separately. + vi.stubEnv("SKILLSPECTOR_PROVIDER", "anthropic"); if (failure === "shared deadline") { const now = Date.now.bind(Date); vi.spyOn(Date, "now").mockImplementation( diff --git a/scripts/security/run-codex-scan-worker-skillspector.test.ts b/scripts/security/run-codex-scan-worker-skillspector.test.ts index 76da32cd5f..6fdb68a30c 100644 --- a/scripts/security/run-codex-scan-worker-skillspector.test.ts +++ b/scripts/security/run-codex-scan-worker-skillspector.test.ts @@ -4,7 +4,8 @@ import { chmod, mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/pr import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, expect, it, vi } from "vitest"; -import { type ClaimedJob, runClawScan } from "./run-codex-scan-worker"; +import * as workerCommand from "../lib/runWorkerCommand"; +import { type ClaimedJob, runBundledSkillSpector, runClawScan } from "./run-codex-scan-worker"; const workspaces: string[] = []; @@ -78,6 +79,139 @@ function writeReport(report: unknown) { printf '%s' '${JSON.stringify(report)}' > "$2"`; } +it("runs two bundled roots together without multiplying the model-call ceiling", async () => { + const roots = ["skills/first", "skills/second"]; + const { run, handoff, workspace } = await fixture( + `root="$2" +if [[ "$root" == *first ]]; then + while [[ ! -e second-completed ]]; do sleep 0.02; done +fi +printf '%s\\n' "\${SKILLSPECTOR_MAX_LLM_CONCURRENCY:-unset}" >> limits +${writeReport(findings)} +if [[ "$root" == *second ]]; then touch second-completed; fi`, + roots, + ); + vi.stubEnv("CODEX_SECURITY_SCAN_CLAWSCAN_TIMEOUT_MS", "1500"); + await expect(run()).rejects.toThrow("exited 23"); + expect((await readFile(join(workspace, "limits"), "utf8")).trim().split("\n")).toEqual([ + "5", + "5", + ]); + expect(JSON.parse(await readFile(handoff, "utf8"))).toMatchObject({ + status: "suspicious", + issue_count: 2, + filtered_findings: roots.map((root) => ({ file: `${root}/SKILL.md` })), + }); +}); + +it("retains serial execution and provider concurrency defaults outside OpenAI", async () => { + const { run, workspace } = await fixture( + `mkdir serial-lock +trap 'rmdir serial-lock' EXIT +printf '%s\\n' "\${SKILLSPECTOR_MAX_LLM_CONCURRENCY:-unset}" >> limits +${writeReport(findings)}`, + ["skills/first", "skills/second"], + ); + vi.stubEnv("SKILLSPECTOR_PROVIDER", "anthropic"); + await expect(run()).rejects.toThrow("exited 23"); + expect((await readFile(join(workspace, "limits"), "utf8")).trim().split("\n")).toEqual([ + "unset", + "unset", + ]); +}); + +it("cancels and joins a sibling after a failed root before admitting another root", async () => { + const originalRun = workerCommand.runWorkerCommand; + let siblingTermination: + | { cancelled: boolean; timedOut: boolean; processGroup: string } + | undefined; + vi.spyOn(workerCommand, "runWorkerCommand").mockImplementation((command, args, options) => + originalRun(command, args, { + ...options, + onTermination: (termination) => { + if (command === "skillspector" && args[1]?.endsWith("second")) { + siblingTermination = { + cancelled: options.signal?.aborted === true, + timedOut: termination.timedOut, + processGroup: termination.processGroup, + }; + } + options.onTermination?.(termination); + }, + }), + ); + const { run, handoff, workspace } = await fixture( + `if [[ "$2" == *first ]]; then + while [[ ! -e second-started ]]; do sleep 0.02; done + exit 7 +fi +sleep 30 & +printf '%s' "$!" > sibling-pid +touch second-started +wait`, + ["skills/first", "skills/second", "skills/third"], + ); + vi.stubEnv("CODEX_SECURITY_SCAN_CLAWSCAN_TIMEOUT_MS", "2000"); + await expect(run()).rejects.toThrow("skillspector exited 7"); + expect(siblingTermination).toEqual({ cancelled: true, timedOut: false, processGroup: "absent" }); + await expect(readFile(handoff)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(readFile(join(workspace, "skillspector-report-2.json"))).rejects.toMatchObject({ + code: "ENOENT", + }); +}); + +it("counts accepted reports independently of the failed root's sorted index", async () => { + const roots = ["skills/first", "skills/second", "skills/third", "skills/fourth"]; + const { workspace } = await fixture("", roots); + let releaseFirst!: () => void; + const firstFailure = new Promise((resolve) => { + releaseFirst = resolve; + }); + const calls: string[] = []; + vi.spyOn(workerCommand, "runWorkerCommand").mockImplementation( + async (_command, args, options) => { + const root = args[1]!; + calls.push(root); + if (root.endsWith("first")) { + await firstFailure; + throw new workerCommand.CommandFailure("fixture root failed", 2, "", "", false); + } + if (root.endsWith("second")) { + await writeFile(args[args.indexOf("--output") + 1]!, JSON.stringify(findings)); + return { stdout: "", stderr: "" }; + } + if (root.endsWith("third")) { + // Admission of this next root proves the second report was accepted while + // the first root remained in flight. Then fail first and cancel third. + releaseFirst(); + return await new Promise((_resolve, reject) => { + const cancel = () => + reject(new workerCommand.CommandFailure("fixture cancellation", null, "", "", false)); + if (options.signal?.aborted) cancel(); + else options.signal?.addEventListener("abort", cancel, { once: true }); + }); + } + throw new Error("Unexpected root admitted after failure"); + }, + ); + const onDiagnostic = vi.fn(); + await expect( + runBundledSkillSpector( + workspace, + roots.map((rootPath) => ({ rootPath, scanPath: join(workspace, "artifact", rootPath) })), + onDiagnostic, + ), + ).rejects.toThrow("fixture root failed"); + expect(calls.map((path) => path.split("/").at(-1)).sort()).toEqual(["first", "second", "third"]); + expect(onDiagnostic).toHaveBeenCalledExactlyOnceWith({ + bundledSkillSpector: expect.objectContaining({ + failed: true, + exitCode: 2, + context: expect.objectContaining({ rootIndex: 0, rootCount: 4, completedRoots: 1 }), + }), + }); +}); + it.each([0, 1])( "accepts the structured report despite 2 MiB stderr (exit %s)", async (exitCode) => { @@ -95,6 +229,38 @@ exit ${exitCode}`); }, ); +it.each(["cleanup", "output_limit", "observer"] as const)( + "does not accept a findings exit when the command failed for %s", + async (failureKind) => { + const message = `Fixture ${failureKind} failure`; + vi.spyOn(workerCommand, "runWorkerCommand").mockImplementationOnce( + async (_command, args, options) => { + await writeFile(args[args.indexOf("--output") + 1]!, JSON.stringify(findings)); + options.onTermination?.({ + elapsedMs: 1, + cleanupMs: 1, + processGroup: failureKind === "cleanup" ? "unknown" : "absent", + exitCode: 1, + timedOut: false, + outputLimitExceeded: failureKind === "output_limit" ? "stdout" : null, + }); + throw new workerCommand.CommandFailure( + message, + 1, + "", + "", + false, + failureKind === "output_limit" ? "stdout" : undefined, + failureKind, + ); + }, + ); + const { run, handoff } = await fixture(writeReport(findings)); + await expect(run()).rejects.toThrow(message); + await expect(readFile(handoff)).rejects.toMatchObject({ code: "ENOENT" }); + }, +); + it.each([1, 7])( "rejects exit %s without a findings report despite noisy stderr", async (exitCode) => { @@ -122,13 +288,13 @@ ${trapExit ? "trap 'exit 1' TERM; while true; do sleep 30; done" : "exec sleep 3 it("gives successive bundled roots their own timeout within the preparation budget", async () => { const roots = ["skills/first", "skills/second", "skills/third"]; const { run, handoff } = await fixture( - `sleep 0.7 + `sleep 0.9 ${writeReport(findings)}`, roots, ); vi.stubEnv("CODEX_SECURITY_SCAN_CLAWSCAN_TIMEOUT_MS", "1500"); - // Three real sequential subprocesses need >= 2.1s together, but each fits - // its 1.5s budget. The old shared 1.5s deadline kills the third root. + // Two waves need >= 1.8s together, but each child fits its 1.5s budget. + // A shared 1.5s deadline would still kill the third root. await expect(run()).rejects.toThrow("exited 23"); expect(JSON.parse(await readFile(handoff, "utf8"))).toMatchObject({ applicable: true, @@ -148,14 +314,14 @@ it.each([ "enforces the total preparation ceiling with $remainingMs ms left", async ({ remainingMs, error }) => { const { run, workspace, handoff } = await fixture( - `if [[ "$2" == *second ]]; then sleep 1; fi + `if [[ "$2" == *third ]]; then sleep 1; fi ${writeReport(findings)}`, - ["skills/first", "skills/second"], + ["skills/first", "skills/second", "skills/third"], ); const firstReport = join(workspace, "skillspector-report-0.json"); const startedAt = Date.now(); // Advance only the budget clock after the first real subprocess writes its - // report. Timers and the second subprocess remain real; no 20-minute sleep. + // report. Timers and the next-wave subprocess remain real; no 20-minute sleep. vi.spyOn(Date, "now").mockImplementation( () => startedAt + @@ -165,13 +331,13 @@ ${writeReport(findings)}`, ); await expect(run()).rejects.toThrow(error); expect(JSON.parse(await readFile(firstReport, "utf8"))).toEqual(findings); - const secondReport = join(workspace, "skillspector-report-1.json"); + const nextReport = join(workspace, "skillspector-report-2.json"); if (remainingMs > 0) { // A started child has a private output slot, never a completed report. - expect(await readFile(secondReport, "utf8")).toBe(""); - expect((await stat(secondReport)).mode & 0o777).toBe(0o600); + expect(await readFile(nextReport, "utf8")).toBe(""); + expect((await stat(nextReport)).mode & 0o777).toBe(0o600); } else { - await expect(readFile(secondReport)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(readFile(nextReport)).rejects.toMatchObject({ code: "ENOENT" }); } await expect(readFile(handoff)).rejects.toMatchObject({ code: "ENOENT" }); }, diff --git a/scripts/security/run-codex-scan-worker.ts b/scripts/security/run-codex-scan-worker.ts index 1cea15c352..6d63f5650a 100644 --- a/scripts/security/run-codex-scan-worker.ts +++ b/scripts/security/run-codex-scan-worker.ts @@ -992,10 +992,15 @@ async function runCommand( onStderrChunk?: (chunk: Buffer) => void; onTermination?: (termination: WorkerCommandTermination) => void; cleanupProcessGroup?: boolean; + signal?: AbortSignal; + skillspectorMaxLlmConcurrency?: number; }, ) { const env = codexEnv(options.cwd); for (const name of options.omitEnv ?? []) delete env[name]; + if (options.skillspectorMaxLlmConcurrency !== undefined) { + env.SKILLSPECTOR_MAX_LLM_CONCURRENCY = String(options.skillspectorMaxLlmConcurrency); + } return await runWorkerCommand(command, args, { cwd: options.cwd, env, @@ -1005,6 +1010,7 @@ async function runCommand( onStderrChunk: options.onStderrChunk, onTermination: options.onTermination, cleanupProcessGroup: options.cleanupProcessGroup, + signal: options.signal, }); } @@ -1688,22 +1694,33 @@ export async function runBundledSkillSpector( scanInputs: BundledSkillSpectorScanInput[], onDiagnostic: (diagnostic: Partial) => void, ) { - const analyses: SkillSpectorAnalysis[] = []; + const analyses: Array = new Array(scanInputs.length); const rootTimeoutMs = clawScanTimeoutMs(); - // Keep roots sequential to avoid multiplying SkillSpector's own API fan-out. - // The total ceiling preserves room for the judge within the hosted job/lease. + // The pinned HTTP scanner allows ten model calls. Share that ceiling between + // two roots instead of multiplying it; retain serial execution for other providers. + const parallel = + scanInputs.length > 1 && + securityScannerEnvironment(process.env).SKILLSPECTOR_PROVIDER === "openai"; + const concurrency = parallel ? 2 : 1; + const controller = new AbortController(); + const diagnostics = new Map>(); + let completedRoots = 0; + let nextIndex = 0; + let firstFailure: { index: number; error: unknown } | undefined; + // Preserve the aggregate ceiling and leave room for the judge in the job/lease. const deadlineMs = Date.now() + MAX_BUNDLED_SKILLSPECTOR_TOTAL_MS; - for (const [index, scanInput] of scanInputs.entries()) { + const runRoot = async (index: number) => { + const scanInput = scanInputs[index]!; const remainingMs = deadlineMs - Date.now(); const context: BundledSkillSpectorFailureContext = { rootIndex: index, rootCount: scanInputs.length, - completedRoots: analyses.length, + completedRoots, childBudgetMs: Math.max(0, Math.min(rootTimeoutMs, remainingMs)), remainingStageBudgetMs: Math.max(0, remainingMs), }; if (remainingMs <= 0) { - onDiagnostic({ + diagnostics.set(index, { bundledSkillSpector: { failed: true, timedOut: true, exitCode: null, context }, }); throw new Error("SkillSpector bundled-skill scan deadline exceeded"); @@ -1721,15 +1738,16 @@ export async function runBundledSkillSpector( // just as the ClawScan adapter and direct diagnostic probe do. discardStderr: true, timeoutMs: context.childBudgetMs, + cleanupProcessGroup: true, + signal: controller.signal, + skillspectorMaxLlmConcurrency: parallel ? 5 : undefined, onTermination: (next) => { termination = next; }, }); - analyses.push( - prefixSkillSpectorFindingPaths( - normalizeSkillSpectorAnalysis(await readFile(resultPath, "utf8"), 0), - scanInput.rootPath, - ), + return prefixSkillSpectorFindingPaths( + normalizeSkillSpectorAnalysis(await readFile(resultPath, "utf8"), 0), + scanInput.rootPath, ); } catch (error) { let acceptedFindings = false; @@ -1742,21 +1760,23 @@ export async function runBundledSkillSpector( scanInput.rootPath, ); const validFindingsExit = + !controller.signal.aborted && + error.failureKind === "exit" && + termination?.processGroup === "absent" && !error.timedOut && error.exitCode === 1 && (analysis.status === "suspicious" || analysis.status === "malicious") && analysis.issueCount > 0 && analysis.issues.length > 0; if (validFindingsExit) { - analyses.push(analysis); acceptedFindings = true; - continue; + return analysis; } } } } finally { if (!acceptedFindings) { - onDiagnostic({ + diagnostics.set(index, { bundledSkillSpector: { failed: true, timedOut: error instanceof CommandFailure && error.timedOut, @@ -1774,8 +1794,50 @@ export async function runBundledSkillSpector( } throw error; } + }; + const worker = async () => { + while (!firstFailure) { + const index = nextIndex++; + if (index >= scanInputs.length) return; + try { + analyses[index] = await runRoot(index); + completedRoots += 1; + } catch (error) { + if (!firstFailure) { + firstFailure = { index, error }; + controller.abort(); + } + return; + } + } + }; + // Join every worker after cancellation so no sibling can outlive the scan. + await Promise.all(Array.from({ length: concurrency }, worker)); + if (firstFailure) { + const diagnostic = diagnostics.get(firstFailure.index); + const context = diagnostic?.bundledSkillSpector?.context; + if (context) { + context.completedRoots = completedRoots; + context.remainingStageBudgetMs = Math.max(0, deadlineMs - Date.now()); + } + onDiagnostic( + diagnostic ?? { + bundledSkillSpector: { + failed: true, + timedOut: firstFailure.error instanceof CommandFailure && firstFailure.error.timedOut, + exitCode: + firstFailure.error instanceof CommandFailure ? firstFailure.error.exitCode : null, + }, + }, + ); + throw firstFailure.error; } - return aggregateSkillSpectorAnalyses(analyses); + if (completedRoots !== scanInputs.length) { + throw new Error("SkillSpector bundled-skill scans did not complete"); + } + return aggregateSkillSpectorAnalyses( + analyses.filter((analysis): analysis is SkillSpectorAnalysis => analysis !== undefined), + ); } async function prepareBundledSkillSpector( diff --git a/specs/security-moderation.md b/specs/security-moderation.md index b4d78742ce..2fb1f1c571 100644 --- a/specs/security-moderation.md +++ b/specs/security-moderation.md @@ -602,6 +602,19 @@ See also: [acceptable-usage.md](./acceptable-usage.md) for the marketplace polic summary. The plugin package root is never a fallback SkillSpector target; packages with no bundled roots provide ClawScan an explicit not-applicable result. This is not a prepublication-worker contract. +- Bundled-skill execution may run at most two OpenAI roots concurrently, with + five process-wide model requests per root, preserving the pinned scanner's + existing total ceiling of ten. Other providers retain serial roots and their + existing defaults. Keep each child deadline and the 20-minute aggregate + deadline. Findings aggregate in canonical root order; every root must finish + before handing off to ClawScan. After a failure, stop admitting roots, cancel + in-flight siblings and join their confirmed process-group cleanup before + returning failure. Only normal exit-1 findings with confirmed process-group + absence can be accepted; cancellation, timeout, output-limit, observer and + cleanup failures cannot supply findings-exit results. Command cancellation + must reject already-cancelled work + without spawning and remove its listener after termination; arbitrary abort + reasons must not enter errors or diagnostics. - Current skill and plugin scans are queued through `securityScanJobs` and completed by the external Codex worker. - VirusTotal is not displayed in the UI: audit pages, the audit directory, version