feat(worker): deduplicate static and runtime findings before exploitation

Parse Agentic SAST SARIF into typed observations, enrich and route those observations, and reconcile them with pentest findings before exploitation.

Publish deterministic exploitation queues with stable lineage, exact-path Git commits, retry-safe manifests, named drop reasons, and confined task formation. Reject duplicate producer IDs before commit and adopt either legal provenance shape after a lost acknowledgement.

apps/worker/src/services/queue-validation.ts

@@ -6,8 +6,9 @@
 
 import { fs, path } from 'zx';
 
-import type { ExploitationDecision, VulnType } from '../types/agents.js';
+import type { ExploitationDecision } from '../types/agents.js';
 import { ErrorCode } from '../types/errors.js';
+import type { ReconciliationClass } from '../types/reconciliation.js';
 import { err, ok, type Result } from '../types/result.js';
 import { asyncPipe } from '../utils/functional.js';
 import { PentestError } from './error-handling.js';
@@ -17,14 +18,15 @@ export type { ExploitationDecision, VulnType } from '../types/agents.js';
 interface VulnTypeConfigItem {
   deliverable: string;
   queue: string;
+  deliverableRequired: boolean;
 }
 
-type VulnTypeConfig = Record<VulnType, VulnTypeConfigItem>;
+type VulnTypeConfig = Record<ReconciliationClass, VulnTypeConfigItem>;
 
-type ErrorMessageResolver = string | ((existence: FileExistence) => string);
+type ErrorMessageResolver = string | ((context: ExistenceContext) => string);
 
 interface ValidationRule {
-  predicate: (existence: FileExistence) => boolean;
+  predicate: (context: ExistenceContext) => boolean;
   errorMessage: ErrorMessageResolver;
   retryable: boolean;
 }
@@ -34,8 +36,13 @@ interface FileExistence {
   queueExists: boolean;
 }
 
