diff --git a/packages/deepsec/src/__tests__/setup-coordinator.test.ts b/packages/deepsec/src/__tests__/setup-coordinator.test.ts index f3b312c0..763845c3 100644 --- a/packages/deepsec/src/__tests__/setup-coordinator.test.ts +++ b/packages/deepsec/src/__tests__/setup-coordinator.test.ts @@ -155,6 +155,7 @@ describe("one-shot setup coordinator", () => { analysisCount: 1, findingCount: 0, errorBatchCount: 0, + skippedCount: 0, })); let fingerprint = "source-v1"; const services = { diff --git a/packages/deepsec/src/commands/process.ts b/packages/deepsec/src/commands/process.ts index 35c609c3..4182b747 100644 --- a/packages/deepsec/src/commands/process.ts +++ b/packages/deepsec/src/commands/process.ts @@ -388,6 +388,15 @@ async function processDirectMode(opts: Parameters[0]) { console.log(`${RED}${result.findingCount} new finding(s) — exiting 1${RESET}`); process.exit(1); } + // Files another run holds are unreviewed, not clean. A forged or stale + // lock must not turn the CI gate green, so fail rather than pass quietly. + if (result.skippedCount > 0) { + console.log(); + console.log( + `${RED}${result.skippedCount} file(s) skipped because another run holds their lock — exiting 1 (not a clean review).${RESET}`, + ); + process.exit(1); + } console.log(); console.log(`${GREEN}No findings.${RESET}`); } diff --git a/packages/deepsec/src/setup/coordinator.ts b/packages/deepsec/src/setup/coordinator.ts index 41f73ccb..378f2f60 100644 --- a/packages/deepsec/src/setup/coordinator.ts +++ b/packages/deepsec/src/setup/coordinator.ts @@ -868,6 +868,7 @@ export async function runSetupWorkflow( analysisCount: 0, findingCount: 0, errorBatchCount: 0, + skippedCount: 0, }; reporter.emit({ type: "phase-skip", diff --git a/packages/processor/src/__tests__/process-revalidate.test.ts b/packages/processor/src/__tests__/process-revalidate.test.ts index 7c5c3c90..a0dfbb61 100644 --- a/packages/processor/src/__tests__/process-revalidate.test.ts +++ b/packages/processor/src/__tests__/process-revalidate.test.ts @@ -2,7 +2,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { defineConfig, type FileRecord, setLoadedConfig } from "@deepsec/core"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { QuotaExhaustedError } from "../agents/shared.js"; import { process as processProject, revalidate } from "../index.js"; import { StubAgent } from "./stub-agent.js"; @@ -76,6 +76,27 @@ function pendingRecord(projectId: string, filePath: string): FileRecord { }; } +function writeRun( + fx: Fixture, + runId: string, + meta: { hostname: string; phase: "running" | "done"; pid: number }, +): void { + const runsDir = path.join(fx.dataRoot, fx.projectId, "runs"); + fs.mkdirSync(runsDir, { recursive: true }); + fs.writeFileSync( + path.join(runsDir, `${runId}.json`), + JSON.stringify({ + runId, + projectId: fx.projectId, + rootPath: fx.targetRoot, + createdAt: new Date().toISOString(), + type: "process", + stats: {}, + ...meta, + }), + ); +} + describe("processor with stub agent", () => { let prevDataRoot: string | undefined; @@ -458,6 +479,308 @@ describe("processor with stub agent", () => { expect(after.lockedByRunId).toBe(liveRunId); }); + it.each([ + ["--reinvestigate", { reinvestigate: true }], + ["a file list", { filePaths: ["live.ts", "dead.ts", "mine.ts", "done.ts"] }], + ])("process() with %s does NOT take a file a still-running other run holds", async (_mode, forceParams) => { + // Both modes select files whatever their status, so they can re-claim + // analyzed files. Only the other run's non-reclaimable lock must survive. + const fx = setupProject({ files: ["live.ts", "dead.ts", "mine.ts", "done.ts"] }); + const liveRunId = "20260101000000-liveforceaaaaaaa"; + const deadRunId = "20260101000000-deadforceaaaaaaa"; + const ownRunId = "20260101000000-ownforceaaaaaaaa"; + + const lock = (filePath: string, runId: string, pid: number) => { + const rec = pendingRecord(fx.projectId, filePath); + rec.status = "processing"; + rec.lockedByRunId = runId; + rec.lockedAt = new Date().toISOString(); + fx.writeRecord(rec); + writeRun(fx, runId, { hostname: os.hostname(), phase: "running", pid }); + }; + lock("live.ts", liveRunId, process.pid); + const liveBefore = fx.readRecord("live.ts"); + lock("dead.ts", deadRunId, 0x7fffffff); + lock("mine.ts", ownRunId, process.pid); + const done = pendingRecord(fx.projectId, "done.ts"); + done.status = "analyzed"; + fx.writeRecord(done); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + + await processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + runId: ownRunId, + ...forceParams, + }); + + const investigated = stub.calls.investigateCalls.flatMap((c) => c.batch.map((r) => r.filePath)); + expect(investigated.sort()).toEqual(["dead.ts", "done.ts", "mine.ts"]); + expect(fx.readRecord("live.ts")).toEqual(liveBefore); + }); + + it.each([ + ["leaves a fresh lock from another host", "processing", "other-host", "running", 0, false], + [ + "takes a lock from another host older than 1 hour", + "processing", + "other-host", + "running", + 61 * 60_000, + true, + ], + [ + "takes a lock from another host whose run is done", + "processing", + "other-host", + "done", + 0, + true, + ], + [ + "leaves a pending file with a fresh lock from another host", + "pending", + "other-host", + "running", + 0, + false, + ], + [ + "leaves an errored file with a fresh lock from another host", + "error", + "other-host", + "running", + 0, + false, + ], + [ + "takes a lock from another host dated a day in the future", + "processing", + "other-host", + "running", + -24 * 60 * 60_000, + true, + ], + [ + "leaves an analyzed file with a live lock on this host", + "analyzed", + os.hostname(), + "running", + 0, + false, + ], + ] as const)("process() with --reinvestigate %s", async (_name, status, hostname, phase, ageMs, taken) => { + // Force mode skips a file exactly when its lock isn't reclaimable, + // whatever the host and whatever the record's status. + const filePath = "locked.ts"; + const fx = setupProject({ files: [filePath] }); + const liveRunId = "20260101000000-liveremoteaaaaaa"; + const rec = pendingRecord(fx.projectId, filePath); + rec.status = status; + rec.lockedByRunId = liveRunId; + rec.lockedAt = new Date(Date.now() - ageMs).toISOString(); + fx.writeRecord(rec); + writeRun(fx, liveRunId, { hostname, phase, pid: process.pid }); + const before = fx.readRecord(filePath); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + const result = await processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + reinvestigate: true, + }); + + const investigated = stub.calls.investigateCalls.flatMap((c) => c.batch.map((r) => r.filePath)); + expect(investigated).toEqual(taken ? [filePath] : []); + expect(result.skippedCount).toBe(taken ? 0 : 1); + if (!taken) expect(fx.readRecord(filePath)).toEqual(before); + }); + + it("process() with --reinvestigate takes a lock whose timestamp is unparseable", async () => { + const fx = setupProject({ files: ["locked.ts"] }); + const liveRunId = "20260101000000-badstampaaaaaaaa"; + const rec = pendingRecord(fx.projectId, "locked.ts"); + rec.status = "processing"; + rec.lockedByRunId = liveRunId; + rec.lockedAt = "not-a-date"; + fx.writeRecord(rec); + writeRun(fx, liveRunId, { hostname: "other-host", phase: "running", pid: process.pid }); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + await processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + reinvestigate: true, + }); + + const investigated = stub.calls.investigateCalls.flatMap((c) => c.batch.map((r) => r.filePath)); + expect(investigated).toEqual(["locked.ts"]); + }); + + it("process() with a file list warns about requested files another run holds", async () => { + const fx = setupProject({ files: ["held.ts", "free.ts"] }); + const otherRunId = "20260101000000-otherwarnaaaaaaa"; + const held = pendingRecord(fx.projectId, "held.ts"); + held.status = "processing"; + held.lockedByRunId = otherRunId; + held.lockedAt = new Date().toISOString(); + fx.writeRecord(held); + fx.writeRecord(pendingRecord(fx.projectId, "free.ts")); + writeRun(fx, otherRunId, { hostname: "other-host", phase: "running", pid: process.pid }); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const result = await processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + filePaths: ["held.ts", "free.ts"], + }); + expect(result.skippedCount).toBe(1); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0][0]).toContain("held.ts"); + expect(warn.mock.calls[0][0]).not.toContain("free.ts"); + } finally { + warn.mockRestore(); + } + const investigated = stub.calls.investigateCalls.flatMap((c) => c.batch.map((r) => r.filePath)); + expect(investigated).toEqual(["free.ts"]); + }); + + it("process() in default mode warns about a processing file another run holds", async () => { + const fx = setupProject({ files: ["held.ts", "free.ts"] }); + const otherRunId = "20260101000000-otherdefaultaaaa"; + const held = pendingRecord(fx.projectId, "held.ts"); + held.status = "processing"; + held.lockedByRunId = otherRunId; + held.lockedAt = new Date().toISOString(); + fx.writeRecord(held); + fx.writeRecord(pendingRecord(fx.projectId, "free.ts")); + writeRun(fx, otherRunId, { hostname: "other-host", phase: "running", pid: process.pid }); + const before = fx.readRecord("held.ts"); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + const warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + try { + const result = await processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + }); + expect(result.skippedCount).toBe(1); + expect(warn).toHaveBeenCalledTimes(1); + expect(warn.mock.calls[0][0]).toContain("held.ts"); + } finally { + warn.mockRestore(); + } + const investigated = stub.calls.investigateCalls.flatMap((c) => c.batch.map((r) => r.filePath)); + expect(investigated).toEqual(["free.ts"]); + expect(fx.readRecord("held.ts")).toEqual(before); + }); + + it("process() with --reinvestigate keeps findings another run wrote after the snapshot loaded", async () => { + // The other run's lock is live when the records are loaded, so the + // snapshot has no findings. The run finishes the file before the claim, + // which clears the lock and makes the file claimable. + const filePath = "shared.ts"; + const fx = setupProject({ files: [filePath] }); + const otherRunId = "20260101000000-othersnapaaaaaaa"; + const rec = pendingRecord(fx.projectId, filePath); + rec.status = "processing"; + rec.lockedByRunId = otherRunId; + rec.lockedAt = new Date().toISOString(); + fx.writeRecord(rec); + writeRun(fx, otherRunId, { hostname: os.hostname(), phase: "running", pid: process.pid }); + + const stub = new StubAgent(); + setLoadedConfig( + defineConfig({ + projects: [{ id: fx.projectId, root: fx.targetRoot }], + plugins: [{ name: "stub", agents: [stub] }], + }), + ); + + // Hold the claim mutex so process() loads its snapshot, then waits. + const lockDir = path.join(fx.dataRoot, fx.projectId, ".process.lock"); + fs.mkdirSync(lockDir); + const run = processProject({ + projectId: fx.projectId, + agentType: "stub", + concurrency: 1, + reinvestigate: true, + }); + + const finished = fx.readRecord(filePath); + finished.status = "analyzed"; + finished.lockedByRunId = undefined; + finished.lockedAt = undefined; + finished.findings = [ + { + severity: "HIGH", + vulnSlug: "auth-bypass", + title: "finding from the other run", + description: "written after the snapshot loaded", + lineNumbers: [1], + recommendation: "fix it", + confidence: "high", + }, + ]; + finished.analysisHistory = [ + { + runId: otherRunId, + investigatedAt: new Date().toISOString(), + durationMs: 1, + agentType: "stub", + model: "stub", + modelConfig: {}, + findingCount: 1, + }, + ]; + fx.writeRecord(finished); + fs.rmSync(lockDir, { recursive: true }); + await run; + + const after = fx.readRecord(filePath); + expect(after.findings.map((f) => f.title)).toContain("finding from the other run"); + expect(after.analysisHistory.map((h) => h.runId)).toContain(otherRunId); + expect(after.analysisHistory).toHaveLength(2); + }); + it("process() captures refusals from the agent into AnalysisEntry", async () => { const fx = setupProject({ files: ["app.ts"] }); fx.writeRecord(pendingRecord(fx.projectId, "app.ts")); diff --git a/packages/processor/src/index.ts b/packages/processor/src/index.ts index 09bd3923..abe5a5c7 100644 --- a/packages/processor/src/index.ts +++ b/packages/processor/src/index.ts @@ -164,6 +164,12 @@ export async function process(params: { * crashed CLI). CI gates on this so a silent fail doesn't pass. */ errorBatchCount: number; + /** + * Files left unclaimed because another run held a live lock on them. A + * run that requested files but claimed none is not a clean review, so + * direct mode (the CI gate) fails on a non-zero count. + */ + skippedCount: number; /** * Set when one of the batches threw `QuotaExhaustedError`. Once any * batch hits this, every subsequent API call against the same credential @@ -293,7 +299,7 @@ export async function process(params: { type: "all_complete", message: `Run ${runId} already completed`, }); - return { runId, analysisCount: 0, findingCount: 0, errorBatchCount: 0 }; + return { runId, analysisCount: 0, findingCount: 0, errorBatchCount: 0, skippedCount: 0 }; } } else { // Create new run @@ -354,6 +360,7 @@ export async function process(params: { // thinking. False reclaims are catastrophic; false rejections only // cost a retry on the next run. const STALE_LOCK_MS = 60 * 60 * 1000; + const FUTURE_SKEW_MS = 5 * 60 * 1000; const localHostname = os.hostname(); const isReclaimableLock = (r: FileRecord): boolean => { if (!r.lockedByRunId) return true; @@ -387,8 +394,11 @@ export async function process(params: { // those as old enough to reclaim so we can recover legacy locked // state. if (!r.lockedAt) return true; - const ageMs = Date.now() - new Date(r.lockedAt).getTime(); - return ageMs >= STALE_LOCK_MS; + // An unparseable or future timestamp would never age out, so it + // can't hold a lock. FUTURE_SKEW_MS allows for clock drift between hosts. + const lockedMs = Date.parse(r.lockedAt); + if (Number.isNaN(lockedMs) || lockedMs > Date.now() + FUTURE_SKEW_MS) return true; + return Date.now() - lockedMs >= STALE_LOCK_MS; }; // Load file records and pick which to process @@ -445,12 +455,10 @@ export async function process(params: { (r) => r.status === "pending" || r.status === "error" || - // Reclaim a `processing` record from another run only when the - // owning lock is genuinely abandoned. The previous version - // reclaimed unconditionally, which let two concurrent runs - // both pick up the same file and clobber each other on - // write. See `isReclaimableLock` for the exact criteria. - (r.status === "processing" && r.lockedByRunId !== runId && isReclaimableLock(r)), + // Another run's `processing` record goes on to the claim step, + // which takes it only if the lock is reclaimable and otherwise + // reports it as skipped. Dropping it here would hide it. + (r.status === "processing" && r.lockedByRunId !== runId), ); } @@ -502,7 +510,7 @@ export async function process(params: { message: "No files to process", }); completeRun(projectId, runId, "done", { filesProcessed: 0 }); - return { runId, analysisCount: 0, findingCount: 0, errorBatchCount: 0 }; + return { runId, analysisCount: 0, findingCount: 0, errorBatchCount: 0, skippedCount: 0 }; } // Apply path filter @@ -510,11 +518,6 @@ export async function process(params: { toProcess = toProcess.filter((r) => r.filePath.startsWith(params.filter!)); } - // Apply limit - if (params.limit && toProcess.length > params.limit) { - toProcess = toProcess.slice(0, params.limit); - } - // Atomic claim under a per-project mutex. Without serializing this // section, two `process()` invocations against the same project both // see the same pending records, both write status="processing" with @@ -526,10 +529,13 @@ export async function process(params: { // primitive and the 1h stale-lock cutoff. const lockedAt = new Date().toISOString(); const claimed: FileRecord[] = []; + const lockedPaths: string[] = []; const inForceMode = !!reinvestigate || params.filePaths !== undefined; const releaseProcessLock = await acquireProcessLock(projectId, runId); try { for (const record of toProcess) { + // The limit counts claimed files, so locked ones don't use up slots. + if (params.limit && claimed.length >= params.limit) break; let current: FileRecord; try { const fresh = readFileRecord(projectId, record.filePath); @@ -540,13 +546,18 @@ export async function process(params: { } const isOurs = current.lockedByRunId === runId; - const isFreelyClaimable = + // Another run's lock decides first, whatever the status says: a fresh + // lock on a `pending` or `error` record is still that run's file. + // Otherwise a pending/error/processing record is free to take, and + // force mode also takes analyzed ones. + const heldByOther = !isOurs && !isReclaimableLock(current); + const isClaimable = + inForceMode || current.status === "pending" || current.status === "error" || - (current.status === "processing" && - current.lockedByRunId !== runId && - isReclaimableLock(current)); - if (!isOurs && !isFreelyClaimable && !inForceMode) { + current.status === "processing"; + if (!isOurs && (heldByOther || !isClaimable)) { + lockedPaths.push(current.filePath); continue; } @@ -555,16 +566,19 @@ export async function process(params: { current.lockedAt = lockedAt; writeFileRecord(current); - // Mutate the in-memory snapshot we'll process from so downstream code - // sees the same lock state. - record.status = current.status; - record.lockedByRunId = current.lockedByRunId; - record.lockedAt = current.lockedAt; - claimed.push(record); + // Process from the record we just claimed, not the startup snapshot: + // another run may have finished this file since the snapshot was + // loaded, and its findings and history live only in `current`. + claimed.push(current); } } finally { releaseProcessLock(); } + if (lockedPaths.length > 0) { + console.warn( + `[deepsec] process: ${lockedPaths.length} file(s) were locked by another run and were skipped: ${lockedPaths.slice(0, 5).join(", ")}${lockedPaths.length > 5 ? "…" : ""}`, + ); + } toProcess = claimed; if (toProcess.length === 0) { emitProgress({ @@ -572,7 +586,13 @@ export async function process(params: { message: "Nothing to claim — another run owned every candidate file.", }); completeRun(projectId, runId, "done", { filesProcessed: 0 }); - return { runId, analysisCount: 0, findingCount: 0, errorBatchCount: 0 }; + return { + runId, + analysisCount: 0, + findingCount: 0, + errorBatchCount: 0, + skippedCount: lockedPaths.length, + }; } const batches = batchCandidates(toProcess, params.batchSize); @@ -851,6 +871,7 @@ export async function process(params: { analysisCount: totalAnalyses, findingCount: totalFindings, errorBatchCount: batchesFailed, + skippedCount: lockedPaths.length, quotaExhausted, totalCostUsd, costLimitReached,