import { setTimeout as delay } from "node:timers/promises"; import type { InsightRun, InsightRunCreateInput, InsightRunFailureClass, InsightRunOutputMetadata, InsightRunTrigger, InsightRunUpdateInput, } from "./insight-types.js"; import { InsightLifecycleError, InsightStore } from "./insight-store.js"; export interface InsightRunAttemptResult { summary?: string | null; insightsCreated: number; insightsUpdated: number; outputMetadata?: InsightRunOutputMetadata; } export interface InsightRunAttemptContext { run: InsightRun; attempt: number; maxAttempts: number; signal: AbortSignal; } export interface InsightRunExecutorOptions { store: InsightStore; projectId: string; input: InsightRunCreateInput; executeAttempt: (ctx: InsightRunAttemptContext) => Promise; timeoutMs?: number; maxAttempts?: number; retryDelayMs?: number; signal?: AbortSignal; } export interface InsightRunExecutorErrorClassification { failureClass: InsightRunFailureClass; retryable: boolean; terminalReason: "cancelled" | "failed" | "timed_out"; terminalCause: string; } function isAbortLike(error: unknown): boolean { return error instanceof DOMException && error.name === "AbortError"; } function asErrorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } export function classifyInsightRunError(error: unknown): InsightRunExecutorErrorClassification { if (isAbortLike(error)) { return { failureClass: "cancelled", retryable: false, terminalReason: "cancelled", terminalCause: asErrorMessage(error), }; } const message = asErrorMessage(error); if (/timeout|timed out|deadline/i.test(message)) { return { failureClass: "timed_out", retryable: true, terminalReason: "timed_out", terminalCause: message, }; } if (/ECONNRESET|ENOTFOUND|EAI_AGAIN|ETIMEDOUT|429|5\d\d/i.test(message)) { return { failureClass: "retryable_transient", retryable: true, terminalReason: "failed", terminalCause: message, }; } return { failureClass: "non_retryable", retryable: false, terminalReason: "failed", terminalCause: message, }; } function composeSignal(timeoutMs: number | undefined, parent?: AbortSignal): { signal: AbortSignal; clear: () => void } { const controller = new AbortController(); const timeoutId = timeoutMs && timeoutMs > 0 ? setTimeout(() => controller.abort(new Error(`Insight run timed out after ${timeoutMs}ms`)), timeoutMs) : undefined; const onAbort = () => { controller.abort(parent?.reason ?? new DOMException("Aborted", "AbortError")); }; if (parent) { if (parent.aborted) onAbort(); else parent.addEventListener("abort", onAbort, { once: true }); } return { signal: controller.signal, clear: () => { if (timeoutId) clearTimeout(timeoutId); if (parent) parent.removeEventListener("abort", onAbort); }, }; } function patchForStatus(status: "completed" | "failed" | "cancelled", patch: InsightRunUpdateInput): InsightRunUpdateInput { if (status === "cancelled") { return { ...patch, cancelledAt: patch.cancelledAt ?? new Date().toISOString(), }; } return patch; } async function executeExistingRun( store: InsightStore, run: InsightRun, options: Omit & { maxAttempts: number; retryDelayMs: number }, ): Promise { const started = store.updateRun(run.id, { status: "running", startedAt: run.startedAt ?? new Date().toISOString(), lifecycle: { ...run.lifecycle, maxAttempts: options.maxAttempts, attempt: run.lifecycle.attempt ?? 1, }, }); let active = started ?? run; store.appendRunEvent(active.id, { type: "status_changed", status: "running", message: "Run started" }); for (let attempt = active.lifecycle.attempt ?? 1; attempt <= options.maxAttempts; attempt += 1) { const { signal, clear } = composeSignal(options.timeoutMs, options.signal); try { if (signal.aborted) { throw signal.reason instanceof Error ? signal.reason : new DOMException("Aborted", "AbortError"); } store.appendRunEvent(active.id, { type: "info", message: `Attempt ${attempt}/${options.maxAttempts}`, metadata: { attempt, maxAttempts: options.maxAttempts }, }); const result = await options.executeAttempt({ run: active, attempt, maxAttempts: options.maxAttempts, signal }); const completed = store.updateRun(active.id, { status: "completed", summary: result.summary ?? null, insightsCreated: result.insightsCreated, insightsUpdated: result.insightsUpdated, outputMetadata: result.outputMetadata, lifecycle: { ...active.lifecycle, attempt, maxAttempts: options.maxAttempts, terminalReason: "completed", retryable: false, }, }); if (!completed) throw new Error(`Run disappeared while completing: ${active.id}`); store.appendRunEvent(completed.id, { type: "status_changed", status: "completed", message: "Run completed" }); return completed; } catch (error) { const classification = classifyInsightRunError(error); const canRetry = classification.retryable && attempt < options.maxAttempts; store.appendRunEvent(active.id, { type: canRetry ? "retry_scheduled" : "error", status: canRetry ? "running" : classification.terminalReason === "cancelled" ? "cancelled" : "failed", classification: classification.failureClass, message: canRetry ? `Attempt ${attempt} failed (${classification.failureClass}); retrying` : `Run failed (${classification.failureClass})`, metadata: { attempt, maxAttempts: options.maxAttempts, error: asErrorMessage(error) }, }); if (canRetry) { active = store.updateRun(active.id, { lifecycle: { ...active.lifecycle, attempt: attempt + 1, maxAttempts: options.maxAttempts, failureClass: classification.failureClass, retryable: true, }, }) ?? active; if (options.retryDelayMs > 0) { await delay(options.retryDelayMs, undefined, { signal: options.signal }); } continue; } const terminalStatus = classification.terminalReason === "cancelled" ? "cancelled" : "failed"; const terminal = store.updateRun(active.id, patchForStatus(terminalStatus, { status: terminalStatus, error: asErrorMessage(error), lifecycle: { ...active.lifecycle, attempt, maxAttempts: options.maxAttempts, terminalReason: classification.terminalReason, terminalCause: classification.terminalCause, failureClass: classification.failureClass, retryable: classification.failureClass === "retryable_transient", timeoutAt: classification.failureClass === "timed_out" ? new Date().toISOString() : active.lifecycle.timeoutAt, }, })); if (!terminal) throw new Error(`Run disappeared while failing: ${active.id}`); return terminal; } finally { clear(); } } const failed = store.updateRun(active.id, { status: "failed", error: "Run exhausted attempts", lifecycle: { ...active.lifecycle, terminalReason: "failed", terminalCause: "Run exhausted attempts", failureClass: "non_retryable", retryable: false, attempt: options.maxAttempts, maxAttempts: options.maxAttempts, }, }); if (!failed) throw new Error(`Run disappeared after attempts exhausted: ${active.id}`); return failed; } export async function executeInsightRunLifecycle(options: InsightRunExecutorOptions): Promise { const maxAttempts = Math.max(1, options.maxAttempts ?? 2); const retryDelayMs = Math.max(0, options.retryDelayMs ?? 250); let run: InsightRun; try { run = options.store.createRunOrThrowConflict(options.projectId, { ...options.input, lifecycle: { ...options.input.lifecycle, attempt: options.input.lifecycle?.attempt ?? 1, maxAttempts, rootRunId: options.input.lifecycle?.rootRunId, }, }); } catch (error) { if (error instanceof InsightLifecycleError && error.code === "active_run_conflict") { throw error; } throw error; } options.store.appendRunEvent(run.id, { type: "status_changed", status: "pending", message: "Run created", }); return executeExistingRun(options.store, run, { ...options, maxAttempts, retryDelayMs, }); } export async function retryInsightRunLifecycle( options: Omit & { runId: string; trigger?: InsightRunTrigger; inputMetadata?: InsightRunCreateInput["inputMetadata"] }, ): Promise<{ run: InsightRun; retryOf: InsightRun }> { const original = options.store.getRun(options.runId); if (!original) { throw new Error(`Insight run not found: ${options.runId}`); } if (original.status !== "failed") { throw new InsightLifecycleError(`Run ${original.id} must be failed to retry`, "not_retryable"); } if (!original.lifecycle.retryable || original.lifecycle.failureClass !== "retryable_transient") { throw new InsightLifecycleError(`Run ${original.id} is non-retryable`, "not_retryable"); } const run = await executeInsightRunLifecycle({ ...options, projectId: original.projectId, input: { trigger: options.trigger ?? original.trigger, inputMetadata: options.inputMetadata ?? original.inputMetadata, lifecycle: { retryOfRunId: original.id, rootRunId: original.lifecycle.rootRunId ?? original.id, attempt: 1, }, }, }); return { run, retryOf: original }; }