Respect another run's file lock in --reinvestigate and direct mode

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 = {

packages/deepsec/src/commands/process.ts

@@ -388,6 +388,15 @@ async function processDirectMode(opts: Parameters<typeof processCommand>[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}`);
 }

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",

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"));