From 321f441f4bb067fbc5a26b22e7c87cd825ff8623 Mon Sep 17 00:00:00 2001 From: ajmallesh Date: Thu, 27 Aug 2026 14:28:15 -0700 Subject: [PATCH] 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. --- CLAUDE.md | 3 +- apps/cli/src/scan/derive.ts | 176 +++++++++++++++--- apps/cli/src/scan/pipeline.ts | 29 +++ apps/cli/src/scan/render.ts | 49 +++-- apps/cli/src/scan/safe-fields.ts | 17 +- .../src/ai/sast/capella/temporal/workflow.ts | 102 +++++++--- apps/worker/src/ai/sast/types.ts | 24 +++ apps/worker/src/audit/workflow-logger.ts | 23 +-- apps/worker/src/temporal/shared.ts | 19 +- apps/worker/src/temporal/workflows.ts | 68 +++++-- 10 files changed, 407 insertions(+), 103 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 86c411af..98baaecb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -103,7 +103,7 @@ Published as `@keygraph/shannon` on npm. Contains Docker orchestration logic plu - `apps/cli/src/index.ts` — CLI dispatcher (`setup`, `start`, `stop`, `reset`, `logs`, `status`, `scans`, `build`, `version`) - `apps/cli/src/temporal-client.ts` — `@temporalio/client` reader for `status`: connects to the frontend on `127.0.0.1:7233` (published by compose), `describeScan` (status + `pendingActivities` → running agents), `queryProgress` (live `getProgress` query → `PipelineState`), `getTerminalOutcome` (workflow `result()`). No worker of its own; scans are visible within Temporal's retention window, which `ensureInfra` (`apps/cli/src/docker.ts`) converges to `168h` (7 days) on every successful `shannon start` — override with `SHANNON_TEMPORAL_RETENTION` (a positive whole-hour value like `72h`) -- `apps/cli/src/scan/` — `status` rendering: `pipeline.ts` (static phase/agent plan + `run*Agent` activity-type→agent map + mirrored `PipelineState`/`AgentMetrics` types; keep in sync with the worker), `render.ts` (one renderer for both the live query state and the terminal result) +- `apps/cli/src/scan/` — `status` rendering: `pipeline.ts` (static phase/agent plan + `run*Agent` activity-type→agent map + mirrored `PipelineState`/`AgentMetrics` types; keep in sync with the worker), `derive.ts` (pure phase/agent state derivation shared by the tree and `--json`), `render.ts` (one renderer for both the live query state and the terminal result). The tree shows model work only: every row is an agent, an Agentic SAST stage, or a report step that is currently running or failed. Reconciliation is model work owned by a class, so its wall time renders as a trailing `+ duration` on that class's exploitation row (its analysis row when `exploit: false`) rather than as a row of its own; deterministic bookkeeping stages (`report:*` renumber/assemble/finalize/surface) never appear once they complete. `DerivedPhase.children` (renders sub-rows) and `DerivedPhase.meta` (`duration` vs a `k/N done` tally) are independent — Agentic SAST lists stages under a duration, exploitation lists classes under a tally - `apps/cli/src/mode.ts` — Auto-detection: local mode if `SHANNON_LOCAL=1` env var is set - `apps/cli/src/docker.ts` — Compose lifecycle, image pull/build, ephemeral `docker run` worker spawning - `apps/cli/src/home.ts` — State directory management (`~/.shannon/` for npx, `./` for local) @@ -163,6 +163,7 @@ Around those phases: ### Supporting Systems - **Configuration** — YAML configs in `apps/worker/configs/` use the closed JSON Schema in `config-schema.json`. Every fresh scan runs the fixed five analysis classes; there is no public class selector. `agentic_sast.enabled` is the only public agentic-SAST setting. Finding reconciliation runs on every scan and has no public setting of its own. Config also supports authentication (MFA/TOTP), URL/code rule scoping (`rules.avoid`/`rules.focus`), `exploit`, free-form `rules_of_engagement`, and post-hoc `report` options (`min_severity`, `min_confidence`, `guidance`, and exploit-only `sarif` output via `apps/worker/src/services/sarif-renderer.ts`, on by default for exploit runs and opt out with `report.sarif: "false"`). `code_path` avoid rules are enforced via the `@gotgenes/pi-permission-system` extension: `apps/worker/src/temporal/activities.ts:syncCodePathDenyRules` writes a global `path` deny config once per workflow (`apps/worker/src/ai/pi/permission-system.ts:syncPermissionSystemConfig`), and the executor loads the extension when that config is present (`apps/worker/src/ai/pi/pi-executor.ts`), so denies fire across every tool and child `task` session. Credential resolution — local mode: env vars → `./.env`; npx mode: env vars → `~/.shannon/config.toml` (via `npx @keygraph/shannon setup`) +- **Agentic SAST progress** — Capella runs as a child workflow, so its activities are absent from the parent's `pendingActivities` and invisible to the CLI. The child signals each stage boundary up via `capellaStageProgress` (`apps/worker/src/temporal/shared.ts`); the parent's handler validates the payload and writes the child-supplied `startedAt` and `durationMs` directly to `operationalStages['agentic-sast:']`, so both the live `getProgress` query and the terminal result carry per-stage rows. Signalling is best-effort and every failure is swallowed — a closed or unreachable parent must never fail a SAST run. `CAPELLA_STAGE_LABELS` in `apps/worker/src/ai/sast/types.ts` is the one label table, shared by the scan log and the status tree; `CAPELLA_PROGRESS_STAGES` omits `export`, which runs no model and so never becomes a row. Scans predating the signal keep the aggregate `agentic-sast` span and render as a bare phase line - **Prompts** — Per-phase templates in `apps/worker/prompts/` with variable substitution (`{{TARGET_URL}}`, `{{CONFIG_CONTEXT}}`). Shared partials in `apps/worker/prompts/shared/` via `apps/worker/src/services/prompt-manager.ts`, including `_code-path-rules.txt` (focus/avoid `[FILE]`/`[GLOB]` routing) and `_rules-of-engagement.txt` (free-text engagement rules). When `exploit: false`, `apps/worker/src/services/findings-renderer.ts` deterministically converts each `*_exploitation_queue.json` into a `*_findings.md` for report assembly — no LLM in the loop - **Agent Harness (pi)** — Uses the **pi harness** (`@earendil-works/pi-coding-agent`, requires Node ≥ 22.19) via `apps/worker/src/ai/pi/pi-executor.ts` (`runPiPrompt` → `createAgentSession`). Retry is split in `apps/worker/src/ai/pi/retry-settings.ts`: pi's agent-level loop is off so Temporal owns agent restarts, while `provider.maxRetries` stays on — pi reads the `provider` block independently of the `enabled` flag — so transport faults are absorbed in-session rather than costing a full agent re-run. `maxRetryDelayMs` is left at pi's 60s default. One model runs every phase, named by `SHANNON_AI_MODEL=:` (default `anthropic:claude-sonnet-4-6`). `apps/worker/src/ai/models.ts` parses the spec — splitting on the **first** colon only, so Bedrock IDs keep theirs — and resolves it through pi's `ModelRuntime`. pi ships the `CredentialStore` interface but no in-memory implementation (its own reads `auth.json` from disk), so `RuntimeCredentialStore` in that file supplies one: credentials arrive as env vars in an ephemeral container and must never touch disk. `createModelRuntime(providerId, apiKey)` builds the runtime; `allowModelNetwork` stays at its default `false` so a scan never blocks on a catalog refresh. `resolveModelSelection()` is **async** because `ModelRuntime.create()` is. Any pi-ai provider id is accepted — `parseModelSpec` no longer rejects against a hardcoded list, so pi's registry is the authority (an unknown provider/model surfaces as a clear "not found in pi registry" error at preflight, which points to the browsable catalogue at `pi.dev/models` — `PI_CATALOG_URL` in `apps/worker/src/ai/models.ts`, appended to the not-found errors and shown in the setup wizard's "Other provider" hint). Four providers are **curated** (`CURATED_PROVIDERS`: `anthropic`, `openai`, `xai`, `amazon-bedrock`) with their own credential variables, config sections, and setup flows; each provider's API key env var is declared once in `PROVIDER_API_KEY_ENV` — Shannon uses each vendor's own variable name (`OPENAI_API_KEY`, `XAI_API_KEY`, …), never an invented one; Bedrock's entry is `AWS_BEARER_TOKEN_BEDROCK`, paired with `AWS_REGION`, which preflight requires separately as provider config rather than a credential. Any other provider uses the **generic** credential path: `SHANNON_AI_API_KEY` (`GENERIC_API_KEY_ENV`) supplies the key for any provider whose credential is a plain API key. Curated providers' own variables take precedence over it, and it also works as a fallback for them — Bedrock is the sole exception (it authenticates through its AWS_ variables, so the generic key never stands in for it). The CLI forwards `SHANNON_AI_API_KEY` in `COMMON_FORWARD_VARS` (it is provider-neutral, binding to whatever `SHANNON_AI_MODEL` names, so the "only one provider configured" guard counts only named credentials), and stores it under a generic `[provider]` config.toml section (`provider.api_key`). `npx @keygraph/shannon setup` exposes this as the "Other provider" option: free-text provider id + model id + key (a curated provider id is rejected there, since it has its own option). `SHANNON_AI_BASE_URL` overrides the endpoint for any provider (proxies/gateways); the credential is unchanged. `pointAtGateway` (`apps/worker/src/ai/models.ts`) applies the one dialect change: behind a base URL, `openai` follows `SHANNON_AI_OPENAI_FORMAT` (`chat-completions` default, or `responses`). On `chat-completions` it switches the API to `openai-completions` and drops the catalogue's Responses-shaped `compat` block so pi's `detectCompat` derives completions settings; on `responses` the descriptor is unchanged but for the endpoint. `resolveGatewayFormat` rejects the variable when the provider is not `openai` or no base URL is set, since it cannot take effect there. All other providers keep their API. The CLI mirrors the accepted values in `apps/cli/src/model-spec.ts`, forwards the variable in `COMMON_FORWARD_VARS`, and maps it to `openai.format` in config.toml. `buildEnvFlags` forwards only the selected provider's credential into the worker container. The CLI mirrors the parse rule and the provider/credential tables in `apps/cli/src/model-spec.ts` (it cannot import from the worker package); the two must stay in sync. pi ships no JSON-schema output or `Task`/`TodoWrite` built-ins, so structured queues are captured via a `submit_exploitation_queue` custom tool (`apps/worker/src/ai/queue-schemas.ts`), and `task` (child sessions scoped to `read`, `grep`, `find`, `ls`, `write`, and `bash` — no nested `task` or collector tools; `CHILD_TOOLS` in `apps/worker/src/ai/pi/task-tool.ts`) + `todo_write` (`apps/worker/src/ai/pi/session-tools.ts`) are provided as custom tools; the per-phase collectors are pi custom tools (TypeBox `defineTool` in `apps/worker/src/collectors/`). Shannon sets no thinking configuration at all — no `thinkingLevel` is passed to any `createAgentSession` call, so pi's own default applies. There is no adaptive-thinking support and no `CLAUDE_ADAPTIVE_THINKING` / `core.adaptive_thinking` setting. Browser automation via `playwright-cli` with session isolation (`-s=`). TOTP generation via `generate-totp` CLI tool. Login flow template at `apps/worker/prompts/shared/login-instructions.txt` supports form, SSO, API, and basic auth. On authenticated whitebox scans, the `validate-authentication` preflight performs the single real login and saves the browser session to `auth-state.json` in the per-session audit directory (path from `authStateFile()` in `apps/worker/src/audit/utils.ts`, derived from `generateAuditPath()`). The validation activity (`apps/worker/src/services/validate-authentication.ts`) removes any stale file from a prior run before the agent runs and verifies the file parses and contains cookies or storage before the preflight is marked complete; `logWorkflowComplete` deletes it when the workflow ends so authenticated cookies don't sit on disk between scans. Agent prompts opt in to session reuse by `@include(shared/_shared-session.txt)` before their `` block — the partial restores the session and falls through to the full login flow if verification fails. `vuln-auth`/`exploit-auth` omit the include and own their own login - **Pi Credential Reuse** — `SHANNON_USE_PI_AUTH=1` opts into reusing the host's Pi login, including an `openai-codex` ChatGPT Plus/Pro subscription selected with `SHANNON_AI_MODEL=openai-codex:`. `apps/cli/src/env.ts` requires `~/.pi/agent/auth.json`; `start.ts` passes its path to `spawnWorker`, which mounts only that file read-write at `/tmp/.pi/agent/auth.json`. The flag itself is not forwarded: the worker detects the file with `piAuthPresent()` and passes its path to `ModelRuntime.create`. CLI and worker API-key presence checks are skipped on this path, but the normal preflight model probe still validates the credential. The image and UID-remapping entrypoint keep `/tmp/.pi/agent` owned by `pentest` so adjacent Pi/Shannon configuration remains writable. Refreshed OAuth state is persisted to the host for subsequent scans. diff --git a/apps/cli/src/scan/derive.ts b/apps/cli/src/scan/derive.ts index 0e5fa6f5..0c1007e5 100644 --- a/apps/cli/src/scan/derive.ts +++ b/apps/cli/src/scan/derive.ts @@ -9,7 +9,9 @@ import type { RunningAgent } from '../temporal-client.js'; import { + AGENTIC_SAST_STAGE_ORDER, agentClass, + isModelBackedOperation, type OperationalStageState, operationFamilyKey, type PipelineState, @@ -31,14 +33,31 @@ export interface DerivedAgent { readonly attempt: number | null; /** The step a running operation row is currently on, merged in from its child activity. */ readonly detail?: string; + /** Reconciliation time for this agent's class, rendered as a trailing `+ duration`. + * Reconciliation is model work that produces this agent's inputs, so it is shown + * attached to the agent it feeds rather than as free-floating background work. */ + readonly attachedMs?: number; + /** This class's findings could not be grouped, so each one became its own task. */ + readonly ungrouped?: boolean; readonly error?: string; } +/** How a phase line summarizes itself: its own wall time, or a k/N tally over its children. */ +export type PhaseMetaKind = 'duration' | 'count'; + export interface DerivedPhase { readonly key: string; readonly label: string; - readonly parallel: boolean; + /** Whether the phase renders its agents as sub-rows. Independent of {@link meta}: + * Agentic SAST lists its stages under a duration, exploitation lists its classes under a tally. */ + readonly children: boolean; + readonly meta: PhaseMetaKind; readonly state: RunState; + /** The phase's own span, when the worker records one for the phase rather than for a single + * agent inside it (Agentic SAST). The phase line presents this exactly like an agent row. */ + readonly summary?: DerivedAgent; + /** Rendered after the phase's summary, e.g. to mark work that overlaps other phases. */ + readonly note?: string; readonly agents: readonly DerivedAgent[]; } @@ -205,7 +224,8 @@ export function derivePipeline(input: RenderInput, now: number): DerivedPhase[] return { key: phase.key, label: phase.label, - parallel: phase.parallel, + children: phase.parallel, + meta: phase.parallel ? ('count' as const) : ('duration' as const), state: phaseGlyphState(agents.map((ag) => ag.state)), agents, }; @@ -248,35 +268,137 @@ export function derivePipeline(input: RenderInput, now: number): DerivedPhase[] }; }); - // The synthetic phase(s) appear only when there is operational work to show, so a scan - // with no recorded operational stages keeps the plain agent tree. - if (operationalAgents.length === 0) return agentPhases; + // Operational rows are not peers of the agents. Each one is either model work that + // belongs to an agent (reconciliation), model work that belongs to the SAST engine + // (its stages), or bookkeeping that only earns a row when it is stuck or broken. + return assemblePhases(agentPhases, operationalAgents); +} - // Agentic SAST is a pluggable analysis engine — a peer to the pentest, not background plumbing — - // so it stands in its own phase; reconciliation and report steps remain under "Background work". - const engineAgents = operationalAgents.filter((agent) => operationFamilyKey(agent.name) === 'agentic-sast'); - const backgroundAgents = operationalAgents.filter((agent) => operationFamilyKey(agent.name) !== 'agentic-sast'); +/** Reconciliation wall time per vulnerability class, plus the classes whose grouping degraded. */ +interface ReconciliationView { + readonly durationByClass: ReadonlyMap; + readonly ungroupedClasses: ReadonlySet; +} - const syntheticPhases: DerivedPhase[] = []; - if (engineAgents.length > 0) { - syntheticPhases.push({ - key: 'analysis-engines', - label: 'Analysis Engines', - parallel: true, - state: phaseGlyphState(engineAgents.map((operation) => operation.state)), - agents: engineAgents, - }); +function reconciliationView(operations: readonly DerivedAgent[]): ReconciliationView { + const durationByClass = new Map(); + const ungroupedClasses = new Set(); + for (const operation of operations) { + if (operationFamilyKey(operation.name) !== 'reconciliation') continue; + const [, vulnerabilityClass] = operation.name.split(':'); + if (vulnerabilityClass === undefined) continue; + if (operation.name.endsWith(':fallback')) { + ungroupedClasses.add(vulnerabilityClass); + continue; + } + if (operation.durationMs !== null) durationByClass.set(vulnerabilityClass, operation.durationMs); } - if (backgroundAgents.length > 0) { - syntheticPhases.push({ - key: 'operational-work', - label: 'Background work', - parallel: true, - state: phaseGlyphState(backgroundAgents.map((operation) => operation.state)), - agents: backgroundAgents, - }); + return { durationByClass, ungroupedClasses }; +} + +/** Attach each class's reconciliation time to the agent row it feeds. */ +function withReconciliation(phase: DerivedPhase, view: ReconciliationView): DerivedPhase { + const agents = phase.agents.map((agent): DerivedAgent => { + const vulnerabilityClass = agentClass(agent.name); + const attachedMs = view.durationByClass.get(vulnerabilityClass); + const ungrouped = view.ungroupedClasses.has(vulnerabilityClass); + return { + ...agent, + ...(attachedMs !== undefined && { attachedMs }), + ...(ungrouped && { ungrouped }), + }; + }); + return { ...phase, agents }; +} + +/** + * Build the Agentic SAST phase from the aggregate span the parent workflow records and the + * per-stage rows the SAST child signals up. Scans that predate stage signalling have the + * aggregate but no stages, and render as a bare phase line rather than an error. + */ +function agenticSastPhase(operations: readonly DerivedAgent[]): DerivedPhase | undefined { + const aggregate = operations.find((operation) => operation.name === 'agentic-sast'); + if (aggregate === undefined) return undefined; + + const byStage = new Map(); + for (const operation of operations) { + const [family, stage] = operation.name.split(':'); + if (family !== 'agentic-sast' || stage === undefined) continue; + // The worker's label is the scan log's Title Case form. These rows sit beside the + // lowercase class rows below them, so they read in the same register here. + byStage.set(stage, { ...operation, label: lowercaseFirst(operation.label) }); } - return [...agentPhases, ...syntheticPhases]; + // Run order, not insertion order: a resumed or replayed run can persist stages out of order. + const stages = AGENTIC_SAST_STAGE_ORDER.map((stage) => byStage.get(stage)).filter( + (stage): stage is DerivedAgent => stage !== undefined, + ); + + return { + key: 'agentic-sast', + label: 'Agentic SAST', + children: stages.length > 0, + meta: 'duration', + state: aggregate.state, + summary: aggregate, + // It shares wall time with the pentest phases below it, so the times do not add up + // in sequence. Saying so is cheaper than a layout that pretends to be two columns. + note: 'concurrent', + agents: stages, + }; +} + +/** + * Bookkeeping rows worth showing. A deterministic stage that has completed says nothing — + * it can only ever read 0s — but one that is still running, or that failed, is exactly what + * an operator needs to see, so those keep a row under the phase they belong to. + */ +function troubledReportSteps(operations: readonly DerivedAgent[]): readonly DerivedAgent[] { + return operations.filter((operation) => { + if (isModelBackedOperation(operation.name)) return false; + if (operationFamilyKey(operation.name) !== 'report') return false; + return operation.state === 'running' || operation.state === 'failed'; + }); +} + +/** + * Fold operational rows into the agent phases. Nothing here becomes a bucket of its own: + * every surviving row is either a SAST stage, time attached to an agent, or a report step + * that is currently in trouble. + */ +function assemblePhases(agentPhases: readonly DerivedPhase[], operations: readonly DerivedAgent[]): DerivedPhase[] { + const view = reconciliationView(operations); + // Reconciliation produces the exploitation queue, so its time belongs on the exploitation + // row it feeds. With exploitation off there is no such row, and it falls back to the + // analysis row for the same class so the time is never silently dropped. + const attachTo = agentPhases.some((phase) => phase.key === 'exploitation') + ? 'exploitation' + : 'vulnerability-analysis'; + const reportSteps = troubledReportSteps(operations); + + const phases = agentPhases.map((phase) => { + if (phase.key === attachTo) return withReconciliation(phase, view); + if (phase.key === 'reporting' && reportSteps.length > 0) { + // The report agent stays on the phase line it already titles; the steps in trouble + // become its children, so nothing is listed twice. + const summary = phase.agents[0]; + return { + ...phase, + children: true, + ...(summary !== undefined && { summary }), + state: phaseGlyphState([...phase.agents, ...reportSteps].map((row) => row.state)), + agents: reportSteps, + }; + } + return phase; + }); + + const sast = agenticSastPhase(operations); + if (sast === undefined) return phases; + + // Agentic SAST starts with the scan and runs alongside the pentest, so it reads after + // the login check rather than appended past Reporting where it never ran. + const afterAuth = phases.findIndex((phase) => phase.key === 'auth-validation') + 1; + return [...phases.slice(0, afterAuth), sast, ...phases.slice(afterAuth)]; } export { agentError }; diff --git a/apps/cli/src/scan/pipeline.ts b/apps/cli/src/scan/pipeline.ts index 7b0a5d4d..52ad772e 100644 --- a/apps/cli/src/scan/pipeline.ts +++ b/apps/cli/src/scan/pipeline.ts @@ -302,6 +302,35 @@ export function operationFamilyKey(stageKey: string): string { return separator === -1 ? stageKey : stageKey.slice(0, separator); } +/** The Capella stages that get a progress row, in run order. Mirrors CAPELLA_PROGRESS_STAGES + * in apps/worker/src/ai/sast/types.ts — the deterministic `export` stage is not among them. */ +export const AGENTIC_SAST_STAGE_ORDER: readonly string[] = [ + 'architecture', + 'threat-model', + 'plan', + 'research', + 'dedupe', + 'review', + 'critic', + 'confirm', + 'calibrate', +]; + +/** + * Whether an operational stage represents model work rather than bookkeeping. + * + * Only the agentic-SAST stages and per-class reconciliation run a model; every other + * operational stage is a git commit or a durable-state write that can only ever record + * sub-second wall time. The progress tree shows model work, so this is what decides + * whether a stage is worth a row at all. + */ +export function isModelBackedOperation(stageKey: string): boolean { + const family = operationFamilyKey(stageKey); + if (family === 'agentic-sast') return true; + // A `reconciliation::fallback` marker records a degradation, not a model span. + return family === 'reconciliation' && !stageKey.endsWith(':fallback'); +} + export interface PipelineSummary { readonly totalCostUsd: number; readonly totalDurationMs: number; // Wall-clock (end - start) diff --git a/apps/cli/src/scan/render.ts b/apps/cli/src/scan/render.ts index b6fe1b84..6aaceeff 100644 --- a/apps/cli/src/scan/render.ts +++ b/apps/cli/src/scan/render.ts @@ -130,6 +130,13 @@ function statusBadge(input: RenderInput, opts: RenderOptions): string { // === Line builders === +/** The parts of a derived row agentMeta reads beyond its state and metrics. */ +interface RowExtras { + readonly runningElapsedMs?: number | null; + readonly attachedMs?: number; + readonly ungrouped?: boolean; +} + function agentMeta( state: RunState, metrics: { durationMs: number } | undefined, @@ -137,15 +144,20 @@ function agentMeta( error: string | undefined, opts: RenderOptions, step?: string, + extras?: RowExtras, ): string { if (state === 'completed') { const duration = metrics?.durationMs != null ? formatDuration(metrics.durationMs) : 'done'; - return paint(duration, COLORS.dim, opts.color); + return paint(`${duration}${attachedSuffix(extras)}`, COLORS.dim, opts.color); } if (state === 'running') { const parts = ['running']; if (step !== undefined) parts.push(step); - if (runner?.startedAt !== undefined) parts.push(formatDuration(opts.now - runner.startedAt)); + // An operational row carries its own elapsed time: it is derived from the persisted stage + // span, and has no pending activity on the parent workflow to read a start time from. + const elapsedMs = + runner?.startedAt !== undefined ? opts.now - runner.startedAt : (extras?.runningElapsedMs ?? null); + if (elapsedMs !== null) parts.push(formatDuration(elapsedMs)); if (runner && runner.attempt > 1) parts.push(`retry ${runner.attempt}`); return paint(parts.join(' · '), COLORS.gold, opts.color); } @@ -157,6 +169,17 @@ function agentMeta( return paint('queued', COLORS.dim, opts.color); } +/** + * Time a reconciliation lane contributed to this agent's class, shown as `+ duration` on the + * row it feeds. `ungrouped` marks a class whose findings could not be grouped, so each one + * was tested separately and duplicates are expected. + */ +function attachedSuffix(extras: RowExtras | undefined): string { + if (extras === undefined) return ''; + const time = extras.attachedMs === undefined ? '' : ` + ${formatDuration(extras.attachedMs)}`; + return extras.ungrouped ? `${time} · ungrouped` : time; +} + function phaseMeta(states: readonly RunState[], inPlay: number, parallel: boolean, opts: RenderOptions): string { if (states.every((s) => s === 'pending')) return paint('pending', COLORS.dim, opts.color); if (states.every((s) => s === 'skipped')) return paint('skipped', COLORS.dim, opts.color); @@ -184,20 +207,24 @@ export function renderScan(input: RenderInput, opts: RenderOptions): string { const phaseRunState = phase.state; const metaFor = (agent: (typeof phase.agents)[number]): string => { const metrics = agent.durationMs === null ? undefined : { durationMs: agent.durationMs }; - return agentMeta(agent.state, metrics, byAgent.get(agent.name), agent.error, opts, agent.detail); + return agentMeta(agent.state, metrics, byAgent.get(agent.name), agent.error, opts, agent.detail, agent); }; - // A single-agent phase carries that agent's own duration/cost on the phase line once it - // starts; a parallel phase gets a "k/N done" summary over the agents in play. + // A phase summarizes itself by wall time or by a "k/N done" tally. A phase with its own + // recorded span (Agentic SAST) presents it like any agent row; otherwise a single-agent + // phase borrows its one agent's duration once that agent starts. const first = phase.agents[0]; const firstState = states[0]; - const phaseMetaStr = - !phase.parallel && first && firstState && inPlay(firstState) - ? metaFor(first) - : phaseMeta(states, playing, phase.parallel, opts); - lines.push(` ${glyph(phaseRunState, opts)} ${phase.label.padEnd(26)}${phaseMetaStr}`); + const borrowed = first && firstState && inPlay(firstState) ? metaFor(first) : undefined; + const durationMeta = phase.summary === undefined ? borrowed : metaFor(phase.summary); + const summaryMeta = + phase.meta === 'duration' && durationMeta !== undefined + ? durationMeta + : phaseMeta(states, playing, phase.meta === 'count', opts); + const note = phase.note === undefined ? '' : paint(` · ${phase.note}`, COLORS.dim, opts.color); + lines.push(` ${glyph(phaseRunState, opts)} ${phase.label.padEnd(26)}${summaryMeta}${note}`); - if (!phase.parallel) continue; + if (!phase.children) continue; for (let i = 0; i < phase.agents.length; i++) { const agent = phase.agents[i]; const state = states[i]; diff --git a/apps/cli/src/scan/safe-fields.ts b/apps/cli/src/scan/safe-fields.ts index 1d0bed65..57b3111a 100644 --- a/apps/cli/src/scan/safe-fields.ts +++ b/apps/cli/src/scan/safe-fields.ts @@ -76,7 +76,18 @@ function isProviderFailureCategory(value: unknown): value is string { const OPERATION_LABELS = new Set([ 'Agentic SAST', - 'Miscellaneous findings', + // Capella stage rows, signalled up from the SAST child workflow. Mirrors + // CAPELLA_STAGE_LABELS in apps/worker/src/ai/sast/types.ts, minus the deterministic + // export stage, which never becomes a row. + 'Architecture', + 'Threat model', + 'Plan', + 'Research', + 'Dedupe', + 'Review', + 'Critique', + 'Confirm', + 'Calibrate', 'Reconcile injection', 'Reconcile xss', 'Reconcile auth', @@ -220,7 +231,9 @@ export function safeOperationKey(value: string): string { /^(?:agentic-sast|miscellaneous-pipeline|report:(?:initialize|assemble|compact|checkpoint|finalize|finalize-degraded|terminal|surface))$/u.test( value, ) || - /^(?:reconciliation|report:renumber):(?:injection|xss|auth|authz|ssrf|miscellaneous)$/u.test(value) + /^agentic-sast:(?:architecture|threat-model|plan|research|dedupe|review|critic|confirm|calibrate)$/u.test(value) || + /^(?:reconciliation|report:renumber):(?:injection|xss|auth|authz|ssrf|miscellaneous)$/u.test(value) || + /^reconciliation:(?:injection|xss|auth|authz|ssrf|miscellaneous):fallback$/u.test(value) ) { return value; } diff --git a/apps/worker/src/ai/sast/capella/temporal/workflow.ts b/apps/worker/src/ai/sast/capella/temporal/workflow.ts index 79331d8a..80ead2a6 100644 --- a/apps/worker/src/ai/sast/capella/temporal/workflow.ts +++ b/apps/worker/src/ai/sast/capella/temporal/workflow.ts @@ -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(); let lastGoodFindings: | { readonly artifact: CapellaFindingActivityInput['findingsArtifact']; @@ -317,78 +322,112 @@ export async function capellaWorkflow(input: CapellaWorkflowInput): Promise { + 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 { + stageStartedAt.set(stage, Date.now()); + await signalStage(stage, 'running'); + return stage; + } + + async function endStage(stage: CapellaStage, result: CapellaActivityResult): Promise { + 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> = { + 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 { diff --git a/apps/worker/src/audit/workflow-logger.ts b/apps/worker/src/audit/workflow-logger.ts index 172037d0..e512a9e8 100644 --- a/apps/worker/src/audit/workflow-logger.ts +++ b/apps/worker/src/audit/workflow-logger.ts @@ -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> = { - 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)`, ); } diff --git a/apps/worker/src/temporal/shared.ts b/apps/worker/src/temporal/shared.ts index a9e1c91d..f5b33df3 100644 --- a/apps/worker/src/temporal/shared.ts +++ b/apps/worker/src/temporal/shared.ts @@ -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('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'); diff --git a/apps/worker/src/temporal/workflows.ts b/apps/worker/src/temporal/workflows.ts index b8c93605..1cf7676e 100644 --- a/apps/worker/src/temporal/workflows.ts +++ b/apps/worker/src/temporal/workflows.ts @@ -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 { + recordCapellaStage(progress); + }); + const activityInput: ActivityInput = { webUrl: input.webUrl, repoPath: input.repoPath, @@ -519,11 +530,6 @@ export async function pentestPipeline(input: PipelineInput): Promise(key: string, label: string, operation: () => Promise): Promise { const startedAt = startOperation(key, label); try { @@ -536,6 +542,46 @@ export async function pentestPipeline(input: PipelineInput): Promise; + 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 { - 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