+interface ExistenceContext {
+  existence: FileExistence;
+  deliverableRequired: boolean;
+}
+
 interface PathsBase {
-  vulnType: VulnType;
+  vulnType: ReconciliationClass;
   deliverable: string;
   queue: string;
   sourceDir: string;
@@ -64,56 +71,87 @@ interface QueueValidationResult {
   error: string | null;
 }
 
+function isRetryableQueueFileSystemError(error: unknown): boolean {
+  if (!(error instanceof Error)) return false;
+  const code = (error as NodeJS.ErrnoException).code;
+  return code !== undefined && !['EACCES', 'EINVAL', 'EISDIR', 'ENAMETOOLONG', 'ENOTDIR', 'EPERM'].includes(code);
+}
+
 /**
  * Result type for safe validation - explicit error handling.
  */
-export type SafeValidationResult = Result<ExploitationDecision, PentestError>;
+export type ReconciliationExploitationDecision<T extends ReconciliationClass = ReconciliationClass> = Omit<
+  ExploitationDecision,
+  'vulnType'
+> & {
+  vulnType: T;
+};
+
+export type SafeValidationResult<T extends ReconciliationClass = ReconciliationClass> = Result<
+  ReconciliationExploitationDecision<T>,
+  PentestError
+>;
 
 // Vulnerability type configuration as immutable data
 const VULN_TYPE_CONFIG: VulnTypeConfig = Object.freeze({
   injection: Object.freeze({
     deliverable: 'injection_analysis_deliverable.md',
     queue: 'injection_exploitation_queue.json',
+    deliverableRequired: true,
   }),
   xss: Object.freeze({
     deliverable: 'xss_analysis_deliverable.md',
     queue: 'xss_exploitation_queue.json',
+    deliverableRequired: true,
   }),
   auth: Object.freeze({
     deliverable: 'auth_analysis_deliverable.md',
     queue: 'auth_exploitation_queue.json',
+    deliverableRequired: true,
   }),
   ssrf: Object.freeze({
     deliverable: 'ssrf_analysis_deliverable.md',
     queue: 'ssrf_exploitation_queue.json',
+    deliverableRequired: true,
   }),
   authz: Object.freeze({
     deliverable: 'authz_analysis_deliverable.md',
     queue: 'authz_exploitation_queue.json',
+    deliverableRequired: true,
+  }),
+  miscellaneous: Object.freeze({
+    deliverable: 'miscellaneous_analysis_deliverable.md',
+    queue: 'miscellaneous_exploitation_queue.json',
+    deliverableRequired: false,
   }),
 }) as VulnTypeConfig;
 
 // Pure function to create validation rule
 function createValidationRule(
-  predicate: (existence: FileExistence) => boolean,
+  predicate: (context: ExistenceContext) => boolean,
   errorMessage: ErrorMessageResolver,
   retryable: boolean = true,
 ): ValidationRule {
   return Object.freeze({ predicate, errorMessage, retryable });
 }
 
-// Symmetric deliverable rules: queue and deliverable must exist together (prevents partial analysis from triggering exploitation)
+// A queue is always required. Analysis-backed classes also require their analysis deliverable;
+// the analysis-less `miscellaneous` class deliberately has no such artifact.
 const fileExistenceRules: readonly ValidationRule[] = Object.freeze([
   createValidationRule(
-    ({ deliverableExists, queueExists }) => deliverableExists && queueExists,
+    ({ existence, deliverableRequired }) =>
+      existence.queueExists && (!deliverableRequired || existence.deliverableExists),
     getExistenceErrorMessage,
   ),
 ]);
 
 // Generate appropriate error message based on which files are missing
-function getExistenceErrorMessage(existence: FileExistence): string {
+function getExistenceErrorMessage({ existence, deliverableRequired }: ExistenceContext): string {
   const { deliverableExists, queueExists } = existence;
 
+  if (!deliverableRequired) {
+    return 'Analysis failed: Queue file missing. A queue is required.';
+  }
   if (!deliverableExists && !queueExists) {
     return 'Analysis failed: Neither deliverable nor queue file exists. Both are required.';
   }
@@ -124,7 +162,7 @@ function getExistenceErrorMessage(existence: FileExistence): string {
 }
 
 // Pure function to create file paths
-const createPaths = (vulnType: VulnType, sourceDir: string): PathsBase | PathsWithError => {
+const createPaths = (vulnType: ReconciliationClass, sourceDir: string): PathsBase | PathsWithError => {
   const config = VULN_TYPE_CONFIG[vulnType];
   if (!config) {
     return {
@@ -144,10 +182,23 @@ const createPaths = (vulnType: VulnType, sourceDir: string): PathsBase | PathsWi
 const checkFileExistence = async (paths: PathsBase | PathsWithError): Promise<PathsWithExistence | PathsWithError> => {
   if ('error' in paths) return paths;
 
-  const [deliverableExists, queueExists] = await Promise.all([
-    fs.pathExists(paths.deliverable),
-    fs.pathExists(paths.queue),
-  ]);
+  let deliverableExists: boolean;
+  let queueExists: boolean;
+  try {
+    [deliverableExists, queueExists] = await Promise.all([
+      fs.pathExists(paths.deliverable),
+      fs.pathExists(paths.queue),
+    ]);
+  } catch (error) {
+    return {
+      error: new PentestError(
+        'Queue validation could not inspect the required files.',
+        'filesystem',
+        isRetryableQueueFileSystemError(error),
+        { vulnType: paths.vulnType },
+      ),
+    };
+  }
 
   return Object.freeze({
     ...paths,
@@ -162,13 +213,15 @@ const validateExistenceRules = (
   if ('error' in pathsWithExistence) return pathsWithExistence;
 
   const { existence, vulnType } = pathsWithExistence;
+  const { deliverableRequired } = VULN_TYPE_CONFIG[vulnType];
+  const context: ExistenceContext = { existence, deliverableRequired };
 
   // Find the first rule that fails
-  const failedRule = fileExistenceRules.find((rule) => !rule.predicate(existence));
+  const failedRule = fileExistenceRules.find((rule) => !rule.predicate(context));
 
   if (failedRule) {
     const message =
-      typeof failedRule.errorMessage === 'function' ? failedRule.errorMessage(existence) : failedRule.errorMessage;
+      typeof failedRule.errorMessage === 'function' ? failedRule.errorMessage(context) : failedRule.errorMessage;
 
     return {
       error: new PentestError(
@@ -177,8 +230,6 @@ const validateExistenceRules = (
         failedRule.retryable,
         {
           vulnType,
-          deliverablePath: pathsWithExistence.deliverable,
-          queuePath: pathsWithExistence.queue,
           existence,
         },
         ErrorCode.DELIVERABLE_NOT_FOUND,
@@ -204,11 +255,11 @@ const validateQueueStructure = (content: string): QueueValidationResult => {
       data: isValid ? (parsed as QueueData) : null,
       error: null,
     });
-  } catch (parseError) {
+  } catch {
     return Object.freeze({
       valid: false,
       data: null,
-      error: parseError instanceof Error ? parseError.message : String(parseError),
+      error: 'invalid_json',
     });
   }
 };
@@ -234,9 +285,6 @@ const validateQueueContent = async (
           true, // retryable
           {
             vulnType: pathsWithExistence.vulnType,
-            queuePath: pathsWithExistence.queue,
-            originalError: queueValidation.error,
-            queueStructure: queueValidation.data ? Object.keys(queueValidation.data) : [],
           },
         ),
       };
@@ -249,13 +297,11 @@ const validateQueueContent = async (
   } catch (readError) {
     return {
       error: new PentestError(
-        `Failed to read queue file for ${pathsWithExistence.vulnType}: ${readError instanceof Error ? readError.message : String(readError)}`,
+        `Queue file for ${pathsWithExistence.vulnType} could not be read.`,
         'filesystem',
-        false,
+        isRetryableQueueFileSystemError(readError),
         {
           vulnType: pathsWithExistence.vulnType,
-          queuePath: pathsWithExistence.queue,
-          originalError: readError instanceof Error ? readError.message : String(readError),
         },
       ),
     };
@@ -263,7 +309,9 @@ const validateQueueContent = async (
 };
 
 // Final decision: skip if queue says no vulns, proceed if vulns found, error otherwise
-const determineExploitationDecision = (validatedData: PathsWithQueue | PathsWithError): ExploitationDecision => {
+const determineExploitationDecision = (
+  validatedData: PathsWithQueue | PathsWithError,
+): ReconciliationExploitationDecision => {
   if ('error' in validatedData) {
     throw validatedData.error;
   }
@@ -281,11 +329,11 @@ const determineExploitationDecision = (validatedData: PathsWithQueue | PathsWith
 };
 
 // Main functional validation pipeline
-export async function validateQueueAndDeliverable(
-  vulnType: VulnType,
+export async function validateQueueAndDeliverable<T extends ReconciliationClass>(
+  vulnType: T,
   sourceDir: string,
-): Promise<ExploitationDecision> {
-  return asyncPipe<ExploitationDecision>(
+): Promise<ReconciliationExploitationDecision<T>> {
+  return asyncPipe<ReconciliationExploitationDecision<T>>(
     createPaths(vulnType, sourceDir),
     checkFileExistence,
     validateExistenceRules,
@@ -298,11 +346,17 @@ export async function validateQueueAndDeliverable(
  * Safely validate queue and deliverable files.
  * Returns Result<ExploitationDecision, PentestError> for explicit error handling.
  */
-export async function validateQueueSafe(vulnType: VulnType, sourceDir: string): Promise<SafeValidationResult> {
+export async function validateQueueSafe<T extends ReconciliationClass>(
+  vulnType: T,
+  sourceDir: string,
+): Promise<SafeValidationResult<T>> {
   try {
     const result = await validateQueueAndDeliverable(vulnType, sourceDir);
     return ok(result);
   } catch (error) {
-    return err(error as PentestError);
+    if (error instanceof PentestError) return err(error);
+    return err(
+      new PentestError('Queue validation failed closed on an internal invariant.', 'unknown', false, { vulnType }),
+    );
   }
 }