fix: resolve parallel workflow race conditions and retry logic bugs

- Fix save_deliverable race condition using closure pattern instead of global variable
- Fix error classification order so OutputValidationError matches before generic validation
- Fix ApplicationFailure re-classification bug by checking instanceof before re-throwing
- Add per-error-type retry limits (3 for output validation, 50 for billing)
- Add fast retry intervals for pipeline testing mode (10s vs 5min)
- Increase worker concurrent activities to 25 for parallel workflows

src/error-handling.ts

@@ -247,8 +247,18 @@ export function classifyErrorForTemporal(error: unknown): TemporalErrorClassific
     return { type: 'PermissionError', retryable: false };
   }
 
+  // === OUTPUT VALIDATION ERRORS (Retryable) ===
+  // Agent didn't produce expected deliverables - retry may succeed
+  // IMPORTANT: Must come BEFORE generic 'validation' check below
+  if (
+    message.includes('failed output validation') ||
+    message.includes('output validation failed')
+  ) {
+    return { type: 'OutputValidationError', retryable: true };
+  }
+
   // Invalid Request (400) - malformed request is permanent
-  // Note: Checked AFTER billing since Anthropic billing is 400
+  // Note: Checked AFTER billing and AFTER output validation
   if (
     message.includes('invalid_request_error') ||
     message.includes('malformed') ||

src/temporal/activities.ts

@@ -25,6 +25,10 @@ import chalk from 'chalk';
 const MAX_ERROR_MESSAGE_LENGTH = 2000;
 const MAX_STACK_TRACE_LENGTH = 1000;
 
+// Max retries for output validation errors (agent didn't save deliverables)
+// Lower than default 50 since this is unlikely to self-heal
+const MAX_OUTPUT_VALIDATION_RETRIES = 3;
+
 /**
  * Truncate error message to prevent buffer overflow in Temporal serialization.
  */
@@ -193,6 +197,16 @@ async function runAgentActivity(
         success: false,
         error: 'Output validation failed',
       });
+
+      // Limit output validation retries (unlikely to self-heal)
+      if (attemptNumber >= MAX_OUTPUT_VALIDATION_RETRIES) {
+        throw ApplicationFailure.nonRetryable(
+          `Agent ${agentName} failed output validation after ${attemptNumber} attempts`,
+          'OutputValidationError',
+          [{ agentName, attemptNumber, elapsed: Date.now() - startTime }]
+        );
+      }
+      // Let Temporal retry (will be classified as OutputValidationError)
       throw new Error(`Agent ${agentName} failed output validation`);
     }
 
@@ -224,6 +238,12 @@ async function runAgentActivity(
       console.error(`Failed to rollback git workspace for ${agentName}:`, rollbackErr);
     }
 
+    // If error is already an ApplicationFailure (e.g., from our retry limit logic),
+    // re-throw it directly without re-classifying
+    if (error instanceof ApplicationFailure) {
+      throw error;
+    }
+
     // Classify error for Temporal retry behavior
     const classified = classifyErrorForTemporal(error);
     // Truncate message to prevent protobuf buffer overflow

src/temporal/worker.ts

@@ -9,7 +9,7 @@
  * Temporal worker for Shannon pentest pipeline.
  *
  * Polls the 'shannon-pipeline' task queue and executes activities.
- * Handles up to 5 concurrent activities to support parallel agent execution.
+ * Handles up to 25 concurrent activities to support multiple parallel workflows.
  *
  * Usage:
  *   npm run temporal:worker
@@ -49,7 +49,7 @@ async function runWorker(): Promise<void> {
     workflowBundle,
     activities,
     taskQueue: 'shannon-pipeline',
-    maxConcurrentActivityTaskExecutions: 5, // Match parallel agent count
+    maxConcurrentActivityTaskExecutions: 25, // Support multiple parallel workflows (5 agents × ~5 workflows)
   });
 
   // Graceful shutdown handling

src/temporal/workflows.ts

@@ -35,25 +35,44 @@ import {
   type PipelineProgress,
 } from './shared.js';
 
