diff --git a/CLAUDE.md b/CLAUDE.md index 80e366a9..45f268f1 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -99,13 +99,13 @@ apps/worker/ — @shannon/worker (private, Temporal worker + pipeline logic) ``` ### CLI Package (`apps/cli/`) -Published as `@keygraph/shannon` on npm. Contains Docker orchestration logic plus a read-only `@temporalio/client` reader (for `status`); no worker/pipeline business logic or prompts. Bundled with tsdown for single-file ESM output (deps stay external). +Published as `@keygraph/shannon` on npm. Contains Docker orchestration and a direct `@temporalio/client` integration for read-only status plus bounded workflow lifecycle operations; no worker/pipeline business logic or prompts. Bundled with tsdown for single-file ESM output (deps stay external). - `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/temporal-client.ts` — `@temporalio/client` integration: connects to the frontend on `127.0.0.1:7233` (published by compose), provides `describeScan` (status + `pendingActivities` → running agents), `queryProgress` (live `getProgress` query → `PipelineState`), `getTerminalOutcome` (workflow `result()`), and bounded lifecycle RPCs for `stop`. `stop` requests cancellation first, waits up to 10 seconds, then requests termination only when necessary and verifies closure within a bounded window. 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), `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/docker.ts` — Compose lifecycle, image pull/build, and ephemeral `docker run` worker spawning. Each worker carries workspace, task-queue, and preselected workflow-ID labels so stop can correlate the local worker with its Temporal execution before `session.json` exists. Before `docker run`, start fsyncs that exact candidate under the workspace's hidden internals and clears it only when `session.json` registers the same ID; stop reconciles any candidate left by an interrupted launch. Start also checks the image's workflow-ID protocol label and refuses a stale worker that would ignore the preselected ID - `apps/cli/src/home.ts` — State directory management (`~/.shannon/` for npx, `./` for local) - `apps/cli/src/env.ts` — `.env` loading, TOML fallback (npx only) via `apps/cli/src/config/resolver.ts`, credential validation, provider-scoped env flag building - `apps/cli/src/model-spec.ts` — `SHANNON_AI_MODEL` (`:`) parsing; mirrors `apps/worker/src/ai/models.ts` @@ -121,7 +121,7 @@ Published as `@keygraph/shannon` on npm. Contains Docker orchestration logic plu - `shannon` — Node.js entry point (`#!/usr/bin/env node`) that delegates to `apps/cli/dist/index.mjs` ### Docker Architecture -Infra (Temporal) runs via `docker-compose.yml`. Workers are ephemeral `docker run --rm` containers, one per scan, each with a unique task queue and isolated volume mounts. +Infra (Temporal) runs via `docker-compose.yml`. Workers are ephemeral `docker run --rm` containers, one per scan, each with a unique task queue, preselected workflow ID, matching identity labels, and isolated volume mounts. `shannon stop --all` takes the union of labeled running workers and Temporal-running workflows, so an orphaned workflow is still stopped after its worker has disappeared. - `docker-compose.yml` — Infra only: `shannon-temporal` (port 7233/8233). Network: `shannon-net` - `Dockerfile` — 2-stage build (builder + Chainguard Wolfi runtime). Uses pnpm. Entrypoint: `CMD ["node", "apps/worker/dist/temporal/worker.js"]` diff --git a/Dockerfile b/Dockerfile index 6f289a49..de062557 100644 --- a/Dockerfile +++ b/Dockerfile @@ -43,6 +43,9 @@ RUN rm -rf node_modules apps/*/node_modules && pnpm install --frozen-lockfile -- # Runtime stage - Minimal production image FROM cgr.dev/chainguard/wolfi-base:latest AS runtime +# Lifecycle protocol consumed by the CLI before it trusts a container workflow-id label. +LABEL shannon.worker-protocol="workflow-id-v1" + # Install only runtime dependencies USER root RUN apk update && apk add --no-cache \ diff --git a/apps/cli/src/commands/start.ts b/apps/cli/src/commands/start.ts index 5fe5934b..ecc4b313 100644 --- a/apps/cli/src/commands/start.ts +++ b/apps/cli/src/commands/start.ts @@ -12,7 +12,7 @@ import { setTimeout as sleep } from 'node:timers/promises'; import * as p from '@clack/prompts'; import { ensureDocker, ensureImage, ensureInfra, randomSuffix, spawnWorker } from '../docker.js'; import { buildEnvFlags, loadEnv, resolveHostPiAuthPath, shouldUsePiAuth, validateCredentials } from '../env.js'; -import { fail } from '../errors.js'; +import { fail, warn } from '../errors.js'; import { getWorkspacesDir, initHome } from '../home.js'; import { commandPrefix, isLocal } from '../mode.js'; import { resolveModelSpec } from '../model-spec.js'; @@ -25,6 +25,7 @@ import { resolveRepo, resolveRunFile, } from '../paths.js'; +import { clearPendingWorkflowIdentity, writePendingWorkflowIdentity } from '../pending-workflow.js'; import { indentFailureSegments } from '../scan/failure.js'; import { resolveWorkflowId } from '../session.js'; import { displayPlainBanner, displaySplash } from '../splash.js'; @@ -214,6 +215,12 @@ export function writeLaunchStateAtomically(internalPath: string, outputDir: stri } } +/** Select the workflow ID before Docker starts so the container can carry it as immutable identity. */ +export function createWorkflowId(workspace: string, isResume: boolean, timestamp: number = Date.now()): string { + if (isResume) return `${workspace}_resume_${timestamp}`; + return /_shannon-\d+$/.test(workspace) ? workspace : `${workspace}_shannon-${timestamp}`; +} + export async function start(args: StartArgs): Promise { // 1. Resolve non-mutating inputs and classify the workspace before changing it. initHome(); @@ -250,6 +257,7 @@ export async function start(args: StartArgs): Promise { const suffix = randomSuffix(); const taskQueue = `shannon-${suffix}`; const containerName = `shannon-worker-${suffix}`; + const workflowId = createWorkflowId(workspace, launchDecision.isResume); // 4. Create writable overlay directories after resume validation has succeeded. // The run dir and its INTERNAL_DIR must be 0o777 so the container user can create audit @@ -294,13 +302,23 @@ export async function start(args: StartArgs): Promise { initialResumeCount = Array.isArray(attempts) ? attempts.length : 0; } - // 8. Spawn the worker container. + // 8. Persist the exact launch candidate before Docker can start the worker. Session + // registration later replaces this bridge as the durable workflow identity. + try { + writePendingWorkflowIdentity(workspacePath, workflowId, taskQueue); + } catch { + spinner.error('Could not record the scan workflow identity'); + process.exit(1); + } + + // 9. Spawn the worker container. const proc = spawnWorker({ version: args.version, url: args.url, repo, workspacesDir, taskQueue, + workflowId, containerName, envFlags: buildEnvFlags(), ...(config && { config }), @@ -366,10 +384,17 @@ export async function start(args: StartArgs): Promise { const resumeAttempts: { workflowId: string }[] = session.session?.resumeAttempts ?? []; // Fresh: session.json appears with originalWorkflowId. Resume: new resumeAttempts entry. - const ready = isResume ? resumeAttempts.length > initialResumeCount : !!session.session?.originalWorkflowId; + const ready = isResume + ? resumeAttempts.slice(initialResumeCount).some((attempt) => attempt.workflowId === workflowId) + : session.session?.originalWorkflowId === workflowId; if (ready) { started = true; + try { + clearPendingWorkflowIdentity(workspacePath, taskQueue); + } catch { + warn(`Scan ${workspace} started, but its launch record could not be removed.`); + } spinner.stop(`Scan started — ${workspace}`); printInfo(args, workspace, repo.hostPath, workspacesDir); if (args.follow) { diff --git a/apps/cli/src/commands/stop.ts b/apps/cli/src/commands/stop.ts index f754a817..3ea2353e 100644 --- a/apps/cli/src/commands/stop.ts +++ b/apps/cli/src/commands/stop.ts @@ -1,5 +1,5 @@ /** - * `shannon stop` command — stop one scan by workspace, or every scan with --all. + * `shannon stop` command: stop one scan by workspace, or every scan with --all. * Never touches infra or data; to wipe Temporal state entirely, use `shannon reset`. */ @@ -7,24 +7,36 @@ import path from 'node:path'; import * as p from '@clack/prompts'; import { confirmOrExit } from '../confirm.js'; import { - anyRunningScanWorkflow, - cancelWorkflow, + type CommandQueryResult, ensureDocker, - isTemporalReady, - isWorkflowRunning, - runningContainers, - runningScanWorkspaces, + type RunningScanContainer, + runningContainersChecked, + runningScanContainersChecked, scanFilter, stopContainers, - terminateWorkflow, WORKER_FILTER, + WORKFLOW_ID_PROTOCOL, } from '../docker.js'; import { fail, failUsage, warn } from '../errors.js'; import { getWorkspacesDir } from '../home.js'; import { commandPrefix } from '../mode.js'; import { resolveRunFile } from '../paths.js'; +import { + clearPendingWorkflowIdentity, + type PendingWorkflowIdentity, + readPendingWorkflowIdentities, +} from '../pending-workflow.js'; import { resolveWorkflowId } from '../session.js'; -import { resolveDefaultWorkspace } from '../workspaces.js'; +import { + describeWorkflowLifecycle, + listRunningScanWorkflows, + type RunningScanWorkflow, + refreshWorkflowLifecycleConnection, + requestWorkflowCancellation, + requestWorkflowTermination, + type WorkflowLifecycleState, +} from '../temporal-client.js'; +import { listWorkspaces, resolveScanIdentity } from '../workspaces.js'; import { appendCancellationFallback } from './logs.js'; export interface StopOptions { @@ -35,187 +47,837 @@ export interface StopOptions { const CANCELLATION_GRACE_MS = 10_000; const CANCELLATION_POLL_MS = 250; +const TERMINATION_VERIFY_MS = 5_000; +const TERMINATION_ATTEMPTS = 2; +const TERMINATION_REASON = 'Stopped after cancellation grace period'; +const CANDIDATE_REGISTRATION_SETTLE_MS = 3_000; +const VISIBILITY_SETTLE_MS = 1_000; +const VISIBILITY_MAX_SETTLE_MS = 5_000; -export interface StopTarget { - readonly workspace: string; - readonly workflowId?: string; - readonly workflowRunning: boolean; -} +export type WorkflowStopOutcome = + | { readonly kind: 'graceful' } + | { readonly kind: 'forced' } + | { readonly kind: 'already-closed' } + | { readonly kind: 'unverified' }; + +export type ContainerStopOutcome = + | { readonly kind: 'stopped'; readonly hadContainers: boolean } + | { readonly kind: 'still-running'; readonly remaining: number } + | { readonly kind: 'unverified' }; export interface StopLifecycle { - readonly cancel: (workflowId: string) => boolean; - readonly isRunning: (workflowId: string) => boolean; - readonly terminate: (workflowId: string) => boolean; - readonly containers: (workspace: string) => string[]; - readonly stopContainers: (ids: string[]) => Promise; + readonly cancel: (workflowId: string) => Promise<'requested' | 'not-found'>; + readonly describe: (workflowId: string) => Promise; + readonly refresh: () => Promise; + readonly terminate: (workflowId: string) => Promise<'requested' | 'not-found'>; + readonly containers: (filter: readonly string[]) => CommandQueryResult; + readonly stopContainers: (ids: readonly string[]) => Promise; readonly appendFallback: (workspace: string) => void; readonly wait: (milliseconds: number) => Promise; + readonly now: () => number; +} + +export interface WorkflowStopTarget { + readonly workflowId: string; + readonly workspace?: string; + readonly containerCandidate: boolean; + /** Safe to synthesize a log marker when this CLI-owned launch never reached session registration. */ + readonly preRegistrationFallback?: boolean; +} + +export interface WorkflowStopResult { + readonly target: WorkflowStopTarget; + readonly outcome: WorkflowStopOutcome; +} + +export interface StopExecutionResult { + readonly workflows: readonly WorkflowStopResult[]; + readonly containers: ContainerStopOutcome; + readonly preRegistrationWorkspaces: readonly string[]; +} + +export interface WorkflowTargetPlan { + readonly targets: readonly WorkflowStopTarget[]; + readonly containersWithoutVerifiedWorkflowId: readonly string[]; +} + +interface PendingWorkflowReference { + readonly workspace: string; + readonly identity: PendingWorkflowIdentity; +} + +interface PendingWorkflowTargets { + readonly byWorkspace: ReadonlyMap; + readonly references: readonly PendingWorkflowReference[]; + readonly unreadableCount: number; } const stopLifecycle: StopLifecycle = { - cancel: cancelWorkflow, - isRunning: isWorkflowRunning, - terminate: (workflowId) => terminateWorkflow(workflowId, 'Stopped after cancellation grace period'), - containers: (workspace) => runningContainers(scanFilter(workspace)), - stopContainers, + cancel: requestWorkflowCancellation, + describe: describeWorkflowLifecycle, + refresh: refreshWorkflowLifecycleConnection, + terminate: (workflowId) => requestWorkflowTermination(workflowId, TERMINATION_REASON), + containers: runningContainersChecked, + stopContainers: (ids) => stopContainers([...ids]), appendFallback: (workspace) => { const logFile = resolveRunFile(path.join(getWorkspacesDir(), workspace), 'workflow.log'); appendCancellationFallback(logFile); }, wait: (milliseconds) => new Promise((resolve) => setTimeout(resolve, milliseconds)), + now: Date.now, }; -/** Cancel first; terminate and write the fallback heading only when graceful closure misses its deadline. */ -export async function stopTargetCancelFirst( - target: StopTarget, - lifecycle: StopLifecycle = stopLifecycle, - graceMs: number = CANCELLATION_GRACE_MS, - pollMs: number = CANCELLATION_POLL_MS, -): Promise<'graceful' | 'forced'> { - let forced = !target.workflowRunning || target.workflowId === undefined; - if (!forced && target.workflowId !== undefined) { - lifecycle.cancel(target.workflowId); - const deadline = Date.now() + graceMs; - while (lifecycle.isRunning(target.workflowId) && Date.now() < deadline) { - await lifecycle.wait(pollMs); - } - forced = lifecycle.isRunning(target.workflowId); - if (forced) lifecycle.terminate(target.workflowId); +function readPendingTargets(workspaces: readonly string[]): PendingWorkflowTargets { + const byWorkspace = new Map(); + const references: PendingWorkflowReference[] = []; + let unreadableCount = 0; + + for (const workspace of new Set(workspaces)) { + const workspacePath = path.join(getWorkspacesDir(), workspace); + const result = readPendingWorkflowIdentities(workspacePath); + unreadableCount += result.unreadableCount; + if (result.identities.length === 0) continue; + byWorkspace.set(workspace, result.identities); + for (const identity of result.identities) references.push({ workspace, identity }); } - await lifecycle.stopContainers(lifecycle.containers(target.workspace)); - if (forced && lifecycle.containers(target.workspace).length === 0) { - lifecycle.appendFallback(target.workspace); - } - return forced ? 'forced' : 'graceful'; + return { byWorkspace, references, unreadableCount }; } -/** Apply the same captured-target lifecycle concurrently for `stop --all`. */ -export function stopTargetsCancelFirst( - targets: readonly StopTarget[], - lifecycle: StopLifecycle = stopLifecycle, - graceMs: number = CANCELLATION_GRACE_MS, - pollMs: number = CANCELLATION_POLL_MS, -): Promise { - return Promise.all(targets.map((target) => stopTargetCancelFirst(target, lifecycle, graceMs, pollMs))); +function clearPendingTargets(references: readonly PendingWorkflowReference[]): number { + let failures = 0; + for (const reference of references) { + try { + clearPendingWorkflowIdentity(path.join(getWorkspacesDir(), reference.workspace), reference.identity.task_queue); + } catch { + failures++; + } + } + return failures; +} + +function workflowClosed(state: WorkflowLifecycleState): boolean { + return state.kind === 'terminal' || state.kind === 'not-found'; +} + +/** Poll direct workflow state until Temporal positively confirms closure or the deadline expires. */ +async function waitForWorkflowClosure( + workflowId: string, + lifecycle: StopLifecycle, + deadline: number, + pollMs: number, +): Promise { + while (true) { + if (deadline - lifecycle.now() <= 0) return false; + try { + if (workflowClosed(await lifecycle.describe(workflowId))) return true; + } catch { + // An unavailable status is unknown, never evidence that the workflow closed. + } + + const remaining = deadline - lifecycle.now(); + if (remaining <= 0) return false; + await lifecycle.wait(Math.min(pollMs, remaining)); + } } /** - * Stop a single scan. Cooperative cancellation gets the first ten seconds so the - * workflow can flush its terminal log; termination and a host-written heading are - * the fallback for the pre-registration window or an unavailable finalizer. + * Request cooperative cancellation, then make at most two termination attempts when the + * workflow does not close during its grace period. Every success is backed by direct state. */ +export async function stopWorkflowCancelFirst( + workflowId: string, + lifecycle: StopLifecycle = stopLifecycle, + graceMs: number = CANCELLATION_GRACE_MS, + pollMs: number = CANCELLATION_POLL_MS, + verifyMs: number = TERMINATION_VERIFY_MS, +): Promise { + try { + if ((await lifecycle.cancel(workflowId)) === 'not-found') return { kind: 'already-closed' }; + } catch { + // The request may have reached Temporal even when its acknowledgement was lost. + } + + if (await waitForWorkflowClosure(workflowId, lifecycle, lifecycle.now() + graceMs, pollMs)) { + return { kind: 'graceful' }; + } + + const verifyPerAttemptMs = Math.max(pollMs, Math.ceil(verifyMs / TERMINATION_ATTEMPTS)); + for (let attempt = 0; attempt < TERMINATION_ATTEMPTS; attempt++) { + if (attempt > 0) { + try { + await lifecycle.refresh(); + } catch { + // The termination call below makes one final bounded connection attempt. + } + } + + try { + if ((await lifecycle.terminate(workflowId)) === 'not-found') return { kind: 'already-closed' }; + } catch { + // A lost acknowledgement is resolved by the direct verification below. + } + if (await waitForWorkflowClosure(workflowId, lifecycle, lifecycle.now() + verifyPerAttemptMs, pollMs)) { + return { kind: 'forced' }; + } + } + + return { kind: 'unverified' }; +} + +/** Apply the same bounded lifecycle concurrently to a captured set of workflow IDs. */ +export async function stopWorkflowsCancelFirst( + workflowIds: readonly string[], + lifecycle: StopLifecycle = stopLifecycle, + graceMs: number = CANCELLATION_GRACE_MS, + pollMs: number = CANCELLATION_POLL_MS, + verifyMs: number = TERMINATION_VERIFY_MS, +): Promise { + const settlements = await Promise.allSettled( + workflowIds.map((workflowId) => stopWorkflowCancelFirst(workflowId, lifecycle, graceMs, pollMs, verifyMs)), + ); + return settlements.map((settlement) => + settlement.status === 'fulfilled' ? settlement.value : { kind: 'unverified' }, + ); +} + +/** Stop exactly the captured workers, then fail closed if any matching worker remains or appears. */ +export async function stopContainersAndVerify( + initialIds: readonly string[], + filter: readonly string[], + lifecycle: StopLifecycle = stopLifecycle, +): Promise { + try { + await lifecycle.stopContainers(initialIds); + } catch { + // The post-stop query below decides whether the operation actually succeeded. + } + + let after: CommandQueryResult; + try { + after = lifecycle.containers(filter); + } catch { + return { kind: 'unverified' }; + } + if (after.kind === 'unavailable') return { kind: 'unverified' }; + if (after.value.length > 0) return { kind: 'still-running', remaining: after.value.length }; + return { kind: 'stopped', hadContainers: initialIds.length > 0 }; +} + +function addWorkflowTarget(targets: Map, candidate: WorkflowStopTarget): void { + const current = targets.get(candidate.workflowId); + if (current === undefined) { + targets.set(candidate.workflowId, candidate); + return; + } + const workspace = current.workspace ?? candidate.workspace; + targets.set(candidate.workflowId, { + workflowId: candidate.workflowId, + ...(workspace !== undefined && { workspace }), + containerCandidate: current.containerCandidate || candidate.containerCandidate, + ...((current.preRegistrationFallback === true || candidate.preRegistrationFallback === true) && { + preRegistrationFallback: true, + }), + }); +} + +/** + * Build the stop union from immutable container candidates, recorded session IDs, + * pre-registration launch records, and Temporal visibility. Visibility supplies positive + * targets but never proves absence. + */ +function verifiedContainerWorkflowId( + container: RunningScanContainer, + visibleWorkflows: readonly RunningScanWorkflow[], +): string | undefined { + if (container.workerProtocol === WORKFLOW_ID_PROTOCOL && container.workflowId !== undefined) { + return container.workflowId; + } + if (container.taskQueue === undefined) return undefined; + const matches = visibleWorkflows.filter((workflow) => workflow.taskQueue === container.taskQueue); + return matches.length === 1 ? matches[0]?.workflowId : undefined; +} + +export function buildWorkflowTargetPlan( + containers: readonly RunningScanContainer[], + recordedByWorkspace: ReadonlyMap, + visibleWorkflows: readonly RunningScanWorkflow[], + pendingByWorkspace: ReadonlyMap = new Map(), +): WorkflowTargetPlan { + const targets = new Map(); + const containersWithoutVerifiedWorkflowId: string[] = []; + + for (const container of containers) { + const verifiedWorkflowId = verifiedContainerWorkflowId(container, visibleWorkflows); + if (verifiedWorkflowId === undefined) containersWithoutVerifiedWorkflowId.push(container.id); + else { + addWorkflowTarget(targets, { + workflowId: verifiedWorkflowId, + ...(container.workspace !== undefined && { workspace: container.workspace }), + containerCandidate: true, + }); + } + } + + for (const [workspace, workflowId] of recordedByWorkspace) { + addWorkflowTarget(targets, { + workflowId, + workspace, + containerCandidate: + containers.some((container) => verifiedContainerWorkflowId(container, visibleWorkflows) === workflowId) || + pendingByWorkspace.get(workspace)?.some((identity) => identity.workflow_id === workflowId) === true, + }); + } + + for (const [workspace, identities] of pendingByWorkspace) { + for (const identity of identities) { + addWorkflowTarget(targets, { + workflowId: identity.workflow_id, + workspace, + containerCandidate: true, + ...(!recordedByWorkspace.has(workspace) && { preRegistrationFallback: true }), + }); + } + } + + for (const workflow of visibleWorkflows) { + const matchingWorkspaces = new Set( + containers + .filter((container) => container.taskQueue === workflow.taskQueue && container.workspace !== undefined) + .flatMap((container) => container.workspace ?? []), + ); + for (const [workspace, identities] of pendingByWorkspace) { + if (identities.some((identity) => identity.task_queue === workflow.taskQueue)) matchingWorkspaces.add(workspace); + } + const workspace = matchingWorkspaces.size === 1 ? [...matchingWorkspaces][0] : undefined; + addWorkflowTarget(targets, { + workflowId: workflow.workflowId, + ...(workspace !== undefined && { workspace }), + containerCandidate: + containers.some((container) => container.taskQueue === workflow.taskQueue) || + [...pendingByWorkspace.values()].some((identities) => + identities.some((identity) => identity.task_queue === workflow.taskQueue), + ), + }); + } + + return { targets: [...targets.values()], containersWithoutVerifiedWorkflowId }; +} + +/** + * Stop known workflows while their workers can finalize, stop the captured workers, then + * re-describe every container candidate. The last pass closes a NotFound-to-started race. + */ +export async function executeStopPlan( + targets: readonly WorkflowStopTarget[], + containers: readonly RunningScanContainer[], + filter: readonly string[], + lifecycle: StopLifecycle = stopLifecycle, + graceMs: number = CANCELLATION_GRACE_MS, + pollMs: number = CANCELLATION_POLL_MS, + verifyMs: number = TERMINATION_VERIFY_MS, + candidateSettleMs: number = CANDIDATE_REGISTRATION_SETTLE_MS, +): Promise { + const initialOutcomes = await stopWorkflowsCancelFirst( + targets.map((target) => target.workflowId), + lifecycle, + graceMs, + pollMs, + verifyMs, + ); + const outcomes = new Map(); + for (let index = 0; index < targets.length; index++) { + const target = targets[index]; + const outcome = initialOutcomes[index]; + if (target !== undefined && outcome !== undefined) outcomes.set(target.workflowId, outcome); + } + + const containerOutcome = await stopContainersAndVerify( + containers.map((container) => container.id), + filter, + lifecycle, + ); + const preRegistrationWorkspaces = new Set(); + + if (containerOutcome.kind === 'stopped') { + const candidates = targets.filter((target) => target.containerCandidate); + + for (const target of candidates) { + const initialOutcome = outcomes.get(target.workflowId) ?? { kind: 'unverified' }; + const settleDeadline = lifecycle.now() + candidateSettleMs; + let onlyObservedNotFound = initialOutcome.kind === 'already-closed' || initialOutcome.kind === 'unverified'; + + while (true) { + try { + const state = await lifecycle.describe(target.workflowId); + if (state.kind === 'open') { + onlyObservedNotFound = false; + const outcome = await stopWorkflowCancelFirst(target.workflowId, lifecycle, graceMs, pollMs, verifyMs); + outcomes.set(target.workflowId, outcome); + if (outcome.kind !== 'already-closed') break; + } + if (state.kind === 'terminal') { + onlyObservedNotFound = false; + if (initialOutcome.kind === 'unverified') outcomes.set(target.workflowId, { kind: 'already-closed' }); + } + if (state.kind === 'unknown') { + onlyObservedNotFound = false; + outcomes.set(target.workflowId, { kind: 'unverified' }); + break; + } + if (initialOutcome.kind === 'unverified') outcomes.set(target.workflowId, { kind: 'already-closed' }); + } catch { + onlyObservedNotFound = false; + outcomes.set(target.workflowId, { kind: 'unverified' }); + break; + } + + const remaining = settleDeadline - lifecycle.now(); + if (remaining <= 0) { + if (onlyObservedNotFound && target.workspace !== undefined && target.preRegistrationFallback === true) { + preRegistrationWorkspaces.add(target.workspace); + } + break; + } + try { + await lifecycle.wait(Math.min(pollMs, remaining)); + } catch { + outcomes.set(target.workflowId, { kind: 'unverified' }); + break; + } + } + } + } + + return { + workflows: targets.map((target) => ({ + target, + outcome: outcomes.get(target.workflowId) ?? { kind: 'unverified' }, + })), + containers: containerOutcome, + preRegistrationWorkspaces: [...preRegistrationWorkspaces], + }; +} + +function appendFallback(workspace: string, lifecycle: StopLifecycle = stopLifecycle): void { + try { + lifecycle.appendFallback(workspace); + } catch { + warn(`scan ${workspace} stopped, but workflow.log could not be marked cancelled.`); + } +} + +function reportContainerFailure(workspace: string | undefined, outcome: ContainerStopOutcome): void { + const target = workspace === undefined ? '--all' : workspace; + if (outcome.kind === 'still-running') console.error(`${outcome.remaining} scan worker(s) did not stop.`); + else console.error('Docker could not verify that every targeted scan worker stopped.'); + console.error(`Retry: ${commandPrefix()} stop ${target}`); +} + +function withRecordedWorkflows(containers: readonly RunningScanContainer[]): Map { + const recorded = new Map(); + for (const container of containers) { + if (container.workspace === undefined || recorded.has(container.workspace)) continue; + const workflowId = resolveWorkflowId(container.workspace); + if (workflowId !== undefined) recorded.set(container.workspace, workflowId); + } + return recorded; +} + +function resolveTargetWorkspaces(targets: readonly WorkflowStopTarget[]): readonly WorkflowStopTarget[] { + return targets.map((target) => { + if (target.workspace !== undefined) return target; + const identity = resolveScanIdentity(target.workflowId); + return identity.kind === 'ok' ? { ...target, workspace: identity.workspace } : target; + }); +} + +function unverifiedWorkflowCount(results: readonly WorkflowStopResult[]): number { + return results.filter((result) => result.outcome.kind === 'unverified').length; +} + +function appendVerifiedFallbacks(result: StopExecutionResult): void { + const workspaces = new Set(result.preRegistrationWorkspaces); + for (const workflow of result.workflows) { + if (workflow.outcome.kind === 'forced' && workflow.target.workspace !== undefined) { + workspaces.add(workflow.target.workspace); + } + } + for (const workspace of workspaces) appendFallback(workspace); +} + +function visibleWorkflowsForWorkspace( + workspace: string, + containers: readonly RunningScanContainer[], + pending: PendingWorkflowTargets, + visible: readonly RunningScanWorkflow[], +): readonly RunningScanWorkflow[] { + const taskQueues = new Set(containers.flatMap((container) => container.taskQueue ?? [])); + for (const identity of pending.byWorkspace.get(workspace) ?? []) taskQueues.add(identity.task_queue); + + return visible.filter((workflow) => { + if (taskQueues.has(workflow.taskQueue)) return true; + const identity = resolveScanIdentity(workflow.workflowId); + return identity.kind === 'ok' && identity.workspace === workspace; + }); +} + +/** Stop one scan while keeping its worker alive long enough to flush a graceful cancellation. */ async function stopSingleScan(workspace: string, yes: boolean): Promise { - const workflowId = resolveWorkflowId(workspace); const filter = scanFilter(workspace); - const temporalUp = isTemporalReady(); + const containerQuery = runningScanContainersChecked(filter); + if (containerQuery.kind === 'unavailable') { + fail(`Could not inspect the scan worker for ${workspace}.`, `Retry: ${commandPrefix()} stop ${workspace}`); + } + const containers = containerQuery.value.map((container) => ({ ...container, workspace })); + const recordedWorkflowId = resolveWorkflowId(workspace); + const recorded = new Map(); + if (recordedWorkflowId !== undefined) recorded.set(workspace, recordedWorkflowId); + const pending = readPendingTargets([workspace]); + const discovery = await discoverRunningWorkflows(); + const visible = + discovery.kind === 'ok' ? visibleWorkflowsForWorkspace(workspace, containers, pending, discovery.workflows) : []; + const plan = buildWorkflowTargetPlan(containers, recorded, visible, pending.byWorkspace); - const initialContainers = runningContainers(filter); - const workflowRunning = Boolean(workflowId && temporalUp && isWorkflowRunning(workflowId)); - - // Resolve what is running before prompting, so we never confirm a no-op. - if (initialContainers.length === 0 && !workflowRunning) { - if (!workflowId) { + if (containers.length === 0) { + if (plan.targets.length === 0) { + if (pending.unreadableCount > 0) { + fail( + `The launch records for ${workspace} could not be read safely.`, + `Retry: ${commandPrefix()} stop ${workspace}`, + ); + } + if (discovery.kind === 'unavailable') { + fail( + `Could not verify whether scan ${workspace} is still running in Temporal.`, + `Retry: ${commandPrefix()} stop ${workspace}`, + ); + } fail(`No scan found for workspace: ${workspace}`); } - console.log(`Nothing was running for ${workspace}.`); - return; + + const onlyRecordedTarget = + recordedWorkflowId !== undefined && + pending.references.length === 0 && + pending.unreadableCount === 0 && + discovery.kind === 'ok' && + plan.targets.every((target) => target.workflowId === recordedWorkflowId); + if (onlyRecordedTarget) { + try { + const state = await describeWorkflowLifecycle(recordedWorkflowId); + if (state.kind === 'terminal' || state.kind === 'not-found') { + console.log(`Nothing was running for ${workspace}.`); + return; + } + if (state.kind === 'unknown') { + fail( + `Temporal returned an unknown lifecycle state for ${workspace}.`, + `Retry: ${commandPrefix()} stop ${workspace}`, + ); + } + } catch { + fail( + `Could not verify whether scan ${workspace} is still running in Temporal.`, + `Retry: ${commandPrefix()} stop ${workspace}`, + ); + } + } } await confirmOrExit('stop', `Stop the scan "${workspace}"?`, yes); - const spinner = p.spinner(); spinner.start(`Stopping scan ${workspace}`); - await stopTargetCancelFirst({ workspace, ...(workflowId !== undefined && { workflowId }), workflowRunning }); + const initialResult = await executeStopPlan(plan.targets, containers, filter); + const visibilitySettle = await stopVisibleWorkflowsUntilSettled( + initialResult.workflows, + stopLifecycle, + VISIBILITY_SETTLE_MS, + VISIBILITY_MAX_SETTLE_MS, + async () => { + const current = await discoverRunningWorkflows(); + return current.kind === 'ok' + ? { + kind: 'ok', + workflows: visibleWorkflowsForWorkspace(workspace, containers, pending, current.workflows), + } + : current; + }, + ); + const result: StopExecutionResult = { + workflows: visibilitySettle.results, + containers: initialResult.containers, + preRegistrationWorkspaces: initialResult.preRegistrationWorkspaces, + }; + const unverified = unverifiedWorkflowCount(result.workflows); + const finalPending = readPendingTargets([workspace]); + const initialPendingKeys = new Set( + pending.references.map((reference) => `${reference.identity.task_queue}\0${reference.identity.workflow_id}`), + ); + const newPendingCount = finalPending.references.filter( + (reference) => !initialPendingKeys.has(`${reference.identity.task_queue}\0${reference.identity.workflow_id}`), + ).length; + const unreadablePendingCount = Math.max(pending.unreadableCount, finalPending.unreadableCount); + const incomplete = + result.containers.kind !== 'stopped' || + plan.containersWithoutVerifiedWorkflowId.length > 0 || + discovery.kind === 'unavailable' || + visibilitySettle.kind !== 'settled' || + unreadablePendingCount > 0 || + newPendingCount > 0 || + unverified > 0; - const stillRunning = runningContainers(filter); - if (stillRunning.length > 0) { - spinner.error(`Scan ${workspace} may still be running`); - console.error(`${stillRunning.length} container(s) did not stop. Retry: ${commandPrefix()} stop ${workspace}`); + if (incomplete) { + spinner.error(`Scan ${workspace} shutdown could not be fully verified`); + if (result.containers.kind !== 'stopped') reportContainerFailure(workspace, result.containers); + if (unverified > 0) console.error(`Temporal could not confirm closure for ${unverified} workflow(s).`); + if (discovery.kind === 'unavailable') console.error('Temporal could not enumerate every running scan workflow.'); + if (visibilitySettle.kind === 'unavailable') { + console.error('Temporal could not complete the final scan workflow check.'); + } + if (visibilitySettle.kind === 'timed-out') { + console.error('Temporal workflow discovery did not settle before its deadline.'); + } + if (unreadablePendingCount > 0) { + console.error(`${unreadablePendingCount} launch record(s) could not be read safely.`); + } + if (newPendingCount > 0) console.error('A new scan launch began while shutdown was running.'); + if (plan.containersWithoutVerifiedWorkflowId.length > 0) { + console.error('A legacy scan worker could not prove its candidate workflow ID.'); + } + console.error(`Retry: ${commandPrefix()} stop ${workspace}`); process.exit(1); } + appendVerifiedFallbacks(result); + const clearFailures = clearPendingTargets(pending.references); + if (clearFailures > 0) { + spinner.error(`Scan ${workspace} stopped, but its launch record could not be cleared`); + console.error(`Retry: ${commandPrefix()} stop ${workspace}`); + process.exit(1); + } spinner.stop(`Stopped scan ${workspace}`); +} - if (workflowId && temporalUp && isWorkflowRunning(workflowId)) { - warn(`scan ${workspace} stopped, but its workflow is still Running in Temporal.`); +export type WorkflowDiscoveryResult = + | { readonly kind: 'ok'; readonly workflows: readonly RunningScanWorkflow[] } + | { readonly kind: 'unavailable' }; + +async function discoverRunningWorkflows(): Promise { + try { + return { kind: 'ok', workflows: await listRunningScanWorkflows() }; + } catch { + return { kind: 'unavailable' }; + } +} + +interface VisibilitySettleResult { + readonly results: readonly WorkflowStopResult[]; + readonly kind: 'settled' | 'unavailable' | 'timed-out'; +} + +/** Re-enumerate visibility until no new open workflow appears during a bounded quiet horizon. */ +export async function stopVisibleWorkflowsUntilSettled( + seed: readonly WorkflowStopResult[], + lifecycle: StopLifecycle = stopLifecycle, + settleMs: number = VISIBILITY_SETTLE_MS, + maxSettleMs: number = VISIBILITY_MAX_SETTLE_MS, + discover: () => Promise = discoverRunningWorkflows, +): Promise { + const results = new Map(seed.map((result) => [result.target.workflowId, result])); + const retriedUnverified = new Set(); + const recheckedAlreadyClosed = new Set(); + let quietSince = lifecycle.now(); + const maxDeadline = quietSince + maxSettleMs; + + while (true) { + const discovery = await discover(); + if (discovery.kind === 'unavailable') return { kind: 'unavailable', results: [...results.values()] }; + + const visibleTargets = resolveTargetWorkspaces( + buildWorkflowTargetPlan([], new Map(), discovery.workflows).targets, + ).map((target) => { + const existingWorkspace = results.get(target.workflowId)?.target.workspace; + return target.workspace === undefined && existingWorkspace !== undefined + ? { ...target, workspace: existingWorkspace } + : target; + }); + const residualTargets = visibleTargets.filter((target) => { + const current = results.get(target.workflowId); + if (current === undefined) return true; + if (current.outcome.kind === 'unverified') return !retriedUnverified.has(target.workflowId); + return current.outcome.kind === 'already-closed' && !recheckedAlreadyClosed.has(target.workflowId); + }); + if (residualTargets.length > 0) { + for (const target of residualTargets) { + if (results.get(target.workflowId)?.outcome.kind === 'unverified') { + retriedUnverified.add(target.workflowId); + } + if (results.get(target.workflowId)?.outcome.kind === 'already-closed') { + recheckedAlreadyClosed.add(target.workflowId); + } + } + const outcomes = await stopWorkflowsCancelFirst( + residualTargets.map((target) => target.workflowId), + lifecycle, + ); + for (let index = 0; index < residualTargets.length; index++) { + const target = residualTargets[index]; + const outcome = outcomes[index]; + if (target !== undefined && outcome !== undefined) results.set(target.workflowId, { target, outcome }); + } + quietSince = lifecycle.now(); + } + + const now = lifecycle.now(); + if (now - quietSince >= settleMs) return { kind: 'settled', results: [...results.values()] }; + if (now >= maxDeadline) return { kind: 'timed-out', results: [...results.values()] }; + try { + await lifecycle.wait(Math.min(CANCELLATION_POLL_MS, settleMs - (now - quietSince))); + } catch { + return { kind: 'timed-out', results: [...results.values()] }; + } } } async function stopAllScans(yes: boolean): Promise { - const temporalUp = isTemporalReady(); - const initial = runningContainers(WORKER_FILTER); - const targets = [...new Set(runningScanWorkspaces())].map((workspace): StopTarget => { - const workflowId = resolveWorkflowId(workspace); - return { - workspace, - ...(workflowId !== undefined && { workflowId }), - workflowRunning: Boolean(workflowId && temporalUp && isWorkflowRunning(workflowId)), - }; - }); + const containerQuery = runningScanContainersChecked(); + if (containerQuery.kind === 'unavailable') { + fail('Could not inspect running scan workers.', `Retry: ${commandPrefix()} stop --all`); + } + const containers = containerQuery.value; + let pending = readPendingTargets(listWorkspaces().map((workspace) => workspace.name)); + const initialDiscovery = await discoverRunningWorkflows(); + let visible = initialDiscovery.kind === 'ok' ? initialDiscovery.workflows : []; + let plan = buildWorkflowTargetPlan(containers, withRecordedWorkflows(containers), visible, pending.byWorkspace); + let targets = resolveTargetWorkspaces(plan.targets); - // Resolve what is running before prompting, so we never confirm a no-op. - if (initial.length === 0) { - console.log('No running scans to stop.'); - return; + if (containers.length === 0 && targets.length === 0) { + if (pending.unreadableCount > 0) { + fail('One or more scan launch records could not be read safely.', `Retry: ${commandPrefix()} stop --all`); + } + if (initialDiscovery.kind === 'unavailable') { + fail('Could not verify whether scan workflows are running in Temporal.', `Retry: ${commandPrefix()} stop --all`); + } + const emptySettleDeadline = stopLifecycle.now() + VISIBILITY_MAX_SETTLE_MS; + while (targets.length === 0) { + const remaining = emptySettleDeadline - stopLifecycle.now(); + if (remaining <= 0) { + console.log('No running scans to stop.'); + return; + } + await stopLifecycle.wait(Math.min(CANCELLATION_POLL_MS, remaining)); + const confirmation = await discoverRunningWorkflows(); + if (confirmation.kind === 'unavailable') { + fail( + 'Could not verify whether scan workflows are running in Temporal.', + `Retry: ${commandPrefix()} stop --all`, + ); + } + visible = confirmation.workflows; + pending = readPendingTargets(listWorkspaces().map((workspace) => workspace.name)); + if (pending.unreadableCount > 0) { + fail('One or more scan launch records could not be read safely.', `Retry: ${commandPrefix()} stop --all`); + } + plan = buildWorkflowTargetPlan(containers, withRecordedWorkflows(containers), visible, pending.byWorkspace); + targets = resolveTargetWorkspaces(plan.targets); + } } await confirmOrExit('stop', 'This will stop all running scans. Continue?', yes); - const spinner = p.spinner(); spinner.start('Stopping all scans'); - await stopTargetsCancelFirst(targets); - // Keep the legacy safety net for a worker whose workspace label was unavailable. - await stopContainers(runningContainers(WORKER_FILTER)); + const initialResult = await executeStopPlan(targets, containers, WORKER_FILTER); + const visibilitySettle = await stopVisibleWorkflowsUntilSettled(initialResult.workflows); + const results = visibilitySettle.results; - const stillRunning = runningContainers(WORKER_FILTER); - if (stillRunning.length > 0) { - spinner.error(`Stopped ${initial.length - stillRunning.length} of ${initial.length} scans`); - console.error(`${stillRunning.length} container(s) did not stop. Retry: ${commandPrefix()} stop --all`); + const combinedResult: StopExecutionResult = { + workflows: results, + containers: initialResult.containers, + preRegistrationWorkspaces: initialResult.preRegistrationWorkspaces, + }; + const unverified = unverifiedWorkflowCount(results); + const finalPending = readPendingTargets(listWorkspaces().map((workspace) => workspace.name)); + const initialPendingKeys = new Set( + pending.references.map( + (reference) => `${reference.workspace}\0${reference.identity.task_queue}\0${reference.identity.workflow_id}`, + ), + ); + const newPendingCount = finalPending.references.filter( + (reference) => + !initialPendingKeys.has( + `${reference.workspace}\0${reference.identity.task_queue}\0${reference.identity.workflow_id}`, + ), + ).length; + const unreadablePendingCount = Math.max(pending.unreadableCount, finalPending.unreadableCount); + const temporalDiscoveryFailed = initialDiscovery.kind === 'unavailable' || visibilitySettle.kind === 'unavailable'; + const temporalDiscoveryTimedOut = visibilitySettle.kind === 'timed-out'; + const incomplete = + combinedResult.containers.kind !== 'stopped' || + plan.containersWithoutVerifiedWorkflowId.length > 0 || + temporalDiscoveryFailed || + temporalDiscoveryTimedOut || + unreadablePendingCount > 0 || + newPendingCount > 0 || + unverified > 0; + + if (incomplete) { + spinner.error('Scan shutdown incomplete'); + if (combinedResult.containers.kind !== 'stopped') reportContainerFailure(undefined, combinedResult.containers); + if (unverified > 0) console.error(`Temporal could not confirm closure for ${unverified} workflow(s).`); + if (temporalDiscoveryFailed) console.error('Temporal could not enumerate every running scan workflow.'); + if (temporalDiscoveryTimedOut) console.error('Temporal workflow discovery did not settle before its deadline.'); + if (unreadablePendingCount > 0) { + console.error(`${unreadablePendingCount} launch record(s) could not be read safely.`); + } + if (newPendingCount > 0) console.error(`${newPendingCount} scan launch(es) began while shutdown was running.`); + if (plan.containersWithoutVerifiedWorkflowId.length > 0) { + console.error( + `${plan.containersWithoutVerifiedWorkflowId.length} legacy worker(s) could not prove a candidate workflow ID.`, + ); + } + console.error(`Retry: ${commandPrefix()} stop --all`); process.exit(1); } - spinner.stop(`Stopped ${initial.length} scan${initial.length === 1 ? '' : 's'}`); - - if (temporalUp && anyRunningScanWorkflow()) { - warn('some scan workflows are still Running in Temporal — check http://localhost:8233'); + appendVerifiedFallbacks(combinedResult); + const clearFailures = clearPendingTargets(pending.references); + if (clearFailures > 0) { + spinner.error('Scans stopped, but one or more launch records could not be cleared'); + console.error(`Retry: ${commandPrefix()} stop --all`); + process.exit(1); } + const stoppedCount = Math.max(containers.length, results.length); + spinner.stop(`Stopped ${stoppedCount} scan${stoppedCount === 1 ? '' : 's'}`); } -/** - * Infer which scan `stop` acts on when neither a workspace nor --all was given: the single - * running scan, announced on stderr so it is never a silent guess. Zero or several running - * scans exit with guidance — there is no most-recent fallback, since stopping a finished - * scan is a no-op. - */ +/** Resolve the omitted target from Docker without turning a failed query into an empty scan list. */ function resolveStopTarget(): string { - const target = resolveDefaultWorkspace({ allowFinished: false }); - if (target.kind === 'ok') { - console.error(`No workspace given; stopping running scan "${target.workspace}".`); - return target.workspace; + const result = runningScanContainersChecked(); + if (result.kind === 'unavailable') { + fail('Could not inspect running scan workers.', `Retry with a workspace: ${commandPrefix()} stop `); } - if (target.kind === 'ambiguous') { - failUsage('Multiple scans are running — specify which one, or use --all:', ` ${target.running.join(', ')}`); + const running = [...new Set(result.value.flatMap((container) => container.workspace ?? []))]; + if (running.length === 1) { + const workspace = running[0] as string; + console.error(`No workspace given; stopping running scan "${workspace}".`); + return workspace; + } + if (running.length > 1) { + failUsage('Multiple scans are running: specify which one, or use --all:', ` ${running.join(', ')}`); + } + if (result.value.length > 0) { + fail('A running scan worker has no workspace label.', `Use ${commandPrefix()} stop --all`); } fail('No running scans to stop.', 'Pass a workspace name to stop a specific scan.'); } export async function stop(opts: StopOptions): Promise { ensureDocker(); + if (opts.all && opts.workspace) failUsage('Pass a workspace name or --all, not both.'); - // Validate the target: exactly one of or --all. - if (opts.all && opts.workspace) { - failUsage('Pass a workspace name or --all, not both.'); - } - - // With no explicit target and no --all, default to the single running scan. const workspace = opts.all ? undefined : (opts.workspace ?? resolveStopTarget()); - - if (workspace) { - await stopSingleScan(workspace, opts.yes); - } else { - await stopAllScans(opts.yes); - } + if (workspace) await stopSingleScan(workspace, opts.yes); + else await stopAllScans(opts.yes); } diff --git a/apps/cli/src/docker.ts b/apps/cli/src/docker.ts index 3875fb45..3640fb2b 100644 --- a/apps/cli/src/docker.ts +++ b/apps/cli/src/docker.ts @@ -27,6 +27,16 @@ const DEV_IMAGE = 'shannon-worker'; /** Docker label stamped on each worker container, mapping it back to its workspace so a single scan can be stopped by name. */ const WORKSPACE_LABEL = 'shannon.workspace'; +/** Docker label that joins a worker container to the Temporal workflow polling its unique task queue. */ +const TASK_QUEUE_LABEL = 'shannon.task-queue'; + +/** Docker label carrying the workflow ID selected before the worker starts. */ +const WORKFLOW_ID_LABEL = 'shannon.workflow-id'; + +/** Image/container protocol proving that the worker honors the preselected workflow ID. */ +const WORKER_PROTOCOL_LABEL = 'shannon.worker-protocol'; +export const WORKFLOW_ID_PROTOCOL = 'workflow-id-v1'; + export function getWorkerImage(version: string): string { return getMode() === 'local' ? DEV_IMAGE : `${NPX_IMAGE_REPO}:${version}`; } @@ -84,9 +94,6 @@ function spawnQuiet(cmd: string, args: string[]): Promise { const TEMPORAL_CONTAINER = 'shannon-temporal'; const TEMPORAL_ADDRESS = 'localhost:7233'; -/** Query matching every running pentest scan workflow. */ -const RUNNING_SCAN_QUERY = "ExecutionStatus = 'Running' AND WorkflowType = 'pentestPipelineWorkflow'"; - /** Build `docker exec` args for a `temporal` CLI command run inside the Temporal container. */ function temporalCmd(...args: string[]): string[] { return ['exec', TEMPORAL_CONTAINER, 'temporal', ...args, '--address', TEMPORAL_ADDRESS]; @@ -256,7 +263,10 @@ export function buildImage(noCache: boolean, version: string): void { export function ensureImage(version: string): void { const image = getWorkerImage(version); const exists = runQuiet('docker', ['image', 'inspect', image]); - if (exists) return; + if (exists) { + ensureWorkerImageProtocol(image); + return; + } if (canBuildImage()) { console.log('Shannon image not found, building...'); @@ -274,6 +284,22 @@ export function ensureImage(version: string): void { } pruneOldImages(version); } + ensureWorkerImageProtocol(image); +} + +/** Refuse a stale worker image that would ignore the CLI-selected workflow ID. */ +function ensureWorkerImageProtocol(image: string): void { + const protocol = runOutput('docker', [ + 'image', + 'inspect', + image, + '--format', + `{{ index .Config.Labels "${WORKER_PROTOCOL_LABEL}" }}`, + ]); + if (protocol === WORKFLOW_ID_PROTOCOL) return; + + const hint = canBuildImage() ? 'Run ./shannon build, then retry.' : 'Reinstall this Shannon version, then retry.'; + fail('The Shannon worker image is incompatible with this CLI.', hint); } /** @@ -377,6 +403,7 @@ export interface WorkerOptions { repo: { hostPath: string; containerPath: string }; workspacesDir: string; taskQueue: string; + workflowId: string; containerName: string; envFlags: string[]; config?: { hostPath: string; containerPath: string }; @@ -399,8 +426,16 @@ export function spawnWorker(opts: WorkerOptions): ChildProcess { } args.push('--name', opts.containerName, '--network', 'shannon-net'); - // Tag with the workspace so `stop ` can target this scan's container - args.push('--label', `${WORKSPACE_LABEL}=${opts.workspace}`); + // Keep the launch identity on the container before session.json exists. The fixed workflow + // ID lets stop verify the pre-registration window without trusting visibility timing. + args.push( + '--label', + `${WORKSPACE_LABEL}=${opts.workspace}`, + '--label', + `${TASK_QUEUE_LABEL}=${opts.taskQueue}`, + '--label', + `${WORKFLOW_ID_LABEL}=${opts.workflowId}`, + ); // Add host flag for Linux args.push(...addHostFlag()); @@ -459,6 +494,7 @@ export function spawnWorker(opts: WorkerOptions): ChildProcess { // Worker command args.push('node', 'apps/worker/dist/temporal/worker.js', opts.url, opts.repo.containerPath); args.push('--task-queue', opts.taskQueue); + args.push('--workflow-id', opts.workflowId); if (opts.config) { args.push('--config', opts.config.containerPath); } @@ -482,6 +518,18 @@ export function spawnWorker(opts: WorkerOptions): ChildProcess { /** `docker ps --filter` args matching every running worker container. */ export const WORKER_FILTER: readonly string[] = ['--filter', 'name=shannon-worker-']; +/** Result of a command-backed query whose unavailable state must not be mistaken for an empty result. */ +export type CommandQueryResult = { kind: 'ok'; value: T } | { kind: 'unavailable' }; + +/** Identity carried by a running scan worker container. Older workers may lack the newer labels. */ +export interface RunningScanContainer { + readonly id: string; + readonly workspace?: string; + readonly taskQueue?: string; + readonly workflowId?: string; + readonly workerProtocol?: string; +} + /** `docker ps --filter` args matching one scan's worker container(s), by workspace label. */ export function scanFilter(workspace: string): readonly string[] { return ['--filter', `label=${WORKSPACE_LABEL}=${workspace}`]; @@ -492,23 +540,85 @@ export function scanFilter(workspace: string): readonly string[] { * the authoritative check for whether containers actually stopped — `docker stop`'s * exit code can't distinguish "already gone" from "failed to stop". */ +export function runningContainersChecked(filter: readonly string[]): CommandQueryResult { + try { + const output = execFileSync('docker', ['ps', '-q', ...filter], { stdio: 'pipe', encoding: 'utf-8' }).trim(); + return { kind: 'ok', value: output.split('\n').filter(Boolean) }; + } catch { + return { kind: 'unavailable' }; + } +} + +/** + * Best-effort counterpart for callers where Docker unavailability is intentionally + * presented as no local running containers. + */ export function runningContainers(filter: readonly string[]): string[] { - const output = runOutput('docker', ['ps', '-q', ...filter]); - return output.split('\n').filter(Boolean); + const result = runningContainersChecked(filter); + return result.kind === 'ok' ? result.value : []; +} + +function normalizedLabel(value: string | undefined): string | undefined { + const normalized = value?.trim(); + return normalized && normalized !== '' ? normalized : undefined; +} + +/** + * Running scan containers with the labels needed to correlate a worker to its Temporal + * workflow. A successful query keeps unlabeled legacy workers in the result by ID. + */ +export function runningScanContainersChecked( + filter: readonly string[] = WORKER_FILTER, +): CommandQueryResult { + try { + const format = `{{.ID}}\t{{ index .Labels "${WORKSPACE_LABEL}" }}\t{{ index .Labels "${TASK_QUEUE_LABEL}" }}\t{{ index .Labels "${WORKFLOW_ID_LABEL}" }}\t{{ index .Labels "${WORKER_PROTOCOL_LABEL}" }}`; + const output = execFileSync('docker', ['ps', ...filter, '--format', format], { + stdio: 'pipe', + encoding: 'utf-8', + }).trim(); + if (!output) return { kind: 'ok', value: [] }; + + const containers: RunningScanContainer[] = []; + for (const line of output.split('\n')) { + const [rawId, rawWorkspace, rawTaskQueue, rawWorkflowId, rawWorkerProtocol] = line.split('\t'); + const id = rawId?.trim(); + if (!id) return { kind: 'unavailable' }; + const workspace = normalizedLabel(rawWorkspace); + const taskQueue = normalizedLabel(rawTaskQueue); + const workflowId = normalizedLabel(rawWorkflowId); + const workerProtocol = normalizedLabel(rawWorkerProtocol); + containers.push({ + id, + ...(workspace !== undefined && { workspace }), + ...(taskQueue !== undefined && { taskQueue }), + ...(workflowId !== undefined && { workflowId }), + ...(workerProtocol !== undefined && { workerProtocol }), + }); + } + return { kind: 'ok', value: containers }; + } catch { + return { kind: 'unavailable' }; + } } /** * Workspace names of every running worker container, read from the shannon.workspace - * label each scan is stamped with at spawn. This is the authoritative running-scan → - * workspace-name map. Best-effort: empty when Docker is unreachable, which is the - * correct answer anyway (no scan can be running without the daemon). + * label each scan is stamped with at spawn. The checked form preserves Docker query + * failures so lifecycle commands do not mistake an unavailable daemon for an empty list. */ +export function runningScanWorkspacesChecked(): CommandQueryResult { + const result = runningScanContainersChecked(); + if (result.kind === 'unavailable') return result; + return { + kind: 'ok', + value: result.value.flatMap((container) => (container.workspace === undefined ? [] : [container.workspace])), + }; +} + +/** Best-effort counterpart for callers that only need the local scan list. */ export function runningScanWorkspaces(): string[] { - const output = runOutput('docker', ['ps', ...WORKER_FILTER, '--format', `{{ index .Labels "${WORKSPACE_LABEL}" }}`]); - return output - .split('\n') - .map((name) => name.trim()) - .filter(Boolean); + const result = runningScanWorkspacesChecked(); + return result.kind === 'ok' ? result.value : []; } /** @@ -520,40 +630,6 @@ export async function stopContainers(ids: string[]): Promise { await Promise.all(ids.map((id) => spawnQuiet('docker', ['stop', id]))); } -/** Request cooperative cancellation so the workflow can run its terminal finalizer. */ -export function cancelWorkflow(workflowId: string): boolean { - return runQuiet('docker', temporalCmd('workflow', 'cancel', '--workflow-id', workflowId)); -} - -/** - * Terminate a Temporal workflow so a stopped scan doesn't linger as a running - * workflow with no worker. Best-effort: returns false if Temporal is unreachable - * or the workflow already closed. Requires Temporal to be up (guard with isTemporalReady). - */ -export function terminateWorkflow(workflowId: string, reason: string): boolean { - return runQuiet('docker', temporalCmd('workflow', 'terminate', '--workflow-id', workflowId, '--reason', reason)); -} - -/** - * Whether a specific workflow is still in the Running state. Re-querying this after - * a terminate verifies it actually took effect, rather than trusting the terminate - * command's exit code. Requires Temporal to be up (guard with isTemporalReady). - */ -export function isWorkflowRunning(workflowId: string): boolean { - const query = `WorkflowId = '${workflowId}' AND ExecutionStatus = 'Running'`; - const output = runOutput('docker', temporalCmd('workflow', 'list', '--query', query)); - return output.includes(workflowId); -} - -/** - * Whether any pentest scan workflow is still Running — the `stop --all` counterpart - * to isWorkflowRunning. Requires Temporal to be up (guard with isTemporalReady). - */ -export function anyRunningScanWorkflow(): boolean { - const output = runOutput('docker', temporalCmd('workflow', 'list', '--query', RUNNING_SCAN_QUERY)); - return output.includes('pentestPipelineWorkflow'); -} - /** * Tear down the compose stack. When `clean` is set, volumes are removed too. */ diff --git a/apps/cli/src/pending-workflow.ts b/apps/cli/src/pending-workflow.ts new file mode 100644 index 00000000..9488f38e --- /dev/null +++ b/apps/cli/src/pending-workflow.ts @@ -0,0 +1,139 @@ +/** Durable CLI-owned workflow candidates that bridge Docker launch and session registration. */ + +import fs from 'node:fs'; +import path from 'node:path'; +import { INTERNAL_DIR } from './paths.js'; + +const SCHEMA_VERSION = 1 as const; +const PENDING_DIR = 'pending-workflows'; + +export interface PendingWorkflowIdentity { + readonly schema_version: typeof SCHEMA_VERSION; + readonly workflow_id: string; + readonly task_queue: string; + readonly created_at: string; +} + +export interface PendingWorkflowReadResult { + readonly identities: readonly PendingWorkflowIdentity[]; + readonly unreadableCount: number; +} + +function pendingDir(workspacePath: string): string { + return path.join(workspacePath, INTERNAL_DIR, PENDING_DIR); +} + +function pendingFile(workspacePath: string, taskQueue: string): string { + return path.join(pendingDir(workspacePath), `launch-${encodeURIComponent(taskQueue)}.json`); +} + +function syncDirectory(directory: string): void { + const descriptor = fs.openSync(directory, 'r'); + try { + fs.fsyncSync(descriptor); + } finally { + fs.closeSync(descriptor); + } +} + +/** Persist the candidate before docker run, so a vanished pre-registration worker remains addressable. */ +export function writePendingWorkflowIdentity(workspacePath: string, workflowId: string, taskQueue: string): void { + const directory = pendingDir(workspacePath); + const directoryAlreadyExisted = fs.existsSync(directory); + fs.mkdirSync(directory, { recursive: true }); + if (!directoryAlreadyExisted) syncDirectory(path.dirname(directory)); + const destination = pendingFile(workspacePath, taskQueue); + const temporary = `${destination}.tmp-${process.pid}-${Date.now()}`; + const identity: PendingWorkflowIdentity = { + schema_version: SCHEMA_VERSION, + workflow_id: workflowId, + task_queue: taskQueue, + created_at: new Date().toISOString(), + }; + + const descriptor = fs.openSync(temporary, 'wx', 0o600); + try { + fs.writeFileSync(descriptor, `${JSON.stringify(identity, null, 2)}\n`, 'utf8'); + fs.fsyncSync(descriptor); + } finally { + fs.closeSync(descriptor); + } + try { + // Link installs the fully-fsynced inode without replacing an existing task-queue record. + fs.linkSync(temporary, destination); + fs.unlinkSync(temporary); + syncDirectory(directory); + } catch (error) { + fs.rmSync(temporary, { force: true }); + throw error; + } +} + +/** Remove one candidate only after session registration or a fully verified stop. */ +export function clearPendingWorkflowIdentity(workspacePath: string, taskQueue: string): void { + const directory = pendingDir(workspacePath); + fs.rmSync(pendingFile(workspacePath, taskQueue), { force: true }); + if (fs.existsSync(directory)) syncDirectory(directory); +} + +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); +} + +function isPendingWorkflowIdentity( + value: unknown, + workspace: string, + expectedFilename: string, +): value is PendingWorkflowIdentity { + if (value === null || typeof value !== 'object' || Array.isArray(value)) return false; + const candidate = value as Record; + const keys = Object.keys(candidate).sort(); + const workflowId = candidate.workflow_id; + const workflowPattern = new RegExp(`^${escapeRegExp(workspace)}_(?:shannon-|resume_)\\d+$`); + const workspaceIsWorkflowId = workflowId === workspace && /_shannon-\d+$/.test(workspace); + return ( + keys.length === 4 && + keys[0] === 'created_at' && + keys[1] === 'schema_version' && + keys[2] === 'task_queue' && + keys[3] === 'workflow_id' && + candidate.schema_version === SCHEMA_VERSION && + typeof workflowId === 'string' && + (workspaceIsWorkflowId || workflowPattern.test(workflowId)) && + typeof candidate.task_queue === 'string' && + /^shannon-[0-9a-f]{8}$/.test(candidate.task_queue) && + expectedFilename === `launch-${encodeURIComponent(candidate.task_queue)}.json` && + typeof candidate.created_at === 'string' && + !Number.isNaN(Date.parse(candidate.created_at)) && + new Date(candidate.created_at).toISOString() === candidate.created_at + ); +} + +/** Read every outstanding launch candidate, preserving corrupt records as an explicit failure count. */ +export function readPendingWorkflowIdentities(workspacePath: string): PendingWorkflowReadResult { + let entries: string[]; + try { + entries = fs.readdirSync(pendingDir(workspacePath)).filter((entry) => entry.endsWith('.json')); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return { identities: [], unreadableCount: 0 }; + return { identities: [], unreadableCount: 1 }; + } + + const identities: PendingWorkflowIdentity[] = []; + let unreadableCount = 0; + for (const entry of entries) { + try { + const value: unknown = JSON.parse(fs.readFileSync(path.join(pendingDir(workspacePath), entry), 'utf8')); + if (!isPendingWorkflowIdentity(value, path.basename(workspacePath), entry)) { + unreadableCount++; + continue; + } + identities.push(value); + } catch { + unreadableCount++; + } + } + // Atomic-write temp files are intentionally ignored: start cannot spawn Docker until the + // final .json rename and fsync above have both completed. + return { identities, unreadableCount }; +} diff --git a/apps/cli/src/temporal-client.ts b/apps/cli/src/temporal-client.ts index b3f63fec..016e2398 100644 --- a/apps/cli/src/temporal-client.ts +++ b/apps/cli/src/temporal-client.ts @@ -1,5 +1,5 @@ /** - * Thin Temporal client for reading one scan's state. + * Thin Temporal client for reading scan state and controlling scan workflow lifecycle. * * A running scan is queried live (getProgress) and read via pendingActivities for * the in-flight agents; a closed scan is read once from its result. Everything goes @@ -13,10 +13,20 @@ import { ACTIVITY_TO_PROGRESS, type PipelineState } from './scan/pipeline.js'; const ADDRESS = '127.0.0.1:7233'; const NAMESPACE = 'default'; +const LIFECYCLE_RPC_DEADLINE_MS = 3_000; +const OPEN_SCAN_WORKFLOW_QUERY = + "WorkflowType = 'pentestPipelineWorkflow' AND (ExecutionStatus = 'Running' OR ExecutionStatus = 'Paused')"; -// 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']); +// WorkflowExecutionStatusName values that positively prove this execution has closed. +// PAUSED is open; UNSPECIFIED and UNKNOWN are not safe closure evidence. +const TERMINAL_STATUSES: ReadonlySet = new Set([ + 'COMPLETED', + 'FAILED', + 'CANCELLED', + 'TERMINATED', + 'CONTINUED_AS_NEW', + 'TIMED_OUT', +]); export interface RunningAgent { readonly agent: string; @@ -66,11 +76,28 @@ export type TerminalOutcome = | { readonly kind: 'success'; readonly state: PipelineState } | { readonly kind: 'failed'; readonly message: string }; +/** + * The authoritative Temporal state used by lifecycle commands. Transport failures deliberately + * remain errors instead of being represented as a closed workflow: callers must not report a + * scan stopped unless Temporal has positively confirmed it. + */ +export type WorkflowLifecycleState = + | { readonly kind: 'open'; readonly status: 'RUNNING' | 'PAUSED' } + | { readonly kind: 'terminal'; readonly status: string } + | { readonly kind: 'unknown'; readonly status: string } + | { readonly kind: 'not-found' }; + +/** A scan workflow returned by Temporal's eventually consistent open-workflow visibility query. */ +export interface RunningScanWorkflow { + readonly workflowId: string; + readonly taskQueue: string; +} + let clientPromise: Promise | null = null; function getClient(): Promise { if (!clientPromise) { - const pending = Connection.connect({ address: ADDRESS }).then( + const pending = Connection.connect({ address: ADDRESS, connectTimeout: LIFECYCLE_RPC_DEADLINE_MS }).then( (connection) => new Client({ connection, namespace: NAMESPACE }), ); // A rejected connect must not be cached forever: clear the memo so the next call rebuilds @@ -95,6 +122,95 @@ function resetClient(only?: Promise): void { previous?.then((client) => client.connection.close()).catch(() => {}); } +/** Close the current channel and establish another before a termination retry. */ +export async function refreshWorkflowLifecycleConnection(): Promise { + const previous = clientPromise; + if (previous !== null) { + if (clientPromise === previous) clientPromise = null; + try { + const client = await previous; + await client.connection.close(); + } catch { + // A failed prior connection is already detached. The new connection below is authoritative. + } + } + await getClient(); +} + +/** + * Run a bounded lifecycle RPC and discard the connection when Temporal did not positively say + * that the workflow is absent. A fresh connection is important after a gRPC timeout or transport + * failure: reusing a wedged channel can turn a recoverable stop into an indefinitely ambiguous one. + */ +async function runLifecycleRpc(operation: (client: Client) => Promise): Promise { + const pending = getClient(); + try { + const client = await pending; + return await client.withDeadline(Date.now() + LIFECYCLE_RPC_DEADLINE_MS, () => operation(client)); + } catch (err) { + if (!(err instanceof WorkflowNotFoundError)) resetClient(pending); + throw err; + } +} + +/** Describe a workflow for lifecycle control without reading its progress or pending activities. */ +export async function describeWorkflowLifecycle(workflowId: string): Promise { + try { + const desc = await runLifecycleRpc((client) => client.workflow.getHandle(workflowId).describe()); + if (desc.status.name === 'RUNNING' || desc.status.name === 'PAUSED') { + return { kind: 'open', status: desc.status.name }; + } + if (TERMINAL_STATUSES.has(desc.status.name)) return { kind: 'terminal', status: desc.status.name }; + return { kind: 'unknown', status: desc.status.name }; + } catch (err) { + if (err instanceof WorkflowNotFoundError) return { kind: 'not-found' }; + throw err; + } +} + +/** Request cooperative cancellation. This confirms request acceptance, not workflow closure. */ +export async function requestWorkflowCancellation(workflowId: string): Promise<'requested' | 'not-found'> { + try { + await runLifecycleRpc((client) => client.workflow.getHandle(workflowId).cancel()); + return 'requested'; + } catch (err) { + if (err instanceof WorkflowNotFoundError) return 'not-found'; + throw err; + } +} + +/** Request forced termination. This confirms request acceptance, not workflow closure. */ +export async function requestWorkflowTermination( + workflowId: string, + reason: string, +): Promise<'requested' | 'not-found'> { + try { + await runLifecycleRpc((client) => client.workflow.getHandle(workflowId).terminate(reason)); + return 'requested'; + } catch (err) { + if (err instanceof WorkflowNotFoundError) return 'not-found'; + throw err; + } +} + +/** List currently open Shannon scan workflows through Temporal visibility. */ +export async function listRunningScanWorkflows(): Promise { + return runLifecycleRpc(async (client) => { + const workflows: RunningScanWorkflow[] = []; + for await (const execution of client.workflow.list({ query: OPEN_SCAN_WORKFLOW_QUERY })) { + // Visibility is eventually consistent. Keep only the open scan rows returned by this page; + // each discovered workflow is described directly before `stop` accepts its closure. + if ( + (execution.status.name === 'RUNNING' || execution.status.name === 'PAUSED') && + execution.type === 'pentestPipelineWorkflow' + ) { + workflows.push({ workflowId: execution.workflowId, taskQueue: execution.taskQueue }); + } + } + return workflows; + }); +} + /** Describe a scan: status, timing, and the agents currently running (from pendingActivities). Null if not found. */ export async function describeScan(workflowId: string): Promise { const client = await getClient(); diff --git a/apps/cli/src/workspaces.ts b/apps/cli/src/workspaces.ts index a41260cd..b3856a1b 100644 --- a/apps/cli/src/workspaces.ts +++ b/apps/cli/src/workspaces.ts @@ -8,8 +8,9 @@ * identity proof are separate steps: `resolveScanIdentity` turns a selected or explicit * string into the one canonical (workspace, workflowId) pair the session records prove. * - * Running scans are identified by Docker label (the authoritative source, shared with - * `stop`); recency for finished scans comes from each run's session.json createdAt, + * Running workers are identified by Docker workspace label for default-target selection. + * `stop` supplements that local discovery with Temporal lifecycle state. Recency for + * finished scans comes from each run's session.json createdAt, * with the workspace directory mtime as the fallback for runs that predate it. */ diff --git a/apps/worker/src/temporal/worker.ts b/apps/worker/src/temporal/worker.ts index f8f057d9..bf2ec553 100644 --- a/apps/worker/src/temporal/worker.ts +++ b/apps/worker/src/temporal/worker.ts @@ -18,6 +18,7 @@ * * Options: * --task-queue Task queue name (required, unique per scan) + * --workflow-id Workflow ID selected by the Shannon CLI * --config Configuration file path * --output Stable mounted path for final customer report copies * --workspace Resume from existing workspace @@ -240,6 +241,7 @@ interface CliArgs { webUrl: string; repoPath: string; taskQueue: string; + workflowId?: string; configPath?: string; customerOutputPath?: string; pipelineTestingMode: boolean; @@ -253,6 +255,7 @@ function showUsage(): void { console.log(' node dist/temporal/worker.js --task-queue [options]\n'); console.log('Options:'); console.log(' --task-queue Task queue name (required)'); + console.log(' --workflow-id Workflow ID selected by the Shannon CLI'); console.log(' --config Configuration file path'); console.log(' --workspace Resume from existing workspace'); console.log(' --output Stable mounted path for final customer report copies'); @@ -268,6 +271,7 @@ function parseCliArgs(argv: string[]): CliArgs { let webUrl: string | undefined; let repoPath: string | undefined; let taskQueue: string | undefined; + let workflowId: string | undefined; let configPath: string | undefined; let customerOutputPath: string | undefined; let pipelineTestingMode = false; @@ -281,6 +285,12 @@ function parseCliArgs(argv: string[]): CliArgs { taskQueue = nextArg; i++; } + } else if (arg === '--workflow-id') { + const nextArg = argv[i + 1]; + if (nextArg && !nextArg.startsWith('-')) { + workflowId = nextArg; + i++; + } } else if (arg === '--config') { const nextArg = argv[i + 1]; if (nextArg && !nextArg.startsWith('-')) { @@ -326,6 +336,7 @@ function parseCliArgs(argv: string[]): CliArgs { webUrl, repoPath, taskQueue, + ...(workflowId && { workflowId }), pipelineTestingMode, ...(configPath && { configPath }), ...(customerOutputPath && { customerOutputPath }), @@ -356,6 +367,17 @@ function isValidWorkspaceName(name: string): boolean { return /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,127}$/.test(name); } +function escapeRegExp(value: string): string { + return value.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); +} + +/** Accept a CLI-owned ID only when it preserves this launch branch's public naming contract. */ +function selectWorkflowId(requested: string | undefined, fallback: string, expected: RegExp): string { + if (requested === undefined) return fallback; + if (!expected.test(requested)) throw new Error('Invalid workflow identity supplied by the Shannon CLI'); + return requested; +} + interface WorkspaceResolution { workflowId: string; sessionId: string; @@ -407,7 +429,12 @@ async function terminateExistingWorkflows(client: Client, workspaceName: string) async function resolveWorkspace(client: Client, args: CliArgs, expectedExploit: boolean): Promise { if (!args.resumeFromWorkspace) { const hostname = sanitizeHostname(args.webUrl); - const workflowId = `${hostname}_shannon-${Date.now()}`; + const fallback = `${hostname}_shannon-${Date.now()}`; + const workflowId = selectWorkflowId( + args.workflowId, + fallback, + new RegExp(`^${escapeRegExp(hostname)}_shannon-\\d+$`), + ); return { workflowId, sessionId: workflowId, @@ -442,8 +469,9 @@ async function resolveWorkspace(client: Client, args: CliArgs, expectedExploit: console.log(`Terminated ${terminatedWorkflows.length} previous scan(s)\n`); } + const fallback = `${workspace}_resume_${Date.now()}`; return { - workflowId: `${workspace}_resume_${Date.now()}`, + workflowId: selectWorkflowId(args.workflowId, fallback, new RegExp(`^${escapeRegExp(workspace)}_resume_\\d+$`)), sessionId: workspace, isResume: true, terminatedWorkflows, @@ -461,7 +489,12 @@ async function resolveWorkspace(client: Client, args: CliArgs, expectedExploit: // If the workspace name already looks like a CLI-generated ID // (ends with _shannon-), use it directly to avoid double _shannon- suffixes - const workflowId = /_shannon-\d+$/.test(workspace) ? workspace : `${workspace}_shannon-${Date.now()}`; + const fallback = /_shannon-\d+$/.test(workspace) ? workspace : `${workspace}_shannon-${Date.now()}`; + const expected = + fallback === workspace + ? new RegExp(`^${escapeRegExp(workspace)}$`) + : new RegExp(`^${escapeRegExp(workspace)}_shannon-\\d+$`); + const workflowId = selectWorkflowId(args.workflowId, fallback, expected); return { workflowId,