onChange(event.target.value)}
/>
),
}));
-import { useSidebarSaveStateStore } from "@/store/SidebarSaveStateStore";
import { usePendingCommitsStore } from "@/store/PendingCommitsStore";
+import { useSidebarSaveStateStore } from "@/store/SidebarSaveStateStore";
import { ConditionalBlockForm } from "./ConditionalBlockForm";
beforeEach(() => {
vi.useFakeTimers();
mockNodes.clear();
updateNodeData.mockReset();
+ isWorkflowBlockNodeMock.mockReset();
+ isWorkflowBlockNodeMock.mockImplementation(
+ (node) => node.type !== "nodeAdder" && node.type !== "start",
+ );
usePendingCommitsStore.setState({ commits: {} });
useSidebarSaveStateStore.getState().reset();
});
@@ -75,7 +103,7 @@ function setConditionalNode(
id: "branch_a",
criteria: {
criteria_type: "jinja2_template",
- expression: "",
+ expression: "{{ total > 100 }}",
description: null,
},
next_block_label: null,
@@ -97,7 +125,7 @@ function setConditionalNode(
});
}
-describe("ConditionalBlockForm (SKY-9361)", () => {
+describe("ConditionalBlockForm", () => {
test("returns null for missing node", () => {
const { container } = render(
);
expect(container.firstChild).toBeNull();
@@ -109,14 +137,121 @@ describe("ConditionalBlockForm (SKY-9361)", () => {
expect(container.firstChild).toBeNull();
});
- test("renders BranchesEditor with the blockId and node data", () => {
+ test("returns null for non-workflow conditional nodes", () => {
+ isWorkflowBlockNodeMock.mockReturnValue(false);
+ setConditionalNode("c1");
+
+ const { container } = render(
);
+
+ expect(container.firstChild).toBeNull();
+ expect(isWorkflowBlockNodeMock).toHaveBeenCalledWith(
+ expect.objectContaining({ type: "conditional" }),
+ );
+ });
+
+ test("renders branch prompts and the default branch in the sidebar", () => {
setConditionalNode("c1");
render(
);
- const editor = screen.getByTestId("branches-editor");
- expect(editor.getAttribute("data-node-id")).toBe("c1");
- expect(editor.getAttribute("data-branches-count")).toBe("2");
- expect(editor.getAttribute("data-active-branch")).toBe("branch_a");
+ expect(screen.getByText("A • If")).toBeDefined();
+ expect(screen.getByText("B • Else")).toBeDefined();
+ expect(screen.getByText("Active")).toBeDefined();
+ expect(screen.getByDisplayValue("{{ total > 100 }}")).toBeDefined();
+ expect(
+ (
+ screen.getByDisplayValue(
+ "Executed when no other condition matches",
+ ) as HTMLTextAreaElement
+ ).disabled,
+ ).toBe(true);
+ });
+
+ test("preserves every default branch when ordering branch prompts", () => {
+ setConditionalNode("c1", {
+ branches: [
+ {
+ id: "branch_a",
+ criteria: {
+ criteria_type: "jinja2_template",
+ expression: "{{ total > 100 }}",
+ description: null,
+ },
+ next_block_label: null,
+ description: null,
+ is_default: false,
+ },
+ {
+ id: "branch_default_1",
+ criteria: null,
+ next_block_label: null,
+ description: null,
+ is_default: true,
+ },
+ {
+ id: "branch_b",
+ criteria: {
+ criteria_type: "jinja2_template",
+ expression: "{{ total > 250 }}",
+ description: null,
+ },
+ next_block_label: null,
+ description: null,
+ is_default: false,
+ },
+ {
+ id: "branch_default_2",
+ criteria: null,
+ next_block_label: null,
+ description: null,
+ is_default: true,
+ },
+ ],
+ });
+
+ render(
);
+
+ expect(screen.getByText("A • If")).toBeDefined();
+ expect(screen.getByText("B • Else If")).toBeDefined();
+ expect(screen.getByText("C • Else")).toBeDefined();
+ expect(screen.getByText("D • Else")).toBeDefined();
+ expect(
+ screen.getAllByDisplayValue("Executed when no other condition matches"),
+ ).toHaveLength(2);
+ });
+
+ test("editing a branch prompt propagates via updateNodeData", () => {
+ setConditionalNode("c1");
+ render(
);
+
+ fireEvent.change(screen.getByDisplayValue("{{ total > 100 }}"), {
+ target: { value: "{{ total > 250 }}" },
+ });
+
+ expect(updateNodeData).toHaveBeenCalledWith("c1", {
+ branches: [
+ expect.objectContaining({
+ id: "branch_a",
+ criteria: expect.objectContaining({
+ expression: "{{ total > 250 }}",
+ }),
+ }),
+ expect.objectContaining({
+ id: "branch_default",
+ criteria: null,
+ }),
+ ],
+ });
+ });
+
+ test("non-editable branch prompts do not propagate edits", () => {
+ setConditionalNode("c1", { editable: false });
+ render(
);
+
+ fireEvent.change(screen.getByDisplayValue("{{ total > 100 }}"), {
+ target: { value: "{{ total > 250 }}" },
+ });
+
+ expect(updateNodeData).not.toHaveBeenCalled();
});
test("registers/unregisters commit on mount/unmount", () => {
@@ -137,19 +272,4 @@ describe("ConditionalBlockForm (SKY-9361)", () => {
});
expect(ok).toBe(true);
});
-
- test("propagates updated activeBranchId to BranchesEditor on rerender", () => {
- setConditionalNode("c1", { activeBranchId: "branch_a" });
- const { rerender } = render(
);
- expect(
- screen.getByTestId("branches-editor").getAttribute("data-active-branch"),
- ).toBe("branch_a");
-
- setConditionalNode("c1", { activeBranchId: "branch_default" });
- rerender(
);
-
- expect(
- screen.getByTestId("branches-editor").getAttribute("data-active-branch"),
- ).toBe("branch_default");
- });
});
diff --git a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/ConditionalBlockForm.tsx b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/ConditionalBlockForm.tsx
index f9985856f..26e826042 100644
--- a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/ConditionalBlockForm.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/ConditionalBlockForm.tsx
@@ -1,38 +1,41 @@
-import { useReactFlow } from "@xyflow/react";
+import { useNodesData } from "@xyflow/react";
import { useEffect, useMemo } from "react";
+import { WorkflowBlockInputTextarea } from "@/components/WorkflowBlockInputTextarea";
+import { Label } from "@/components/ui/label";
import { usePendingCommitsStore } from "@/store/PendingCommitsStore";
-import { AppNode, isWorkflowBlockNode } from "../../nodes";
-import { BranchesEditor } from "../../nodes/ConditionalNode/BranchesEditor";
+import { type AppNode, isWorkflowBlockNode } from "../../nodes";
+import {
+ getConditionLabel,
+ orderBranchesWithDefaultsLast,
+} from "../../nodes/ConditionalNode/branchDisplayUtils";
import {
- type ConditionalNode,
type ConditionalNodeData,
+ defaultBranchCriteria,
} from "../../nodes/ConditionalNode/types";
+import { useUpdate } from "../../useUpdate";
import { useDebouncedSidebarSave } from "../useDebouncedSidebarSave";
function ConditionalBlockForm({ blockId }: { blockId: string }) {
- const rf = useReactFlow
();
- const node = rf.getNode(blockId);
- if (!node || !isWorkflowBlockNode(node) || node.type !== "conditional") {
+ const nodeSlice = useNodesData(blockId);
+ if (
+ !nodeSlice ||
+ !isWorkflowBlockNode(nodeSlice as AppNode) ||
+ nodeSlice.type !== "conditional"
+ ) {
return null;
}
- return (
-
- );
+ return ;
}
function ConditionalBlockFormBody({
blockId,
- node,
+ data,
}: {
blockId: string;
- node: ConditionalNode;
+ data: ConditionalNodeData;
}) {
- const data = node.data;
const {
branches,
activeBranchId,
@@ -40,6 +43,14 @@ function ConditionalBlockFormBody({
continueOnFailure,
nextLoopOnFailure,
} = data;
+ const orderedBranches = useMemo(
+ () => orderBranchesWithDefaultsLast(branches),
+ [branches],
+ );
+ const update = useUpdate({
+ id: blockId,
+ editable: data.editable,
+ });
const value = useMemo(
() => ({
@@ -68,9 +79,66 @@ function ConditionalBlockFormBody({
return () => store.unregister(blockId);
}, [blockId, commit]);
+ const handleExpressionChange = (
+ branchId: string,
+ expression: string,
+ ): void => {
+ const targetBranch = branches.find((branch) => branch.id === branchId);
+ if (!targetBranch || targetBranch.is_default) {
+ return;
+ }
+
+ update({
+ branches: branches.map((branch) => {
+ if (branch.id !== branchId) {
+ return branch;
+ }
+ return {
+ ...branch,
+ criteria: {
+ ...(branch.criteria ?? { ...defaultBranchCriteria }),
+ expression,
+ },
+ };
+ }),
+ });
+ };
+
return (
-
-
+
+ {orderedBranches.map((branch, index) => {
+ const isDefaultBranch = branch.is_default;
+ const branchExpression = isDefaultBranch
+ ? "Executed when no other condition matches"
+ : (branch.criteria?.expression ?? "");
+ return (
+
+
+
+ {branch.id === activeBranchId && (
+
+ Active
+
+ )}
+
+
+ handleExpressionChange(branch.id, nextValue)
+ }
+ placeholder="Enter condition to evaluate (Jinja, natural language, or both)"
+ className="nopan text-xs"
+ />
+
+ );
+ })}
);
}
diff --git a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.test.tsx b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.test.tsx
index 37f08ba61..c0e6fa704 100644
--- a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.test.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.test.tsx
@@ -35,14 +35,16 @@ vi.mock("@/components/WorkflowBlockInput", () => ({
WorkflowBlockInput: ({
value,
onChange,
+ "data-testid": dataTestId,
}: {
value: string;
onChange: (v: string) => void;
nodeId: string;
className?: string;
+ "data-testid"?: string;
}) => (
onChange(e.target.value)}
/>
@@ -114,11 +116,15 @@ afterEach(() => {
function setLoopNode(
id: string,
data: Partial<{
+ loopKind: "for_each" | "while";
loopVariableReference: string;
dataSchema: string;
completeIfEmpty: boolean;
continueOnFailure: boolean;
nextLoopOnFailure: boolean | undefined;
+ whileConditionExpression: string;
+ whileConditionCriteriaType: "jinja2_template" | "prompt";
+ whileConditionDescription: string | null;
editable: boolean;
}> = {},
) {
@@ -140,9 +146,11 @@ function setLoopNode(
// SKY-8771 introduced loopKind / whileCondition* fields. Default the
// fixture to the for-each branch so the existing test cases continue
// to exercise the for-each form fields.
- loopKind: "for_each",
- whileConditionExpression: "{{ true }}",
- whileConditionCriteriaType: "jinja2_template",
+ loopKind: data.loopKind ?? "for_each",
+ whileConditionExpression: data.whileConditionExpression ?? "{{ true }}",
+ whileConditionCriteriaType:
+ data.whileConditionCriteriaType ?? "jinja2_template",
+ whileConditionDescription: data.whileConditionDescription ?? null,
},
});
}
@@ -206,6 +214,28 @@ describe("LoopBlockForm (SKY-9361)", () => {
});
});
+ test("renders while loop condition and edits propagate via updateNodeData", () => {
+ setLoopNode("p1", {
+ loopKind: "while",
+ whileConditionExpression: "{{ count < 3 }}",
+ });
+ render(
);
+
+ expect(screen.getByText("Loop Condition")).toBeDefined();
+ expect(
+ (screen.getByTestId("while-condition-input") as HTMLInputElement).value,
+ ).toBe("{{ count < 3 }}");
+
+ fireEvent.change(screen.getByTestId("while-condition-input"), {
+ target: { value: "{{ count < 5 }}" },
+ });
+
+ expect(updateNodeData).toHaveBeenCalledWith("p1", {
+ whileConditionExpression: "{{ count < 5 }}",
+ whileConditionCriteriaType: "jinja2_template",
+ });
+ });
+
test("editing dataSchema propagates via updateNodeData", () => {
setLoopNode("p1");
render(
);
diff --git a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.tsx b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.tsx
index b2ee11b50..c6417863a 100644
--- a/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/panels/BlockConfigForm/LoopBlockForm.tsx
@@ -23,27 +23,39 @@ function LoopBlockFormBody({
node: LoopNode;
}) {
const {
+ loopKind,
loopVariableReference,
dataSchema,
completeIfEmpty,
continueOnFailure,
nextLoopOnFailure,
+ whileConditionExpression,
+ whileConditionCriteriaType,
+ whileConditionDescription,
} = node.data;
const value = useMemo(
() => ({
+ loopKind,
loopVariableReference,
dataSchema,
completeIfEmpty,
continueOnFailure,
nextLoopOnFailure,
+ whileConditionExpression,
+ whileConditionCriteriaType,
+ whileConditionDescription,
}),
[
+ loopKind,
loopVariableReference,
dataSchema,
completeIfEmpty,
continueOnFailure,
nextLoopOnFailure,
+ whileConditionExpression,
+ whileConditionCriteriaType,
+ whileConditionDescription,
],
);
const { commit } = useDebouncedSidebarSave({
diff --git a/skyvern-frontend/src/routes/workflows/editor/types.ts b/skyvern-frontend/src/routes/workflows/editor/types.ts
index bfef05db5..7b9864aa5 100644
--- a/skyvern-frontend/src/routes/workflows/editor/types.ts
+++ b/skyvern-frontend/src/routes/workflows/editor/types.ts
@@ -1,4 +1,7 @@
-import { WorkflowParameterValueType } from "../types/workflowTypes";
+import {
+ CredentialSelectionStrategy,
+ WorkflowParameterValueType,
+} from "../types/workflowTypes";
// Keys the Login block auto-generates when a user picks a credential directly
// (`credentials`, `credentials_1`, …). Used as the provenance signal that tells
@@ -20,6 +23,8 @@ export type SkyvernCredential = {
description?: string | null;
parameterType: "credential";
credentialId: string;
+ credentialIds?: Array
| null;
+ selectionStrategy?: CredentialSelectionStrategy | null;
dataType?: typeof WorkflowParameterValueType.CredentialId;
};
diff --git a/skyvern-frontend/src/routes/workflows/editor/utils.ts b/skyvern-frontend/src/routes/workflows/editor/utils.ts
index a85a9b3a5..b22d2bc22 100644
--- a/skyvern-frontend/src/routes/workflows/editor/utils.ts
+++ b/skyvern-frontend/src/routes/workflows/editor/utils.ts
@@ -69,6 +69,8 @@ const getInitialParameters = (workflow: WorkflowApiResponse) => {
key: parameter.key,
parameterType: WorkflowEditorParameterTypes.Credential,
credentialId: parameter.credential_id,
+ credentialIds: parameter.credential_ids ?? null,
+ selectionStrategy: parameter.selection_strategy ?? null,
description: parameter.description,
};
} else if (
diff --git a/skyvern-frontend/src/routes/workflows/editor/workflowEditorUtils.ts b/skyvern-frontend/src/routes/workflows/editor/workflowEditorUtils.ts
index b91e1b954..14f05b361 100644
--- a/skyvern-frontend/src/routes/workflows/editor/workflowEditorUtils.ts
+++ b/skyvern-frontend/src/routes/workflows/editor/workflowEditorUtils.ts
@@ -4055,6 +4055,8 @@ function convertParametersToParameterYAML(
...base,
parameter_type: WorkflowParameterTypes.Credential,
credential_id: parameter.credential_id,
+ credential_ids: parameter.credential_ids ?? null,
+ selection_strategy: parameter.selection_strategy ?? null,
};
}
case WorkflowParameterTypes.OnePassword: {
diff --git a/skyvern-frontend/src/routes/workflows/hooks/useWorkflowRunsQuery.ts b/skyvern-frontend/src/routes/workflows/hooks/useWorkflowRunsQuery.ts
index 248dbccfb..f190a1034 100644
--- a/skyvern-frontend/src/routes/workflows/hooks/useWorkflowRunsQuery.ts
+++ b/skyvern-frontend/src/routes/workflows/hooks/useWorkflowRunsQuery.ts
@@ -23,6 +23,8 @@ type Props = {
search?: string;
// ANDed with the internal gating (workflow id + globalWorkflows loaded).
enabled?: boolean;
+ createdAtStart?: string;
+ createdAtEnd?: string;
} & UseQueryOptions;
function useWorkflowRunsQuery({
@@ -32,6 +34,8 @@ function useWorkflowRunsQuery({
pageSize,
search,
enabled,
+ createdAtStart,
+ createdAtEnd,
...queryOptions
}: Props) {
const { data: globalWorkflows } = useGlobalWorkflowsQuery();
@@ -48,6 +52,8 @@ function useWorkflowRunsQuery({
page,
pageSize,
search,
+ createdAtStart,
+ createdAtEnd,
],
activeOrgQueryKeyScope,
),
@@ -72,6 +78,12 @@ function useWorkflowRunsQuery({
if (search) {
params.append("search_key", search);
}
+ if (createdAtStart) {
+ params.append("created_at_start", createdAtStart);
+ }
+ if (createdAtEnd) {
+ params.append("created_at_end", createdAtEnd);
+ }
return client
.get(`/workflows/${workflowPermanentId}/runs`, {
diff --git a/skyvern-frontend/src/routes/workflows/resolveRunWindow.test.ts b/skyvern-frontend/src/routes/workflows/resolveRunWindow.test.ts
new file mode 100644
index 000000000..72678a201
--- /dev/null
+++ b/skyvern-frontend/src/routes/workflows/resolveRunWindow.test.ts
@@ -0,0 +1,50 @@
+import { describe, expect, it } from "vitest";
+
+import { resolveRunWindow } from "./resolveRunWindow";
+
+const NOW = new Date("2026-06-08T12:00:00.000Z");
+
+describe("resolveRunWindow", () => {
+ it("returns an empty window when no period is set (pure-OSS default)", () => {
+ expect(resolveRunWindow(new URLSearchParams(), NOW)).toEqual({});
+ });
+
+ it("maps a preset to a midnight-snapped start with no upper bound", () => {
+ expect(resolveRunWindow(new URLSearchParams("period=7d"), NOW)).toEqual({
+ createdAtStart: "2026-06-01T00:00:00.000Z",
+ });
+ });
+
+ it("maps 30d to 30 calendar days (table approximation of the billing period)", () => {
+ expect(resolveRunWindow(new URLSearchParams("period=30d"), NOW)).toEqual({
+ createdAtStart: "2026-05-09T00:00:00.000Z",
+ });
+ });
+
+ it("resolves a valid custom range to inclusive whole-day bounds", () => {
+ expect(
+ resolveRunWindow(
+ new URLSearchParams("period=custom&from=2026-05-01&to=2026-05-03"),
+ NOW,
+ ),
+ ).toEqual({
+ createdAtStart: "2026-05-01T00:00:00.000Z",
+ createdAtEnd: "2026-05-04T00:00:00.000Z",
+ });
+ });
+
+ it("returns an empty window for an invalid custom range", () => {
+ expect(
+ resolveRunWindow(
+ new URLSearchParams("period=custom&from=2026-05-10&to=2026-05-01"),
+ NOW,
+ ),
+ ).toEqual({});
+ });
+
+ it("returns an empty window for an unknown preset", () => {
+ expect(resolveRunWindow(new URLSearchParams("period=14d"), NOW)).toEqual(
+ {},
+ );
+ });
+});
diff --git a/skyvern-frontend/src/routes/workflows/resolveRunWindow.ts b/skyvern-frontend/src/routes/workflows/resolveRunWindow.ts
new file mode 100644
index 000000000..93408fc02
--- /dev/null
+++ b/skyvern-frontend/src/routes/workflows/resolveRunWindow.ts
@@ -0,0 +1,62 @@
+export type RunWindow = {
+ createdAtStart?: string;
+ createdAtEnd?: string;
+};
+
+const PRESET_DAYS: Record = {
+ "7d": 7,
+ "30d": 30,
+ "90d": 90,
+ "365d": 365,
+};
+
+const ISO_DATE_RE = /^\d{4}-\d{2}-\d{2}$/;
+
+function isValidIsoDate(value: string): boolean {
+ if (!ISO_DATE_RE.test(value)) {
+ return false;
+ }
+ const parsed = new Date(`${value}T00:00:00Z`);
+ return (
+ !Number.isNaN(parsed.getTime()) &&
+ parsed.toISOString().slice(0, 10) === value
+ );
+}
+
+/**
+ * Derives the runs-table created-at window from the shared ?period= URL contract.
+ *
+ * Returns {} when no period is set so pure-OSS pages list all runs.
+ */
+export function resolveRunWindow(
+ searchParams: URLSearchParams,
+ now: Date = new Date(),
+): RunWindow {
+ const period = searchParams.get("period");
+ if (!period) {
+ return {};
+ }
+
+ if (period === "custom") {
+ const from = searchParams.get("from") ?? "";
+ const to = searchParams.get("to") ?? "";
+ if (!isValidIsoDate(from) || !isValidIsoDate(to) || from > to) {
+ return {};
+ }
+ const end = new Date(`${to}T00:00:00Z`);
+ end.setUTCDate(end.getUTCDate() + 1); // inclusive of the whole `to` day
+ return {
+ createdAtStart: new Date(`${from}T00:00:00Z`).toISOString(),
+ createdAtEnd: end.toISOString(),
+ };
+ }
+
+ const days = PRESET_DAYS[period];
+ if (!days) {
+ return {};
+ }
+ const start = new Date(now);
+ start.setUTCHours(0, 0, 0, 0);
+ start.setUTCDate(start.getUTCDate() - days);
+ return { createdAtStart: start.toISOString() };
+}
diff --git a/skyvern-frontend/src/routes/workflows/types/workflowTypes.ts b/skyvern-frontend/src/routes/workflows/types/workflowTypes.ts
index bb9ce1c7f..c566a4feb 100644
--- a/skyvern-frontend/src/routes/workflows/types/workflowTypes.ts
+++ b/skyvern-frontend/src/routes/workflows/types/workflowTypes.ts
@@ -85,11 +85,15 @@ export type AzureVaultCredentialParameter = WorkflowParameterBase & {
deleted_at: string | null;
};
+export type CredentialSelectionStrategy = "round_robin" | "random";
+
export type CredentialParameter = WorkflowParameterBase & {
parameter_type: "credential";
workflow_id: string;
credential_parameter_id: string;
credential_id: string;
+ credential_ids?: Array | null;
+ selection_strategy?: CredentialSelectionStrategy | null;
created_at: string;
modified_at: string;
deleted_at: string | null;
diff --git a/skyvern-frontend/src/routes/workflows/types/workflowYamlTypes.ts b/skyvern-frontend/src/routes/workflows/types/workflowYamlTypes.ts
index 68b1ac2aa..4607e7d85 100644
--- a/skyvern-frontend/src/routes/workflows/types/workflowYamlTypes.ts
+++ b/skyvern-frontend/src/routes/workflows/types/workflowYamlTypes.ts
@@ -1,6 +1,7 @@
import { ProxyLocation, RunEngine } from "@/api/types";
import {
CodeBlockStep,
+ CredentialSelectionStrategy,
WorkflowBlockType,
WorkflowModel,
} from "./workflowTypes";
@@ -129,6 +130,8 @@ export type OutputParameterYAML = ParameterYAMLBase & {
export type CredentialParameterYAML = ParameterYAMLBase & {
parameter_type: "credential";
credential_id: string;
+ credential_ids?: Array | null;
+ selection_strategy?: CredentialSelectionStrategy | null;
};
export type BlockYAML =
diff --git a/skyvern-frontend/src/store/PageSlots.ts b/skyvern-frontend/src/store/PageSlots.ts
new file mode 100644
index 000000000..e40723420
--- /dev/null
+++ b/skyvern-frontend/src/store/PageSlots.ts
@@ -0,0 +1,17 @@
+import { createContext, useContext } from "react";
+
+export type WorkflowAnalyticsPanelProps = {
+ workflowPermanentId: string;
+};
+
+export type PageSlots = {
+ workflowAnalyticsPanel?: React.ComponentType;
+};
+
+const PageSlotsContext = createContext({});
+
+export const PageSlotsProvider = PageSlotsContext.Provider;
+
+export function usePageSlots(): PageSlots {
+ return useContext(PageSlotsContext);
+}
diff --git a/skyvern/client/types/credential_parameter.py b/skyvern/client/types/credential_parameter.py
index 6d4705bdf..e06bb2d75 100644
--- a/skyvern/client/types/credential_parameter.py
+++ b/skyvern/client/types/credential_parameter.py
@@ -13,6 +13,8 @@ class CredentialParameter(UniversalBaseModel):
credential_parameter_id: str
workflow_id: str
credential_id: str
+ credential_ids: typing.Optional[typing.List[str]] = None
+ selection_strategy: typing.Optional[str] = None
created_at: dt.datetime
modified_at: dt.datetime
deleted_at: typing.Optional[dt.datetime] = None
diff --git a/skyvern/client/types/credential_parameter_yaml.py b/skyvern/client/types/credential_parameter_yaml.py
index aa8a22e10..bc5ca5ecf 100644
--- a/skyvern/client/types/credential_parameter_yaml.py
+++ b/skyvern/client/types/credential_parameter_yaml.py
@@ -10,6 +10,8 @@ class CredentialParameterYaml(UniversalBaseModel):
key: str
description: typing.Optional[str] = None
credential_id: str
+ credential_ids: typing.Optional[typing.List[str]] = None
+ selection_strategy: typing.Optional[str] = None
if IS_PYDANTIC_V2:
model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict(extra="allow", frozen=True) # type: ignore # Pydantic v2
diff --git a/skyvern/client/types/workflow_definition_parameters_item.py b/skyvern/client/types/workflow_definition_parameters_item.py
index a2cf40f66..527152aca 100644
--- a/skyvern/client/types/workflow_definition_parameters_item.py
+++ b/skyvern/client/types/workflow_definition_parameters_item.py
@@ -183,6 +183,8 @@ class WorkflowDefinitionParametersItem_Credential(UniversalBaseModel):
credential_parameter_id: str
workflow_id: str
credential_id: str
+ credential_ids: typing.Optional[typing.List[str]] = None
+ selection_strategy: typing.Optional[str] = None
created_at: dt.datetime
modified_at: dt.datetime
deleted_at: typing.Optional[dt.datetime] = None
diff --git a/skyvern/client/types/workflow_definition_yaml_parameters_item.py b/skyvern/client/types/workflow_definition_yaml_parameters_item.py
index 35603ba6d..2363b0275 100644
--- a/skyvern/client/types/workflow_definition_yaml_parameters_item.py
+++ b/skyvern/client/types/workflow_definition_yaml_parameters_item.py
@@ -128,6 +128,8 @@ class WorkflowDefinitionYamlParametersItem_Credential(UniversalBaseModel):
key: str
description: typing.Optional[str] = None
credential_id: str
+ credential_ids: typing.Optional[typing.List[str]] = None
+ selection_strategy: typing.Optional[str] = None
if IS_PYDANTIC_V2:
model_config: typing.ClassVar[pydantic.ConfigDict] = pydantic.ConfigDict(extra="allow", frozen=True) # type: ignore # Pydantic v2
diff --git a/skyvern/forge/sdk/copilot/output_policy.py b/skyvern/forge/sdk/copilot/output_policy.py
index b29e0c308..08581034f 100644
--- a/skyvern/forge/sdk/copilot/output_policy.py
+++ b/skyvern/forge/sdk/copilot/output_policy.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import re
+from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import StrEnum
from typing import Any, cast
@@ -22,7 +23,7 @@ from skyvern.forge.sdk.copilot.secret_redaction import (
)
from skyvern.forge.sdk.copilot.workflow_credential_utils import (
block_credential_ids,
- credential_params,
+ credential_param_ids,
parse_workflow_yaml,
url_origin,
workflow_blocks,
@@ -837,7 +838,7 @@ def _workflow_broadens_credential_scope(parsed_workflow: dict[str, Any], request
if not isinstance(workflow_definition, dict):
return False
- credential_params_by_key = credential_params(workflow_definition.get("parameters"))
+ credential_params_by_key = credential_param_ids(workflow_definition.get("parameters"))
if not credential_params_by_key:
return False
@@ -861,7 +862,7 @@ def _approved_origins_by_id(request_policy: RequestPolicy) -> dict[str, set[str]
def _block_broadens_credential_scope(
block: dict[str, Any],
- credential_params_by_key: dict[str, str],
+ credential_params_by_key: Mapping[str, str | set[str]],
approved_origins: dict[str, set[str]],
) -> bool:
credential_ids = block_credential_ids(block, credential_params_by_key)
diff --git a/skyvern/forge/sdk/copilot/tools/credentials.py b/skyvern/forge/sdk/copilot/tools/credentials.py
index faf023447..9119db9ea 100644
--- a/skyvern/forge/sdk/copilot/tools/credentials.py
+++ b/skyvern/forge/sdk/copilot/tools/credentials.py
@@ -73,6 +73,8 @@ def _extract_credential_ids_from_workflow_parameters(parameters: Any) -> list[st
if slot_field is None:
continue
found.extend(_extract_credential_ids_from_tool_value(parameter.get(slot_field)))
+ if slot_field == "credential_id":
+ found.extend(_extract_credential_ids_from_tool_value(parameter.get("credential_ids")))
return list(dict.fromkeys(found))
@@ -133,8 +135,11 @@ def _credential_id_misbinding_findings(workflow_yaml: str | None) -> list[dict[s
if not isinstance(parameter, dict):
return
legal_slot_field = _credential_parameter_slot_field(parameter)
+ legal_slot_fields = {legal_slot_field} if legal_slot_field else set()
+ if legal_slot_field == "credential_id":
+ legal_slot_fields.add("credential_ids")
for field_name, field_value in parameter.items():
- if field_name == legal_slot_field:
+ if field_name in legal_slot_fields:
continue
_scan_value(field_value, location, str(field_name))
diff --git a/skyvern/forge/sdk/copilot/tools/workflow_update.py b/skyvern/forge/sdk/copilot/tools/workflow_update.py
index 8fd25df5d..5003e9fdc 100644
--- a/skyvern/forge/sdk/copilot/tools/workflow_update.py
+++ b/skyvern/forge/sdk/copilot/tools/workflow_update.py
@@ -105,7 +105,11 @@ from skyvern.forge.sdk.copilot.turn_halt import (
blocker_signal_is_genuinely_terminal,
stash_turn_halt_from_blocker_signal,
)
-from skyvern.forge.sdk.copilot.workflow_credential_utils import credential_params, parse_workflow_yaml, workflow_blocks
+from skyvern.forge.sdk.copilot.workflow_credential_utils import (
+ credential_param_ids,
+ parse_workflow_yaml,
+ workflow_blocks,
+)
from skyvern.forge.sdk.routes.workflow_copilot import _process_workflow_yaml
from skyvern.forge.sdk.workflow.exceptions import BaseWorkflowHTTPException, InsecureCodeDetected
from skyvern.forge.sdk.workflow.models.block import CodeBlock
@@ -3564,7 +3568,7 @@ def _reconcile_synthesized_parameters(
existing_by_key = {
str(param.get("key")): param for param in parameters if isinstance(param, dict) and param.get("key")
}
- existing_credentials = credential_params(parameters)
+ existing_credentials = credential_param_ids(parameters)
parameter_keys: list[str] = []
violations: list[str] = []
aliases: dict[str, str] = {}
@@ -3614,7 +3618,7 @@ def _reconcile_synthesized_parameters(
typed_length = typed_length or scout_typed_length
if credential_id:
if existing is not None:
- if existing_credentials.get(key) != credential_id:
+ if credential_id not in existing_credentials.get(key, set()):
violations.append(
f"Unable to bind synthesized credential parameter `{key}`: submitted credential binding does not match scout metadata."
)
@@ -4794,7 +4798,7 @@ def _credentialed_code_block_scout_gate_errors(
workflow_definition = parsed.get("workflow_definition")
if not isinstance(workflow_definition, dict):
return []
- credential_params_by_key = credential_params(workflow_definition.get("parameters"))
+ credential_params_by_key = credential_param_ids(workflow_definition.get("parameters"))
if not credential_params_by_key:
return []
scout_trajectory = getattr(ctx, "scout_trajectory", None)
@@ -4808,25 +4812,28 @@ def _credentialed_code_block_scout_gate_errors(
code = str(block.get("code") or "")
if not code.strip():
continue
- required_fields_by_credential: dict[str, set[str]] = {}
+ required_fields_by_parameter: dict[str, tuple[set[str], set[str]]] = {}
for access in _credential_field_accesses(code):
if not access.requires_live_scout:
continue
- credential_id = credential_params_by_key.get(access.parameter_key)
- if credential_id:
- required_fields_by_credential.setdefault(credential_id, set()).add(access.field)
- if not required_fields_by_credential:
+ credential_ids = credential_params_by_key.get(access.parameter_key)
+ if credential_ids:
+ allowed_credential_ids, required_fields = required_fields_by_parameter.setdefault(
+ access.parameter_key, (credential_ids, set())
+ )
+ required_fields.add(access.field)
+ if not required_fields_by_parameter:
continue
matched_fill_indexes: list[int] = []
matched_source_urls: set[str] = set()
missing_fields: list[str] = []
- for credential_id, required_fields in required_fields_by_credential.items():
+ for allowed_credential_ids, required_fields in required_fields_by_parameter.values():
matched_fields: set[str] = set()
for index, interaction in enumerate(scout_trajectory):
if str(interaction.get("tool_name") or "").strip() != "fill_credential_field":
continue
- if str(interaction.get("credential_id") or "").strip() != credential_id:
+ if str(interaction.get("credential_id") or "").strip() not in allowed_credential_ids:
continue
field = str(interaction.get("credential_field") or "").strip()
if field not in required_fields:
diff --git a/skyvern/forge/sdk/copilot/workflow_credential_utils.py b/skyvern/forge/sdk/copilot/workflow_credential_utils.py
index f70ad3d64..cbdc0edf8 100644
--- a/skyvern/forge/sdk/copilot/workflow_credential_utils.py
+++ b/skyvern/forge/sdk/copilot/workflow_credential_utils.py
@@ -1,6 +1,7 @@
from __future__ import annotations
import re
+from collections.abc import Mapping
from typing import Any
from urllib.parse import urlparse
@@ -58,6 +59,36 @@ def credential_params(parameters: Any) -> dict[str, str]:
return out
+def credential_param_ids(parameters: Any) -> dict[str, set[str]]:
+ if not isinstance(parameters, list):
+ return {}
+ out: dict[str, set[str]] = {}
+ for parameter in parameters:
+ if not isinstance(parameter, dict):
+ continue
+ key = parameter.get("key")
+ if not isinstance(key, str):
+ continue
+ parameter_type = str(parameter.get("parameter_type") or "").lower()
+ workflow_parameter_type = str(parameter.get("workflow_parameter_type") or "").lower()
+ if parameter_type == "credential":
+ ids: set[str] = set()
+ credential_ids = parameter.get("credential_ids")
+ if isinstance(credential_ids, list):
+ ids.update(item for item in credential_ids if isinstance(item, str))
+ if isinstance(parameter.get("credential_id"), str):
+ ids.add(parameter["credential_id"])
+ if ids:
+ out[key] = ids
+ elif (
+ parameter_type == "workflow"
+ and workflow_parameter_type == "credential_id"
+ and isinstance(parameter.get("default_value"), str)
+ ):
+ out[key] = {parameter["default_value"]}
+ return out
+
+
def workflow_blocks(parsed: dict[str, Any]) -> list[dict[str, Any]]:
workflow_definition = parsed.get("workflow_definition")
if not isinstance(workflow_definition, dict):
@@ -97,13 +128,17 @@ def workflow_blocks(parsed: dict[str, Any]) -> list[dict[str, Any]]:
return collected
-def block_credential_ids(block: dict[str, Any], credential_params_by_key: dict[str, str]) -> set[str]:
+def block_credential_ids(block: dict[str, Any], credential_params_by_key: Mapping[str, str | set[str]]) -> set[str]:
credential_ids: set[str] = set()
parameter_keys = block.get("parameter_keys")
if isinstance(parameter_keys, list):
for key in parameter_keys:
if isinstance(key, str) and key in credential_params_by_key:
- credential_ids.add(credential_params_by_key[key])
+ ids = credential_params_by_key[key]
+ if isinstance(ids, str):
+ credential_ids.add(ids)
+ else:
+ credential_ids.update(ids)
direct_credential_id = block.get("credential_id")
if isinstance(direct_credential_id, str):
credential_ids.add(direct_credential_id)
@@ -124,8 +159,10 @@ def workflow_credential_ids_from_parsed(parsed: dict[str, Any]) -> set[str]:
if not isinstance(workflow_definition, dict):
return set()
- credential_params_by_key = credential_params(workflow_definition.get("parameters"))
- credential_ids = set(credential_params_by_key.values())
+ credential_params_by_key = credential_param_ids(workflow_definition.get("parameters"))
+ credential_ids: set[str] = set()
+ for ids in credential_params_by_key.values():
+ credential_ids.update(ids)
for block in workflow_blocks(parsed):
credential_ids.update(block_credential_ids(block, credential_params_by_key))
return credential_ids
@@ -145,7 +182,7 @@ def workflow_credential_origins_from_parsed(parsed: dict[str, Any]) -> dict[str,
if not isinstance(workflow_definition, dict):
return {}
- credential_params_by_key = credential_params(workflow_definition.get("parameters"))
+ credential_params_by_key = credential_param_ids(workflow_definition.get("parameters"))
origins_by_id: dict[str, set[str]] = {}
for block in workflow_blocks(parsed):
credential_ids = block_credential_ids(block, credential_params_by_key)
diff --git a/skyvern/forge/sdk/db/agent_db.py b/skyvern/forge/sdk/db/agent_db.py
index 1fa842a61..a382049e7 100644
--- a/skyvern/forge/sdk/db/agent_db.py
+++ b/skyvern/forge/sdk/db/agent_db.py
@@ -33,6 +33,9 @@ from skyvern.forge.sdk.db.repositories.scripts import ScriptsRepository
from skyvern.forge.sdk.db.repositories.tags import TagsRepository
from skyvern.forge.sdk.db.repositories.tasks import TasksRepository
from skyvern.forge.sdk.db.repositories.workflow_parameters import WorkflowParametersRepository
+from skyvern.forge.sdk.db.repositories.workflow_run_credential_selections import (
+ WorkflowRunCredentialSelectionsRepository,
+)
from skyvern.forge.sdk.db.repositories.workflow_runs import WorkflowRunsRepository
from skyvern.forge.sdk.db.repositories.workflows import WorkflowsRepository
from skyvern.forge.sdk.db.utils import (
@@ -366,6 +369,9 @@ class AgentDB(BaseAlchemyDB):
self.tasks = TasksRepository(self.Session, debug_enabled, self.is_retryable_error)
self.workflows = WorkflowsRepository(self.Session, debug_enabled, self.is_retryable_error)
self.workflow_params = WorkflowParametersRepository(self.Session, debug_enabled, self.is_retryable_error)
+ self.workflow_run_credential_selections = WorkflowRunCredentialSelectionsRepository(
+ self.Session, debug_enabled, self.is_retryable_error
+ )
self.credentials = CredentialRepository(self.Session, debug_enabled, self.is_retryable_error)
self.credential_folders = CredentialFoldersRepository(self.Session, debug_enabled, self.is_retryable_error)
self.otp = OTPRepository(self.Session, debug_enabled, self.is_retryable_error)
diff --git a/skyvern/forge/sdk/db/id.py b/skyvern/forge/sdk/db/id.py
index 2e7a33b31..1dfdd572e 100644
--- a/skyvern/forge/sdk/db/id.py
+++ b/skyvern/forge/sdk/db/id.py
@@ -75,6 +75,7 @@ ORGANIZATION_BILLING_PREFIX = "ob"
WORKFLOW_COPILOT_CHAT_PREFIX = "wcc"
WORKFLOW_COPILOT_CHAT_MESSAGE_PREFIX = "wccm"
WORKFLOW_COPILOT_COMPLETION_CRITERIA_SET_PREFIX = "wccs"
+WORKFLOW_RUN_CREDENTIAL_SELECTION_PREFIX = "wrcs"
SCRIPT_FALLBACK_EPISODE_PREFIX = "sfe"
WORKFLOW_SCHEDULE_PREFIX = "wfs"
TAG_EVENT_PREFIX = "tge"
@@ -322,6 +323,11 @@ def generate_workflow_copilot_completion_criteria_set_id() -> str:
return f"{WORKFLOW_COPILOT_COMPLETION_CRITERIA_SET_PREFIX}_{int_id}"
+def generate_workflow_run_credential_selection_id() -> str:
+ int_id = generate_id()
+ return f"{WORKFLOW_RUN_CREDENTIAL_SELECTION_PREFIX}_{int_id}"
+
+
def generate_script_fallback_episode_id() -> str:
int_id = generate_id()
return f"{SCRIPT_FALLBACK_EPISODE_PREFIX}_{int_id}"
diff --git a/skyvern/forge/sdk/db/models.py b/skyvern/forge/sdk/db/models.py
index e650c2d18..5fcdfdaa1 100644
--- a/skyvern/forge/sdk/db/models.py
+++ b/skyvern/forge/sdk/db/models.py
@@ -68,6 +68,7 @@ from skyvern.forge.sdk.db.id import (
generate_workflow_parameter_id,
generate_workflow_permanent_id,
generate_workflow_run_block_id,
+ generate_workflow_run_credential_selection_id,
generate_workflow_run_id,
generate_workflow_schedule_id,
generate_workflow_script_id,
@@ -823,12 +824,37 @@ class CredentialParameterModel(Base):
description = Column(String, nullable=True)
credential_id = Column(String, nullable=False)
+ credential_ids = Column(JSON, nullable=True)
+ selection_strategy = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
modified_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow, nullable=False)
deleted_at = Column(DateTime, nullable=True)
+class WorkflowRunCredentialSelectionModel(Base):
+ __tablename__ = "workflow_run_credential_selections"
+ __table_args__ = (
+ UniqueConstraint("workflow_run_id", "parameter_key", name="uq_wrcs_workflow_run_parameter_key"),
+ Index(
+ "idx_wrcs_lru_lookup",
+ "organization_id",
+ "workflow_permanent_id",
+ "parameter_key",
+ "credential_id",
+ "created_at",
+ ),
+ )
+
+ selection_id = Column(String, primary_key=True, default=generate_workflow_run_credential_selection_id)
+ organization_id = Column(String, nullable=False)
+ workflow_run_id = Column(String, nullable=False)
+ workflow_permanent_id = Column(String, nullable=False)
+ parameter_key = Column(String, nullable=False)
+ credential_id = Column(String, nullable=False)
+ created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
+
+
class OnePasswordCredentialParameterModel(Base):
__tablename__ = "onepassword_credential_parameters"
diff --git a/skyvern/forge/sdk/db/repositories/workflow_parameters.py b/skyvern/forge/sdk/db/repositories/workflow_parameters.py
index f6ac7f355..77c256664 100644
--- a/skyvern/forge/sdk/db/repositories/workflow_parameters.py
+++ b/skyvern/forge/sdk/db/repositories/workflow_parameters.py
@@ -227,6 +227,8 @@ class WorkflowParametersRepository(BaseRepository):
key=parameter.key,
description=parameter.description,
credential_id=parameter.credential_id,
+ credential_ids=parameter.credential_ids,
+ selection_strategy=parameter.selection_strategy,
deleted_at=parameter.deleted_at,
)
elif isinstance(parameter, OnePasswordCredentialParameter):
diff --git a/skyvern/forge/sdk/db/repositories/workflow_run_credential_selections.py b/skyvern/forge/sdk/db/repositories/workflow_run_credential_selections.py
new file mode 100644
index 000000000..d0b0cb608
--- /dev/null
+++ b/skyvern/forge/sdk/db/repositories/workflow_run_credential_selections.py
@@ -0,0 +1,164 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from skyvern.forge.sdk.db._error_handling import db_operation
+from skyvern.forge.sdk.db.base_repository import BaseRepository
+from skyvern.forge.sdk.db.models import WorkflowRunCredentialSelectionModel
+
+
+class WorkflowRunCredentialSelectionsRepository(BaseRepository):
+ @db_operation("get_workflow_run_credential_selection")
+ async def get_selection(self, workflow_run_id: str, parameter_key: str) -> str | None:
+ async with self.Session() as session:
+ selection = (
+ await session.scalars(
+ select(WorkflowRunCredentialSelectionModel)
+ .where(WorkflowRunCredentialSelectionModel.workflow_run_id == workflow_run_id)
+ .where(WorkflowRunCredentialSelectionModel.parameter_key == parameter_key)
+ )
+ ).first()
+ return selection.credential_id if selection else None
+
+ async def _get_selection(self, session: AsyncSession, workflow_run_id: str, parameter_key: str) -> str | None:
+ selection = (
+ await session.scalars(
+ select(WorkflowRunCredentialSelectionModel)
+ .where(WorkflowRunCredentialSelectionModel.workflow_run_id == workflow_run_id)
+ .where(WorkflowRunCredentialSelectionModel.parameter_key == parameter_key)
+ )
+ ).first()
+ return selection.credential_id if selection else None
+
+ @db_operation("get_latest_workflow_run_credential_selections")
+ async def get_latest_selections(
+ self,
+ *,
+ organization_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ ) -> dict[str, datetime]:
+ if not credential_ids:
+ return {}
+ async with self.Session() as session:
+ rows = (
+ await session.execute(
+ select(
+ WorkflowRunCredentialSelectionModel.credential_id,
+ func.max(WorkflowRunCredentialSelectionModel.created_at),
+ )
+ .where(WorkflowRunCredentialSelectionModel.organization_id == organization_id)
+ .where(WorkflowRunCredentialSelectionModel.workflow_permanent_id == workflow_permanent_id)
+ .where(WorkflowRunCredentialSelectionModel.parameter_key == parameter_key)
+ .where(WorkflowRunCredentialSelectionModel.credential_id.in_(credential_ids))
+ .group_by(WorkflowRunCredentialSelectionModel.credential_id)
+ )
+ ).all()
+ return {credential_id: created_at for credential_id, created_at in rows if created_at is not None}
+
+ async def _get_latest_selections(
+ self,
+ session: AsyncSession,
+ *,
+ organization_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ ) -> dict[str, datetime]:
+ if not credential_ids:
+ return {}
+ rows = (
+ await session.execute(
+ select(
+ WorkflowRunCredentialSelectionModel.credential_id,
+ func.max(WorkflowRunCredentialSelectionModel.created_at),
+ )
+ .where(WorkflowRunCredentialSelectionModel.organization_id == organization_id)
+ .where(WorkflowRunCredentialSelectionModel.workflow_permanent_id == workflow_permanent_id)
+ .where(WorkflowRunCredentialSelectionModel.parameter_key == parameter_key)
+ .where(WorkflowRunCredentialSelectionModel.credential_id.in_(credential_ids))
+ .group_by(WorkflowRunCredentialSelectionModel.credential_id)
+ )
+ ).all()
+ return {credential_id: created_at for credential_id, created_at in rows if created_at is not None}
+
+ async def _take_rotation_advisory_lock(self, session: AsyncSession, lock_key: str) -> None:
+ bind = session.get_bind()
+ dialect_name = bind.dialect.name if bind is not None else "postgresql"
+ if dialect_name not in {"postgresql", "postgres"}:
+ return
+ await session.execute(select(func.pg_advisory_xact_lock(func.hashtext(lock_key))))
+
+ @db_operation("create_round_robin_workflow_run_credential_selection", log_errors=False)
+ async def create_round_robin_selection(
+ self,
+ *,
+ organization_id: str,
+ workflow_run_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ ) -> str:
+ async with self.Session() as session:
+ lock_key = f"wrcs:{organization_id}:{workflow_permanent_id}:{parameter_key}"
+ # The lock, idempotency check, LRU read, and insert must stay in this transaction.
+ await self._take_rotation_advisory_lock(session, lock_key)
+
+ existing = await self._get_selection(
+ session,
+ workflow_run_id=workflow_run_id,
+ parameter_key=parameter_key,
+ )
+ if existing:
+ return existing
+
+ latest_selections = await self._get_latest_selections(
+ session,
+ organization_id=organization_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_ids=credential_ids,
+ )
+ unseen = next((candidate for candidate in credential_ids if candidate not in latest_selections), None)
+ credential_id = (
+ unseen
+ if unseen is not None
+ else min(credential_ids, key=lambda candidate: latest_selections[candidate])
+ )
+
+ selection = WorkflowRunCredentialSelectionModel(
+ organization_id=organization_id,
+ workflow_run_id=workflow_run_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_id=credential_id,
+ )
+ session.add(selection)
+ await session.commit()
+ return credential_id
+
+ @db_operation("create_workflow_run_credential_selection", log_errors=False)
+ async def create_selection(
+ self,
+ *,
+ organization_id: str,
+ workflow_run_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_id: str,
+ ) -> str:
+ async with self.Session() as session:
+ selection = WorkflowRunCredentialSelectionModel(
+ organization_id=organization_id,
+ workflow_run_id=workflow_run_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_id=credential_id,
+ )
+ session.add(selection)
+ await session.commit()
+ return credential_id
diff --git a/skyvern/forge/sdk/db/repositories/workflow_runs.py b/skyvern/forge/sdk/db/repositories/workflow_runs.py
index b186bb16d..af7cff558 100644
--- a/skyvern/forge/sdk/db/repositories/workflow_runs.py
+++ b/skyvern/forge/sdk/db/repositories/workflow_runs.py
@@ -1155,6 +1155,8 @@ class WorkflowRunsRepository(BaseRepository):
search_key: str | None = None,
error_code: str | None = None,
exclude_child_runs: bool = False,
+ created_at_start: datetime | None = None,
+ created_at_end: datetime | None = None,
) -> list[WorkflowRun]:
"""
Get runs for a workflow, with optional `search_key` on run ID, parameter key/description/value,
@@ -1175,6 +1177,10 @@ class WorkflowRunsRepository(BaseRepository):
query = self._apply_error_code_filter(query, error_code)
if status:
query = query.filter(WorkflowRunModel.status.in_(status))
+ if created_at_start is not None:
+ query = query.filter(WorkflowRunModel.created_at >= created_at_start)
+ if created_at_end is not None:
+ query = query.filter(WorkflowRunModel.created_at < created_at_end)
query = query.order_by(WorkflowRunModel.created_at.desc()).limit(page_size).offset(db_page * page_size)
workflow_runs_and_titles_tuples = (await session.execute(query)).all()
workflow_runs = [
diff --git a/skyvern/forge/sdk/routes/agent_protocol.py b/skyvern/forge/sdk/routes/agent_protocol.py
index 9b1bbfa85..4ad79b3c4 100644
--- a/skyvern/forge/sdk/routes/agent_protocol.py
+++ b/skyvern/forge/sdk/routes/agent_protocol.py
@@ -3866,6 +3866,8 @@ async def _get_workflow_runs_by_id(
search_key: str | None,
error_code: str | None,
exclude_child_runs: bool,
+ created_at_start: datetime | None = None,
+ created_at_end: datetime | None = None,
) -> list[WorkflowRun]:
analytics.capture("skyvern-oss-agent-workflow-runs-get")
return await app.WORKFLOW_SERVICE.get_workflow_runs_for_workflow_permanent_id(
@@ -3877,6 +3879,8 @@ async def _get_workflow_runs_by_id(
search_key=search_key,
error_code=error_code,
exclude_child_runs=exclude_child_runs,
+ created_at_start=created_at_start,
+ created_at_end=created_at_end,
)
@@ -3921,6 +3925,14 @@ async def get_workflow_runs_by_id(
),
examples=["INVALID_CREDENTIALS", "LOGIN_FAILED", "CAPTCHA_DETECTED"],
),
+ created_at_start: Annotated[
+ datetime | None,
+ Query(description="Only include runs created at or after this UTC timestamp (ISO 8601)."),
+ ] = None,
+ created_at_end: Annotated[
+ datetime | None,
+ Query(description="Only include runs created strictly before this UTC timestamp (ISO 8601)."),
+ ] = None,
current_org: Organization = Depends(org_auth_service.get_current_org),
) -> list[WorkflowRun]:
"""
@@ -3937,6 +3949,8 @@ async def get_workflow_runs_by_id(
status=status,
search_key=search_key,
error_code=error_code,
+ created_at_start=created_at_start,
+ created_at_end=created_at_end,
exclude_child_runs=True,
)
@@ -3982,6 +3996,14 @@ async def get_workflow_runs_by_id_legacy(
),
examples=["INVALID_CREDENTIALS", "LOGIN_FAILED", "CAPTCHA_DETECTED"],
),
+ created_at_start: Annotated[
+ datetime | None,
+ Query(description="Only include runs created at or after this UTC timestamp (ISO 8601)."),
+ ] = None,
+ created_at_end: Annotated[
+ datetime | None,
+ Query(description="Only include runs created strictly before this UTC timestamp (ISO 8601)."),
+ ] = None,
current_org: Organization = Depends(org_auth_service.get_current_org),
) -> list[WorkflowRun]:
"""
@@ -3997,6 +4019,8 @@ async def get_workflow_runs_by_id_legacy(
status=status,
search_key=search_key,
error_code=error_code,
+ created_at_start=created_at_start,
+ created_at_end=created_at_end,
exclude_child_runs=False,
)
diff --git a/skyvern/forge/sdk/workflow/context_manager.py b/skyvern/forge/sdk/workflow/context_manager.py
index 4bd290865..bf03d0cab 100644
--- a/skyvern/forge/sdk/workflow/context_manager.py
+++ b/skyvern/forge/sdk/workflow/context_manager.py
@@ -33,6 +33,7 @@ from skyvern.forge.sdk.schemas.organizations import Organization
from skyvern.forge.sdk.schemas.tasks import TaskStatus
from skyvern.forge.sdk.services.bitwarden import BitwardenConstants, BitwardenService
from skyvern.forge.sdk.services.credentials import AzureVaultConstants, OnePasswordConstants, normalize_totp_config
+from skyvern.forge.sdk.workflow.credential_selection import select_credential_for_run
from skyvern.forge.sdk.workflow.exceptions import MissingJinjaVariables, OutputParameterKeyCollisionError
from skyvern.forge.sdk.workflow.models.parameter import (
PARAMETER_TYPE,
@@ -237,6 +238,7 @@ class WorkflowRunContext:
self.browser_session_id: str | None = None
self.include_secrets_in_templates: bool = False
self.credential_totp_identifiers: dict[str, str] = {}
+ self.resolved_credential_parameter_ids: dict[str, str] = {}
def set_workflow(self, workflow: "Workflow") -> None:
"""
@@ -708,7 +710,9 @@ class WorkflowRunContext:
LOG.info("Fetching credential parameter value", parameter_key=parameter.key)
credential_id = None
- if parameter.credential_id:
+ if parameter.credential_ids:
+ credential_id = await self.resolve_credential_parameter_id(parameter, organization.organization_id)
+ elif parameter.credential_id:
if self.has_parameter(parameter.credential_id) and self.has_value(parameter.credential_id):
credential_id = self.values[parameter.credential_id]
else:
@@ -720,6 +724,28 @@ class WorkflowRunContext:
await self._register_credential_parameter_value(credential_id, parameter, organization)
+ async def resolve_credential_parameter_id(
+ self,
+ parameter: CredentialParameter,
+ organization_id: str,
+ ) -> str:
+ cached = self.resolved_credential_parameter_ids.get(parameter.key)
+ if cached:
+ return cached
+ if not parameter.credential_ids:
+ self.resolved_credential_parameter_ids[parameter.key] = parameter.credential_id
+ return parameter.credential_id
+ credential_id = await select_credential_for_run(
+ workflow_run_id=self.workflow_run_id,
+ organization_id=organization_id,
+ workflow_permanent_id=self.workflow_permanent_id,
+ parameter_key=parameter.key,
+ credential_ids=parameter.credential_ids,
+ selection_strategy=parameter.selection_strategy,
+ )
+ self.resolved_credential_parameter_ids[parameter.key] = credential_id
+ return credential_id
+
async def register_aws_secret_parameter_value(
self,
parameter: AWSSecretParameter,
diff --git a/skyvern/forge/sdk/workflow/credential_selection.py b/skyvern/forge/sdk/workflow/credential_selection.py
new file mode 100644
index 000000000..5e8da3366
--- /dev/null
+++ b/skyvern/forge/sdk/workflow/credential_selection.py
@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import random
+
+import structlog
+from sqlalchemy.exc import IntegrityError
+
+from skyvern.forge import app
+
+LOG = structlog.get_logger()
+
+ROUND_ROBIN = "round_robin"
+RANDOM = "random"
+VALID_SELECTION_STRATEGIES = frozenset({ROUND_ROBIN, RANDOM})
+
+
+def normalize_selection_strategy(selection_strategy: str | None) -> str:
+ return selection_strategy or ROUND_ROBIN
+
+
+async def select_credential_for_run(
+ workflow_run_id: str,
+ organization_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ selection_strategy: str | None,
+) -> str:
+ existing = await app.DATABASE.workflow_run_credential_selections.get_selection(
+ workflow_run_id=workflow_run_id,
+ parameter_key=parameter_key,
+ )
+ if existing:
+ return existing
+
+ strategy = normalize_selection_strategy(selection_strategy)
+ try:
+ if strategy == RANDOM:
+ selected = await app.DATABASE.workflow_run_credential_selections.create_selection(
+ organization_id=organization_id,
+ workflow_run_id=workflow_run_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_id=random.choice(credential_ids),
+ )
+ else:
+ selected = await app.DATABASE.workflow_run_credential_selections.create_round_robin_selection(
+ organization_id=organization_id,
+ workflow_run_id=workflow_run_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_ids=credential_ids,
+ )
+ except IntegrityError:
+ existing_selection = await app.DATABASE.workflow_run_credential_selections.get_selection(
+ workflow_run_id=workflow_run_id,
+ parameter_key=parameter_key,
+ )
+ if not existing_selection:
+ raise
+ selected = existing_selection
+
+ LOG.info(
+ "Selected workflow run credential",
+ workflow_run_id=workflow_run_id,
+ parameter_key=parameter_key,
+ credential_id=selected,
+ strategy=strategy,
+ )
+ return selected
diff --git a/skyvern/forge/sdk/workflow/models/parameter.py b/skyvern/forge/sdk/workflow/models/parameter.py
index 61577867b..0ce962523 100644
--- a/skyvern/forge/sdk/workflow/models/parameter.py
+++ b/skyvern/forge/sdk/workflow/models/parameter.py
@@ -125,6 +125,8 @@ class CredentialParameter(Parameter):
workflow_id: str
credential_id: str
+ credential_ids: list[str] | None = None
+ selection_strategy: str | None = None
created_at: datetime
modified_at: datetime
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index 32518eb16..83641cdf6 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -77,6 +77,11 @@ from skyvern.forge.sdk.workflow.browser_profile_key import (
build_workflow_browser_session_storage_key,
render_browser_profile_key,
)
+from skyvern.forge.sdk.workflow.credential_selection import (
+ VALID_SELECTION_STRATEGIES,
+ normalize_selection_strategy,
+ select_credential_for_run,
+)
from skyvern.forge.sdk.workflow.exceptions import (
InvalidWorkflowDefinition,
WorkflowVersionConflict,
@@ -983,6 +988,38 @@ class WorkflowService:
if missing:
raise InvalidCredentialId(", ".join(missing))
+ async def _validate_and_normalize_credential_rotation_parameters(
+ self,
+ parameters: list[Any],
+ organization: Organization,
+ ) -> None:
+ credential_ids_to_validate: list[str] = []
+ for parameter in parameters:
+ credential_ids = getattr(parameter, "credential_ids", None)
+ if credential_ids is None:
+ continue
+ key = getattr(parameter, "key", None) or ""
+ if not credential_ids:
+ raise SkyvernHTTPException(
+ message=f"credential_ids for credential parameter {key} must be non-empty.",
+ status_code=400,
+ )
+ credential_ids = list(dict.fromkeys(credential_ids))
+ parameter.credential_ids = credential_ids
+ selection_strategy = getattr(parameter, "selection_strategy", None)
+ if normalize_selection_strategy(selection_strategy) not in VALID_SELECTION_STRATEGIES:
+ raise SkyvernHTTPException(
+ message=(
+ f"selection_strategy for credential parameter {key} must be one of: "
+ f"{', '.join(sorted(VALID_SELECTION_STRATEGIES))}."
+ ),
+ status_code=400,
+ )
+ parameter.credential_id = credential_ids[0]
+ credential_ids_to_validate.extend(credential_ids)
+
+ await self._validate_credential_ids(credential_ids_to_validate, organization)
+
async def validate_schedule_parameters(
self,
workflow: Workflow,
@@ -3055,6 +3092,7 @@ class WorkflowService:
block=block,
workflow_run_id=workflow_run_id,
organization_id=organization_id,
+ workflow_permanent_id=workflow_run.workflow_permanent_id,
)
# Save the original navigation goal before any mutation so
# retries don't stack the browser-session prefix repeatedly.
@@ -3726,6 +3764,7 @@ class WorkflowService:
block=block,
workflow_run_id=None,
organization_id=organization_id,
+ workflow_permanent_id=None,
)
async def _resolve_login_block_browser_profile_id(
@@ -3733,6 +3772,7 @@ class WorkflowService:
block: Block,
workflow_run_id: str | None,
organization_id: str | None,
+ workflow_permanent_id: str | None,
) -> str | None:
"""Inspect the block-level parameters and return the browser_profile_id
from the credential parameter bound to this specific block."""
@@ -3743,6 +3783,8 @@ class WorkflowService:
credential_ids = await self._resolve_login_block_credential_ids(
block=block,
workflow_run_id=workflow_run_id,
+ organization_id=organization_id,
+ workflow_permanent_id=workflow_permanent_id,
)
for credential_id in credential_ids:
@@ -3787,6 +3829,8 @@ class WorkflowService:
self,
block: Block,
workflow_run_id: str | None,
+ organization_id: str | None = None,
+ workflow_permanent_id: str | None = None,
) -> list[str]:
"""Return credential ids bound to this block, preserving parameter order."""
params = block.parameters
@@ -3800,7 +3844,12 @@ class WorkflowService:
# Style 1: CredentialParameter (has credential_id directly)
if isinstance(param, CredentialParameter):
- credential_id = param.credential_id
+ credential_id = await self._resolve_credential_parameter_id(
+ parameter=param,
+ workflow_run_id=workflow_run_id,
+ organization_id=organization_id,
+ workflow_permanent_id=workflow_permanent_id,
+ )
# Style 2: WorkflowParameter with type CREDENTIAL_ID
elif (
@@ -3842,6 +3891,30 @@ class WorkflowService:
return credential_ids
+ async def _resolve_credential_parameter_id(
+ self,
+ *,
+ parameter: CredentialParameter,
+ workflow_run_id: str | None,
+ organization_id: str | None,
+ workflow_permanent_id: str | None,
+ ) -> str:
+ if not parameter.credential_ids or not workflow_run_id or not organization_id or not workflow_permanent_id:
+ return parameter.credential_id
+
+ workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.workflow_run_contexts.get(workflow_run_id)
+ if workflow_run_context:
+ return await workflow_run_context.resolve_credential_parameter_id(parameter, organization_id)
+
+ return await select_credential_for_run(
+ workflow_run_id=workflow_run_id,
+ organization_id=organization_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter.key,
+ credential_ids=parameter.credential_ids,
+ selection_strategy=parameter.selection_strategy,
+ )
+
async def _apply_login_block_credential_proxy_pin(
self,
*,
@@ -3859,6 +3932,8 @@ class WorkflowService:
credential_ids = await self._resolve_login_block_credential_ids(
block=block,
workflow_run_id=workflow_run_id,
+ organization_id=organization_id,
+ workflow_permanent_id=getattr(workflow_run, "workflow_permanent_id", None),
)
for credential_id in credential_ids:
if not organization_id:
@@ -4786,6 +4861,13 @@ class WorkflowService:
edited_by: str | None | object = _UNSET,
) -> Workflow:
if workflow_definition is not None:
+ if organization_id is not None:
+ organization = await app.DATABASE.organizations.get_organization(organization_id=organization_id)
+ if organization is not None:
+ await self._validate_and_normalize_credential_rotation_parameters(
+ workflow_definition.parameters,
+ organization,
+ )
updated_workflow = await app.DATABASE.workflows.update_workflow_and_reconcile_definition_params(
workflow_id=workflow_id,
title=title,
@@ -5076,6 +5158,8 @@ class WorkflowService:
search_key: str | None = None,
error_code: str | None = None,
exclude_child_runs: bool = False,
+ created_at_start: datetime | None = None,
+ created_at_end: datetime | None = None,
) -> list[WorkflowRun]:
return await app.DATABASE.workflow_runs.get_workflow_runs_for_workflow_permanent_id(
workflow_permanent_id=workflow_permanent_id,
@@ -5086,6 +5170,8 @@ class WorkflowService:
search_key=search_key,
error_code=error_code,
exclude_child_runs=exclude_child_runs,
+ created_at_start=created_at_start,
+ created_at_end=created_at_end,
)
async def get_workflow_runs_for_browser_session(
@@ -6942,6 +7028,10 @@ class WorkflowService:
organization_id=organization_id,
title=title,
)
+ await self._validate_and_normalize_credential_rotation_parameters(
+ request.workflow_definition.parameters,
+ organization,
+ )
new_workflow_id: str | None = None
refresh_schedule_runtime_limits = False
effective_max_elapsed_time_minutes: int | None = None
diff --git a/skyvern/forge/sdk/workflow/workflow_definition_converter.py b/skyvern/forge/sdk/workflow/workflow_definition_converter.py
index a51de60d3..ad156fae3 100644
--- a/skyvern/forge/sdk/workflow/workflow_definition_converter.py
+++ b/skyvern/forge/sdk/workflow/workflow_definition_converter.py
@@ -168,6 +168,8 @@ def convert_workflow_definition(
key=parameter.key,
description=parameter.description,
credential_id=parameter.credential_id,
+ credential_ids=parameter.credential_ids,
+ selection_strategy=parameter.selection_strategy,
created_at=now,
modified_at=now,
)
diff --git a/skyvern/schemas/workflows.py b/skyvern/schemas/workflows.py
index f16f4fef2..b984c067f 100644
--- a/skyvern/schemas/workflows.py
+++ b/skyvern/schemas/workflows.py
@@ -596,6 +596,8 @@ class BitwardenLoginCredentialParameterYAML(ParameterYAML):
class CredentialParameterYAML(ParameterYAML):
parameter_type: Literal[ParameterType.CREDENTIAL] = ParameterType.CREDENTIAL # type: ignore
credential_id: str
+ credential_ids: list[str] | None = None
+ selection_strategy: str | None = None
class BitwardenSensitiveInformationParameterYAML(ParameterYAML):
diff --git a/tests/unit/forge/sdk/db/test_workflow_runs_created_at_filter.py b/tests/unit/forge/sdk/db/test_workflow_runs_created_at_filter.py
new file mode 100644
index 000000000..bc3ac18ff
--- /dev/null
+++ b/tests/unit/forge/sdk/db/test_workflow_runs_created_at_filter.py
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from skyvern.forge.sdk.db.repositories.workflow_runs import WorkflowRunsRepository
+
+
+class _SessionContext:
+ def __init__(self, session: MagicMock) -> None:
+ self._session = session
+
+ async def __aenter__(self) -> MagicMock:
+ return self._session
+
+ async def __aexit__(self, exc_type, exc, tb) -> bool:
+ return False
+
+
+class _Result:
+ def all(self):
+ return []
+
+
+def _make_repo(captured: dict[str, Any]) -> WorkflowRunsRepository:
+ async def _execute(query):
+ captured["query"] = query
+ return _Result()
+
+ session = MagicMock()
+ session.execute = AsyncMock(side_effect=_execute)
+ return WorkflowRunsRepository(session_factory=lambda: _SessionContext(session), debug_enabled=False)
+
+
+@pytest.mark.asyncio
+async def test_get_workflow_runs_filters_by_created_at_window() -> None:
+ captured: dict[str, Any] = {}
+ repo = _make_repo(captured)
+
+ await repo.get_workflow_runs_for_workflow_permanent_id(
+ workflow_permanent_id="wpid_test",
+ organization_id="o_test",
+ created_at_start=datetime(2026, 6, 1, tzinfo=timezone.utc),
+ created_at_end=datetime(2026, 6, 8, tzinfo=timezone.utc),
+ )
+
+ where = str(captured["query"].whereclause)
+ assert "created_at >=" in where
+ assert "created_at <" in where
+
+
+@pytest.mark.asyncio
+async def test_get_workflow_runs_omits_created_at_filter_when_unset() -> None:
+ captured: dict[str, Any] = {}
+ repo = _make_repo(captured)
+
+ await repo.get_workflow_runs_for_workflow_permanent_id(
+ workflow_permanent_id="wpid_test",
+ organization_id="o_test",
+ )
+
+ where = str(captured["query"].whereclause)
+ assert "created_at >=" not in where
+ assert "created_at <" not in where
diff --git a/tests/unit/test_copilot_tools_credential_id_misbinding.py b/tests/unit/test_copilot_tools_credential_id_misbinding.py
index af0810cf1..df3da1935 100644
--- a/tests/unit/test_copilot_tools_credential_id_misbinding.py
+++ b/tests/unit/test_copilot_tools_credential_id_misbinding.py
@@ -85,6 +85,61 @@ def test_credential_id_bound_through_block_credential_parameter_is_allowed() ->
assert _credential_id_misbinding_findings(yaml) == []
+def test_credential_ids_bound_through_credential_parameter_are_allowed() -> None:
+ yaml = _yaml(
+ """
+ title: Sign in
+ workflow_definition:
+ parameters:
+ - key: login_credentials
+ parameter_type: credential
+ credential_id: cred_527971855302737592
+ credential_ids:
+ - cred_527971855302737592
+ - cred_827971855302737593
+ blocks:
+ - block_type: login
+ label: login_to_portal
+ url: https://authenticationtest.com/loginUserAndPassword/
+ parameter_keys: [login_credentials]
+ """
+ )
+
+ assert _credential_id_misbinding_findings(yaml) == []
+
+
+def test_credential_parameter_unrelated_field_with_credential_id_is_still_flagged() -> None:
+ yaml = _yaml(
+ """
+ title: Sign in
+ workflow_definition:
+ parameters:
+ - key: login_credentials
+ parameter_type: credential
+ credential_id: cred_527971855302737592
+ credential_ids:
+ - cred_527971855302737592
+ - cred_827971855302737593
+ description: Use cred_927971855302737594 for this login.
+ blocks:
+ - block_type: login
+ label: login_to_portal
+ url: https://authenticationtest.com/loginUserAndPassword/
+ parameter_keys: [login_credentials]
+ """
+ )
+
+ findings = _credential_id_misbinding_findings(yaml)
+
+ assert findings == [
+ {
+ "location": "workflow",
+ "field": "description",
+ "credential_id": "cred_927971855302737594",
+ }
+ ]
+
+
def test_credential_id_in_parameter_keys_list_is_flagged() -> None:
yaml = _yaml(
"""
diff --git a/tests/unit/test_credential_rotation.py b/tests/unit/test_credential_rotation.py
new file mode 100644
index 000000000..a52697405
--- /dev/null
+++ b/tests/unit/test_credential_rotation.py
@@ -0,0 +1,423 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import func, select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+from skyvern.exceptions import InvalidCredentialId, SkyvernHTTPException
+from skyvern.forge.sdk.copilot.output_policy import OutputPolicyReason, evaluate_output_policy
+from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
+from skyvern.forge.sdk.db.agent_db import AgentDB, _build_engine
+from skyvern.forge.sdk.db.models import Base, WorkflowRunCredentialSelectionModel
+from skyvern.forge.sdk.db.repositories.workflow_run_credential_selections import (
+ WorkflowRunCredentialSelectionsRepository,
+)
+from skyvern.forge.sdk.workflow.credential_selection import select_credential_for_run
+from skyvern.forge.sdk.workflow.models.parameter import CredentialParameter
+from skyvern.forge.sdk.workflow.service import WorkflowService
+from skyvern.forge.sdk.workflow.workflow_definition_converter import convert_workflow_definition
+from skyvern.schemas.workflows import CredentialParameterYAML, WorkflowDefinitionYAML
+
+
+def _credential_parameter(
+ *,
+ credential_id: str = "cred_a",
+ credential_ids: list[str] | None = None,
+ selection_strategy: str | None = None,
+) -> CredentialParameter:
+ now = datetime.now(timezone.utc)
+ return CredentialParameter(
+ key="login_cred",
+ credential_parameter_id="cp_test",
+ workflow_id="wf_test",
+ credential_id=credential_id,
+ credential_ids=credential_ids,
+ selection_strategy=selection_strategy,
+ created_at=now,
+ modified_at=now,
+ )
+
+
+class _SelectionRepo:
+ def __init__(
+ self,
+ *,
+ existing: dict[tuple[str, str], str] | None = None,
+ latest: dict[str, datetime] | None = None,
+ raise_on_create: bool = False,
+ ) -> None:
+ self.existing = existing or {}
+ self.latest = latest or {}
+ self.raise_on_create = raise_on_create
+ self.created: list[dict[str, str]] = []
+
+ async def get_selection(self, workflow_run_id: str, parameter_key: str) -> str | None:
+ return self.existing.get((workflow_run_id, parameter_key))
+
+ async def get_latest_selections(
+ self,
+ *,
+ organization_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ ) -> dict[str, datetime]:
+ return {
+ credential_id: self.latest[credential_id]
+ for credential_id in credential_ids
+ if credential_id in self.latest
+ }
+
+ async def create_selection(
+ self,
+ *,
+ organization_id: str,
+ workflow_run_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_id: str,
+ ) -> str:
+ if self.raise_on_create:
+ self.existing[(workflow_run_id, parameter_key)] = "cred_winner"
+ raise IntegrityError("insert", {}, Exception("duplicate"))
+ self.created.append(
+ {
+ "organization_id": organization_id,
+ "workflow_run_id": workflow_run_id,
+ "workflow_permanent_id": workflow_permanent_id,
+ "parameter_key": parameter_key,
+ "credential_id": credential_id,
+ }
+ )
+ self.existing[(workflow_run_id, parameter_key)] = credential_id
+ return credential_id
+
+ async def create_round_robin_selection(
+ self,
+ *,
+ organization_id: str,
+ workflow_run_id: str,
+ workflow_permanent_id: str,
+ parameter_key: str,
+ credential_ids: list[str],
+ ) -> str:
+ existing = await self.get_selection(workflow_run_id=workflow_run_id, parameter_key=parameter_key)
+ if existing:
+ return existing
+
+ latest_selections = await self.get_latest_selections(
+ organization_id=organization_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_ids=credential_ids,
+ )
+ unseen = next((candidate for candidate in credential_ids if candidate not in latest_selections), None)
+ credential_id = (
+ unseen if unseen is not None else min(credential_ids, key=lambda candidate: latest_selections[candidate])
+ )
+ return await self.create_selection(
+ organization_id=organization_id,
+ workflow_run_id=workflow_run_id,
+ workflow_permanent_id=workflow_permanent_id,
+ parameter_key=parameter_key,
+ credential_id=credential_id,
+ )
+
+
+@pytest_asyncio.fixture
+async def sqlite_engine() -> AsyncEngine:
+ engine = _build_engine("sqlite+aiosqlite:///:memory:")
+ async with engine.begin() as conn:
+ await conn.run_sync(Base.metadata.create_all)
+ try:
+ yield engine
+ finally:
+ await engine.dispose()
+
+
+@pytest_asyncio.fixture
+async def sqlite_db(sqlite_engine: AsyncEngine) -> AgentDB:
+ return AgentDB("sqlite+aiosqlite:///:memory:", db_engine=sqlite_engine)
+
+
+async def _select(repo: _SelectionRepo, credential_ids: list[str], strategy: str | None = None) -> str:
+ with patch("skyvern.forge.sdk.workflow.credential_selection.app") as mock_app:
+ mock_app.DATABASE.workflow_run_credential_selections = repo
+ return await select_credential_for_run(
+ workflow_run_id="wr_test",
+ organization_id="org_test",
+ workflow_permanent_id="wpid_test",
+ parameter_key="login_cred",
+ credential_ids=credential_ids,
+ selection_strategy=strategy,
+ )
+
+
+@pytest.mark.asyncio
+async def test_round_robin_picks_unseen_first() -> None:
+ repo = _SelectionRepo(latest={"cred_a": datetime.now(timezone.utc)})
+
+ selected = await _select(repo, ["cred_a", "cred_b", "cred_c"])
+
+ assert selected == "cred_b"
+ assert repo.created[0]["credential_id"] == "cred_b"
+
+
+@pytest.mark.asyncio
+async def test_round_robin_picks_oldest_last_used_and_ties_by_list_order() -> None:
+ now = datetime.now(timezone.utc)
+ repo = _SelectionRepo(latest={"cred_a": now, "cred_b": now - timedelta(minutes=5), "cred_c": now})
+
+ selected = await _select(repo, ["cred_a", "cred_b", "cred_c"])
+
+ assert selected == "cred_b"
+
+ tied_repo = _SelectionRepo(latest={"cred_a": now, "cred_b": now, "cred_c": now})
+ tied_selected = await _select(tied_repo, ["cred_a", "cred_b", "cred_c"])
+
+ assert tied_selected == "cred_a"
+
+
+@pytest.mark.asyncio
+async def test_selection_is_idempotent_for_run_and_key() -> None:
+ repo = _SelectionRepo(existing={("wr_test", "login_cred"): "cred_a"})
+
+ first = await _select(repo, ["cred_a", "cred_b"])
+ second = await _select(repo, ["cred_a", "cred_b"])
+
+ assert first == "cred_a"
+ assert second == "cred_a"
+ assert repo.created == []
+
+
+@pytest.mark.asyncio
+async def test_random_selection_returns_member(monkeypatch: pytest.MonkeyPatch) -> None:
+ repo = _SelectionRepo()
+ monkeypatch.setattr("skyvern.forge.sdk.workflow.credential_selection.random.choice", lambda ids: ids[-1])
+
+ selected = await _select(repo, ["cred_a", "cred_b"], "random")
+
+ assert selected == "cred_b"
+ assert selected in {"cred_a", "cred_b"}
+
+
+@pytest.mark.asyncio
+async def test_duplicate_insert_race_returns_existing_winner() -> None:
+ repo = _SelectionRepo(raise_on_create=True)
+
+ selected = await _select(repo, ["cred_a", "cred_b"])
+
+ assert selected == "cred_winner"
+
+
+@pytest.mark.asyncio
+async def test_round_robin_repository_serialized_path_picks_distinct_credentials(sqlite_db: AgentDB) -> None:
+ repo = sqlite_db.workflow_run_credential_selections
+
+ first = await repo.create_round_robin_selection(
+ organization_id="org_test",
+ workflow_run_id="wr_one",
+ workflow_permanent_id="wpid_test",
+ parameter_key="login_cred",
+ credential_ids=["cred_a", "cred_b"],
+ )
+ second = await repo.create_round_robin_selection(
+ organization_id="org_test",
+ workflow_run_id="wr_two",
+ workflow_permanent_id="wpid_test",
+ parameter_key="login_cred",
+ credential_ids=["cred_a", "cred_b"],
+ )
+
+ assert first == "cred_a"
+ assert second == "cred_b"
+
+
+@pytest.mark.asyncio
+async def test_round_robin_repository_idempotent_recall_returns_existing(sqlite_db: AgentDB) -> None:
+ repo = sqlite_db.workflow_run_credential_selections
+
+ first = await repo.create_round_robin_selection(
+ organization_id="org_test",
+ workflow_run_id="wr_one",
+ workflow_permanent_id="wpid_test",
+ parameter_key="login_cred",
+ credential_ids=["cred_a", "cred_b"],
+ )
+ second = await repo.create_round_robin_selection(
+ organization_id="org_test",
+ workflow_run_id="wr_one",
+ workflow_permanent_id="wpid_test",
+ parameter_key="login_cred",
+ credential_ids=["cred_a", "cred_b"],
+ )
+
+ async with sqlite_db.Session() as session:
+ count = (
+ await session.execute(select(func.count()).select_from(WorkflowRunCredentialSelectionModel))
+ ).scalar_one()
+
+ assert first == "cred_a"
+ assert second == "cred_a"
+ assert count == 1
+
+
+@pytest.mark.asyncio
+async def test_rotation_advisory_lock_skips_non_postgres_dialect() -> None:
+ repo = WorkflowRunCredentialSelectionsRepository(MagicMock())
+ session = MagicMock()
+ session.get_bind.return_value = SimpleNamespace(dialect=SimpleNamespace(name="sqlite"))
+ session.execute = AsyncMock()
+
+ await repo._take_rotation_advisory_lock(session, "wrcs:org:wpid:login_cred")
+
+ session.execute.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_workflow_save_validation_rejects_empty_credential_ids() -> None:
+ service = WorkflowService()
+ org = SimpleNamespace(organization_id="org_test")
+ parameter = _credential_parameter(credential_ids=[])
+
+ with pytest.raises(SkyvernHTTPException, match="credential_ids"):
+ await service._validate_and_normalize_credential_rotation_parameters([parameter], org)
+
+
+@pytest.mark.asyncio
+async def test_workflow_save_validation_rejects_unknown_credential_id() -> None:
+ service = WorkflowService()
+ org = SimpleNamespace(organization_id="org_test")
+ parameter = _credential_parameter(credential_ids=["cred_missing"])
+
+ with patch("skyvern.forge.sdk.workflow.service.app") as mock_app:
+ mock_app.DATABASE.credentials.get_credentials_by_ids = AsyncMock(return_value=[])
+ with pytest.raises(InvalidCredentialId):
+ await service._validate_and_normalize_credential_rotation_parameters([parameter], org)
+
+
+@pytest.mark.asyncio
+async def test_workflow_save_validation_rejects_bad_strategy() -> None:
+ service = WorkflowService()
+ org = SimpleNamespace(organization_id="org_test")
+ parameter = _credential_parameter(credential_ids=["cred_a"], selection_strategy="newest")
+
+ with pytest.raises(SkyvernHTTPException, match="selection_strategy"):
+ await service._validate_and_normalize_credential_rotation_parameters([parameter], org)
+
+
+@pytest.mark.asyncio
+async def test_workflow_save_validation_normalizes_credential_id_to_first_rotating_id() -> None:
+ service = WorkflowService()
+ org = SimpleNamespace(organization_id="org_test")
+ parameter = _credential_parameter(credential_id="cred_stale", credential_ids=["cred_a", "cred_b"])
+ existing = [SimpleNamespace(credential_id="cred_a"), SimpleNamespace(credential_id="cred_b")]
+
+ with patch("skyvern.forge.sdk.workflow.service.app") as mock_app:
+ mock_app.DATABASE.credentials.get_credentials_by_ids = AsyncMock(return_value=existing)
+ await service._validate_and_normalize_credential_rotation_parameters([parameter], org)
+
+ assert parameter.credential_id == "cred_a"
+
+
+@pytest.mark.asyncio
+async def test_workflow_save_validation_dedupes_credential_ids_preserving_order() -> None:
+ service = WorkflowService()
+ org = SimpleNamespace(organization_id="org_test")
+ parameter = _credential_parameter(
+ credential_id="cred_stale",
+ credential_ids=["cred_a", "cred_b", "cred_a", "cred_c", "cred_b"],
+ )
+ existing = [
+ SimpleNamespace(credential_id="cred_a"),
+ SimpleNamespace(credential_id="cred_b"),
+ SimpleNamespace(credential_id="cred_c"),
+ ]
+
+ with patch("skyvern.forge.sdk.workflow.service.app") as mock_app:
+ mock_get_credentials = AsyncMock(return_value=existing)
+ mock_app.DATABASE.credentials.get_credentials_by_ids = mock_get_credentials
+ await service._validate_and_normalize_credential_rotation_parameters([parameter], org)
+
+ assert parameter.credential_ids == ["cred_a", "cred_b", "cred_c"]
+ assert parameter.credential_id == "cred_a"
+ mock_get_credentials.assert_awaited_once_with(["cred_a", "cred_b", "cred_c"], organization_id="org_test")
+
+
+def test_output_policy_origin_broadening_checks_non_first_rotating_credential() -> None:
+ workflow_yaml = """
+title: Login
+workflow_definition:
+ parameters:
+ - parameter_type: credential
+ key: login_cred
+ credential_id: cred_first
+ credential_ids:
+ - cred_first
+ - cred_second
+ blocks:
+ - block_type: login
+ label: Login
+ url: https://portal.example.com/login
+ parameter_keys:
+ - login_cred
+"""
+ request_policy = RequestPolicy(
+ resolved_credentials=[
+ SimpleNamespace(credential_id="cred_first", tested_url="https://portal.example.com/login"),
+ SimpleNamespace(credential_id="cred_second", tested_url="https://other.example.com/login"),
+ ]
+ )
+
+ verdict = evaluate_output_policy(request_policy=request_policy, workflow_yaml=workflow_yaml)
+
+ assert OutputPolicyReason.CREDENTIAL_SCOPE_BROADENED in verdict.reason_codes
+
+
+def test_yaml_to_credential_parameter_round_trip_preserves_rotation_fields() -> None:
+ yaml_definition = WorkflowDefinitionYAML(
+ parameters=[
+ CredentialParameterYAML(
+ key="login_cred",
+ credential_id="cred_a",
+ credential_ids=["cred_a", "cred_b"],
+ selection_strategy="round_robin",
+ )
+ ],
+ blocks=[],
+ )
+
+ definition = convert_workflow_definition(yaml_definition, workflow_id="wf_test")
+ parameter = definition.parameters[0]
+
+ assert isinstance(parameter, CredentialParameter)
+ assert parameter.credential_id == "cred_a"
+ assert parameter.credential_ids == ["cred_a", "cred_b"]
+ assert parameter.selection_strategy == "round_robin"
+
+
+@pytest.mark.asyncio
+async def test_resolve_login_block_credential_ids_returns_selected_rotating_id() -> None:
+ service = WorkflowService()
+ parameter = _credential_parameter(credential_ids=["cred_a", "cred_b"])
+ context = MagicMock()
+ context.resolve_credential_parameter_id = AsyncMock(return_value="cred_b")
+ block = SimpleNamespace(parameters=[parameter])
+
+ with patch("skyvern.forge.sdk.workflow.service.app") as mock_app:
+ mock_app.WORKFLOW_CONTEXT_MANAGER.workflow_run_contexts = {"wr_test": context}
+ credential_ids = await service._resolve_login_block_credential_ids(
+ block=block,
+ workflow_run_id="wr_test",
+ organization_id="org_test",
+ workflow_permanent_id="wpid_test",
+ )
+
+ assert credential_ids == ["cred_b"]
+ context.resolve_credential_parameter_id.assert_awaited_once_with(parameter, "org_test")
diff --git a/tests/unit/test_execute_step_exception_handling.py b/tests/unit/test_execute_step_exception_handling.py
new file mode 100644
index 000000000..53afa118a
--- /dev/null
+++ b/tests/unit/test_execute_step_exception_handling.py
@@ -0,0 +1,311 @@
+"""Characterization tests for ``ForgeAgent.execute_step`` exception handling (SKY-11786).
+
+These pin the CURRENT observable behavior of the exception handlers at the tail of
+``skyvern/forge/agent.py::ForgeAgent.execute_step``. For each exception type they pin:
+
+* whether ``fail_task`` runs,
+* whether ``clean_up_task`` runs (and, for the conditional handlers, whether that is
+ gated on ``fail_task`` reporting the task as failed),
+* the webhook decision (``need_call_webhook``) and the final-screenshot decision
+ (``need_final_screenshot``) handed to ``clean_up_task``.
+
+They gate the SKY-11743 restructure (SKY-11787), which collapses the nine cleanup
+handlers behind shared per-exception configuration; the suite must pass against current
+main with no change to ``execute_step``.
+
+Each handler is reached by making the first in-``try`` await
+(``app.AGENT_FUNCTION.validate_step_execution``) raise the target exception, which lands
+control directly in the matching ``except`` clause. ``fail_task`` and ``clean_up_task``
+are mocked, so the assertions read the decisions off their call args and never touch a
+real browser or database. Webhook/screenshot are asserted as *effective* values
+(kwarg-or-default), so a restructure that makes the current defaults explicit still
+passes — only a change in behavior fails.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any, Callable
+from unittest.mock import AsyncMock
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from skyvern.exceptions import (
+ FailedToNavigateToUrl,
+ FailedToParseActionInstruction,
+ FailedToSendWebhook,
+ InvalidTaskStatusTransition,
+ MissingBrowserStatePage,
+ ScrapingFailed,
+ StepTerminationError,
+ StepUnableToExecuteError,
+ TaskAlreadyCanceled,
+ TaskAlreadyTimeout,
+ UnsupportedActionType,
+ UnsupportedTaskType,
+)
+from skyvern.forge.agent import ForgeAgent
+from skyvern.forge.sdk.core import skyvern_context
+from skyvern.forge.sdk.core.skyvern_context import SkyvernContext
+from skyvern.forge.sdk.models import StepStatus
+from tests.unit.helpers import make_organization, make_step, make_task
+
+# ``clean_up_task``'s real signature defaults; handlers that omit these kwargs get them.
+_CLEANUP_WEBHOOK_DEFAULT = True
+_CLEANUP_SCREENSHOT_DEFAULT = True
+
+
+@dataclass
+class ExecuteStepOutcome:
+ returned: tuple | None
+ raised: BaseException | None
+ fail_task: AsyncMock
+ clean_up_task: AsyncMock
+
+ @property
+ def fail_task_called(self) -> bool:
+ return self.fail_task.await_count > 0
+
+ @property
+ def cleanup_called(self) -> bool:
+ return self.clean_up_task.await_count > 0
+
+ def _cleanup_kwarg(self, name: str, default: Any) -> Any:
+ assert self.cleanup_called, "clean_up_task was not called"
+ return self.clean_up_task.await_args.kwargs.get(name, default)
+
+ @property
+ def effective_webhook(self) -> bool:
+ return bool(self._cleanup_kwarg("need_call_webhook", _CLEANUP_WEBHOOK_DEFAULT))
+
+ @property
+ def effective_final_screenshot(self) -> bool:
+ return bool(self._cleanup_kwarg("need_final_screenshot", _CLEANUP_SCREENSHOT_DEFAULT))
+
+
+async def _drive_execute_step(
+ monkeypatch: pytest.MonkeyPatch,
+ exc: BaseException,
+ *,
+ fail_task_result: bool = True,
+) -> ExecuteStepOutcome:
+ """Run ``execute_step`` so that ``exc`` is raised at the first in-``try`` await."""
+ agent = ForgeAgent()
+ now = datetime.now(UTC)
+ organization = make_organization(now)
+ task = make_task(now, organization)
+ step = make_step(now, task, step_id="step-0", status=StepStatus.running, order=0, output=None)
+
+ fail_task_mock = AsyncMock(return_value=fail_task_result)
+ clean_up_task_mock = AsyncMock(return_value=None)
+ agent.fail_task = fail_task_mock # type: ignore[method-assign]
+ agent.clean_up_task = clean_up_task_mock # type: ignore[method-assign]
+
+ # Pre-``try`` DB reads in execute_step: keep them inert so we reach the try body.
+ monkeypatch.setattr("skyvern.forge.agent.app.DATABASE.tasks.get_task", AsyncMock(return_value=None))
+ monkeypatch.setattr("skyvern.forge.agent.app.DATABASE.tasks.update_task", AsyncMock(return_value=task))
+ # First in-``try`` await; raising here routes straight to the matching except clause.
+ monkeypatch.setattr(
+ "skyvern.forge.agent.app.AGENT_FUNCTION.validate_step_execution",
+ AsyncMock(side_effect=exc),
+ )
+
+ context = SkyvernContext(
+ organization_id=organization.organization_id,
+ task_id=task.task_id,
+ step_id=None,
+ tz_info=ZoneInfo("UTC"),
+ )
+ skyvern_context.set(context)
+ returned: tuple | None = None
+ raised: BaseException | None = None
+ try:
+ returned = await agent.execute_step(
+ organization=organization,
+ task=task,
+ step=step,
+ api_key="api-key",
+ download_baseline_files=[],
+ )
+ except BaseException as caught: # noqa: BLE001 - we characterize which exceptions propagate
+ raised = caught
+ finally:
+ skyvern_context.reset()
+
+ return ExecuteStepOutcome(
+ returned=returned,
+ raised=raised,
+ fail_task=fail_task_mock,
+ clean_up_task=clean_up_task_mock,
+ )
+
+
+@dataclass(frozen=True)
+class CleanupHandlerCase:
+ """Pinned contract for one cleanup ``except`` clause in ``execute_step``."""
+
+ id: str
+ exc_factory: Callable[[], BaseException]
+ fail_task_called: bool
+ effective_webhook: bool
+ effective_final_screenshot: bool
+ # True => clean_up_task runs only when fail_task reports the task as failed.
+ cleanup_gated_on_fail_task: bool
+
+
+# The nine cleanup ``except`` clauses (SKY-11743's "nine cleanup handlers"). The
+# unsupported-* clause catches three exception types and is exercised by all three below.
+CLEANUP_CASES: list[CleanupHandlerCase] = [
+ CleanupHandlerCase(
+ id="task_already_timeout",
+ exc_factory=lambda: TaskAlreadyTimeout("task-123"),
+ fail_task_called=False,
+ effective_webhook=True,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="step_termination",
+ exc_factory=lambda: StepTerminationError("terminated", step_id="step-0", task_id="task-123"),
+ fail_task_called=True,
+ effective_webhook=True,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=True,
+ ),
+ CleanupHandlerCase(
+ id="failed_to_navigate",
+ exc_factory=lambda: FailedToNavigateToUrl("https://example.com", "boom"),
+ fail_task_called=True,
+ effective_webhook=True,
+ effective_final_screenshot=False, # the only handler that suppresses the final screenshot
+ cleanup_gated_on_fail_task=True,
+ ),
+ CleanupHandlerCase(
+ id="task_already_canceled",
+ exc_factory=lambda: TaskAlreadyCanceled("failed", "task-123"),
+ fail_task_called=False,
+ effective_webhook=False,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="invalid_status_transition",
+ exc_factory=lambda: InvalidTaskStatusTransition("running", "failed", "task-123"),
+ fail_task_called=False,
+ effective_webhook=False,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="unsupported_action_type",
+ exc_factory=lambda: UnsupportedActionType("MYSTERY_ACTION"),
+ fail_task_called=True,
+ effective_webhook=False,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="unsupported_task_type",
+ exc_factory=lambda: UnsupportedTaskType("mystery_task"),
+ fail_task_called=True,
+ effective_webhook=False,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="failed_to_parse_action",
+ exc_factory=lambda: FailedToParseActionInstruction("bad", "ValueError"),
+ fail_task_called=True,
+ effective_webhook=False,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="scraping_failed",
+ exc_factory=lambda: ScrapingFailed(reason="page gone"),
+ fail_task_called=True,
+ effective_webhook=True,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="missing_browser_state_page",
+ exc_factory=lambda: MissingBrowserStatePage(task_id="task-123"),
+ fail_task_called=True,
+ effective_webhook=True,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=False,
+ ),
+ CleanupHandlerCase(
+ id="generic_exception",
+ exc_factory=lambda: RuntimeError("something unexpected"),
+ fail_task_called=True,
+ effective_webhook=True,
+ effective_final_screenshot=True,
+ cleanup_gated_on_fail_task=True,
+ ),
+]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("case", CLEANUP_CASES, ids=[c.id for c in CLEANUP_CASES])
+async def test_cleanup_handler_pins_webhook_screenshot_and_fail_task(
+ monkeypatch: pytest.MonkeyPatch, case: CleanupHandlerCase
+) -> None:
+ """Each cleanup handler runs clean_up_task with a pinned webhook/screenshot decision.
+
+ Driven with fail_task reporting the task as failed, so the gated handlers also clean up.
+ """
+ outcome = await _drive_execute_step(monkeypatch, case.exc_factory(), fail_task_result=True)
+
+ assert outcome.raised is None, f"{case.id} unexpectedly propagated {outcome.raised!r}"
+ assert outcome.fail_task_called is case.fail_task_called
+ assert outcome.cleanup_called is True
+ assert outcome.effective_webhook is case.effective_webhook
+ assert outcome.effective_final_screenshot is case.effective_final_screenshot
+ # Every cleanup handler returns (step, detailed_output, next_step); nothing advanced here.
+ assert outcome.returned is not None
+ assert outcome.returned[0].step_id == "step-0"
+ assert outcome.returned[2] is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("case", CLEANUP_CASES, ids=[c.id for c in CLEANUP_CASES])
+async def test_cleanup_gating_when_fail_task_reports_not_failed(
+ monkeypatch: pytest.MonkeyPatch, case: CleanupHandlerCase
+) -> None:
+ """Pin whether clean_up_task is skipped when fail_task reports the task was NOT failed.
+
+ Only three handlers (step-termination, failed-to-navigate, generic Exception) gate
+ cleanup on that result; the rest clean up unconditionally.
+ """
+ outcome = await _drive_execute_step(monkeypatch, case.exc_factory(), fail_task_result=False)
+
+ assert outcome.raised is None
+ assert outcome.cleanup_called is not case.cleanup_gated_on_fail_task
+
+
+@pytest.mark.asyncio
+async def test_step_unable_to_execute_reraises_without_cleanup(monkeypatch: pytest.MonkeyPatch) -> None:
+ """StepUnableToExecuteError propagates out of execute_step; no fail_task, no cleanup."""
+ outcome = await _drive_execute_step(monkeypatch, StepUnableToExecuteError("step-0", "cannot run"))
+
+ assert isinstance(outcome.raised, StepUnableToExecuteError)
+ assert outcome.returned is None
+ assert outcome.fail_task_called is False
+ assert outcome.cleanup_called is False
+
+
+@pytest.mark.asyncio
+async def test_failed_to_send_webhook_is_swallowed_without_cleanup(monkeypatch: pytest.MonkeyPatch) -> None:
+ """FailedToSendWebhook is swallowed (step returned); it never fails or cleans up the task."""
+ outcome = await _drive_execute_step(monkeypatch, FailedToSendWebhook(task_id="task-123"))
+
+ assert outcome.raised is None
+ assert outcome.fail_task_called is False
+ assert outcome.cleanup_called is False
+ assert outcome.returned is not None
+ assert outcome.returned[0].step_id == "step-0"
+ assert outcome.returned[2] is None
diff --git a/tests/unit/test_runs_v2_route.py b/tests/unit/test_runs_v2_route.py
index 076368373..c7904cd37 100644
--- a/tests/unit/test_runs_v2_route.py
+++ b/tests/unit/test_runs_v2_route.py
@@ -131,6 +131,8 @@ async def test_get_workflow_runs_by_id_child_filter_depends_on_route(
search_key="login",
error_code="LOGIN_FAILED",
exclude_child_runs=expected_exclude_child_runs,
+ created_at_start=None,
+ created_at_end=None,
)