fix(SKY-10886): suppress benign asyncio driver-pipe teardown noise (#6730)

skyvern-frontend/src/api/types.ts

@@ -263,6 +263,19 @@ export type OnePasswordItemsApiResponse = {
   items: Array<OnePasswordItemApiResponse>;
 };
 
+export type BitwardenItemApiResponse = {
+  item_id: string;
+  title: string;
+  collection_id?: string | null;
+  credential_type: "password" | "credit_card" | "secret";
+  url?: string | null;
+};
+
+export type BitwardenItemsApiResponse = {
+  configured: boolean;
+  items: Array<BitwardenItemApiResponse>;
+};
+
 export type CreateOnePasswordTokenRequest = {
   token: string;
 };

skyvern-frontend/src/hooks/useBitwardenCredential.ts

@@ -39,9 +39,10 @@ export function useBitwardenCredential() {
         .then((response) => response.data as BitwardenCredentialResponse);
     },
     onSuccess: () => {
-      queryClient.invalidateQueries({
+      void queryClient.invalidateQueries({
         queryKey: ["bitwardenOrganizationAuthToken"],
       });
+      void queryClient.invalidateQueries({ queryKey: ["bitwardenItems"] });
       toast({
         title: "Success",
         description: "Bitwarden credential updated successfully",
@@ -64,6 +65,7 @@ export function useBitwardenCredential() {
   const clearCredentialMutation = useClearOrganizationAuthToken({
     providerPath: "bitwarden",
     queryKey: "bitwardenOrganizationAuthToken",
+    invalidateQueryKeys: ["bitwardenItems"],
     successDescription: "Bitwarden credential cleared successfully",
     errorDescription: "Failed to clear Bitwarden credential",
   });

skyvern-frontend/src/hooks/useEditorOnboardingTour.test.tsx

@@ -374,7 +374,7 @@ describe("useEditorOnboardingTour", () => {
       expect(lastInstance().drive).toHaveBeenCalledOnce();
     });
 
-    it("ignores manual re-trigger when the experiment flag is disabled", () => {
+    it("starts manual re-trigger even when the experiment flag is disabled", () => {
       flagState.variant = false;
       const ctx = makeContext({
         isNewUser: false,
@@ -391,6 +391,19 @@ describe("useEditorOnboardingTour", () => {
         useProductTourStore.getState().requestTour();
       });
 
+      expect(driverMock()).toHaveBeenCalledOnce();
+      expect(lastInstance().drive).toHaveBeenCalledOnce();
+    });
+
+    it("ignores manual re-trigger when no onboarding provider is mounted", () => {
+      renderHook(() => useEditorOnboardingTour(), {
+        wrapper: ({ children }) => <>{children}</>,
+      });
+
+      act(() => {
+        useProductTourStore.getState().requestTour();
+      });
+
       expect(driverMock()).not.toHaveBeenCalled();
     });
 

skyvern-frontend/src/hooks/useEditorOnboardingTour.ts

@@ -259,16 +259,17 @@ function useEditorOnboardingTour(): UseEditorOnboardingTourReturn {
     if (requestedAt === null) return;
     clearRequest();
     // No provider (non-cloud app) means dismissals/completions cannot persist, so
-    // never start an untracked tour - mirrors the auto-start guard.
+    // never start an untracked tour - mirrors the auto-start guard. The experiment
+    // arm only gates auto-start; the menu item and Shift+? are always-available
+    // affordances, so the manual trigger must run regardless of flag enrollment.
     if (onboarding === null) return;
-    if (!isABVariant(flagVariant)) return;
     if (tourActiveRef.current) {
       driverRef.current?.destroy();
       tourActiveRef.current = false;
       setTourActive(false);
     }
     startDriver();
-  }, [requestedAt, clearRequest, startDriver, flagVariant, onboarding]);
+  }, [requestedAt, clearRequest, startDriver, onboarding]);
 
   // Step 2 ("Add blocks") is gated: opening the block library while it is the
   // active step satisfies the gate once and advances. Once satisfied the Next

skyvern-frontend/src/routes/credentials/BitwardenCredentialsList.tsx

@@ -0,0 +1,128 @@
+import { BitwardenIcon } from "@/components/icons/BitwardenIcon";
+import { getHostname } from "@/util/getHostname";
+import { useBitwardenItemsQuery } from "@/routes/workflows/hooks/useBitwardenItemsQuery";
+
+type Props = {
+  search?: string;
+  folderId?: string | null;
+};
+
+function BitwardenCredentialsList({ search, folderId }: Props) {
+  const { data, isLoading, isError } = useBitwardenItemsQuery();
+  const normalizedSearch = search?.trim().toLowerCase() ?? "";
+  const muted = "text-sm text-neutral-600 dark:text-slate-400";
+  const filteredItems = (data?.items ?? []).filter(
+    (item) =>
+      item.credential_type === "password" &&
+      (!normalizedSearch ||
+        item.title.toLowerCase().includes(normalizedSearch)),
+  );
+  const hasNonPasswordItems = (data?.items ?? []).some(
+    (item) => item.credential_type !== "password",
+  );
+
+  const header = (
+    <div className="flex items-center gap-2">
+      <BitwardenIcon className="size-5 shrink-0" />
+      <div className="min-w-0 space-y-1">
+        <p className="text-sm font-medium">Bitwarden</p>
+        <p className={muted}>Read-only — managed in your Bitwarden account</p>
+      </div>
+    </div>
+  );
+
+  if (folderId || isLoading) {
+    return null;
+  }
+
+  if (isError) {
+    return (
+      <div className="space-y-4">
+        {header}
+        <div className="rounded-lg bg-slate-elevation2 p-4 text-sm text-neutral-600 dark:text-slate-400">
+          Couldn&apos;t load Bitwarden items.
+        </div>
+      </div>
+    );
+  }
+
+  if (!data?.configured) {
+    return null;
+  }
+
+  if (filteredItems.length === 0) {
+    if (!hasNonPasswordItems) {
+      return null;
+    }
+    return (
+      <div className="space-y-4">
+        {header}
+        <p className={muted}>
+          Credit cards and secrets are available in the workflow editor.
+        </p>
+      </div>
+    );
+  }
+
+  return (
+    <div className="space-y-4">
+      {header}
+      <div className="space-y-5">
+        {filteredItems.map((item) => {
+          const rows = [
+            item.url && {
+              label: "Website",
+              value: getHostname(item.url) ?? item.url,
+            },
+            item.collection_id && {
+              label: "Collection ID",
+              value: item.collection_id,
+            },
+            { label: "Item ID", value: item.item_id },
+          ].filter(Boolean) as Array<{ label: string; value: string }>;
+
+          return (
+            <div
+              className="flex gap-5 rounded-lg bg-slate-elevation2 p-4"
+              key={item.item_id}
+            >
+              <div className="flex w-48 items-center gap-2">
+                <BitwardenIcon className="size-5 shrink-0" />
+                <div className="min-w-0 space-y-1">
+                  <p className="truncate" title={item.title}>
+                    {item.title}
+                  </p>
+                  {item.collection_id && (
+                    <p
+                      className={`truncate ${muted}`}
+                      title={item.collection_id}
+                    >
+                      {item.collection_id}
+                    </p>
+                  )}
+                </div>
+              </div>
+              <div className="flex gap-5 border-l pl-5">
+                {rows.map((row) => (
+                  <div className="min-w-0 space-y-2" key={row.label}>
+                    <p className={muted}>{row.label}</p>
+                    <p className="truncate text-sm" title={row.value}>
+                      {row.value}
+                    </p>
+                  </div>
+                ))}
+              </div>
+            </div>
+          );
+        })}
+      </div>
+      {hasNonPasswordItems && (
+        <p className={muted}>
+          Credit cards and secrets are available in the workflow editor.
+        </p>
+      )}
+    </div>
+  );
+}
+
+export { BitwardenCredentialsList };

skyvern-frontend/src/routes/credentials/CredentialsPage.tsx

@@ -9,6 +9,7 @@ import {
 } from "./useCredentialModalState";
 import { CredentialsModal } from "./CredentialsModal";
 import { CredentialsList } from "./CredentialsList";
+import { BitwardenCredentialsList } from "./BitwardenCredentialsList";
 import { OnePasswordCredentialsList } from "./OnePasswordCredentialsList";
 import { useBackgroundCredentialTest } from "./useBackgroundCredentialTest";
 import {
@@ -322,6 +323,10 @@ function CredentialsPage() {
             search={debouncedSearch}
             folderId={selectedFolderId}
           />
+          <BitwardenCredentialsList
+            search={debouncedSearch}
+            folderId={selectedFolderId}
+          />
         </TabsContent>
 
         <TabsContent value="creditCards" className="space-y-4">

skyvern-frontend/src/routes/workflows/components/BitwardenCredentialUi.test.tsx

@@ -0,0 +1,126 @@
+// @vitest-environment jsdom
+
+import { cleanup, render, screen } from "@testing-library/react";
+import type { ReactNode } from "react";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import { BitwardenCredentialsList } from "@/routes/credentials/BitwardenCredentialsList";
+import { useBitwardenItemsQuery } from "../hooks/useBitwardenItemsQuery";
+import { BitwardenItemSelector } from "./BitwardenItemSelector";
+
+vi.mock("@/components/ui/select", () => {
+  const Pass = ({ children }: { children?: ReactNode }) => <>{children}</>;
+  return {
+    Select: Pass,
+    SelectContent: Pass,
+    SelectItemText: Pass,
+    SelectTrigger: Pass,
+    SelectValue: ({ placeholder }: { placeholder?: string }) => (
+      <span>{placeholder}</span>
+    ),
+    CustomSelectItem: Pass,
+  };
+});
+
+vi.mock("../hooks/useBitwardenItemsQuery", () => ({
+  useBitwardenItemsQuery: vi.fn(),
+}));
+
+const mockedUseBitwardenItemsQuery = vi.mocked(useBitwardenItemsQuery);
+const items = [
+  {
+    item_id: "password-item",
+    title: "Acme Login",
+    collection_id: "collection-1",
+    credential_type: "password" as const,
+    url: "https://app.acme.test",
+  },
+  {
+    item_id: "card-item",
+    title: "Acme Card",
+    collection_id: "collection-2",
+    credential_type: "credit_card" as const,
+    url: null,
+  },
+  {
+    item_id: "card-without-collection",
+    title: "Personal Card",
+    collection_id: null,
+    credential_type: "credit_card" as const,
+    url: null,
+  },
+];
+
+function mockItems(configured = true, itemOverrides = items) {
+  mockedUseBitwardenItemsQuery.mockReturnValue({
+    data: { configured, items: itemOverrides },
+    isLoading: false,
+    isError: false,
+  } as ReturnType<typeof useBitwardenItemsQuery>);
+}
+
+afterEach(() => {
+  cleanup();
+  vi.clearAllMocks();
+});
+
+describe("Bitwarden credential UI", () => {
+  it("renders read-only password items on the credentials page", () => {
+    mockItems();
+    render(<BitwardenCredentialsList />);
+
+    expect(screen.getByText("Bitwarden")).toBeTruthy();
+    expect(screen.getByText("Acme Login")).toBeTruthy();
+    expect(screen.getByText("app.acme.test")).toBeTruthy();
+    expect(
+      screen.getByText(
+        "Credit cards and secrets are available in the workflow editor.",
+      ),
+    ).toBeTruthy();
+    expect(screen.queryByText("Acme Card")).toBeNull();
+  });
+
+  it("shows the Bitwarden footnote when only non-password items are available", () => {
+    mockItems(true, [items[1]!, items[2]!]);
+    render(<BitwardenCredentialsList />);
+
+    expect(screen.getByText("Bitwarden")).toBeTruthy();
+    expect(
+      screen.getByText(
+        "Credit cards and secrets are available in the workflow editor.",
+      ),
+    ).toBeTruthy();
+    expect(screen.queryByText("Acme Card")).toBeNull();
+  });
+
+  it("filters workflow-picker credit cards to items with collection IDs", () => {
+    mockItems();
+    render(
+      <BitwardenItemSelector
+        itemId=""
+        credentialDataType="creditCard"
+        onSelect={vi.fn()}
+      />,
+    );
+
+    expect(screen.getByText("Acme Card")).toBeTruthy();
+    expect(screen.queryByText("Acme Login")).toBeNull();
+    expect(screen.queryByText("Personal Card")).toBeNull();
+  });
+
+  it("explains when credit cards exist but none are collection-scoped", () => {
+    mockItems(true, [items[2]!]);
+    render(
+      <BitwardenItemSelector
+        itemId=""
+        credentialDataType="creditCard"
+        onSelect={vi.fn()}
+      />,
+    );
+
+    expect(
+      screen.getByText("No collection-scoped Bitwarden credit cards found"),
+    ).toBeTruthy();
+    expect(screen.queryByText("Personal Card")).toBeNull();
+  });
+});

skyvern-frontend/src/routes/workflows/components/BitwardenItemSelector.tsx

@@ -0,0 +1,91 @@
+import { BitwardenIcon } from "@/components/icons/BitwardenIcon";
+import {
+  CustomSelectItem,
+  Select,
+  SelectContent,
+  SelectItemText,
+  SelectTrigger,
+  SelectValue,
+} from "@/components/ui/select";
+import { Skeleton } from "@/components/ui/skeleton";
+import { useBitwardenItemsQuery } from "../hooks/useBitwardenItemsQuery";
+
+type Props = {
+  itemId: string;
+  onSelect: (collectionId: string | null, itemId: string) => void;
+  credentialDataType: "password" | "creditCard";
+};
+
+function BitwardenItemSelector({
+  itemId,
+  onSelect,
+  credentialDataType,
+}: Props) {
+  const { data, isLoading, isError } = useBitwardenItemsQuery();
+  const bitwardenItems = data?.items ?? [];
+  const items = bitwardenItems.filter((item) =>
+    credentialDataType === "password"
+      ? item.credential_type === "password"
+      : item.credential_type === "credit_card" && item.collection_id,
+  );
+  const hasUnselectableCreditCards =
+    credentialDataType === "creditCard" &&
+    bitwardenItems.some(
+      (item) => item.credential_type === "credit_card" && !item.collection_id,
+    );
+
+  if (isLoading) {
+    return <Skeleton className="h-10 w-full" />;
+  }
+
+  const selectedValue = items.some((item) => item.item_id === itemId)
+    ? itemId
+    : "";
+  const message = isError
+    ? "Couldn't load Bitwarden items. Enter a value manually instead."
+    : !data?.configured
+      ? "Connect Bitwarden in Settings to list items"
+      : items.length === 0
+        ? hasUnselectableCreditCards
+          ? "No collection-scoped Bitwarden credit cards found"
+          : "No Bitwarden items found"
+        : null;
+
+  return (
+    <Select
+      value={selectedValue}
+      onValueChange={(value) => {
+        const item = items.find((item) => item.item_id === value);
+        if (item) {
+          onSelect(item.collection_id ?? null, item.item_id);
+        }
+      }}
+    >
+      <SelectTrigger>
+        <SelectValue placeholder="Select a Bitwarden item" />
+      </SelectTrigger>
+      <SelectContent>
+        {message && (
+          <div className="px-2 py-1.5 text-xs text-slate-400">{message}</div>
+        )}
+        {items.map((item) => (
+          <CustomSelectItem key={item.item_id} value={item.item_id}>
+            <div className="flex min-w-0 items-center gap-2">
+              <BitwardenIcon className="size-4" />
+              <span className="min-w-0 flex-1 truncate text-sm font-medium">
+                <SelectItemText>{item.title}</SelectItemText>
+              </span>
+              {item.collection_id && (
+                <span className="shrink-0 rounded bg-slate-800 px-1.5 py-0.5 text-[10px] text-slate-400">
+                  {item.collection_id}
+                </span>
+              )}
+            </div>
+          </CustomSelectItem>
+        ))}
+      </SelectContent>
+    </Select>
+  );
+}
+
+export { BitwardenItemSelector };

skyvern-frontend/src/routes/workflows/editor/panels/WorkflowParameterEditPanel.helpers.ts

@@ -4,8 +4,8 @@ import {
 } from "../../types/workflowTypes";
 import {
   ParametersState,
-  parameterIsAzureVaultCredential,
   parameterIsBitwardenCredential,
+  parameterIsAzureVaultCredential,
   parameterIsSkyvernCredential,
 } from "../types";
 
@@ -64,6 +64,35 @@ export function detectInitialCredentialSource(
   return isCloud ? "skyvern" : "bitwarden";
 }
 
+export function detectInitialBitwardenManualEntry(
+  initialValues: ParametersState[number] | undefined,
+): boolean {
+  if (!initialValues) return false;
+
+  if (
+    initialValues.parameterType === "credential" &&
+    parameterIsBitwardenCredential(initialValues)
+  ) {
+    const itemId = initialValues.itemId ?? "";
+    const collectionId = initialValues.collectionId ?? "";
+    return (
+      !itemId ||
+      Boolean(initialValues.urlParameterKey) ||
+      itemId.includes("{{") ||
+      collectionId.includes("{{")
+    );
+  }
+
+  if (initialValues.parameterType === "creditCardData") {
+    return (
+      initialValues.collectionId.includes("{{") ||
+      initialValues.itemId.includes("{{")
+    );
+  }
+
+  return false;
+}
+
 export function header(type: WorkflowEditorParameterType, isEdit: boolean) {
   const prefix = isEdit ? "Edit" : "Add";
   if (type === "workflow") {

skyvern-frontend/src/routes/workflows/editor/panels/WorkflowParameterEditPanel.test.ts

@@ -1,11 +1,13 @@
 import { describe, expect, it } from "vitest";
 import { ParametersState } from "../types";
 import {
+  detectInitialBitwardenManualEntry,
   detectInitialCredentialDataType,
   detectInitialCredentialSource,
   detectInitialParameterTypeSelection,
   header,
 } from "./WorkflowParameterEditPanel.helpers";
+import { validateBitwardenLoginCredential } from "./util";
 
 type Parameter = ParametersState[number];
 
@@ -184,3 +186,70 @@ describe("detectInitialCredentialSource", () => {
     );
   });
 });
+
+describe("validateBitwardenLoginCredential", () => {
+  it("requires a collection or item reference", () => {
+    expect(validateBitwardenLoginCredential(null, null, null)).toBe(
+      "Collection ID or Item ID is required",
+    );
+  });
+
+  it("allows picker-selected items with collection IDs and no URL lookup key", () => {
+    expect(
+      validateBitwardenLoginCredential("collection-id", "item-id", null),
+    ).toBeNull();
+  });
+
+  it("still requires URL lookup key for collection-only login credentials", () => {
+    expect(validateBitwardenLoginCredential("collection-id", null, null)).toBe(
+      "URL Input Key is required when collection ID is used",
+    );
+  });
+
+  it("allows item-only login credentials", () => {
+    expect(validateBitwardenLoginCredential(null, "item-id", null)).toBeNull();
+  });
+});
+
+describe("detectInitialBitwardenManualEntry", () => {
+  it("keeps picker-selected password items with collection IDs in picker mode", () => {
+    expect(detectInitialBitwardenManualEntry(bitwardenLogin)).toBe(false);
+  });
+
+  it("uses manual entry for URL-based Bitwarden password lookup", () => {
+    expect(
+      detectInitialBitwardenManualEntry({
+        ...bitwardenLogin,
+        itemId: null,
+        urlParameterKey: "url",
+      }),
+    ).toBe(true);
+  });
+
+  it("uses manual entry for templated Bitwarden password item IDs", () => {
+    expect(
+      detectInitialBitwardenManualEntry({
+        ...bitwardenLogin,
+        itemId: "{{ item_id }}",
+      }),
+    ).toBe(true);
+  });
+
+  it("uses manual entry for templated Bitwarden password collection IDs", () => {
+    expect(
+      detectInitialBitwardenManualEntry({
+        ...bitwardenLogin,
+        collectionId: "{{ collection_id }}",
+      }),
+    ).toBe(true);
+  });
+
+  it("uses manual entry for templated Bitwarden credit-card IDs", () => {
+    expect(
+      detectInitialBitwardenManualEntry({
+        ...creditCardCredential,
+        collectionId: "{{ collection_id }}",
+      }),
+    ).toBe(true);
+  });
+});

skyvern-frontend/src/routes/workflows/editor/panels/WorkflowParameterEditPanel.tsx

@@ -16,6 +16,7 @@ import { cn } from "@/util/utils";
 import CloudContext from "@/store/CloudContext";
 import { CodeIcon, Cross2Icon } from "@radix-ui/react-icons";
 import { useContext, useEffect, useRef, useState } from "react";
+import { BitwardenItemSelector } from "../../components/BitwardenItemSelector";
 import { CredentialParameterSourceSelector } from "../../components/CredentialParameterSourceSelector";
 import { OnePasswordItemSelector } from "../../components/OnePasswordItemSelector";
 import { SourceParameterKeySelector } from "../../components/SourceParameterKeySelector";
@@ -40,6 +41,7 @@ import { useCustomCredentialServiceConfig } from "@/hooks/useCustomCredentialSer
 import {
   CredentialDataType,
   CredentialSource,
+  detectInitialBitwardenManualEntry,
   detectInitialCredentialDataType,
   detectInitialCredentialSource,
   detectInitialParameterTypeSelection,
@@ -140,6 +142,63 @@ function validateParameterKey(key: string): string | null {
   return null;
 }
 
+function BitwardenItemFieldHeader({
+  manualEntry,
+  onToggle,
+  tooltip,
+}: {
+  manualEntry: boolean;
+  onToggle: () => void;
+  tooltip: string;
+}) {
+  return (
+    <div className="flex items-center justify-between gap-2">
+      <div className="flex items-center gap-2">
+        <Label className="text-xs text-slate-300">Bitwarden Item</Label>
+        <HelpTooltip content={tooltip} />
+      </div>
+      <button
+        type="button"
+        aria-pressed={manualEntry}
+        title={
+          manualEntry
+            ? "Pick from your Bitwarden items"
+            : "Enter a custom value"
+        }
+        className={cn(
+          "rounded p-1 text-slate-400 transition-colors hover:text-slate-200",
+          manualEntry && "bg-slate-700 text-slate-100",
+        )}
+        onClick={onToggle}
+      >
+        <CodeIcon className="size-4" />
+      </button>
+    </div>
+  );
+}
+
+function BitwardenManualInput({
+  label,
+  onChange,
+  tooltip,
+  value,
+}: {
+  label: string;
+  onChange: (value: string) => void;
+  tooltip: string;
+  value: string;
+}) {
+  return (
+    <div className="space-y-1">
+      <div className="flex gap-2">
+        <Label className="text-xs text-slate-300">{label}</Label>
+        <HelpTooltip content={tooltip} />
+      </div>
+      <Input value={value} onChange={(e) => onChange(e.target.value)} />
+    </div>
+  );
+}
+
 function WorkflowParameterEditPanel({
   type,
   onClose,
@@ -268,6 +327,9 @@ function WorkflowParameterEditPanel({
 
   const [bitwardenLoginCredentialItemId, setBitwardenLoginCredentialItemId] =
     useState(isBitwardenCredential ? (initialValues?.itemId ?? "") : "");
+  const [bitwardenManualEntry, setBitwardenManualEntry] = useState(
+    detectInitialBitwardenManualEntry(initialValues),
+  );
 
   const [azureVaultName, setAzureVaultName] = useState(
     isAzureVaultCredential ? initialValues.vaultName : "",
@@ -287,6 +349,11 @@ function WorkflowParameterEditPanel({
     setCredentialDataType(newDataType);
     setOpVaultId("");
     setOpItemId("");
+    setBitwardenLoginCredentialItemId("");
+    setBitwardenCollectionId("");
+    setUrlParameterKey("");
+    setSensitiveInformationItemId("");
+    setBitwardenManualEntry(false);
     const availableSources = getAvailableSourcesForDataType(
       newDataType,
       isCloud,
@@ -587,46 +654,54 @@ function WorkflowParameterEditPanel({
 
           {/* Bitwarden Password Fields */}
           {showBitwardenPasswordFields && (
-            <>
-              <div className="space-y-1">
-                <div className="flex gap-2">
-                  <Label className="text-xs text-slate-300">
-                    URL Parameter Key
-                  </Label>
-                  <HelpTooltip content="Optional. The agent input key that holds the URL. If provided, Skyvern will match the credential based on this URL." />
-                </div>
-                <Input
-                  value={urlParameterKey}
-                  onChange={(e) => setUrlParameterKey(e.target.value)}
-                />
-              </div>
-              <div className="space-y-1">
-                <div className="flex gap-2">
-                  <Label className="text-xs text-slate-300">
-                    Bitwarden Collection ID
-                  </Label>
-                  <HelpTooltip content="Find in the Bitwarden collection URL. Supports agent inputs." />
-                </div>
-                <Input
-                  value={bitwardenCollectionId}
-                  onChange={(e) => setBitwardenCollectionId(e.target.value)}
-                />
-              </div>
-              <div className="space-y-1">
-                <div className="flex gap-2">
-                  <Label className="text-xs text-slate-300">
-                    Bitwarden Item ID
-                  </Label>
-                  <HelpTooltip content="Find in /#/vault?itemId=[ITEM_ID]. Supports agent inputs." />
-                </div>
-                <Input
-                  value={bitwardenLoginCredentialItemId}
-                  onChange={(e) =>
-                    setBitwardenLoginCredentialItemId(e.target.value)
+            <div className="space-y-1">
+              <BitwardenItemFieldHeader
+                manualEntry={bitwardenManualEntry}
+                tooltip="Pick an item from your connected Bitwarden account. Click the </> button to pass in a custom value instead, for example a dynamic {{ input }} reference or raw Bitwarden IDs."
+                onToggle={() => {
+                  if (bitwardenManualEntry) {
+                    if (bitwardenLoginCredentialItemId.includes("{{")) {
+                      setBitwardenLoginCredentialItemId("");
+                    }
+                    setUrlParameterKey("");
+                    setBitwardenCollectionId("");
                   }
+                  setBitwardenManualEntry((prev) => !prev);
+                }}
+              />
+              {bitwardenManualEntry ? (
+                <div className="space-y-3">
+                  <BitwardenManualInput
+                    label="URL Parameter Key"
+                    value={urlParameterKey}
+                    onChange={setUrlParameterKey}
+                    tooltip="Optional. The agent input key that holds the URL. If provided, Skyvern will match the credential based on this URL."
+                  />
+                  <BitwardenManualInput
+                    label="Bitwarden Collection ID"
+                    value={bitwardenCollectionId}
+                    onChange={setBitwardenCollectionId}
+                    tooltip="Find in the Bitwarden collection URL. Supports agent inputs."
+                  />
+                  <BitwardenManualInput
+                    label="Bitwarden Item ID"
+                    value={bitwardenLoginCredentialItemId}
+                    onChange={setBitwardenLoginCredentialItemId}
+                    tooltip="Find in /#/vault?itemId=[ITEM_ID]. Supports agent inputs."
+                  />
+                </div>
+              ) : (
+                <BitwardenItemSelector
+                  itemId={bitwardenLoginCredentialItemId}
+                  credentialDataType="password"
+                  onSelect={(collectionId, itemId) => {
+                    setBitwardenLoginCredentialItemId(itemId);
+                    setBitwardenCollectionId(collectionId ?? "");
+                    setUrlParameterKey("");
+                  }}
                 />
-              </div>
-            </>
+              )}
+            </div>
           )}
 
           {/* Bitwarden Secret Fields */}
@@ -672,34 +747,48 @@ function WorkflowParameterEditPanel({
 
           {/* Bitwarden Credit Card Fields */}
           {showBitwardenCreditCardFields && (
-            <>
-              <div className="space-y-1">
-                <div className="flex gap-2">
-                  <Label className="text-xs text-slate-300">
-                    Bitwarden Collection ID
-                  </Label>
-                  <HelpTooltip content="Collection containing the credit card. Supports agent inputs." />
-                </div>
-                <Input
-                  value={bitwardenCollectionId}
-                  onChange={(e) => setBitwardenCollectionId(e.target.value)}
-                />
-              </div>
-              <div className="space-y-1">
-                <div className="flex gap-2">
-                  <Label className="text-xs text-slate-300">
-                    Bitwarden Item ID
-                  </Label>
-                  <HelpTooltip content="Credit card item ID. Supports agent inputs." />
-                </div>
-                <Input
-                  value={sensitiveInformationItemId}
-                  onChange={(e) =>
-                    setSensitiveInformationItemId(e.target.value)
+            <div className="space-y-1">
+              <BitwardenItemFieldHeader
+                manualEntry={bitwardenManualEntry}
+                tooltip="Pick a credit card from your connected Bitwarden account. Click the </> button to pass in custom collection/item IDs or dynamic {{ input }} references."
+                onToggle={() => {
+                  if (
+                    bitwardenManualEntry &&
+                    (bitwardenCollectionId.includes("{{") ||
+                      sensitiveInformationItemId.includes("{{"))
+                  ) {
+                    setBitwardenCollectionId("");
+                    setSensitiveInformationItemId("");
                   }
+                  setBitwardenManualEntry((prev) => !prev);
+                }}
+              />
+              {bitwardenManualEntry ? (
+                <div className="space-y-3">
+                  <BitwardenManualInput
+                    label="Bitwarden Collection ID"
+                    value={bitwardenCollectionId}
+                    onChange={setBitwardenCollectionId}
+                    tooltip="Collection containing the credit card. Supports agent inputs."
+                  />
+                  <BitwardenManualInput
+                    label="Bitwarden Item ID"
+                    value={sensitiveInformationItemId}
+                    onChange={setSensitiveInformationItemId}
+                    tooltip="Credit card item ID. Supports agent inputs."
+                  />
+                </div>
+              ) : (
+                <BitwardenItemSelector
+                  itemId={sensitiveInformationItemId}
+                  credentialDataType="creditCard"
+                  onSelect={(collectionId, itemId) => {
+                    setBitwardenCollectionId(collectionId ?? "");
+                    setSensitiveInformationItemId(itemId);
+                  }}
                 />
-              </div>
-            </>
+              )}
+            </div>
           )}
 
           {/* 1Password Fields */}

skyvern-frontend/src/routes/workflows/editor/panels/util.ts

@@ -6,7 +6,7 @@ export function validateBitwardenLoginCredential(
   if (!collectionId && !itemId) {
     return "Collection ID or Item ID is required";
   }
-  if (collectionId && !urlParameterKey) {
+  if (collectionId && !itemId && !urlParameterKey) {
     return "URL Input Key is required when collection ID is used";
   }
   return null;

skyvern-frontend/src/routes/workflows/hooks/useBitwardenItemsQuery.ts

@@ -0,0 +1,32 @@
+import { getClient } from "@/api/AxiosClient";
+import { BitwardenItemsApiResponse } from "@/api/types";
+import { useCredentialGetter } from "@/hooks/useCredentialGetter";
+import { useWorkflowScopeReadOnly } from "@/routes/workflows/editor/WorkflowScopeContext";
+import { useQuery } from "@tanstack/react-query";
+
+type QueryReturnType = BitwardenItemsApiResponse;
+type UseQueryOptions = Omit<
+  Parameters<typeof useQuery<QueryReturnType>>[0],
+  "queryKey" | "queryFn"
+>;
+
+function useBitwardenItemsQuery(props: UseQueryOptions = {}) {
+  const credentialGetter = useCredentialGetter();
+  const scopeReadOnly = useWorkflowScopeReadOnly();
+
+  return useQuery<BitwardenItemsApiResponse>({
+    queryKey: ["bitwardenItems"],
+    queryFn: async () => {
+      const client = await getClient(credentialGetter, "sans-api-v1");
+      return client
+        .get("/credentials/bitwarden/items")
+        .then((res) => res.data as BitwardenItemsApiResponse);
+    },
+    // Bitwarden items change rarely; avoid refetching every time the panel remounts.
+    staleTime: 60_000,
+    ...props,
+    enabled: props.enabled !== false && !scopeReadOnly,
+  });
+}
+
+export { useBitwardenItemsQuery };

skyvern/forge/sdk/forge_log.py

@@ -28,6 +28,24 @@ LOGGING_LEVEL_MAP: dict[str, int] = {
 # Resolved once at setup time and injected into every log event.
 _entrypoint: str = "unknown"
 
+_DRIVER_PIPE_CLOSED_ERROR = "Connection closed while reading from the driver"
+_ORPHANED_FUTURE_MESSAGE = "Future exception was never retrieved"
+
+
+class _DriverPipeNoiseFilter(logging.Filter):
+    """Drop asyncio's orphaned-future noise from a torn-down Playwright driver pipe"""
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        message = record.getMessage()
+        if _ORPHANED_FUTURE_MESSAGE not in message:
+            return True
+        if _DRIVER_PIPE_CLOSED_ERROR in message:
+            return False
+        exc = record.exc_info[1] if record.exc_info and len(record.exc_info) > 1 else None
+        if exc is not None and _DRIVER_PIPE_CLOSED_ERROR in str(exc):
+            return False
+        return True
+
 
 def _get_entrypoint() -> str:
     """Derive a human-readable entrypoint name for the current process.
@@ -479,3 +497,10 @@ def setup_logger() -> None:
 
     # Anthropic Bedrock SDK emits high-volume WARN noise; keep only its errors.
     logging.getLogger("anthropic").setLevel(logging.ERROR)
+
+    # Drop asyncio's orphaned-future noise from torn-down Playwright driver pipes (logged at
+    # ERROR but non-actionable). setup_logger may run more than once (uvicorn reload), so keep
+    # exactly one instance instead of stacking duplicates.
+    asyncio_logger = logging.getLogger("asyncio")
+    asyncio_logger.filters = [f for f in asyncio_logger.filters if not isinstance(f, _DriverPipeNoiseFilter)]
+    asyncio_logger.addFilter(_DriverPipeNoiseFilter())

skyvern/forge/sdk/routes/credentials.py

@@ -62,6 +62,7 @@ from skyvern.forge.sdk.routes.code_samples import (
 from skyvern.forge.sdk.routes.routers import base_router, legacy_base_router
 from skyvern.forge.sdk.routes.trigger_type import workflow_run_trigger_type_from_user_agent
 from skyvern.forge.sdk.schemas.credentials import (
+    BitwardenItemsResponse,
     CancelTestResponse,
     CreateCredentialRequest,
     Credential,
@@ -1936,6 +1937,50 @@ async def list_onepassword_items(
         raise HTTPException(status_code=502, detail="Failed to list 1Password items") from e
 
 
+@base_router.get(
+    "/credentials/bitwarden/items",
+    response_model=BitwardenItemsResponse,
+    summary="List Bitwarden item metadata",
+    description="Lists Bitwarden item metadata for the current organization.",
+    include_in_schema=False,
+)
+@base_router.get(
+    "/credentials/bitwarden/items/",
+    response_model=BitwardenItemsResponse,
+    include_in_schema=False,
+)
+async def list_bitwarden_items(
+    current_org: Organization = Depends(org_auth_service.get_current_org),
+) -> BitwardenItemsResponse:
+    org_auth_token = await app.DATABASE.organizations.get_valid_org_auth_token(
+        current_org.organization_id,
+        OrganizationAuthTokenType.bitwarden_credential.value,
+    )
+    # Org-scoped only: never fall back to global Bitwarden credentials here. In a shared deployment that
+    # would let an org without its own Bitwarden credential browse metadata from the instance/global vault.
+    if not org_auth_token:
+        return BitwardenItemsResponse(configured=False, items=[])
+
+    try:
+        items = await BitwardenService.list_item_overviews(
+            client_id=None,
+            client_secret=None,
+            master_password=org_auth_token.credential.master_password,
+            bw_organization_id=current_org.bw_organization_id,
+            bw_collection_ids=current_org.bw_collection_ids,
+            email=str(org_auth_token.credential.email),
+        )
+        return BitwardenItemsResponse(configured=True, items=items)
+    except Exception as e:
+        LOG.error(
+            "Failed to list Bitwarden items",
+            organization_id=current_org.organization_id,
+            error=str(e),
+            exc_info=True,
+        )
+        raise HTTPException(status_code=502, detail="Failed to list Bitwarden items") from e
+
+
 @base_router.post(
     "/credentials/onepassword/create",
     response_model=CreateOnePasswordTokenResponse,

skyvern/forge/sdk/schemas/credentials.py

@@ -255,6 +255,26 @@ class OnePasswordItemsResponse(BaseModel):
     items: list[OnePasswordItemOverview] = Field(..., description="The available 1Password item metadata")
 
 
+class BitwardenItemOverview(BaseModel):
+    """Response model for Bitwarden item metadata."""
+
+    item_id: str = Field(..., description="The Bitwarden item ID")
+    title: str = Field(..., description="The Bitwarden item title")
+    collection_id: str | None = Field(
+        default=None,
+        description="The ID of a collection containing the item, if available",
+    )
+    credential_type: CredentialType = Field(..., description="The item's credential type")
+    url: str | None = Field(default=None, description="The primary website URL associated with the item, if any")
+
+
+class BitwardenItemsResponse(BaseModel):
+    """Response model for listing Bitwarden item metadata."""
+
+    configured: bool = Field(..., description="Whether Bitwarden credentials are configured")
+    items: list[BitwardenItemOverview] = Field(..., description="The available Bitwarden item metadata")
+
+
 class Credential(BaseModel):
     """Database model for credentials."""
 

skyvern/forge/sdk/services/bitwarden.py

@@ -27,6 +27,7 @@ from skyvern.exceptions import (
 from skyvern.forge.sdk.api.aws import get_aws_client
 from skyvern.forge.sdk.core.aiohttp_helper import aiohttp_delete, aiohttp_get_json, aiohttp_post
 from skyvern.forge.sdk.schemas.credentials import (
+    BitwardenItemOverview,
     CredentialItem,
     CredentialType,
     CreditCardBillingAddress,
@@ -185,6 +186,55 @@ def get_list_response_item_from_bitwarden_item(item: dict) -> CredentialItem:
         raise BitwardenGetItemError(f"Unsupported item type: {item['type']}")
 
 
+def get_bitwarden_item_overview_from_bitwarden_item(
+    item: dict,
+    allowed_collection_ids: list[str] | None = None,
+    preferred_collection_id: str | None = None,
+) -> BitwardenItemOverview | None:
+    item_id, title, item_type = item.get("id"), item.get("name"), item.get("type")
+    if not isinstance(item_type, int):
+        return None
+    try:
+        credential_type = {
+            BitwardenItemType.LOGIN: CredentialType.PASSWORD,
+            BitwardenItemType.CREDIT_CARD: CredentialType.CREDIT_CARD,
+            BitwardenItemType.SECURE_NOTE: CredentialType.SECRET,
+            BitwardenItemType.IDENTITY: CredentialType.SECRET,
+        }[BitwardenItemType(item_type)]
+    except (TypeError, ValueError):
+        return None
+    if not isinstance(item_id, str) or not isinstance(title, str):
+        return None
+
+    collection_ids = [str(collection_id) for collection_id in item.get("collectionIds") or [] if collection_id]
+    allowed_collection_ids = allowed_collection_ids or []
+    # Collection-scoped CLI queries can omit collectionIds, so keep the queried collection as a fallback.
+    collection_id = (
+        preferred_collection_id
+        if preferred_collection_id and preferred_collection_id in collection_ids
+        else next((cid for cid in collection_ids if cid in allowed_collection_ids), None)
+        or (collection_ids[0] if collection_ids else preferred_collection_id)
+    )
+    login = item.get("login")
+    uris = login.get("uris") if isinstance(login, dict) else []
+    url = next(
+        (
+            uri.get("uri")
+            for uri in uris or []
+            if isinstance(uri, dict) and isinstance(uri.get("uri"), str) and uri.get("uri")
+        ),
+        None,
+    )
+
+    return BitwardenItemOverview(
+        item_id=item_id,
+        title=title,
+        collection_id=collection_id,
+        credential_type=credential_type,
+        url=url,
+    )
+
+
 def is_valid_email(email: str | None) -> bool:
     if not email:
         return False
@@ -233,6 +283,8 @@ class RunCommandResult(BaseModel):
 
 
 class BitwardenService:
+    _cli_session_lock: asyncio.Lock = asyncio.Lock()
+
     @staticmethod
     def _is_ignorable_login_stderr(stderr: str) -> bool:
         lines = [line.strip() for line in stderr.splitlines() if line.strip()]
@@ -322,6 +374,98 @@ class BitwardenService:
 
         return None
 
+    @staticmethod
+    async def _list_items_using_cli(
+        session_key: str,
+        bw_organization_id: str | None = None,
+        collection_id: str | None = None,
+        timeout: int = 60,
+    ) -> list[dict]:
+        list_command = ["bw", "list", "items", "--session", session_key]
+        if bw_organization_id:
+            list_command.extend(["--organizationid", bw_organization_id])
+        if collection_id:
+            list_command.extend(["--collectionid", collection_id])
+
+        items_result = await BitwardenService.run_command(list_command, timeout=timeout)
+        if items_result.returncode != 0:
+            raise BitwardenListItemsError(f"Failed to list Bitwarden items. Error: {items_result.stderr}")
+
+        try:
+            items = json.loads(items_result.stdout)
+            if isinstance(items, list):
+                return items
+        except json.JSONDecodeError:
+            pass
+        raise BitwardenListItemsError("Failed to parse items JSON")
+
+    @staticmethod
+    async def list_item_overviews(
+        client_id: str | None,
+        client_secret: str | None,
+        master_password: str,
+        bw_organization_id: str | None,
+        bw_collection_ids: list[str] | None,
+        email: str,
+        timeout: int = settings.BITWARDEN_TIMEOUT_SECONDS,
+    ) -> list[BitwardenItemOverview]:
+        if not email or not master_password:
+            raise BitwardenLoginError("Bitwarden item listing requires org-scoped email and master password")
+
+        await BitwardenService._apply_jitter()
+        async with asyncio.timeout(timeout):
+            async with BitwardenService._cli_session_lock:
+                try:
+                    # The Bitwarden CLI stores active login state globally, so the whole session workflow is locked.
+                    await BitwardenService.logout()
+                    await BitwardenService.login(client_id, client_secret, email=email, master_password=master_password)
+                    await BitwardenService.sync()
+                    session_key = await BitwardenService.unlock(master_password)
+                    raw_items_with_preferred_collection: list[tuple[dict, str | None]] = []
+
+                    if bw_organization_id:
+                        raw_items = await BitwardenService._list_items_using_cli(
+                            session_key=session_key,
+                            bw_organization_id=bw_organization_id,
+                            timeout=timeout,
+                        )
+                        allowed_collection_ids = set(bw_collection_ids or [])
+                        if allowed_collection_ids:
+                            raw_items = [
+                                item
+                                for item in raw_items
+                                if any(cid in allowed_collection_ids for cid in item.get("collectionIds") or [])
+                            ]
+                        raw_items_with_preferred_collection = [(item, None) for item in raw_items]
+                    elif bw_collection_ids:
+                        for collection_id in bw_collection_ids:
+                            raw_items = await BitwardenService._list_items_using_cli(
+                                session_key=session_key,
+                                collection_id=collection_id,
+                                timeout=timeout,
+                            )
+                            raw_items_with_preferred_collection.extend((item, collection_id) for item in raw_items)
+                    else:
+                        raw_items = await BitwardenService._list_items_using_cli(
+                            session_key=session_key, timeout=timeout
+                        )
+                        raw_items_with_preferred_collection = [(item, None) for item in raw_items]
+
+                    overviews: list[BitwardenItemOverview] = []
+                    seen_item_ids: set[str] = set()
+                    for item, preferred_collection_id in raw_items_with_preferred_collection:
+                        overview = get_bitwarden_item_overview_from_bitwarden_item(
+                            item,
+                            allowed_collection_ids=bw_collection_ids,
+                            preferred_collection_id=preferred_collection_id,
+                        )
+                        if overview is not None and overview.item_id not in seen_item_ids:
+                            seen_item_ids.add(overview.item_id)
+                            overviews.append(overview)
+                    return overviews
+                finally:
+                    await BitwardenService.logout()
+
     @staticmethod
     async def get_secret_value_from_url(
         client_id: str | None,
@@ -352,18 +496,19 @@ class BitwardenService:
             timeout = (i + 1) * timeout
             try:
                 async with asyncio.timeout(timeout):
-                    return await BitwardenService._get_secret_value_from_url(
-                        client_id=client_id,
-                        client_secret=client_secret,
-                        master_password=master_password,
-                        bw_organization_id=bw_organization_id,
-                        bw_collection_ids=bw_collection_ids,
-                        url=url,
-                        collection_id=collection_id,
-                        item_id=item_id,
-                        timeout=timeout,
-                        email=email,
-                    )
+                    async with BitwardenService._cli_session_lock:
+                        return await BitwardenService._get_secret_value_from_url(
+                            client_id=client_id,
+                            client_secret=client_secret,
+                            master_password=master_password,
+                            bw_organization_id=bw_organization_id,
+                            bw_collection_ids=bw_collection_ids,
+                            url=url,
+                            collection_id=collection_id,
+                            item_id=item_id,
+                            timeout=timeout,
+                            email=email,
+                        )
             except BitwardenAccessDeniedError as e:
                 raise e
             except Exception as e:
@@ -548,17 +693,18 @@ class BitwardenService:
             await BitwardenService._apply_jitter()
         try:
             async with asyncio.timeout(timeout):
-                return await BitwardenService._get_sensitive_information_from_identity(
-                    client_id=client_id,
-                    client_secret=client_secret,
-                    master_password=master_password,
-                    bw_organization_id=bw_organization_id,
-                    bw_collection_ids=bw_collection_ids,
-                    collection_id=collection_id,
-                    identity_key=identity_key,
-                    identity_fields=identity_fields,
-                    email=email,
-                )
+                async with BitwardenService._cli_session_lock:
+                    return await BitwardenService._get_sensitive_information_from_identity(
+                        client_id=client_id,
+                        client_secret=client_secret,
+                        master_password=master_password,
+                        bw_organization_id=bw_organization_id,
+                        bw_collection_ids=bw_collection_ids,
+                        collection_id=collection_id,
+                        identity_key=identity_key,
+                        identity_fields=identity_fields,
+                        email=email,
+                    )
         except BitwardenAccessDeniedError as e:
             raise e
         except Exception as e:
@@ -852,16 +998,17 @@ class BitwardenService:
             await BitwardenService._apply_jitter()
         try:
             async with asyncio.timeout(settings.BITWARDEN_TIMEOUT_SECONDS):
-                return await BitwardenService._get_credit_card_data(
-                    client_id=client_id,
-                    client_secret=client_secret,
-                    master_password=master_password,
-                    bw_organization_id=bw_organization_id,
-                    bw_collection_ids=bw_collection_ids,
-                    collection_id=collection_id,
-                    item_id=item_id,
-                    email=email,
-                )
+                async with BitwardenService._cli_session_lock:
+                    return await BitwardenService._get_credit_card_data(
+                        client_id=client_id,
+                        client_secret=client_secret,
+                        master_password=master_password,
+                        bw_organization_id=bw_organization_id,
+                        bw_collection_ids=bw_collection_ids,
+                        collection_id=collection_id,
+                        item_id=item_id,
+                        email=email,
+                    )
         except BitwardenAccessDeniedError as e:
             raise e
         except Exception as e:

skyvern/forge/sdk/workflow/models/block.py

@@ -4735,8 +4735,9 @@ class FileUploadBlock(Block):
     continue_on_empty: bool = Field(
         default=False,
         description=(
-            "When the run download directory has no files, fail the block (False, default) instead "
-            "of reporting success. Set True to allow an empty upload (success with no files uploaded)."
+            "When the run download directory has no files, allow the empty upload only after confirming no registered, "
+            "browser-session, or alternate candidate downloads exist (False, default). Set True to always allow an "
+            "empty upload."
         ),
     )
 
@@ -4863,6 +4864,194 @@ class FileUploadBlock(Block):
             azure_blob_name=blob_name,
         )
 
+    @staticmethod
+    def _candidate_download_signal_run_ids(
+        *,
+        context: SkyvernContext | None,
+        workflow_run_id: str,
+        run_download_id: str | None,
+    ) -> list[str]:
+        candidate_run_ids: list[str] = []
+        for candidate in (
+            run_download_id,
+            workflow_run_id,
+            context.run_id if context else None,
+            context.workflow_run_id if context else None,
+        ):
+            if candidate and candidate not in candidate_run_ids:
+                candidate_run_ids.append(candidate)
+        if not candidate_run_ids and context and context.task_id:
+            candidate_run_ids.append(context.task_id)
+        return candidate_run_ids
+
+    def _get_files_to_upload_from_download_dir(
+        self,
+        *,
+        download_files_path: str,
+        max_file_count: int,
+    ) -> list[str]:
+        files_to_upload = []
+        if os.path.isdir(download_files_path):
+            files = os.listdir(download_files_path)
+            if len(files) > max_file_count:
+                raise ValueError(f"Too many files in the directory, not uploading. Max: {max_file_count}")
+            for file in files:
+                if os.path.isdir(os.path.join(download_files_path, file)):
+                    LOG.warning("FileUploadBlock Skipping directory", file=file)
+                    continue
+                files_to_upload.append(os.path.join(download_files_path, file))
+        return files_to_upload
+
+    def _get_files_in_alternate_candidate_download_dirs(
+        self,
+        *,
+        context: SkyvernContext | None,
+        workflow_run_id: str,
+        run_download_id: str | None,
+        download_files_path: str,
+        max_file_count: int,
+    ) -> tuple[list[str] | None, str]:
+        """Return alternate local files plus a failure-reason count label.
+
+        None means the local evidence could not be trusted, so execute() fails closed rather than no-oping.
+        """
+        for candidate_run_id in self._candidate_download_signal_run_ids(
+            context=context,
+            workflow_run_id=workflow_run_id,
+            run_download_id=run_download_id,
+        ):
+            candidate_download_files_path = str(get_path_for_workflow_download_directory(candidate_run_id).absolute())
+            if candidate_download_files_path == download_files_path:
+                continue
+            try:
+                alternate_files = self._get_files_to_upload_from_download_dir(
+                    download_files_path=candidate_download_files_path,
+                    max_file_count=max_file_count,
+                )
+            except ValueError:
+                LOG.warning(
+                    "FileUploadBlock found too many files in an alternate candidate download directory",
+                    workflow_run_id=workflow_run_id,
+                    candidate_run_id=candidate_run_id,
+                    download_files_path=download_files_path,
+                    candidate_download_files_path=candidate_download_files_path,
+                    exc_info=True,
+                )
+                return None, "too_many"
+            if alternate_files:
+                LOG.warning(
+                    "FileUploadBlock found files in an alternate candidate download directory",
+                    workflow_run_id=workflow_run_id,
+                    candidate_run_id=candidate_run_id,
+                    download_files_path=download_files_path,
+                    candidate_download_files_path=candidate_download_files_path,
+                    file_count=len(alternate_files),
+                )
+                return alternate_files, str(len(alternate_files))
+        return [], "0"
+
+    async def _get_browser_session_downloaded_files_for_empty_scan(
+        self,
+        *,
+        organization_id: str | None,
+        workflow_run_id: str,
+        workflow_run_block_id: str,
+        browser_session_id: str | None,
+    ) -> list[str] | None:
+        if not browser_session_id:
+            # No persistent-session namespace exists to inspect, so this signal cannot contain downloads.
+            return []
+        if not organization_id:
+            LOG.warning(
+                "FileUploadBlock cannot check browser-session downloads without organization_id",
+                workflow_run_id=workflow_run_id,
+                workflow_run_block_id=workflow_run_block_id,
+                browser_session_id=browser_session_id,
+            )
+            return None
+
+        try:
+            async with asyncio.timeout(GET_DOWNLOADED_FILES_TIMEOUT):
+                return await app.STORAGE.list_downloaded_files_in_browser_session(
+                    organization_id=organization_id,
+                    browser_session_id=browser_session_id,
+                )
+        except asyncio.TimeoutError:
+            LOG.warning(
+                "Timeout checking browser-session downloads for empty FileUploadBlock scan",
+                workflow_run_id=workflow_run_id,
+                workflow_run_block_id=workflow_run_block_id,
+                browser_session_id=browser_session_id,
+            )
+            return None
+        except Exception:
+            LOG.warning(
+                "Failed to check browser-session downloads for empty FileUploadBlock scan",
+                workflow_run_id=workflow_run_id,
+                workflow_run_block_id=workflow_run_block_id,
+                browser_session_id=browser_session_id,
+                exc_info=True,
+            )
+            return None
+
+    async def _get_registered_downloaded_files_for_empty_scan(
+        self,
+        *,
+        organization_id: str | None,
+        workflow_run_id: str,
+        workflow_run_block_id: str,
+        run_download_id: str | None,
+        context: SkyvernContext | None,
+    ) -> list[FileInfo] | None:
+        """Return registered downloads, or None when the signal is unknown.
+
+        A timeout on any candidate stays unknown even if later candidates might be empty; an empty later lookup cannot
+        prove that the timed-out candidate had no downloads, so the caller fails closed.
+        """
+        if not organization_id:
+            LOG.warning(
+                "FileUploadBlock cannot check registered downloads without organization_id",
+                workflow_run_id=workflow_run_id,
+                workflow_run_block_id=workflow_run_block_id,
+            )
+            return None
+
+        registered_downloaded_files: list[FileInfo] = []
+        for candidate_run_id in self._candidate_download_signal_run_ids(
+            context=context,
+            workflow_run_id=workflow_run_id,
+            run_download_id=run_download_id,
+        ):
+            try:
+                async with asyncio.timeout(GET_DOWNLOADED_FILES_TIMEOUT):
+                    registered_downloaded_files.extend(
+                        await app.STORAGE.get_downloaded_files(
+                            organization_id=organization_id,
+                            run_id=candidate_run_id,
+                        )
+                    )
+                    if registered_downloaded_files:
+                        return registered_downloaded_files
+            except asyncio.TimeoutError:
+                LOG.warning(
+                    "Timeout checking registered downloads for empty FileUploadBlock scan",
+                    workflow_run_id=workflow_run_id,
+                    workflow_run_block_id=workflow_run_block_id,
+                    candidate_run_id=candidate_run_id,
+                )
+                return None
+            except Exception:
+                LOG.warning(
+                    "Failed to check registered downloads for empty FileUploadBlock scan",
+                    workflow_run_id=workflow_run_id,
+                    workflow_run_block_id=workflow_run_block_id,
+                    candidate_run_id=candidate_run_id,
+                    exc_info=True,
+                )
+                return None
+
+        return registered_downloaded_files
+
     async def execute(
         self,
         workflow_run_id: str,
@@ -4923,38 +5112,88 @@ class FileUploadBlock(Block):
             )
 
         context = skyvern_context.current()
-        download_files_path = str(
-            get_path_for_workflow_download_directory(
-                resolve_run_download_id(context, fallback_run_id=workflow_run_id)
-            ).absolute()
-        )
+        run_download_id = resolve_run_download_id(context, fallback_run_id=workflow_run_id)
+        download_files_path = str(get_path_for_workflow_download_directory(run_download_id).absolute())
 
-        uploaded_uris = []
+        uploaded_uris: list[str] = []
         try:
             workflow_run_context = self.get_workflow_run_context(workflow_run_id)
             files_to_upload = []
-            if os.path.isdir(download_files_path):
-                files = os.listdir(download_files_path)
-                max_file_count = (
-                    MAX_UPLOAD_FILE_COUNT
-                    if self.storage_type == FileStorageType.S3
-                    else AZURE_BLOB_STORAGE_MAX_UPLOAD_FILE_COUNT
-                )
-                if len(files) > max_file_count:
-                    raise ValueError(f"Too many files in the directory, not uploading. Max: {max_file_count}")
-                for file in files:
-                    if os.path.isdir(os.path.join(download_files_path, file)):
-                        LOG.warning("FileUploadBlock Skipping directory", file=file)
-                        continue
-                    files_to_upload.append(os.path.join(download_files_path, file))
+            max_file_count = (
+                MAX_UPLOAD_FILE_COUNT
+                if self.storage_type == FileStorageType.S3
+                else AZURE_BLOB_STORAGE_MAX_UPLOAD_FILE_COUNT
+            )
+            files_to_upload = self._get_files_to_upload_from_download_dir(
+                download_files_path=download_files_path,
+                max_file_count=max_file_count,
+            )
 
             if not files_to_upload and not self.continue_on_empty:
-                # Never green on a zero-file upload by default: a download-dir/scan-dir desync
-                # would otherwise be silent data loss (SKY-11153). Opt out with continue_on_empty.
+                (
+                    (alternate_files, alternate_file_count),
+                    browser_session_downloaded_files,
+                    registered_downloaded_files,
+                ) = await asyncio.gather(
+                    asyncio.to_thread(
+                        self._get_files_in_alternate_candidate_download_dirs,
+                        context=context,
+                        workflow_run_id=workflow_run_id,
+                        run_download_id=run_download_id,
+                        download_files_path=download_files_path,
+                        max_file_count=max_file_count,
+                    ),
+                    self._get_browser_session_downloaded_files_for_empty_scan(
+                        organization_id=organization_id or workflow_run_context.organization_id,
+                        workflow_run_id=workflow_run_id,
+                        workflow_run_block_id=workflow_run_block_id,
+                        browser_session_id=browser_session_id or (context.browser_session_id if context else None),
+                    ),
+                    self._get_registered_downloaded_files_for_empty_scan(
+                        organization_id=organization_id or workflow_run_context.organization_id,
+                        workflow_run_id=workflow_run_id,
+                        workflow_run_block_id=workflow_run_block_id,
+                        run_download_id=run_download_id,
+                        context=context,
+                    ),
+                )
+                if (
+                    registered_downloaded_files == []
+                    and alternate_files == []
+                    and browser_session_downloaded_files == []
+                ):
+                    LOG.info(
+                        "FileUploadBlock empty scan has no registered downloads; treating as no-op",
+                        workflow_run_id=workflow_run_id,
+                        workflow_run_block_id=workflow_run_block_id,
+                        download_files_path=download_files_path,
+                        storage_type=self.storage_type,
+                    )
+                    await self.record_output_parameter_value(workflow_run_context, workflow_run_id, uploaded_uris)
+                    return await self.build_block_result(
+                        success=True,
+                        failure_reason=None,
+                        output_parameter_value=uploaded_uris,
+                        status=BlockStatus.completed,
+                        workflow_run_block_id=workflow_run_block_id,
+                        organization_id=organization_id,
+                    )
+
+                registered_download_count = (
+                    str(len(registered_downloaded_files)) if registered_downloaded_files is not None else "unknown"
+                )
+                browser_session_download_count = (
+                    str(len(browser_session_downloaded_files))
+                    if browser_session_downloaded_files is not None
+                    else "unknown"
+                )
                 return await self.build_block_result(
                     success=False,
                     failure_reason=(
                         f"No files found to upload in the run download directory ({download_files_path}); "
+                        f"registered_download_count={registered_download_count}; "
+                        f"alternate_file_count={alternate_file_count}; "
+                        f"browser_session_download_count={browser_session_download_count}; "
                         f"nothing was sent to {self.storage_type}."
                     ),
                     output_parameter_value=None,

skyvern/forge/sdk/workflow/workflow_definition_converter.py

@@ -202,7 +202,11 @@ def convert_workflow_definition(
                     workflow_parameter_key=parameter.key,
                     required_value="bitwarden_collection_id or bitwarden_item_id",
                 )
-            if parameter.bitwarden_collection_id and not parameter.url_parameter_key:
+            if (
+                parameter.bitwarden_collection_id
+                and not parameter.bitwarden_item_id
+                and not parameter.url_parameter_key
+            ):
                 raise WorkflowParameterMissingRequiredValue(
                     workflow_parameter_type=ParameterType.BITWARDEN_LOGIN_CREDENTIAL,
                     workflow_parameter_key=parameter.key,

tests/unit/test_bitwarden_null_credentials.py

@@ -12,6 +12,7 @@ spurious "invalid credentials" errors during runs because the real value never g
 typed into the login form.
 """
 
+import asyncio
 import json
 
 import pytest
@@ -191,3 +192,58 @@ async def test_get_credit_card_data_includes_billing_custom_fields(
     assert result["billing_address_country_code"] == "US"
     assert result["billing_email"] == "billing@example.com"
     assert result["metadata_customer_id"] == "cus_123"
+
+
+@pytest.mark.asyncio
+async def test_list_item_overviews_serializes_cli_session_workflows(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first_list_entered = asyncio.Event()
+    release_first_list = asyncio.Event()
+    second_list_entered = asyncio.Event()
+    list_entries: list[str] = []
+
+    async def fake_list_items_using_cli(**_: object) -> list[dict]:
+        if not list_entries:
+            list_entries.append("first")
+            first_list_entered.set()
+            await release_first_list.wait()
+        else:
+            list_entries.append("second")
+            second_list_entered.set()
+        return []
+
+    monkeypatch.setattr(BitwardenService, "_list_items_using_cli", fake_list_items_using_cli)
+
+    first_request = asyncio.create_task(
+        BitwardenService.list_item_overviews(
+            client_id=None,
+            client_secret=None,
+            master_password="master-password",
+            bw_organization_id="org-id",
+            bw_collection_ids=None,
+            email="first@example.com",
+            timeout=5,
+        )
+    )
+    await first_list_entered.wait()
+
+    second_request = asyncio.create_task(
+        BitwardenService.list_item_overviews(
+            client_id=None,
+            client_secret=None,
+            master_password="master-password",
+            bw_organization_id="org-id",
+            bw_collection_ids=None,
+            email="second@example.com",
+            timeout=5,
+        )
+    )
+
+    with pytest.raises(asyncio.TimeoutError):
+        await asyncio.wait_for(second_list_entered.wait(), timeout=0.05)
+
+    release_first_list.set()
+    await asyncio.gather(first_request, second_request)
+
+    assert list_entries == ["first", "second"]

tests/unit/test_download_dir_rebind.py

@@ -328,11 +328,63 @@ async def test_block_adoption_prefers_context_run_id_over_workflow_run_id() -> N
 
 
 @pytest.mark.asyncio
-async def test_file_upload_block_fails_when_no_files_found(tmp_path) -> None:
-    """SKY-11153 regression: an empty download dir must fail the block, not silently report
-    success with zero files uploaded."""
+async def test_file_upload_block_empty_scan_without_registered_downloads_succeeds(tmp_path) -> None:
+    """SKY-11225: zero downloads during the run is a successful no-op."""
     from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
-    from skyvern.schemas.workflows import FileStorageType
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
+
+    block = FileUploadBlock.model_construct(
+        label="upload",
+        storage_type=FileStorageType.AZURE,
+        azure_storage_account_name="account",
+        azure_storage_account_key="key",
+        azure_blob_container_name="container",
+        path=None,
+        continue_on_empty=False,
+    )
+    empty_dir = tmp_path / "wr_empty"
+    empty_dir.mkdir()
+    sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
+
+    with (
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
+        patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
+        patch.object(
+            FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
+        ) as mock_result,
+        patch(
+            "skyvern.forge.sdk.workflow.models.block.get_path_for_workflow_download_directory",
+            return_value=empty_dir,
+        ),
+        patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=None),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+    ):
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(return_value=[])
+        mock_app.AGENT_FUNCTION.upload_file_to_customer_storage = AsyncMock()
+        result = await block.execute(
+            workflow_run_id="wr_empty",
+            workflow_run_block_id="wrb_x",
+            organization_id="org_1",
+        )
+
+    assert result is sentinel
+    assert mock_result.await_args.kwargs["success"] is True
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.completed
+    assert mock_result.await_args.kwargs["failure_reason"] is None
+    assert mock_result.await_args.kwargs["output_parameter_value"] == []
+    mock_record.assert_awaited_once()
+    mock_app.AGENT_FUNCTION.upload_file_to_customer_storage.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_file_upload_block_empty_scan_with_registered_downloads_fails(tmp_path) -> None:
+    """SKY-11153/SKY-11225: downloaded files with an empty scan dir still fail loudly."""
+    from skyvern.forge.sdk.schemas.files import FileInfo
+    from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
 
     block = FileUploadBlock.model_construct(
         label="upload",
@@ -346,10 +398,13 @@ async def test_file_upload_block_fails_when_no_files_found(tmp_path) -> None:
     empty_dir = tmp_path / "wr_empty"
     empty_dir.mkdir()
     sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
 
     with (
-        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=MagicMock()),
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
         patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
         patch.object(
             FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
         ) as mock_result,
@@ -358,7 +413,11 @@ async def test_file_upload_block_fails_when_no_files_found(tmp_path) -> None:
             return_value=empty_dir,
         ),
         patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=None),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
     ):
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(
+            return_value=[FileInfo(url="https://example.com/invoice.pdf", filename="invoice.pdf")]
+        )
         result = await block.execute(
             workflow_run_id="wr_empty",
             workflow_run_block_id="wrb_x",
@@ -367,6 +426,233 @@ async def test_file_upload_block_fails_when_no_files_found(tmp_path) -> None:
 
     assert result is sentinel
     assert mock_result.await_args.kwargs["success"] is False
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.failed
+    assert "registered_download_count=1" in mock_result.await_args.kwargs["failure_reason"]
+    mock_record.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_file_upload_block_empty_scan_with_alternate_download_dir_files_fails(tmp_path) -> None:
+    """SKY-11225: local files in a sibling candidate dir still indicate a download-dir desync."""
+    from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
+
+    block = FileUploadBlock.model_construct(
+        label="upload",
+        storage_type=FileStorageType.S3,
+        s3_bucket="bucket",
+        aws_access_key_id="ak",
+        aws_secret_access_key="sk",
+        path=None,
+        continue_on_empty=False,
+    )
+    scan_dir = tmp_path / "run_ctx"
+    alternate_dir = tmp_path / "wr_empty"
+    scan_dir.mkdir()
+    alternate_dir.mkdir()
+    (alternate_dir / "invoice.pdf").write_text("pdf")
+    sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
+    context = SkyvernContext(run_id="run_ctx", workflow_run_id="wr_empty")
+
+    def get_download_dir_for_run_id(run_id: str | None):
+        return {"run_ctx": scan_dir, "wr_empty": alternate_dir}[run_id]
+
+    with (
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
+        patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
+        patch.object(
+            FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
+        ) as mock_result,
+        patch(
+            "skyvern.forge.sdk.workflow.models.block.get_path_for_workflow_download_directory",
+            side_effect=get_download_dir_for_run_id,
+        ),
+        patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=context),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+    ):
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(return_value=[])
+        mock_app.AGENT_FUNCTION.upload_file_to_customer_storage = AsyncMock()
+        result = await block.execute(
+            workflow_run_id="wr_empty",
+            workflow_run_block_id="wrb_x",
+            organization_id="org_1",
+        )
+
+    assert result is sentinel
+    assert mock_result.await_args.kwargs["success"] is False
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.failed
+    assert "alternate_file_count=1" in mock_result.await_args.kwargs["failure_reason"]
+    mock_record.assert_not_awaited()
+    mock_app.AGENT_FUNCTION.upload_file_to_customer_storage.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_file_upload_block_empty_scan_with_too_many_alternate_files_reports_too_many(tmp_path) -> None:
+    """SKY-11225: oversized alternate dirs fail closed with a specific diagnostic."""
+    from skyvern.constants import MAX_UPLOAD_FILE_COUNT
+    from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
+
+    block = FileUploadBlock.model_construct(
+        label="upload",
+        storage_type=FileStorageType.S3,
+        s3_bucket="bucket",
+        aws_access_key_id="ak",
+        aws_secret_access_key="sk",
+        path=None,
+        continue_on_empty=False,
+    )
+    scan_dir = tmp_path / "run_ctx"
+    alternate_dir = tmp_path / "wr_empty"
+    scan_dir.mkdir()
+    alternate_dir.mkdir()
+    for index in range(MAX_UPLOAD_FILE_COUNT + 1):
+        (alternate_dir / f"invoice_{index}.pdf").write_text("pdf")
+    sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
+    context = SkyvernContext(run_id="run_ctx", workflow_run_id="wr_empty")
+
+    def get_download_dir_for_run_id(run_id: str | None):
+        return {"run_ctx": scan_dir, "wr_empty": alternate_dir}[run_id]
+
+    with (
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
+        patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
+        patch.object(
+            FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
+        ) as mock_result,
+        patch(
+            "skyvern.forge.sdk.workflow.models.block.get_path_for_workflow_download_directory",
+            side_effect=get_download_dir_for_run_id,
+        ),
+        patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=context),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+    ):
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(return_value=[])
+        mock_app.AGENT_FUNCTION.upload_file_to_customer_storage = AsyncMock()
+        result = await block.execute(
+            workflow_run_id="wr_empty",
+            workflow_run_block_id="wrb_x",
+            organization_id="org_1",
+        )
+
+    assert result is sentinel
+    assert mock_result.await_args.kwargs["success"] is False
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.failed
+    assert "alternate_file_count=too_many" in mock_result.await_args.kwargs["failure_reason"]
+    mock_record.assert_not_awaited()
+    mock_app.AGENT_FUNCTION.upload_file_to_customer_storage.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_file_upload_block_empty_scan_with_browser_session_downloads_fails(tmp_path) -> None:
+    """SKY-11225: unclaimed browser-session downloads are not a benign empty run."""
+    from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
+
+    block = FileUploadBlock.model_construct(
+        label="upload",
+        storage_type=FileStorageType.S3,
+        s3_bucket="bucket",
+        aws_access_key_id="ak",
+        aws_secret_access_key="sk",
+        path=None,
+        continue_on_empty=False,
+    )
+    empty_dir = tmp_path / "wr_empty"
+    empty_dir.mkdir()
+    sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
+
+    with (
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
+        patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
+        patch.object(
+            FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
+        ) as mock_result,
+        patch(
+            "skyvern.forge.sdk.workflow.models.block.get_path_for_workflow_download_directory",
+            return_value=empty_dir,
+        ),
+        patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=None),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+    ):
+        mock_app.STORAGE.list_downloaded_files_in_browser_session = AsyncMock(
+            return_value=["s3://downloads/session/invoice.pdf"]
+        )
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(return_value=[])
+        mock_app.AGENT_FUNCTION.upload_file_to_customer_storage = AsyncMock()
+        result = await block.execute(
+            workflow_run_id="wr_empty",
+            workflow_run_block_id="wrb_x",
+            organization_id="org_1",
+            browser_session_id="pbs_1",
+        )
+
+    assert result is sentinel
+    assert mock_result.await_args.kwargs["success"] is False
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.failed
+    assert "browser_session_download_count=1" in mock_result.await_args.kwargs["failure_reason"]
+    mock_record.assert_not_awaited()
+    mock_app.AGENT_FUNCTION.upload_file_to_customer_storage.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_file_upload_block_empty_scan_registered_download_timeout_fails_with_unknown_count(tmp_path) -> None:
+    """SKY-11225: unknown registered-download state fails closed with a readable failure reason."""
+    import asyncio
+
+    from skyvern.forge.sdk.workflow.models.block import FileUploadBlock
+    from skyvern.schemas.workflows import BlockStatus, FileStorageType
+
+    block = FileUploadBlock.model_construct(
+        label="upload",
+        storage_type=FileStorageType.S3,
+        s3_bucket="bucket",
+        aws_access_key_id="ak",
+        aws_secret_access_key="sk",
+        path=None,
+        continue_on_empty=False,
+    )
+    empty_dir = tmp_path / "wr_empty"
+    empty_dir.mkdir()
+    sentinel = object()
+    workflow_run_context = MagicMock()
+    workflow_run_context.organization_id = "org_1"
+
+    with (
+        patch.object(FileUploadBlock, "get_workflow_run_context", return_value=workflow_run_context),
+        patch.object(FileUploadBlock, "format_potential_template_parameters", return_value=None),
+        patch.object(FileUploadBlock, "record_output_parameter_value", new_callable=AsyncMock) as mock_record,
+        patch.object(
+            FileUploadBlock, "build_block_result", new_callable=AsyncMock, return_value=sentinel
+        ) as mock_result,
+        patch(
+            "skyvern.forge.sdk.workflow.models.block.get_path_for_workflow_download_directory",
+            return_value=empty_dir,
+        ),
+        patch("skyvern.forge.sdk.workflow.models.block.skyvern_context.current", return_value=None),
+        patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+    ):
+        mock_app.STORAGE.get_downloaded_files = AsyncMock(side_effect=asyncio.TimeoutError)
+        result = await block.execute(
+            workflow_run_id="wr_empty",
+            workflow_run_block_id="wrb_x",
+            organization_id="org_1",
+        )
+
+    assert result is sentinel
+    assert mock_result.await_args.kwargs["success"] is False
+    assert mock_result.await_args.kwargs["status"] == BlockStatus.failed
+    assert "registered_download_count=unknown" in mock_result.await_args.kwargs["failure_reason"]
+    mock_record.assert_not_awaited()
 
 
 @pytest.mark.asyncio

tests/unit/test_forge_log_driver_pipe_noise.py

@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterator
+
+import pytest
+
+from skyvern.forge.sdk.forge_log import _DriverPipeNoiseFilter, setup_logger
+
+
+def _orphaned_future_record(exc_text: str) -> logging.LogRecord:
+    exc = Exception(exc_text)
+    message = f"Future exception was never retrieved\nfuture: <Future finished exception=Exception('{exc_text}')>"
+    return logging.LogRecord(
+        name="asyncio",
+        level=logging.ERROR,
+        pathname=__file__,
+        lineno=1,
+        msg=message,
+        args=(),
+        exc_info=(type(exc), exc, None),
+    )
+
+
+def test_filter_drops_driver_pipe_future_noise() -> None:
+    record = _orphaned_future_record("Connection closed while reading from the driver")
+    assert _DriverPipeNoiseFilter().filter(record) is False
+
+
+def test_filter_matches_via_exc_info_when_message_lacks_repr() -> None:
+    exc = Exception("Connection closed while reading from the driver")
+    record = logging.LogRecord(
+        name="asyncio",
+        level=logging.ERROR,
+        pathname=__file__,
+        lineno=1,
+        msg="Future exception was never retrieved",
+        args=(),
+        exc_info=(type(exc), exc, None),
+    )
+    assert _DriverPipeNoiseFilter().filter(record) is False
+
+
+def test_filter_keeps_other_orphaned_future_exceptions() -> None:
+    record = _orphaned_future_record("some other unrelated failure")
+    assert _DriverPipeNoiseFilter().filter(record) is True
+
+
+def test_filter_keeps_real_driver_errors_not_from_orphaned_future() -> None:
+    exc = Exception("Connection closed while reading from the driver")
+    record = logging.LogRecord(
+        name="asyncio",
+        level=logging.ERROR,
+        pathname=__file__,
+        lineno=1,
+        msg="Task exception was never retrieved",
+        args=(),
+        exc_info=(type(exc), exc, None),
+    )
+    assert _DriverPipeNoiseFilter().filter(record) is True
+
+
+def test_filter_keeps_unrelated_asyncio_logs() -> None:
+    record = logging.LogRecord(
+        name="asyncio",
+        level=logging.DEBUG,
+        pathname=__file__,
+        lineno=1,
+        msg="Using selector: %s",
+        args=("KqueueSelector",),
+        exc_info=None,
+    )
+    assert _DriverPipeNoiseFilter().filter(record) is True
+
+
+@pytest.fixture
+def _restore_asyncio_filters() -> Iterator[None]:
+    asyncio_logger = logging.getLogger("asyncio")
+    saved = list(asyncio_logger.filters)
+    yield
+    asyncio_logger.filters = saved
+
+
+def test_setup_logger_installs_single_driver_pipe_filter(_restore_asyncio_filters: None) -> None:
+    setup_logger()
+    setup_logger()
+    asyncio_logger = logging.getLogger("asyncio")
+    installed = [f for f in asyncio_logger.filters if isinstance(f, _DriverPipeNoiseFilter)]
+    assert len(installed) == 1
+
+
+def test_setup_logger_filter_suppresses_emitted_noise(_restore_asyncio_filters: None) -> None:
+    setup_logger()
+    captured: list[logging.LogRecord] = []
+
+    class _Capture(logging.Handler):
+        def emit(self, record: logging.LogRecord) -> None:
+            captured.append(record)
+
+    asyncio_logger = logging.getLogger("asyncio")
+    handler = _Capture()
+    asyncio_logger.addHandler(handler)
+    try:
+        asyncio_logger.handle(_orphaned_future_record("Connection closed while reading from the driver"))
+        asyncio_logger.handle(_orphaned_future_record("a genuinely different error"))
+    finally:
+        asyncio_logger.removeHandler(handler)
+
+    messages = [r.getMessage() for r in captured]
+    assert not any("Connection closed while reading from the driver" in m for m in messages)
+    assert any("a genuinely different error" in m for m in messages)

tests/unit/workflow/test_bitwarden_workflow_definition_converter.py

@@ -0,0 +1,63 @@
+import pytest
+
+from skyvern.forge.sdk.workflow.exceptions import WorkflowParameterMissingRequiredValue
+from skyvern.forge.sdk.workflow.models.parameter import BitwardenLoginCredentialParameter
+from skyvern.forge.sdk.workflow.workflow_definition_converter import convert_workflow_definition
+from skyvern.schemas.workflows import BitwardenLoginCredentialParameterYAML, WorkflowDefinitionYAML
+
+
+def _bitwarden_login_parameter(
+    *,
+    bitwarden_collection_id: str | None,
+    bitwarden_item_id: str | None,
+    url_parameter_key: str | None,
+) -> BitwardenLoginCredentialParameterYAML:
+    return BitwardenLoginCredentialParameterYAML(
+        key="login",
+        bitwarden_client_id_aws_secret_key="client_id_secret",
+        bitwarden_client_secret_aws_secret_key="client_secret_secret",
+        bitwarden_master_password_aws_secret_key="master_password_secret",
+        bitwarden_collection_id=bitwarden_collection_id,
+        bitwarden_item_id=bitwarden_item_id,
+        url_parameter_key=url_parameter_key,
+    )
+
+
+def test_bitwarden_login_item_id_with_collection_id_does_not_require_url_parameter_key() -> None:
+    workflow_definition = WorkflowDefinitionYAML(
+        parameters=[
+            _bitwarden_login_parameter(
+                bitwarden_collection_id="collection-id",
+                bitwarden_item_id="item-id",
+                url_parameter_key=None,
+            )
+        ],
+        blocks=[],
+    )
+
+    converted = convert_workflow_definition(workflow_definition_yaml=workflow_definition, workflow_id="workflow-id")
+
+    parameter = converted.parameters[0]
+    assert isinstance(parameter, BitwardenLoginCredentialParameter)
+    assert parameter.bitwarden_collection_id == "collection-id"
+    assert parameter.bitwarden_item_id == "item-id"
+    assert parameter.url_parameter_key is None
+
+
+def test_bitwarden_login_collection_only_still_requires_url_parameter_key() -> None:
+    workflow_definition = WorkflowDefinitionYAML(
+        parameters=[
+            _bitwarden_login_parameter(
+                bitwarden_collection_id="collection-id",
+                bitwarden_item_id=None,
+                url_parameter_key=None,
+            )
+        ],
+        blocks=[],
+    )
+
+    with pytest.raises(WorkflowParameterMissingRequiredValue) as exc_info:
+        convert_workflow_definition(workflow_definition_yaml=workflow_definition, workflow_id="workflow-id")
+
+    assert "workflow_parameter_key: login" in str(exc_info.value)
+    assert "Required value: url_parameter_key" in str(exc_info.value)