-// Activity proxy with retry configuration
+// Retry configuration for production (long intervals for billing recovery)
+const PRODUCTION_RETRY = {
+  initialInterval: '5 minutes',
+  maximumInterval: '30 minutes',
+  backoffCoefficient: 2,
+  maximumAttempts: 50,
+  nonRetryableErrorTypes: [
+    'AuthenticationError',
+    'PermissionError',
+    'InvalidRequestError',
+    'RequestTooLargeError',
+    'ConfigurationError',
+    'InvalidTargetError',
+    'ExecutionLimitError',
+  ],
+};
+
+// Retry configuration for pipeline testing (fast iteration)
+const TESTING_RETRY = {
+  initialInterval: '10 seconds',
+  maximumInterval: '30 seconds',
+  backoffCoefficient: 2,
+  maximumAttempts: 5,
+  nonRetryableErrorTypes: PRODUCTION_RETRY.nonRetryableErrorTypes,
+};
+
+// Activity proxy with production retry configuration (default)
 const acts = proxyActivities<typeof activities>({
   startToCloseTimeout: '2 hours',
   heartbeatTimeout: '30 seconds',
-  retry: {
-    initialInterval: '5 minutes',
-    maximumInterval: '30 minutes',
-    backoffCoefficient: 2,
-    maximumAttempts: 50,
-    nonRetryableErrorTypes: [
-      'AuthenticationError',
-      'PermissionError',
-      'InvalidRequestError',
-      'RequestTooLargeError',
-      'ConfigurationError',
-      'InvalidTargetError',
-      'ExecutionLimitError',
-    ],
-  },
+  retry: PRODUCTION_RETRY,
+});
+
+// Activity proxy with testing retry configuration (fast)
+const testActs = proxyActivities<typeof activities>({
+  startToCloseTimeout: '10 minutes',
+  heartbeatTimeout: '30 seconds',
+  retry: TESTING_RETRY,
 });
 
 export async function pentestPipelineWorkflow(
@@ -61,6 +80,10 @@ export async function pentestPipelineWorkflow(
 ): Promise<PipelineState> {
   const { workflowId } = workflowInfo();
 
+  // Select activity proxy based on testing mode
+  // Pipeline testing uses fast retry intervals (10s) for quick iteration
+  const a = input.pipelineTestingMode ? testActs : acts;
+
   // Workflow state (queryable)
   const state: PipelineState = {
     status: 'running',
@@ -99,13 +122,13 @@ export async function pentestPipelineWorkflow(
     state.currentPhase = 'pre-recon';
     state.currentAgent = 'pre-recon';
     state.agentMetrics['pre-recon'] =
-      await acts.runPreReconAgent(activityInput);
+      await a.runPreReconAgent(activityInput);
     state.completedAgents.push('pre-recon');
 
     // === Phase 2: Reconnaissance ===
     state.currentPhase = 'recon';
     state.currentAgent = 'recon';
-    state.agentMetrics['recon'] = await acts.runReconAgent(activityInput);
+    state.agentMetrics['recon'] = await a.runReconAgent(activityInput);
     state.completedAgents.push('recon');
 
     // === Phase 3: Vulnerability Analysis (Parallel) ===
@@ -113,11 +136,11 @@ export async function pentestPipelineWorkflow(
     state.currentAgent = 'vuln-agents';
 
     const vulnResults = await Promise.all([
-      acts.runInjectionVulnAgent(activityInput),
-      acts.runXssVulnAgent(activityInput),
-      acts.runAuthVulnAgent(activityInput),
-      acts.runSsrfVulnAgent(activityInput),
-      acts.runAuthzVulnAgent(activityInput),
+      a.runInjectionVulnAgent(activityInput),
+      a.runXssVulnAgent(activityInput),
+      a.runAuthVulnAgent(activityInput),
+      a.runSsrfVulnAgent(activityInput),
+      a.runAuthzVulnAgent(activityInput),
     ]);
 
     const vulnAgents = [
@@ -141,11 +164,11 @@ export async function pentestPipelineWorkflow(
     state.currentAgent = 'exploit-agents';
 
     const exploitResults = await Promise.all([
-      acts.runInjectionExploitAgent(activityInput),
-      acts.runXssExploitAgent(activityInput),
-      acts.runAuthExploitAgent(activityInput),
-      acts.runSsrfExploitAgent(activityInput),
-      acts.runAuthzExploitAgent(activityInput),
+      a.runInjectionExploitAgent(activityInput),
+      a.runXssExploitAgent(activityInput),
+      a.runAuthExploitAgent(activityInput),
+      a.runSsrfExploitAgent(activityInput),
+      a.runAuthzExploitAgent(activityInput),
     ]);
 
     const exploitAgents = [
@@ -169,10 +192,10 @@ export async function pentestPipelineWorkflow(
     state.currentAgent = 'report';
 
     // First, assemble the concatenated report from exploitation evidence files
-    await acts.assembleReportActivity(activityInput);
+    await a.assembleReportActivity(activityInput);
 
     // Then run the report agent to add executive summary and clean up
-    state.agentMetrics['report'] = await acts.runReportAgent(activityInput);
+    state.agentMetrics['report'] = await a.runReportAgent(activityInput);
     state.completedAgents.push('report');
 
     // === Complete ===