feat(cli)!: rebuild scan status around model work

- show Capella stages beneath the concurrent Agentic SAST phase
- attach reconciliation time to the class row it feeds
- hide completed bookkeeping and the duplicate miscellaneous wrapper
- carry validated child-workflow progress into durable parent state
- derive the terminal tree and status JSON from the same phase shape

BREAKING CHANGE: `status --json` replaces phase `parallel` with `children` and `meta`, adds phase summaries and notes plus agent attachment fields, and removes the `analysis-engines` and `operational-work` phases.
This commit is contained in:
ajmallesh
2026-08-27 14:28:15 -07:00
parent e3c6e8df16
commit 321f441f4b
10 changed files with 407 additions and 103 deletions
@@ -16,17 +16,21 @@ import {
ActivityCancellationType,
ApplicationFailure,
ChildWorkflowCancellationType,
getExternalWorkflowHandle,
isCancellation,
proxyActivities,
workflowInfo,
} from '@temporalio/workflow';
import { capellaStageProgress } from '../../../../temporal/shared.js';
import { isProviderFailureCategory } from '../../../../types/errors.js';
import type {
AgenticSastFallbackReduction,
AgenticSastReduction,
CapellaRecoveredFailure,
CapellaRunResult,
CapellaStage,
CapellaUsage,
import {
type AgenticSastFallbackReduction,
type AgenticSastReduction,
CAPELLA_PROGRESS_STAGES,
type CapellaRecoveredFailure,
type CapellaRunResult,
type CapellaStage,
type CapellaUsage,
} from '../../types.js';
import { capellaSafeFailureMessage } from '../safe-failures.js';
import { usageAccountingWarning } from '../types.js';
@@ -309,6 +313,7 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
reductions: [],
};
let currentStage: CapellaStage = 'architecture';
const stageStartedAt = new Map<CapellaStage, number>();
let lastGoodFindings:
| {
readonly artifact: CapellaFindingActivityInput['findingsArtifact'];
@@ -317,78 +322,112 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
}
| undefined;
// Capella's activities live in this child's history, so the parent cannot observe them.
// Each stage boundary is signalled up instead, which is what puts stage rows in
// `shannon status`. Export is skipped: it runs no model, so the parent drops it anyway.
const parent = workflowInfo().parent;
async function signalStage(stage: CapellaStage, status: 'running' | 'completed' | 'failed'): Promise<void> {
if (parent === undefined || !CAPELLA_PROGRESS_STAGES.includes(stage)) return;
const startedAt = stageStartedAt.get(stage) ?? Date.now();
try {
await getExternalWorkflowHandle(parent.workflowId, parent.runId).signal(capellaStageProgress, {
stage,
status,
startedAt,
...(status !== 'running' && { durationMs: Date.now() - startedAt }),
});
} catch {
// Progress reporting is cosmetic. A parent that has already closed, or a signal that
// cannot be delivered, must never take down a SAST run that is otherwise fine.
}
}
/** Opens a stage's span and returns it, so the caller's `currentStage` cursor is a
* visible assignment rather than a hidden write from inside this closure. */
async function beginStage(stage: CapellaStage): Promise<CapellaStage> {
stageStartedAt.set(stage, Date.now());
await signalStage(stage, 'running');
return stage;
}
async function endStage<T>(stage: CapellaStage, result: CapellaActivityResult<T>): Promise<void> {
acceptStage(accumulator, stage, result);
await signalStage(stage, 'completed');
}
try {
currentStage = await beginStage('architecture');
const architecture = await architectureActivities.capellaArchitecture(
baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaArchitecture),
);
acceptStage(accumulator, 'architecture', architecture);
await endStage('architecture', architecture);
currentStage = 'threat-model';
currentStage = await beginStage('threat-model');
const threatModelInput: CapellaThreatModelActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaThreatModel),
architectureArtifact: architecture.artifact,
};
const threatModel = await threatModelActivities.capellaThreatModel(threatModelInput);
acceptStage(accumulator, 'threat-model', threatModel);
await endStage('threat-model', threatModel);
currentStage = 'plan';
currentStage = await beginStage('plan');
const planInput: CapellaPlanActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaPlan),
architectureArtifact: architecture.artifact,
threatModelArtifact: threatModel.artifact,
};
const plan = await planActivities.capellaPlan(planInput);
acceptStage(accumulator, 'plan', plan);
await endStage('plan', plan);
if (plan.value.investigationCount === 0) {
// Nothing to research: still run export so the scan always ends with a valid,
// empty SARIF artifact rather than an absent one.
currentStage = 'export';
currentStage = await beginStage('export');
const exported = await exportActivities.capellaExport(exportInput(input));
acceptStage(accumulator, 'export', exported);
await endStage('export', exported);
return succeededResult(startedAt, accumulator, exported);
}
currentStage = 'research';
currentStage = await beginStage('research');
const researchInput: CapellaResearchActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaResearch),
architectureArtifact: architecture.artifact,
planArtifact: plan.artifact,
};
const research = await researchActivities.capellaResearch(researchInput);
acceptStage(accumulator, 'research', research);
await endStage('research', research);
lastGoodFindings = { artifact: research.artifact, stage: 'research', findingCount: research.value.findingCount };
if (research.value.findingCount === 0) {
currentStage = 'export';
currentStage = await beginStage('export');
const exported = await exportActivities.capellaExport(exportInput(input));
acceptStage(accumulator, 'export', exported);
await endStage('export', exported);
return succeededResult(startedAt, accumulator, exported);
}
currentStage = 'dedupe';
currentStage = await beginStage('dedupe');
const dedupeInput: CapellaFindingActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaDedupe),
findingsArtifact: research.artifact,
};
const dedupe = await dedupeActivities.capellaDedupe(dedupeInput);
acceptStage(accumulator, 'dedupe', dedupe);
await endStage('dedupe', dedupe);
lastGoodFindings = { artifact: dedupe.artifact, stage: 'dedupe', findingCount: dedupe.value.findingCount };
currentStage = 'review';
currentStage = await beginStage('review');
const reviewInput: CapellaFindingActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaReview),
findingsArtifact: dedupe.artifact,
};
const review = await reviewActivities.capellaReview(reviewInput);
acceptStage(accumulator, 'review', review);
await endStage('review', review);
lastGoodFindings = { artifact: review.artifact, stage: 'review', findingCount: review.value.findingCount };
let exportArtifact = review.artifact;
let exportStage: CapellaExportSourceStage = 'review';
const reviewedSurvivors = review.value.validCount + review.value.provisionalCount;
if (reviewedSurvivors > 0) {
currentStage = 'critic';
currentStage = await beginStage('critic');
const criticInput: CapellaKnowledgeFindingActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaCritic),
findingsArtifact: review.artifact,
@@ -396,19 +435,19 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
threatModelArtifact: threatModel.artifact,
};
const critic = await criticActivities.capellaCritic(criticInput);
acceptStage(accumulator, 'critic', critic);
await endStage('critic', critic);
lastGoodFindings = { artifact: critic.artifact, stage: 'critic', findingCount: critic.value.findingCount };
currentStage = 'confirm';
currentStage = await beginStage('confirm');
const confirmInput: CapellaFindingActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaConfirm),
findingsArtifact: critic.artifact,
};
const confirm = await confirmActivities.capellaConfirm(confirmInput);
acceptStage(accumulator, 'confirm', confirm);
await endStage('confirm', confirm);
lastGoodFindings = { artifact: confirm.artifact, stage: 'confirm', findingCount: confirm.value.findingCount };
currentStage = 'calibrate';
currentStage = await beginStage('calibrate');
const calibrateInput: CapellaKnowledgeFindingActivityInput = {
...baseInput(input, CAPELLA_ACTIVITY_POLICIES.capellaCalibrate),
findingsArtifact: confirm.artifact,
@@ -416,7 +455,7 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
threatModelArtifact: threatModel.artifact,
};
const calibrate = await calibrateActivities.capellaCalibrate(calibrateInput);
acceptStage(accumulator, 'calibrate', calibrate);
await endStage('calibrate', calibrate);
lastGoodFindings = {
artifact: calibrate.artifact,
stage: 'calibrate',
@@ -426,9 +465,9 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
exportStage = 'calibrate';
}
currentStage = 'export';
currentStage = await beginStage('export');
const exported = await exportActivities.capellaExport(exportInput(input, exportArtifact, exportStage));
acceptStage(accumulator, 'export', exported);
await endStage('export', exported);
return succeededResult(startedAt, accumulator, exported);
} catch (error) {
// Cancellation must escape: absorbing it into a failed result would make a
@@ -438,6 +477,7 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
if (hasCancellationInCauseChain(error)) throw error;
const failedStage = currentStage;
await signalStage(failedStage, 'failed');
const details = acceptFailureDetails(accumulator, error);
const safeError = capellaSafeFailureMessage(applicationFailure(error)?.type);
if (failedStage === 'export') {
@@ -464,7 +504,7 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise<Cape
const fallbackExport = await exportActivities.capellaExport(
exportInput(input, lastGoodFindings?.artifact, lastGoodFindings?.stage, fallbackReduction, fallbackFailure),
);
acceptStage(accumulator, 'export', fallbackExport);
await endStage('export', fallbackExport);
const recoveredFailure: CapellaRecoveredFailure = {
failedStage,
error: safeError,
+24
View File
@@ -32,6 +32,30 @@ export function isCapellaStage(value: string): value is CapellaStage {
return CAPELLA_STAGE_SET.has(value);
}
/**
* The one human-facing name per stage, shared by the scan log and the `shannon status`
* progress tree so an operator reads the same word in both places. This module imports
* nothing, so the parent workflow can use it inside the Temporal sandbox.
*/
export const CAPELLA_STAGE_LABELS: Readonly<Record<CapellaStage, string>> = {
architecture: 'Architecture',
'threat-model': 'Threat model',
plan: 'Plan',
research: 'Research',
dedupe: 'Dedupe',
review: 'Review',
critic: 'Critique',
confirm: 'Confirm',
calibrate: 'Calibrate',
export: 'Export',
};
/**
* Export writes artifacts but runs no model, so it is the one stage the progress tree
* leaves out: a row that can only ever read 0s tells an operator nothing.
*/
export const CAPELLA_PROGRESS_STAGES: readonly CapellaStage[] = CAPELLA_STAGES.filter((stage) => stage !== 'export');
export type CapellaFailurePoint = CapellaStage | 'workflow';
export interface CapellaUsage {
+5 -18
View File
@@ -9,7 +9,7 @@
import { promises as fsPromises } from 'node:fs';
import path from 'node:path';
import { isCapellaSafeFailureMessage, isCapellaTerminalStageLabel } from '../ai/sast/capella/safe-failures.js';
import type { CapellaStage } from '../ai/sast/types.js';
import { CAPELLA_STAGE_LABELS, type CapellaStage } from '../ai/sast/types.js';
import { type ErrorCode, isProviderFailureCategory } from '../types/errors.js';
import { isPartialReason, type PartialReasonView, projectPartialReasons } from '../types/run-state.js';
import { formatDuration, formatTimestamp } from '../utils/formatting.js';
@@ -88,19 +88,6 @@ export interface WorkflowSummary {
export type ChildTaskFailureCode = 'CANCELLED' | 'CHILD_TASK_FAILED';
const AGENTIC_SAST_STAGE_LABELS: Readonly<Record<CapellaStage, string>> = {
architecture: 'Architecture',
'threat-model': 'Threat model',
plan: 'Planning',
research: 'Audit wave',
dedupe: 'Deduplication',
review: 'Review',
critic: 'Critic',
confirm: 'Confirmation',
calibrate: 'Calibration',
export: 'Export',
};
function isSafeCount(value: number): boolean {
return Number.isSafeInteger(value) && value >= 0 && value <= 1_000_000_000;
}
@@ -361,7 +348,7 @@ export class WorkflowLogger {
await WorkflowLogger.writeStageStructuralLine(
workflowLogPath,
stage,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${AGENTIC_SAST_STAGE_LABELS[stage]}: Starting (attempt ${safeAttempt} of ${safeMaximum})`,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${CAPELLA_STAGE_LABELS[stage]}: Starting (attempt ${safeAttempt} of ${safeMaximum})`,
);
}
@@ -382,7 +369,7 @@ export class WorkflowLogger {
await WorkflowLogger.writeStageStructuralLine(
workflowLogPath,
stage,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${AGENTIC_SAST_STAGE_LABELS[stage]}: Completed (${details.join(', ')})`,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${CAPELLA_STAGE_LABELS[stage]}: Completed (${details.join(', ')})`,
);
}
@@ -402,7 +389,7 @@ export class WorkflowLogger {
await WorkflowLogger.writeStageStructuralLine(
workflowLogPath,
stage,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${AGENTIC_SAST_STAGE_LABELS[stage]}: ${outcome} (attempt ${safeAttempt} of ${safeMaximum}, ${safeCode})`,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${CAPELLA_STAGE_LABELS[stage]}: ${outcome} (attempt ${safeAttempt} of ${safeMaximum}, ${safeCode})`,
);
}
@@ -418,7 +405,7 @@ export class WorkflowLogger {
await WorkflowLogger.writeStageStructuralLine(
workflowLogPath,
stage,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${AGENTIC_SAST_STAGE_LABELS[stage]}: Cancelled (attempt ${safeAttempt} of ${safeMaximum}, CANCELLED)`,
`[${new Date().toISOString().replace('T', ' ').slice(0, 19)}] [AGENTIC-SAST] ${CAPELLA_STAGE_LABELS[stage]}: Cancelled (attempt ${safeAttempt} of ${safeMaximum}, CANCELLED)`,
);
}
+18 -1
View File
@@ -1,4 +1,4 @@
import { defineQuery } from '@temporalio/workflow';
import { defineQuery, defineSignal } from '@temporalio/workflow';
export type { AgentMetrics } from '../types/metrics.js';
@@ -230,3 +230,20 @@ export interface VulnExploitPipelineResult {
}
export const getProgress = defineQuery<PipelineProgress>('getProgress');
/**
* One Capella stage transition, reported by the SAST child workflow to its parent.
*
* Capella runs as a child workflow, so its activities never appear in the parent's
* pending activities and the CLI cannot observe them. This signal is how per-stage
* progress reaches the parent's durable `operationalStages`, which is what both the
* live `getProgress` query and the terminal result render from.
*/
export interface CapellaStageProgress {
readonly stage: CapellaStage;
readonly status: 'running' | 'completed' | 'failed';
readonly startedAt: number;
readonly durationMs?: number;
}
export const capellaStageProgress = defineSignal<[CapellaStageProgress]>('capellaStageProgress');
+56 -12
View File
@@ -31,7 +31,12 @@ import type { StageMetrics } from '../ai/reconciliation/stage-contracts.js';
import { capellaTerminalStageLabel, isCapellaSafeFailureMessage } from '../ai/sast/capella/safe-failures.js';
import type { CapellaWorkflowInput } from '../ai/sast/capella/temporal/activity-types.js';
import { CAPELLA_CHILD_WORKFLOW_OPTIONS, capellaWorkflow } from '../ai/sast/capella/temporal/workflow.js';
import type { CapellaRunResult, SarifRef } from '../ai/sast/types.js';
import {
CAPELLA_PROGRESS_STAGES,
CAPELLA_STAGE_LABELS,
type CapellaRunResult,
type SarifRef,
} from '../ai/sast/types.js';
import type { WorkflowPhase } from '../audit/safe-fields.js';
import type { AgentName, VulnType } from '../types/agents.js';
import { ALL_AGENTS } from '../types/agents.js';
@@ -59,6 +64,8 @@ import {
} from './reconcile-activity-types.js';
import {
type AgentMetrics,
type CapellaStageProgress,
capellaStageProgress,
type DurableStateSummary,
type FinalizeReportActivityResult,
getProgress,
@@ -429,6 +436,10 @@ export async function pentestPipeline(input: PipelineInput): Promise<PipelineSta
}),
);
setHandler(capellaStageProgress, (progress: CapellaStageProgress): void => {
recordCapellaStage(progress);
});
const activityInput: ActivityInput = {
webUrl: input.webUrl,
repoPath: input.repoPath,
@@ -519,11 +530,6 @@ export async function pentestPipeline(input: PipelineInput): Promise<PipelineSta
};
}
/** A stage an earlier run already settled. It records no span, so it contributes no wall time. */
function skipOperation(key: string, label: string): void {
state.operationalStages[key] = { key, label, status: 'skipped' };
}
async function runOperation<T>(key: string, label: string, operation: () => Promise<T>): Promise<T> {
const startedAt = startOperation(key, label);
try {
@@ -536,6 +542,46 @@ export async function pentestPipeline(input: PipelineInput): Promise<PipelineSta
}
}
/**
* Record one Capella stage transition signalled by the SAST child workflow.
*
* The payload crosses a workflow boundary, so every field is revalidated here rather
* than trusted: only model-backed stages and valid state/timing payloads are accepted.
*/
function recordCapellaStage(progress: unknown): void {
if (typeof progress !== 'object' || progress === null) return;
const candidate = progress as Record<string, unknown>;
const stageValue = candidate.stage;
if (typeof stageValue !== 'string') return;
const stage = CAPELLA_PROGRESS_STAGES.find((value) => value === stageValue);
if (stage === undefined) return;
const status = candidate.status;
if (status !== 'running' && status !== 'completed' && status !== 'failed') return;
const startedAt = candidate.startedAt;
if (!Number.isSafeInteger(startedAt) || (startedAt as number) < 0) return;
const key = `${CAPELLA_OPERATION_KEY}:${stage}`;
const label = CAPELLA_STAGE_LABELS[stage];
if (status === 'running') {
state.operationalStages[key] = { key, label, status: 'running', startedAt: startedAt as number };
return;
}
// Trust the child's own span for duration: the signal may be delivered after the stage
// ended, so measuring from the parent's clock here would inflate every stage.
const durationMs = candidate.durationMs;
if (!Number.isSafeInteger(durationMs) || (durationMs as number) < 0) {
return;
}
state.operationalStages[key] = {
key,
label,
status,
startedAt: startedAt as number,
durationMs: durationMs as number,
...(status === 'failed' && { error: OPERATION_FAILURE }),
};
}
function addReconciliationMetrics(
vulnerabilityClass: ReconciliationClass,
stage: 'enrich' | 'form',
@@ -1013,17 +1059,17 @@ export async function pentestPipeline(input: PipelineInput): Promise<PipelineSta
* the class was ever admitted for exploitation, rather than re-deciding admission from scratch.
*/
async function runMiscellaneousPipeline(effectiveSarif: SarifRef): Promise<void> {
const key = 'miscellaneous-pipeline';
const label = 'Miscellaneous findings';
// This lane records no operational stage of its own. It is a span around work that
// already reports itself -- `reconcileClass('miscellaneous')` and the miscellaneous
// exploit agent -- so a row here would count both a second time.
//
// An earlier run already settled this class. Re-deciding admission would ask durable state to
// move backwards, which fails closed and would be recorded as a class failure that never
// happened; re-running the lane would also repeat work that run already paid for.
if (miscellaneousLaneIsSettled(miscellaneousOutcome)) {
if (miscellaneousOutcome === 'completed') markCompleted('miscellaneous-exploit');
skipOperation(key, label);
return;
}
const startedAt = startOperation(key, label);
let reconciliationCompleted = false;
try {
await seedMiscellaneousActs.seedEmptyProducerQueue({ sessionId });
@@ -1045,11 +1091,9 @@ export async function pentestPipeline(input: PipelineInput): Promise<PipelineSta
}
const persisted = await deterministicReportActs.persistMiscellaneousOutcome(activityInput, outcome);
applyDurableSummary(persisted);
completeOperation(key, label, startedAt);
} catch (error) {
if (hasCancellationInCauseChain(error)) throw error;
const message = reconciliationCompleted ? MISCELLANEOUS_PIPELINE_FAILURE : CLASS_RECONCILIATION_FAILURE;
failOperation(key, label, startedAt, message);
if (!reconciliationCompleted) {
state.failedReconciliations.push({ vulnerabilityClass: 'miscellaneous', error: message });
addPartialReason({ code: 'class_reconciliation_failed', vulnerabilityClass: 'miscellaneous' });