Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions packages/workflows/src/extension/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1795,7 +1795,11 @@ export function makeExecuteWorkflowTool(
const isPaused =
run?.status === "paused" ||
(run?.stages.some((s) => s.status === "paused") ?? false);
if (!isPaused && run?.status === "failed" && run.endedAt !== undefined && run.resumable !== false) {
const isResumableContinuation = run !== undefined && !isPaused && (
(run.status === "failed" && run.endedAt !== undefined && run.resumable !== false) ||
(run.endedAt === undefined && run.resumable === true && run.failureRecoverability === "recoverable")
);
if (isResumableContinuation) {
const continuation = activeRuntime.resumeFailedRun(stageRunId, stage.stageId, { policy });
return {
action: "resume",
Expand Down Expand Up @@ -3133,7 +3137,11 @@ function factory(pi: ExtensionAPI): void {
const isPaused =
run?.status === "paused" ||
(run?.stages.some((s) => s.status === "paused") ?? false);
if (!isPaused && run?.status === "failed" && run.endedAt !== undefined && run.resumable !== false) {
const isResumableContinuation = run !== undefined && !isPaused && (
(run.status === "failed" && run.endedAt !== undefined && run.resumable !== false) ||
(run.endedAt === undefined && run.resumable === true && run.failureRecoverability === "recoverable")
);
if (isResumableContinuation) {
const continuation = runtimeForContext(ctx).resumeFailedRun(stageRunId, stageId, { policy });
if (continuation.ok) {
print(continuation.message);
Expand Down
38 changes: 35 additions & 3 deletions packages/workflows/src/extension/runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ import {
import { validateWorkflowModels } from "../runs/shared/model-fallback.js";
import { runDetached } from "../runs/background/runner.js";
import type { JobTracker } from "../runs/background/job-tracker.js";
import { appendRunEnd } from "../shared/persistence-session-entries.js";
import { classifyWorkflowFailure } from "../shared/workflow-failures.js";

// ---------------------------------------------------------------------------
Expand Down Expand Up @@ -389,13 +390,39 @@ export function createExtensionRuntime(opts: ExtensionRuntimeOpts = {}): Extensi
return { ok: true, stageId: failedStageId };
}

function finalizeResumedActiveBlockedSourceRun(source: RunSnapshot, continuationRunId: string): void {
const errorMessage = source.error ?? source.failureMessage ?? `workflow resumed in new run ${continuationRunId}`;
const metadata = {
...(source.failureKind !== undefined ? { failureKind: source.failureKind } : {}),
...(source.failureCode !== undefined ? { failureCode: source.failureCode } : {}),
failureRecoverability: "non_recoverable",
failureDisposition: "terminal_killed",
...(source.failureMessage !== undefined ? { failureMessage: source.failureMessage } : {}),
...(source.failedStageId !== undefined ? { failedStageId: source.failedStageId } : {}),
resumable: false,
...(source.retryAfterMs !== undefined ? { retryAfterMs: source.retryAfterMs } : {}),
} as const;
const recorded = activeStore.recordRunEnd(source.id, "killed", undefined, errorMessage, metadata);
if (recorded && persistence !== undefined) {
appendRunEnd(persistence, {
runId: source.id,
status: "killed",
error: errorMessage,
...metadata,
ts: Date.now(),
});
}
}

function resumeFailedRun(sourceRunId: string, stageId?: string, options?: RuntimeDispatchOptions): ResumeFailedRunResult {
const source = activeStore.runs().find((run) => run.id === sourceRunId);
if (source === undefined) {
return { ok: false, reason: "run_not_found", message: `run not found: ${sourceRunId}` };
}
if (source.status !== "failed" || source.endedAt === undefined || source.resumable === false) {
return { ok: false, reason: "not_resumable", message: `run ${sourceRunId} is not a failed resumable workflow run` };
const isTerminalFailedResumable = source.status === "failed" && source.endedAt !== undefined && source.resumable !== false;
const isActiveBlockedResumable = source.endedAt === undefined && source.resumable === true && source.failureRecoverability === "recoverable";
if (!isTerminalFailedResumable && !isActiveBlockedResumable) {
return { ok: false, reason: "not_resumable", message: `run ${sourceRunId} is not a resumable workflow run` };
}
const def = registry.get(source.name);
if (def === undefined) {
Expand All @@ -415,12 +442,17 @@ export function createExtensionRuntime(opts: ExtensionRuntimeOpts = {}): Extensi
...runOptions({ workflow: def.name, inputs: sourceInputs }, options?.policy),
continuation: { source, resumeFromStageId: resolvedStage.stageId },
});
if (isActiveBlockedResumable) {
finalizeResumedActiveBlockedSourceRun(source, accepted.runId);
}
return {
ok: true,
runId: accepted.runId,
sourceRunId: source.id,
resumeFromStageId: resolvedStage.stageId,
message: `Resuming failed workflow "${def.name}" from run ${source.id.slice(0, 8)} at stage ${resolvedStage.stageId.slice(0, 8)} (new run ${accepted.runId}).`,
message: isActiveBlockedResumable
? `Resuming blocked workflow "${def.name}" from run ${source.id.slice(0, 8)} at stage ${resolvedStage.stageId.slice(0, 8)} (new run ${accepted.runId}).`
: `Resuming failed workflow "${def.name}" from run ${source.id.slice(0, 8)} at stage ${resolvedStage.stageId.slice(0, 8)} (new run ${accepted.runId}).`,
};
}

Expand Down
58 changes: 52 additions & 6 deletions packages/workflows/src/runs/background/status.ts
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,14 @@ export interface RunDetail {
readonly stages: readonly RunSnapshot["stages"][number][];
readonly result?: WorkflowOutputValues;
readonly error?: string;
readonly failureKind?: RunSnapshot["failureKind"];
readonly failureCode?: RunSnapshot["failureCode"];
readonly failureRecoverability?: RunSnapshot["failureRecoverability"];
readonly failureDisposition?: RunSnapshot["failureDisposition"];
readonly failedStageId?: string;
readonly resumable?: boolean;
readonly retryAfterMs?: number;
readonly blockedAt?: number;
}

export type InspectRunResult =
Expand Down Expand Up @@ -146,11 +154,26 @@ export function killRun(
const previousStatus = run.status;

// Abort active executor (no-op if not registered)
opts?.cancellation?.abort(runId, "workflow killed");

const recorded = activeStore.recordRunEnd(runId, "killed", undefined, "workflow killed");
const errorMessage = "workflow killed";
opts?.cancellation?.abort(runId, errorMessage);

const metadata = {
failureKind: "cancelled",
failureCode: "cancelled",
failureRecoverability: "non_recoverable",
failureDisposition: "terminal_killed",
failureMessage: errorMessage,
resumable: false,
} as const;
const recorded = activeStore.recordRunEnd(runId, "killed", undefined, errorMessage, metadata);
if (recorded && opts?.persistence) {
appendRunEnd(opts.persistence, { runId, status: "killed", ts: Date.now() });
appendRunEnd(opts.persistence, {
runId,
status: "killed",
error: errorMessage,
...metadata,
ts: Date.now(),
});
}

return { ok: true, runId, previousStatus };
Expand Down Expand Up @@ -247,14 +270,29 @@ export function resumeRun(
// Return a deep copy of the snapshot for safe consumption
const snapshot = structuredClone(run);
const resumedCopy = structuredClone(resumed);
if (run.status === "failed" && run.endedAt !== undefined && run.resumable === false) {
if (run.status === "killed" || run.resumable === false) {
return {
ok: true,
runId,
snapshot,
resumed: resumedCopy,
mode: "not_resumable",
message: "This failed workflow is not resumable; inspect the snapshot and rerun the workflow when ready.",
message: "This workflow is not resumable; inspect the snapshot and start a new workflow run when ready.",
};
}
if (
run.endedAt === undefined &&
run.resumable === true &&
run.failureRecoverability === "recoverable" &&
run.failedStageId !== undefined
) {
return {
ok: true,
runId,
snapshot,
resumed: resumedCopy,
mode: resumedCopy.length > 0 ? "paused" : "snapshot",
message: `Workflow is blocked on a recoverable ${run.failureCode ?? run.failureKind ?? "workflow"} failure at stage ${run.failedStageId}; retry/resume after the issue clears.`,
};
}
return {
Expand Down Expand Up @@ -441,6 +479,14 @@ export function inspectRun(
stages: expandedStages.map((stage) => structuredClone(stage)),
result: copy.result,
error: copy.error,
failureKind: copy.failureKind,
failureCode: copy.failureCode,
failureRecoverability: copy.failureRecoverability,
failureDisposition: copy.failureDisposition,
failedStageId: copy.failedStageId,
resumable: copy.resumable,
retryAfterMs: copy.retryAfterMs,
blockedAt: copy.blockedAt,
};

return { ok: true, runId: copy.id, detail };
Expand Down
Loading
Loading