|
| 1 | +import type { ServerConfig } from "./config.js"; |
| 2 | +import { terminateProcessTree } from "./process-platform.js"; |
| 3 | +import { |
| 4 | + createWorkflowStore, |
| 5 | + type WorkflowStore, |
| 6 | +} from "./workflow-store.js"; |
| 7 | +import { |
| 8 | + WORKFLOW_CANCEL_HARD_MS, |
| 9 | + WORKFLOW_HEARTBEAT_MS, |
| 10 | + type WorkflowRunRecord, |
| 11 | +} from "./workflow-types.js"; |
| 12 | + |
| 13 | +const DEFAULT_TERM_WAIT_MS = 1_000; |
| 14 | +const DEFAULT_POLL_MS = 100; |
| 15 | +const DEFAULT_REAPER_INTERVAL_MS = WORKFLOW_HEARTBEAT_MS * 2; |
| 16 | +const DEFAULT_STALE_AFTER_MS = WORKFLOW_HEARTBEAT_MS * 3; |
| 17 | + |
| 18 | +const ACTIVE_STATUSES = new Set(["starting", "running"]); |
| 19 | + |
| 20 | +export interface WorkflowLifecycleRuntime { |
| 21 | + sleep(ms: number): Promise<void>; |
| 22 | + terminate(pid: number, signal: NodeJS.Signals): void; |
| 23 | +} |
| 24 | + |
| 25 | +const defaultRuntime: WorkflowLifecycleRuntime = { |
| 26 | + sleep: (ms) => new Promise((resolve) => setTimeout(resolve, ms)), |
| 27 | + terminate: (pid, signal) => { |
| 28 | + terminateProcessTree( |
| 29 | + { |
| 30 | + pid, |
| 31 | + kill: (requestedSignal = signal) => { |
| 32 | + try { |
| 33 | + process.kill(pid, requestedSignal); |
| 34 | + return true; |
| 35 | + } catch (error) { |
| 36 | + if ((error as NodeJS.ErrnoException).code === "ESRCH") return false; |
| 37 | + throw error; |
| 38 | + } |
| 39 | + }, |
| 40 | + }, |
| 41 | + signal, |
| 42 | + true, |
| 43 | + ); |
| 44 | + }, |
| 45 | +}; |
| 46 | + |
| 47 | +export interface CancelWorkflowRunOptions { |
| 48 | + graceMs?: number; |
| 49 | + termWaitMs?: number; |
| 50 | + pollMs?: number; |
| 51 | + runtime?: WorkflowLifecycleRuntime; |
| 52 | +} |
| 53 | + |
| 54 | +/** |
| 55 | + * Shared CLI/MCP cancellation path: cooperative flag, grace period, process |
| 56 | + * tree termination, then an atomic terminal fallback in the journal. |
| 57 | + */ |
| 58 | +export async function cancelWorkflowRun( |
| 59 | + store: WorkflowStore, |
| 60 | + runId: string, |
| 61 | + options: CancelWorkflowRunOptions = {}, |
| 62 | +): Promise<WorkflowRunRecord> { |
| 63 | + const requested = store.requestCancelResult(runId); |
| 64 | + if (requested.isErr()) throw requested.error; |
| 65 | + if (!isActive(requested.value)) return requested.value; |
| 66 | + |
| 67 | + const runtime = options.runtime ?? defaultRuntime; |
| 68 | + const graceMs = Math.max(0, options.graceMs ?? WORKFLOW_CANCEL_HARD_MS); |
| 69 | + const termWaitMs = Math.max(0, options.termWaitMs ?? DEFAULT_TERM_WAIT_MS); |
| 70 | + const pollMs = Math.max(1, options.pollMs ?? DEFAULT_POLL_MS); |
| 71 | + |
| 72 | + const cooperative = await waitForTerminal(store, runId, graceMs, pollMs, runtime); |
| 73 | + if (cooperative && !isActive(cooperative)) return cooperative; |
| 74 | + |
| 75 | + let current = store.getRun(runId); |
| 76 | + if (!current) throw new Error(`Unknown workflow run: ${runId}`); |
| 77 | + if (!isActive(current)) return current; |
| 78 | + |
| 79 | + if (current.pid) { |
| 80 | + safelyTerminate(runtime, current.pid, "SIGTERM"); |
| 81 | + const afterTerm = await waitForTerminal(store, runId, termWaitMs, pollMs, runtime); |
| 82 | + if (afterTerm && !isActive(afterTerm)) return afterTerm; |
| 83 | + |
| 84 | + current = store.getRun(runId) ?? current; |
| 85 | + if (isActive(current) && current.pid) { |
| 86 | + safelyTerminate(runtime, current.pid, "SIGKILL"); |
| 87 | + } |
| 88 | + } |
| 89 | + |
| 90 | + const cancelled = store.cancelRunResult(runId, "cancelled by workflow supervisor"); |
| 91 | + if (cancelled.isErr()) throw cancelled.error; |
| 92 | + return cancelled.value; |
| 93 | +} |
| 94 | + |
| 95 | +export function reapStaleWorkflows( |
| 96 | + store: WorkflowStore, |
| 97 | + staleAfterMs = DEFAULT_STALE_AFTER_MS, |
| 98 | +): WorkflowRunRecord[] { |
| 99 | + return store.reapStale(staleAfterMs); |
| 100 | +} |
| 101 | + |
| 102 | +export interface WorkflowReaperHandle { |
| 103 | + close(): void; |
| 104 | +} |
| 105 | + |
| 106 | +export function startWorkflowReaper( |
| 107 | + config: ServerConfig, |
| 108 | + options: { |
| 109 | + intervalMs?: number; |
| 110 | + staleAfterMs?: number; |
| 111 | + onError?: (error: unknown) => void; |
| 112 | + } = {}, |
| 113 | +): WorkflowReaperHandle { |
| 114 | + const intervalMs = Math.max(1, options.intervalMs ?? DEFAULT_REAPER_INTERVAL_MS); |
| 115 | + const staleAfterMs = Math.max(1, options.staleAfterMs ?? DEFAULT_STALE_AFTER_MS); |
| 116 | + |
| 117 | + const tick = (): void => { |
| 118 | + const store = createWorkflowStore(config); |
| 119 | + try { |
| 120 | + reapStaleWorkflows(store, staleAfterMs); |
| 121 | + } catch (error) { |
| 122 | + options.onError?.(error); |
| 123 | + } finally { |
| 124 | + store.close(); |
| 125 | + } |
| 126 | + }; |
| 127 | + |
| 128 | + tick(); |
| 129 | + const timer = setInterval(tick, intervalMs); |
| 130 | + timer.unref(); |
| 131 | + return { |
| 132 | + close(): void { |
| 133 | + clearInterval(timer); |
| 134 | + }, |
| 135 | + }; |
| 136 | +} |
| 137 | + |
| 138 | +async function waitForTerminal( |
| 139 | + store: WorkflowStore, |
| 140 | + runId: string, |
| 141 | + waitMs: number, |
| 142 | + pollMs: number, |
| 143 | + runtime: WorkflowLifecycleRuntime, |
| 144 | +): Promise<WorkflowRunRecord | undefined> { |
| 145 | + const deadline = Date.now() + waitMs; |
| 146 | + let current = store.getRun(runId); |
| 147 | + while (current && isActive(current) && Date.now() < deadline) { |
| 148 | + await runtime.sleep(Math.min(pollMs, Math.max(1, deadline - Date.now()))); |
| 149 | + current = store.getRun(runId); |
| 150 | + } |
| 151 | + return current; |
| 152 | +} |
| 153 | + |
| 154 | +function safelyTerminate( |
| 155 | + runtime: WorkflowLifecycleRuntime, |
| 156 | + pid: number, |
| 157 | + signal: NodeJS.Signals, |
| 158 | +): void { |
| 159 | + try { |
| 160 | + runtime.terminate(pid, signal); |
| 161 | + } catch (error) { |
| 162 | + if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; |
| 163 | + } |
| 164 | +} |
| 165 | + |
| 166 | +function isActive(run: WorkflowRunRecord): boolean { |
| 167 | + return ACTIVE_STATUSES.has(run.status); |
| 168 | +} |
0 commit comments