mirror of
https://github.com/KeygraphHQ/shannon.git
synced 2026-08-26 13:12:40 +02:00
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
This commit is contained in:
@@ -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<void> {
|
||||
export function tailUntilComplete(logFile: string, opts: TailOptions = {}): Promise<TailResult> {
|
||||
return new Promise((resolve) => {
|
||||
let position = 0;
|
||||
let done = false;
|
||||
let sawFailure = false;
|
||||
const controller = new AbortController();
|
||||
let watcher: ReturnType<typeof watch> | 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));
|
||||
}
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<never> {
|
||||
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<nev
|
||||
if (stdoutIsTerminal()) {
|
||||
console.error('\n Following scan log (Ctrl-C to stop watching):\n');
|
||||
}
|
||||
await tailUntilComplete(logFile);
|
||||
|
||||
const workflowId = resolveWorkflowId(workspace);
|
||||
let temporalUnreachable = false;
|
||||
const { sawFailure } = await tailUntilComplete(logFile, {
|
||||
...(workflowId && { workflowId }),
|
||||
onUnreachable: () => {
|
||||
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}`);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
printCommandHelp(topic);
|
||||
} else {
|
||||
const bare = command === undefined;
|
||||
if (bare) displaySplash(isLocal() ? undefined : getVersion());
|
||||
if (bare && stdoutIsTerminal()) displaySplash(isLocal() ? undefined : getVersion());
|
||||
showHelp(bare);
|
||||
}
|
||||
return;
|
||||
|
||||
@@ -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(' — ');
|
||||
}
|
||||
@@ -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(
|
||||
|
||||
@@ -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<string> = 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<WatchEnd> {
|
||||
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<TerminalOutcome> {
|
||||
const client = await getClient();
|
||||
|
||||
Reference in New Issue
Block a user