From 3bdcfac85dadc3db31dc21118144d8e1fb201d7e Mon Sep 17 00:00:00 2001 From: ajmallesh Date: Wed, 26 Aug 2026 19:57:07 -0700 Subject: [PATCH] perf: overlap static analysis and the Miscellaneous lane with the pentest Run Agentic SAST alongside vulnerability analysis and run Miscellaneous exploitation alongside the specialist exploitation lanes. Keep reconciliation dependent on the completed static-analysis result while preserving parallel work everywhere that has no data dependency. --- apps/worker/src/temporal/workflows.ts | 179 ++++++++++++++++++++++---- 1 file changed, 157 insertions(+), 22 deletions(-) diff --git a/apps/worker/src/temporal/workflows.ts b/apps/worker/src/temporal/workflows.ts index 8e717a6b..0336f923 100644 --- a/apps/worker/src/temporal/workflows.ts +++ b/apps/worker/src/temporal/workflows.ts @@ -175,6 +175,20 @@ const MAX_CONCURRENT_PIPELINES = 5; const MAX_PIPELINE_ERROR_MESSAGE_LENGTH = 2000; const MAX_NON_FATAL_FAILURES = 32; +const CAPELLA_OPERATION_KEY = 'agentic-sast'; +const CAPELLA_OPERATION_LABEL = 'Agentic SAST'; +const CAPELLA_INFRASTRUCTURE_FAILURE = 'Agentic SAST infrastructure failed before producing a usable result.'; +const CAPELLA_UNFINISHED = 'Agentic SAST had not finished when the scan stopped.'; + +/** + * The single Capella outcome every vulnerability class joins on. Agentic SAST overlaps the + * pentest, so its result is settled once and read by all five classes: a usable SARIF, no + * SARIF, or the original cancellation that every waiter rethrows unchanged. + */ +type CapellaSettlement = + | { readonly outcome: 'settled'; readonly sarif?: SarifRef } + | { readonly outcome: 'cancelled'; readonly error: unknown }; + function truncatePipelineErrorMessage(message: string): string { if (message.length <= MAX_PIPELINE_ERROR_MESSAGE_LENGTH) return message; return `${message.slice(0, MAX_PIPELINE_ERROR_MESSAGE_LENGTH - 20)}\n[truncated]`; @@ -426,6 +440,13 @@ export async function pentestPipeline(input: PipelineInput): Promise | null = null; + // Latched when a stopped run has recorded the terminal Capella state. The hard-failure path + // does not wait for the child, so the child can still return while the terminal activity + // yields; from that point the recorded state is final and no continuation may rewrite it. + let capellaTerminallyProjected = false; function applyDurableSummary(summary: DurableStateSummary): void { state.expectedAgents = [...summary.expectedAgents]; @@ -666,7 +687,7 @@ export async function pentestPipeline(input: PipelineInput): Promise Promise, runExploitAgent: () => Promise, - effectiveSarif?: SarifRef, + capella: Promise, ): Promise { const vulnAgentName = `${vulnType}-vuln` as AgentName; const exploitAgentName = `${vulnType}-exploit` as AgentName; @@ -684,8 +705,13 @@ export async function pentestPipeline(input: PipelineInput): Promise { + try { + const sarif = await runCapella(); + if (sarif === undefined) return { outcome: 'settled' }; + return { outcome: 'settled', sarif }; + } catch (error) { + if (hasCancellationInCauseChain(error)) return { outcome: 'cancelled', error }; + // `runCapella` projects every failure it can see, so an escape means that projection + // itself failed partway; only the part it never reached is recovered here, and it is one + // shared Capella outcome, never five separate class failures. Nothing in this recovery + // may reject: on some paths no class ever joins, and an unobserved rejection in the + // workflow VM is escalated rather than dropped. + const running = state.agenticSast; + try { + if (running.status === 'running') projectCapellaInfrastructureFailure(running.startedAt); + } catch (projectionError) { + try { + log.warn('Capella failure projection did not complete', { + error: projectionError instanceof Error ? projectionError.message : String(projectionError), + }); + } catch { + // Even the warning is best-effort. A log that cannot be written must not turn the + // settlement every class joins into a rejected promise. + } + } + return { outcome: 'settled' }; + } + } + + /** + * The Capella child runs under wait-for-cancellation, so a cancelled parent observes its + * settlement before projecting the terminal state. Waiting never replaces the cancellation + * this path is already reporting. + */ + async function awaitCapellaSettlement(settlement: Promise): Promise { + try { + await settlement; + } catch (waitError) { + log.warn('Capella settlement did not resolve while the scan was stopping', { + error: waitError instanceof Error ? waitError.message : String(waitError), + }); + } + } + + /** + * A Capella child that never returned recorded no complete operational metric, so a run that + * stops while it is still running reports the stage as failed and its spend as incomplete + * rather than inventing either. A stopped run carries no partial reasons, so none is added. + */ + function projectUnfinishedCapella(): void { + const running = state.agenticSast; + if (running.status !== 'running') return; + projectCapellaWorkflowFailure(CAPELLA_UNFINISHED, running.startedAt); + capellaTerminallyProjected = true; + operationalSpendMissing = true; + } + /** * The internal `miscellaneous` class: findings outside the five fixed vulnerability classes, * carried through the same reconciliation and exploitation-decision path those classes use. @@ -931,6 +1041,17 @@ export async function pentestPipeline(input: PipelineInput): Promise): Promise { + const settled = await capella; + if (settled.outcome === 'cancelled') throw settled.error; + if (settled.sarif !== undefined) await runMiscellaneousPipeline(settled.sarif); + } + function recordAssemblyOmissions(failedClasses: readonly ReconciliationClass[]): void { for (const vulnerabilityClass of failedClasses) { // The append rules drop the omission when the class already carries an upstream reason. @@ -1149,7 +1270,10 @@ export async function pentestPipeline(input: PipelineInput): Promise () => runVulnExploitPipeline(config.vulnType, config.runVuln, config.runExploit, effectiveSarif), + (config) => () => runVulnExploitPipeline(config.vulnType, config.runVuln, config.runExploit, settlement), ); + // Launch the Miscellaneous lane concurrently with the five fixed classes; it shares the same + // settled SARIF and joins the common barrier below. + const miscellaneousLane = runMiscellaneousExploitLane(settlement); const pipelineResults = await runWithConcurrencyLimit(pipelineThunks, MAX_CONCURRENT_PIPELINES); + // Join the sixth lane before aggregation so its outcome is always observed (never a + // dropped rejection in the workflow VM) and a cancellation from either path propagates. + await miscellaneousLane; aggregatePipelineResults(pipelineResults); if (state.failedPipelines.length > 0) { activityInput.failedClasses = state.failedPipelines.map((failure) => failure.vulnType); } await a.logPhaseTransition(activityInput, 'vulnerability-exploitation', 'complete'); - if (effectiveSarif !== undefined) await runMiscellaneousPipeline(effectiveSarif); } await finalizeReportPipeline(); @@ -1182,6 +1311,10 @@ export async function pentestPipeline(input: PipelineInput): Promise