From b13788d8ef1148bdd92867ee77855b334dc4dad9 Mon Sep 17 00:00:00 2001 From: ezl-keygraph Date: Mon, 24 Aug 2026 20:04:47 +0530 Subject: [PATCH] fix: terminate failed scans in Temporal and surface the reason when following (#429) * fix(cli): skip splash screen off a TTY (e.g. CI) * fix: terminate failed scans in Temporal and surface the reason when following * fix(cli): indent embedded newlines within failure-error segments * fix(worker): omit the Agent Breakdown section when no agents completed * fix(cli): don't reprint the failure reason when the log already showed it * fix(worker): indent embedded newlines within the workflow.log error block --- apps/cli/src/commands/logs.ts | 130 ++++++++++++++++------- apps/cli/src/commands/start.ts | 44 ++++++-- apps/cli/src/index.ts | 3 +- apps/cli/src/scan/failure.ts | 31 ++++++ apps/cli/src/scan/render.ts | 4 +- apps/cli/src/temporal-client.ts | 74 +++++++++++++ apps/worker/src/audit/workflow-logger.ts | 32 +++--- apps/worker/src/temporal/pipeline.ts | 1 - apps/worker/src/temporal/shared.ts | 15 --- apps/worker/src/temporal/workflows.ts | 8 +- 10 files changed, 261 insertions(+), 81 deletions(-) create mode 100644 apps/cli/src/scan/failure.ts diff --git a/apps/cli/src/commands/logs.ts b/apps/cli/src/commands/logs.ts index ed6bcec..84989c3 100644 --- a/apps/cli/src/commands/logs.ts +++ b/apps/cli/src/commands/logs.ts @@ -1,21 +1,23 @@ /** * `shannon logs` command — tail a scan's live log. * - * Uses chokidar for reliable cross-platform file watching and - * bounded synchronous reads to prevent duplicate output. + * The log file is streamed for its content; completion is decided by Temporal (the + * workflow's status), so a worker that dies mid-run can't leave the tail hanging. Uses + * chokidar for reliable cross-platform file watching and bounded synchronous reads to + * prevent duplicate output. */ import fs from 'node:fs'; import path from 'node:path'; +import { setTimeout as sleep } from 'node:timers/promises'; import { watch } from 'chokidar'; import { fail } from '../errors.js'; import { getWorkspacesDir } from '../home.js'; import { resolveRunFile } from '../paths.js'; +import { resolveWorkflowId } from '../session.js'; +import { waitForWorkflowClose } from '../temporal-client.js'; import { stdoutIsTerminal } from '../tty.js'; -// Match the exact line the worker writes — anchored to prevent false positives from agent output -const COMPLETION_PATTERN = /^Scan (COMPLETED|FAILED)$/m; - /** Read a byte range from a file and return it as a UTF-8 string. */ function readRange(filePath: string, start: number, end: number): string { const length = end - start; @@ -62,60 +64,114 @@ export function resolveLogFile(workspaceId: string): string { ); } +export interface TailOptions { + /** Workflow whose Temporal status decides when the tail stops. Without it, only Ctrl-C ends the tail. */ + readonly workflowId?: string; + /** Called if the tail ends because Temporal became unreachable, with the captured error. */ + readonly onUnreachable?: (lastError: string) => void; +} + +/** Outcome of a tail: whether the streamed log already contained the worker's `Scan FAILED` block. */ +export interface TailResult { + readonly sawFailure: boolean; +} + +// The worker writes this exact line at the head of its terminal failure summary. +const FAILURE_MARKER = /^Scan FAILED$/m; + /** - * Tail a scan's log until it reports completion, resolving when the completion marker appears - * (or the file is gone, or Ctrl-C stops it). Never exits the process, so the caller decides what - * happens next: plain `logs` exits 0; `start --follow` reads the workflow outcome first. + * Stream a scan's log to the terminal until the workflow closes (completion comes from Temporal, + * or Ctrl-C). A Temporal outage is warned about and, if sustained, ends the tail with a diagnostic. + * Never exits the process: plain `logs` exits; `start --follow` reads the workflow outcome first. + * Reports whether the log already showed the failure, so a caller need not print it a second time. */ -export function tailUntilComplete(logFile: string): Promise { +export function tailUntilComplete(logFile: string, opts: TailOptions = {}): Promise { return new Promise((resolve) => { let position = 0; + let done = false; + let sawFailure = false; + const controller = new AbortController(); + let watcher: ReturnType | undefined; - /** - * Output any new content appended since the last read. - * Returns true when the workflow completion marker is detected. - */ - function flush(): boolean { + /** Output any new content appended since the last read. */ + function flush(): void { try { const { size } = fs.statSync(logFile); - if (size <= position) return false; - + if (size <= position) return; const data = readRange(logFile, position, size); process.stdout.write(data); position = size; - - return COMPLETION_PATTERN.test(data); + if (!sawFailure && FAILURE_MARKER.test(data)) { + sawFailure = true; + } } catch { - // File deleted or unreadable — treat as done - return true; + // File not present yet or transiently unreadable — nothing to flush this round. } } - // 1. Output existing content - if (flush()) { - resolve(); - return; + function finish(): void { + if (done) return; + done = true; + controller.abort(); + if (watcher) { + watcher.close().finally(() => resolve({ sawFailure })); + // Safety net — resolve anyway if watcher.close() stalls. + setTimeout(() => resolve({ sawFailure }), 1000).unref(); + } else { + resolve({ sawFailure }); + } } - // 2. Watch for appended content via chokidar - const watcher = watch(logFile, { persistent: true }); + // 1. Output existing content, then stream anything appended. + flush(); + watcher = watch(logFile, { persistent: true }); + watcher.on('change', () => flush()); - const stop = (): void => { - watcher.close().finally(() => resolve()); - // Safety net — resolve anyway if watcher.close() stalls - setTimeout(() => resolve(), 1000).unref(); - }; + // 2. Ctrl-C stops watching. + process.on('SIGINT', finish); - watcher.on('change', () => { - if (flush()) stop(); - }); - - process.on('SIGINT', stop); + // 3. Temporal decides completion. Without a workflow id, the tail relies on Ctrl-C alone. + if (opts.workflowId) { + waitForWorkflowClose(opts.workflowId, { + signal: controller.signal, + onConnectionTrouble: (lastError) => { + if (!done) console.error(`\n⚠ Lost contact with Temporal, retrying… (${lastError})`); + }, + onReconnected: () => { + if (!done) console.error(' Reconnected to Temporal.'); + }, + }) + .then(async (end) => { + if (done) return; + // Flush, let a just-written final summary land, then flush the tail once more. + flush(); + await sleep(750).catch(() => {}); + flush(); + if (end.reason === 'unreachable') { + console.error('\nScan watch aborted: lost contact with Temporal.'); + console.error(` Last error: ${end.lastError}`); + console.error(' Temporal may have crashed — check `docker compose logs temporal`.'); + opts.onUnreachable?.(end.lastError); + } + finish(); + }) + .catch(() => { + // waitForWorkflowClose never rejects; guard only against an aborted race. + }); + } }); } export function logs(workspaceId: string): void { const logFile = resolveLogFile(workspaceId); + const workflowId = resolveWorkflowId(workspaceId); console.error(stdoutIsTerminal() ? `Tailing scan log: ${logFile}` : 'Tailing scan log'); - tailUntilComplete(logFile).finally(() => process.exit(0)); + + let unreachable = false; + tailUntilComplete(logFile, { + ...(workflowId ? { workflowId } : {}), + onUnreachable: () => { + unreachable = true; + }, + }).finally(() => process.exit(unreachable ? 1 : 0)); } diff --git a/apps/cli/src/commands/start.ts b/apps/cli/src/commands/start.ts index 97de116..9f114fc 100644 --- a/apps/cli/src/commands/start.ts +++ b/apps/cli/src/commands/start.ts @@ -24,6 +24,7 @@ import { resolveRepo, resolveRunFile, } from '../paths.js'; +import { indentFailureSegments } from '../scan/failure.js'; import { resolveWorkflowId } from '../session.js'; import { displaySplash } from '../splash.js'; import { getTerminalOutcome } from '../temporal-client.js'; @@ -81,7 +82,10 @@ export async function start(args: StartArgs): Promise { const config = args.config ? resolveConfig(args.config) : undefined; // Inputs are valid — show the splash before the Docker/Temporal setup work. - displaySplash(isLocal() ? undefined : args.version); + // Skip it off a real terminal (e.g. CI) so piped/logged output stays clean. + if (stdoutIsTerminal()) { + displaySplash(isLocal() ? undefined : args.version); + } // 4. Ensure workspaces dir is writable by container user (UID 1001) const workspacesDir = getWorkspacesDir(); @@ -237,12 +241,14 @@ export async function start(args: StartArgs): Promise { } /** - * Follow a just-started scan (for `--follow`, aimed at CI): stream its log to completion, then - * exit on the workflow outcome — 0 if the assessment ran, 1 if the scan failed. That tracks - * whether the pipeline ran, not whether vulnerabilities were found. + * Follow a just-started scan (for `--follow`, aimed at CI): stream its log while Temporal drives + * completion, then exit on the workflow outcome — 0 if the assessment ran, 1 if the scan failed. + * That tracks whether the pipeline ran, not whether vulnerabilities were found. On failure the + * root-cause message is printed so a red CI build says why. */ async function followScan(workspace: string, workspacesDir: string): Promise { const logFile = resolveRunFile(path.join(workspacesDir, workspace), 'workflow.log'); + const workflowId = resolveWorkflowId(workspace); // The worker creates workflow.log as it starts; wait briefly so the first read doesn't // mistake a not-yet-created file for an already-finished scan. @@ -253,18 +259,38 @@ async function followScan(workspace: string, workspacesDir: string): Promise { + temporalUnreachable = true; + }, + }); + + // The tail already printed the diagnostic; reading the outcome would only fail the same way. + if (temporalUnreachable) { + process.exit(1); + } + if (!workflowId) { fail('Scan finished but its workflow id could not be resolved from session.json.'); } try { const outcome = await getTerminalOutcome(workflowId); - process.exit(outcome.kind === 'success' ? 0 : 1); - } catch { - fail('Could not reach Temporal at 127.0.0.1:7233 to read the scan outcome.'); + if (outcome.kind === 'failed') { + // Print the reason only when the streamed log didn't already show the worker's failure + // summary — otherwise the worker crashed before writing it, and this is the only report. + if (!sawFailure) { + console.error(`\nScan failed:\n${indentFailureSegments(outcome.message)}`); + } + process.exit(1); + } + process.exit(0); + } catch (err) { + const detail = err instanceof Error ? err.message : String(err); + fail('Could not read the scan outcome from Temporal at 127.0.0.1:7233.', ` ${detail}`); } } diff --git a/apps/cli/src/index.ts b/apps/cli/src/index.ts index 640dded..20d5e81 100644 --- a/apps/cli/src/index.ts +++ b/apps/cli/src/index.ts @@ -23,6 +23,7 @@ import { availableCommands, isHelpableCommand, printCommandHelp, START_OPTIONS } import { commandPrefix, getMode, isLocal, type Mode } from './mode.js'; import { displaySplash } from './splash.js'; import { closestMatch } from './suggest.js'; +import { stdoutIsTerminal } from './tty.js'; import { getVersion, getVersionLine } from './version.js'; function blockSudo(): void { @@ -174,7 +175,7 @@ async function main(): Promise { printCommandHelp(topic); } else { const bare = command === undefined; - if (bare) displaySplash(isLocal() ? undefined : getVersion()); + if (bare && stdoutIsTerminal()) displaySplash(isLocal() ? undefined : getVersion()); showHelp(bare); } return; diff --git a/apps/cli/src/scan/failure.ts b/apps/cli/src/scan/failure.ts new file mode 100644 index 0000000..2bf4563 --- /dev/null +++ b/apps/cli/src/scan/failure.ts @@ -0,0 +1,31 @@ +/** + * Rendering for the worker's '|'-delimited failure string. + * + * `formatWorkflowError` in the worker joins error segments — phase context, error type, + * message, and remediation hint — with '|' as a delimiter. These helpers turn that raw + * string into readable output for the CLI's own surfaces. + */ + +/** + * Split the failure string into trimmed, non-empty lines. Segments are delimited by '|', and a + * segment's own embedded newlines (e.g. a multi-line validation message) become their own lines so + * each aligns with the rest of the block. + */ +export function parseFailureSegments(message: string): string[] { + return message + .split(/[|\n]/) + .map((segment) => segment.trim()) + .filter((segment) => segment.length > 0); +} + +/** Multi-line block: one segment per indented line (the caller prints the header). */ +export function indentFailureSegments(message: string, indent = ' '): string { + return parseFailureSegments(message) + .map((segment) => `${indent}${segment}`) + .join('\n'); +} + +/** Single-line summary for compact contexts like the status footer. */ +export function inlineFailureReason(message: string): string { + return parseFailureSegments(message).join(' — '); +} diff --git a/apps/cli/src/scan/render.ts b/apps/cli/src/scan/render.ts index 605b874..7e3ac62 100644 --- a/apps/cli/src/scan/render.ts +++ b/apps/cli/src/scan/render.ts @@ -11,6 +11,7 @@ import { BOLD, DIM, GOLD, paint, RED, YELLOW } from '../colors.js'; import { commandPrefix } from '../mode.js'; import type { RunningAgent } from '../temporal-client.js'; import { agentError, deriveAgentStates, isTerminal, phaseGlyphState, type RunState, scanElapsedMs } from './derive.js'; +import { inlineFailureReason } from './failure.js'; import { PIPELINE, type PipelineState } from './pipeline.js'; export interface RenderInput { @@ -229,7 +230,8 @@ function footerLines(input: RenderInput, opts: RenderOptions): string[] { const temporalValue = temporalDashboardUrl(input.workflowId); if (isTerminal(input.temporalStatus)) { - const reason = input.failureMessage ?? input.state?.error ?? 'no result recorded'; + const rawReason = input.failureMessage ?? input.state?.error; + const reason = rawReason ? inlineFailureReason(rawReason) : 'no result recorded'; return [ footerDivider(opts), paint( diff --git a/apps/cli/src/temporal-client.ts b/apps/cli/src/temporal-client.ts index c7f3fdc..5dd4e8d 100644 --- a/apps/cli/src/temporal-client.ts +++ b/apps/cli/src/temporal-client.ts @@ -7,12 +7,17 @@ * publishes — so this needs Temporal up, but no worker of its own. */ +import { setTimeout as sleep } from 'node:timers/promises'; import { Client, Connection, WorkflowFailedError, WorkflowNotFoundError } from '@temporalio/client'; import { ACTIVITY_TO_AGENT, type PipelineState } from './scan/pipeline.js'; const ADDRESS = '127.0.0.1:7233'; const NAMESPACE = 'default'; +// WorkflowExecutionStatusName values that mean the scan has closed. RUNNING (and the unused +// CONTINUED_AS_NEW) are the only non-terminal states. +const TERMINAL_STATUSES: ReadonlySet = new Set(['COMPLETED', 'FAILED', 'CANCELLED', 'TERMINATED', 'TIMED_OUT']); + export interface RunningAgent { readonly agent: string; readonly attempt: number; @@ -111,6 +116,75 @@ function rootFailureMessage(err: WorkflowFailedError): string { return message; } +/** How a {@link waitForWorkflowClose} watch ended. */ +export type WatchEnd = { readonly reason: 'closed' } | { readonly reason: 'unreachable'; readonly lastError: string }; + +export interface WatchOptions { + /** Poll interval in ms (default 3000). */ + readonly pollMs?: number; + /** Consecutive connection failures before giving up (default 10 → ~30s at the default interval). */ + readonly maxConnectFailures?: number; + /** Consecutive connection failures before {@link onConnectionTrouble} fires once (default 3). */ + readonly warnAfterFailures?: number; + /** Abort the watch (the caller stopped for another reason, e.g. Ctrl-C). */ + readonly signal?: AbortSignal; + /** Called once when contact is first lost, so a live follower's log isn't silent during the outage. */ + readonly onConnectionTrouble?: (lastError: string) => void; + /** Called once when contact is regained after {@link onConnectionTrouble} fired. */ + readonly onReconnected?: () => void; +} + +/** + * Resolve once the scan is no longer running, using the workflow's Temporal status as the + * completion signal. Ends on a terminal status, a not-found workflow (closed past retention), or + * maxConnectFailures consecutive unreachable polls (a scan can't progress while its Temporal is + * down, so sustained no-contact is a safe stop). Never rejects; connection errors surface via the + * callbacks and the returned {@link WatchEnd}. + */ +export async function waitForWorkflowClose(workflowId: string, opts: WatchOptions = {}): Promise { + const pollMs = opts.pollMs ?? 3000; + const maxConnectFailures = opts.maxConnectFailures ?? 10; + const warnAfterFailures = opts.warnAfterFailures ?? 3; + const signal = opts.signal; + + let connectFailures = 0; + let lastError = ''; + let warned = false; + + while (!signal?.aborted) { + try { + const desc = await describeScan(workflowId); + if (desc === null || TERMINAL_STATUSES.has(desc.status)) { + return { reason: 'closed' }; + } + // Reachable and still RUNNING — reset the failure streak and note any recovery. + if (warned) { + warned = false; + opts.onReconnected?.(); + } + connectFailures = 0; + } catch (err) { + connectFailures++; + lastError = err instanceof Error ? err.message : String(err); + if (!warned && connectFailures >= warnAfterFailures) { + warned = true; + opts.onConnectionTrouble?.(lastError); + } + if (connectFailures >= maxConnectFailures) { + return { reason: 'unreachable', lastError }; + } + } + + try { + await sleep(pollMs, undefined, { signal }); + } catch { + break; // Aborted mid-wait by the caller. + } + } + + return { reason: 'closed' }; +} + /** Final state of a closed scan: success carries the full PipelineState, failure carries the message. */ export async function getTerminalOutcome(workflowId: string): Promise { const client = await getClient(); diff --git a/apps/worker/src/audit/workflow-logger.ts b/apps/worker/src/audit/workflow-logger.ts index 8437675..eb5a9b6 100644 --- a/apps/worker/src/audit/workflow-logger.ts +++ b/apps/worker/src/audit/workflow-logger.ts @@ -303,13 +303,17 @@ export class WorkflowLogger { * Output: "Error: phase context\n ErrorType\n ..." */ private formatErrorBlock(errorString: string): string { - const segments = errorString.split('|'); const label = 'Error: '; const indent = ' '.repeat(label.length); - const lines = segments.map((segment, i) => (i === 0 ? `${label}${segment.trim()}` : `${indent}${segment.trim()}`)); + // Segments are delimited by '|'; a segment's own embedded newlines (e.g. a multi-line + // validation message) become their own lines so each aligns under the label. + const lines = errorString + .split(/[|\n]/) + .map((segment) => segment.trim()) + .filter((segment) => segment.length > 0); - return `${lines.join('\n')}\n`; + return `${lines.map((line, i) => (i === 0 ? `${label}${line}` : `${indent}${line}`)).join('\n')}\n`; } /** @@ -336,17 +340,19 @@ export class WorkflowLogger { lines.push(this.formatErrorBlock(summary.error).trimEnd()); } - lines.push(''); - lines.push('Agent Breakdown:'); + if (summary.completedAgents.length > 0) { + lines.push(''); + lines.push('Agent Breakdown:'); - for (const agentName of summary.completedAgents) { - const metrics = summary.agentMetrics[agentName]; - if (metrics) { - const duration = formatDuration(metrics.durationMs); - const cost = metrics.costUsd !== null ? `$${metrics.costUsd.toFixed(4)}` : 'N/A'; - lines.push(` - ${agentName} (${duration}, ${cost})`); - } else { - lines.push(` - ${agentName}`); + for (const agentName of summary.completedAgents) { + const metrics = summary.agentMetrics[agentName]; + if (metrics) { + const duration = formatDuration(metrics.durationMs); + const cost = metrics.costUsd !== null ? `$${metrics.costUsd.toFixed(4)}` : 'N/A'; + lines.push(` - ${agentName} (${duration}, ${cost})`); + } else { + lines.push(` - ${agentName}`); + } } } diff --git a/apps/worker/src/temporal/pipeline.ts b/apps/worker/src/temporal/pipeline.ts index 5e17560..911ffc3 100644 --- a/apps/worker/src/temporal/pipeline.ts +++ b/apps/worker/src/temporal/pipeline.ts @@ -14,5 +14,4 @@ export type { ResumeState, VulnExploitPipelineResult, } from './shared.js'; -export { PipelineExecutionError } from './shared.js'; export { pentestPipeline } from './workflows.js'; diff --git a/apps/worker/src/temporal/shared.ts b/apps/worker/src/temporal/shared.ts index 355783d..2004c05 100644 --- a/apps/worker/src/temporal/shared.ts +++ b/apps/worker/src/temporal/shared.ts @@ -60,21 +60,6 @@ export interface PipelineState { summary: PipelineSummary | null; } -/** - * Thrown by pentestPipeline() when the run fails, carrying the fully-populated - * PipelineState (real agentMetrics, completedAgents, summary) so a consumer can - * report actual spend instead of synthesizing a zeroed failed state. `cause` - * preserves the original error for classification and Temporal failure reporting. - */ -export class PipelineExecutionError extends Error { - override name = 'PipelineExecutionError' as const; - readonly state: PipelineState; - constructor(message: string, state: PipelineState, options?: { cause?: unknown }) { - super(message, options); - this.state = state; - } -} - // Extended state returned by getProgress query (includes computed fields) export interface PipelineProgress extends PipelineState { workflowId: string; diff --git a/apps/worker/src/temporal/workflows.ts b/apps/worker/src/temporal/workflows.ts index b04e83a..124e6b9 100644 --- a/apps/worker/src/temporal/workflows.ts +++ b/apps/worker/src/temporal/workflows.ts @@ -41,7 +41,6 @@ import type { ActivityInput } from './activities.js'; import { type AgentMetrics, getProgress, - PipelineExecutionError, type PipelineInput, type PipelineProgress, type PipelineState, @@ -718,9 +717,10 @@ export async function pentestPipeline(input: PipelineInput): Promise