fix(SKY-11173): support email OTP credentials in Copilot (#6716)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-09 16:09:13 +00:00 · 2026-06-19 13:55:02 -07:00 · 2026-06-19 13:55:02 -07:00 · 44b2e959b9
commit 44b2e959b9
parent a0f2c5d60e
36 changed files with 1652 additions and 127 deletions
--- a/skyvern-frontend/src/hooks/useGoogleOAuthCredentials.ts
+++ b/skyvern-frontend/src/hooks/useGoogleOAuthCredentials.ts
@ -24,6 +24,14 @@ function broadcastCredentialsChanged() {
  credentialBroadcastChannel?.postMessage("invalidate");
 }

+// Falls back to the first credential even when none are valid, so a single
+// needs-reconnect account is still selected rather than left blank.
+export function getDefaultGoogleOAuthCredentialId(
+  credentials: GoogleOAuthCredential[],
+): string | undefined {
+  return credentials.find((c) => c.valid)?.id ?? credentials[0]?.id;
+}
+
 type ApiError = { response?: { data?: { detail?: string } } } & Error;

 function extractApiErrorMessage(error: unknown, fallback: string): string {
@ -31,7 +39,9 @@ function extractApiErrorMessage(error: unknown, fallback: string): string {
  return err?.response?.data?.detail || err?.message || fallback;
 }

-export function useGoogleOAuthCredentials() {
+export function useGoogleOAuthCredentials({
+  enabled = true,
+}: { enabled?: boolean } = {}) {
  const credentialGetter = useCredentialGetter();
  const queryClient = useQueryClient();
  const { toast } = useToast();
@ -54,6 +64,7 @@ export function useGoogleOAuthCredentials() {
    error,
  } = useQuery<GoogleOAuthCredential[]>({
    queryKey: ["googleOAuthCredentials"],
+    enabled,
    queryFn: async () => {
      const client = await getClient(credentialGetter);
      const response = await client.get("/google/oauth/credentials");
--- a/skyvern-frontend/src/routes/workflows/components/GoogleOAuthCredentialSelector.tsx
+++ b/skyvern-frontend/src/routes/workflows/components/GoogleOAuthCredentialSelector.tsx
@ -10,7 +10,10 @@ import {
 } from "@/components/ui/select";
 import { Skeleton } from "@/components/ui/skeleton";
 import { WorkflowBlockInputTextarea } from "@/components/WorkflowBlockInputTextarea";
-import { useGoogleOAuthCredentials } from "@/hooks/useGoogleOAuthCredentials";
+import {
+  getDefaultGoogleOAuthCredentialId,
+  useGoogleOAuthCredentials,
+} from "@/hooks/useGoogleOAuthCredentials";
 import { PlusIcon } from "@radix-ui/react-icons";

 type Props = {
@ -40,8 +43,7 @@ function GoogleOAuthCredentialSelector({

  const hasCredentials = credentials.length > 0;
  const isKnownCredential = credentials.some((c) => c.id === value);
-  const firstValidId =
-    credentials.find((c) => c.valid)?.id ?? credentials[0]?.id;
+  const firstValidId = getDefaultGoogleOAuthCredentialId(credentials);
  const needsAutoFill = !value;

  useEffect(() => {
--- a/skyvern-frontend/src/routes/workflows/copilot/WorkflowCopilotChat.composerMode.test.tsx
+++ b/skyvern-frontend/src/routes/workflows/copilot/WorkflowCopilotChat.composerMode.test.tsx
@ -191,13 +191,13 @@ afterEach(() => {
 });

 describe("WorkflowCopilotChat — composer default mode variant", () => {
-  it("defaults to Build with code OFF when the variant is unset (new baseline)", async () => {
+  it("leaves code_block unset for backend fallback when the default variant is unset", async () => {
    await renderChat({ copilotV2: true, codeBlockMode: true });
    await submit("build me a workflow");
    await waitFor(() => expect(postStreaming).toHaveBeenCalledTimes(1));

    expect(streamCalls[0]?.body.mode).toBe("build");
-    expect(streamCalls[0]?.body.code_block).toBe(false);
+    expect(streamCalls[0]?.body.code_block).toBe(null);
  });

  it("sends code_block=true for the build_code override variant", async () => {
--- a/skyvern-frontend/src/routes/workflows/copilot/WorkflowCopilotChat.tsx
+++ b/skyvern-frontend/src/routes/workflows/copilot/WorkflowCopilotChat.tsx
@ -98,6 +98,23 @@ function normalizeComposerDefaultVariant(
  return "build";
 }

+function defaultVariantUsesCode(variant: ComposerDefaultVariant): boolean {
+  return variant === "build_code" || variant === "ask_code";
+}
+
+function defaultCodeBlockRequestOverride(
+  variant: string | undefined,
+): boolean | null {
+  if (variant === "build_code") {
+    return true;
+  }
+  if (variant === "build" || variant === "build_no_code") {
+    return false;
+  }
+  // Ask-only variants, including ask_code, do not send a build request override.
+  return null;
+}
+
 function formatElapsedSeconds(ms: number): string {
  const seconds = Math.max(0, Math.round(ms / 1000));
  const m = Math.floor(seconds / 60);
@ -405,11 +422,12 @@ export function WorkflowCopilotChat({
      ? "ask"
      : "build",
  );
-  const [codeWorkflow, setCodeWorkflow] = useState(
-    () =>
-      effectiveDefaultVariant === "build_code" ||
-      effectiveDefaultVariant === "ask_code",
+  const [codeWorkflow, setCodeWorkflow] = useState(() =>
+    defaultVariantUsesCode(effectiveDefaultVariant),
  );
+  const [codeBlockRequestOverride, setCodeBlockRequestOverride] = useState<
+    boolean | null
+  >(() => defaultCodeBlockRequestOverride(defaultModeVariant));
  // Flags arrive asynchronously from /customer; seed the default once they resolve, never again.
  const composerSeededRef = useRef(false);
  const flagsResolved =
@ -427,11 +445,11 @@ export function WorkflowCopilotChat({
        ? "ask"
        : "build",
    );
-    setCodeWorkflow(
-      effectiveDefaultVariant === "build_code" ||
-        effectiveDefaultVariant === "ask_code",
+    setCodeWorkflow(defaultVariantUsesCode(effectiveDefaultVariant));
+    setCodeBlockRequestOverride(
+      defaultCodeBlockRequestOverride(defaultModeVariant),
    );
-  }, [flagsResolved, effectiveDefaultVariant]);
+  }, [flagsResolved, effectiveDefaultVariant, defaultModeVariant]);
  // Build can never be active unless the V2 flag is on.
  const isBuild = copilotV2Enabled && composerMode === "build";
  const codeToggleAllowed = effectiveDefaultVariant !== "build_no_code";
@ -1385,7 +1403,8 @@ export function WorkflowCopilotChat({
            audio_artifact_id: audioArtifactId,
            workflow_yaml: workflowYaml,
            mode: copilotV2Enabled ? composerMode : null,
-            code_block: isBuild && codeBlockModeEnabled ? codeWorkflow : null,
+            code_block:
+              isBuild && codeBlockModeEnabled ? codeBlockRequestOverride : null,
            cancel_token: cancelToken,
            target_block_label: targetBlockLabel,
          } as WorkflowCopilotChatRequest,
@ -1496,7 +1515,7 @@ export function WorkflowCopilotChat({
      applyWorkflowUpdate,
      autoAccept,
      codeBlockModeEnabled,
-      codeWorkflow,
+      codeBlockRequestOverride,
      composerMode,
      copilotV2Enabled,
      credentialGetter,
@ -1542,6 +1561,7 @@ export function WorkflowCopilotChat({
    blockBuildTargetLabelRef.current = pendingBlockBuild.blockLabel;
    setComposerMode("build");
    setCodeWorkflow(true);
+    setCodeBlockRequestOverride(true);
    setBlockBuildArmNonce((nonce) => nonce + 1);
    clearPendingBlockBuild();
  }, [pendingBlockBuild, clearPendingBlockBuild]);
@ -2229,6 +2249,7 @@ export function WorkflowCopilotChat({
                    onSelect={() => {
                      setComposerMode("ask");
                      setCodeWorkflow(false);
+                      setCodeBlockRequestOverride(null);
                    }}
                    className="flex items-start gap-2.5"
                  >
@ -2248,6 +2269,7 @@ export function WorkflowCopilotChat({
                    onSelect={() => {
                      setComposerMode("build");
                      setCodeWorkflow(false);
+                      setCodeBlockRequestOverride(false);
                    }}
                    className="flex items-start gap-2.5"
                  >
@ -2269,6 +2291,7 @@ export function WorkflowCopilotChat({
                      onSelect={() => {
                        setComposerMode("build");
                        setCodeWorkflow(true);
+                        setCodeBlockRequestOverride(true);
                      }}
                      className="flex items-start gap-2.5"
                    >
--- a/skyvern-frontend/src/routes/workflows/editor/FlowRenderer.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/FlowRenderer.tsx
@ -119,6 +119,7 @@ import {
 import { toast } from "@/components/ui/use-toast";
 import { useAutoPan } from "./useAutoPan";
 import { useAutoGenerateWorkflowTitle } from "../hooks/useAutoGenerateWorkflowTitle";
+import { useResolveDefaultGoogleSheetsCredential } from "./hooks/useResolveDefaultGoogleSheetsCredential";
 import { SortableBlockScope } from "./sortable/SortableBlockScope";
 import {
  TOP_LEVEL_SCOPE,
@ -1459,6 +1460,7 @@ function FlowRenderer({

  useAutoPan(editorElementRef, nodes);
  useAutoGenerateWorkflowTitle(nodes, edges, readOnly);
+  useResolveDefaultGoogleSheetsCredential(nodes, readOnly);

  useEffect(() => {
    doLayout(nodes, edges);
--- a/skyvern-frontend/src/routes/workflows/editor/hooks/useResolveDefaultGoogleSheetsCredential.test.tsx
+++ b/skyvern-frontend/src/routes/workflows/editor/hooks/useResolveDefaultGoogleSheetsCredential.test.tsx
@ -0,0 +1,219 @@
+// @vitest-environment jsdom
+
+import { cleanup, render } from "@testing-library/react";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
+
+import type { GoogleOAuthCredential } from "@/api/types";
+
+import { useResolveDefaultGoogleSheetsCredential } from "./useResolveDefaultGoogleSheetsCredential";
+
+const updateNodeData = vi.fn();
+const setHasChanges = vi.fn();
+
+vi.mock("@xyflow/react", async () => {
+  const actual =
+    await vi.importActual<typeof import("@xyflow/react")>("@xyflow/react");
+  return {
+    ...actual,
+    useReactFlow: () => ({ updateNodeData }),
+  };
+});
+
+vi.mock("@/store/WorkflowHasChangesStore", () => ({
+  useWorkflowHasChangesStore: (selector: (s: unknown) => unknown) =>
+    selector({ setHasChanges }),
+}));
+
+let mockCredentials: GoogleOAuthCredential[] = [];
+let mockIsLoading = false;
+let mockIsFetching = false;
+
+vi.mock("@/hooks/useGoogleOAuthCredentials", async () => {
+  const actual = await vi.importActual<
+    typeof import("@/hooks/useGoogleOAuthCredentials")
+  >("@/hooks/useGoogleOAuthCredentials");
+  return {
+    ...actual,
+    useGoogleOAuthCredentials: () => ({
+      credentials: mockCredentials,
+      isLoading: mockIsLoading,
+      isFetching: mockIsFetching,
+    }),
+  };
+});
+
+function credential(id: string, valid: boolean = true): GoogleOAuthCredential {
+  return {
+    id,
+    organization_id: "o_1",
+    credential_name: id,
+    scopes: null,
+    valid,
+    created_at: "",
+    modified_at: "",
+  };
+}
+
+// Minimal node shapes; only `type`, `id`, and `data.{editable,credentialId}`
+// are read by the hook + the real node type guards.
+function writeNode(id: string, credentialId: string, editable = true) {
+  return {
+    id,
+    type: "googleSheetsWrite",
+    data: { editable, credentialId },
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  } as any;
+}
+
+function readNode(id: string, credentialId: string, editable = true) {
+  return {
+    id,
+    type: "googleSheetsRead",
+    data: { editable, credentialId },
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  } as any;
+}
+
+function taskNode(id: string) {
+  return {
+    id,
+    type: "task",
+    data: { editable: true },
+    // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  } as any;
+}
+
+function Harness({
+  nodes,
+  readOnly = false,
+}: {
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  nodes: any[];
+  readOnly?: boolean;
+}) {
+  useResolveDefaultGoogleSheetsCredential(nodes, readOnly);
+  return null;
+}
+
+beforeEach(() => {
+  updateNodeData.mockReset();
+  setHasChanges.mockReset();
+  mockCredentials = [];
+  mockIsLoading = false;
+  mockIsFetching = false;
+});
+
+afterEach(() => {
+  cleanup();
+});
+
+describe("useResolveDefaultGoogleSheetsCredential (SKY-11219)", () => {
+  test("fills the default account into a write block with no credential", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).toHaveBeenCalledWith("g1", {
+      credentialId: "cred_default",
+    });
+  });
+
+  // Deferred past the synchronous effect flush so it lands after Workspace's
+  // mount initializer resets the flag (setHasChanges(false)).
+  test("marks the workflow dirty (deferred) so the fill is persisted", async () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(setHasChanges).not.toHaveBeenCalled();
+    await Promise.resolve();
+    expect(setHasChanges).toHaveBeenCalledWith(true);
+  });
+
+  test("does not mark dirty when nothing needs filling", async () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "cred_existing")]} />);
+
+    await Promise.resolve();
+    expect(setHasChanges).not.toHaveBeenCalled();
+  });
+
+  test("does not fill while a credentials refetch is in flight", () => {
+    mockCredentials = [credential("cred_default")];
+    mockIsFetching = true;
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("fills read blocks too", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[readNode("r1", "")]} />);
+
+    expect(updateNodeData).toHaveBeenCalledWith("r1", {
+      credentialId: "cred_default",
+    });
+  });
+
+  test("prefers the first valid credential over an invalid one", () => {
+    mockCredentials = [
+      credential("cred_invalid", false),
+      credential("cred_valid", true),
+    ];
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).toHaveBeenCalledWith("g1", {
+      credentialId: "cred_valid",
+    });
+  });
+
+  test("falls back to the only (invalid) credential rather than leaving it blank", () => {
+    mockCredentials = [credential("cred_only", false)];
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).toHaveBeenCalledWith("g1", {
+      credentialId: "cred_only",
+    });
+  });
+
+  test("leaves an already-configured credential untouched", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "cred_existing")]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("does nothing when no Google account is connected", () => {
+    mockCredentials = [];
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("does not fill while credentials are still loading", () => {
+    mockCredentials = [credential("cred_default")];
+    mockIsLoading = true;
+    render(<Harness nodes={[writeNode("g1", "")]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("does not fill in read-only canvases", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "")]} readOnly />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("does not fill non-editable blocks", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[writeNode("g1", "", false)]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+
+  test("ignores non-Google-Sheets blocks", () => {
+    mockCredentials = [credential("cred_default")];
+    render(<Harness nodes={[taskNode("t1")]} />);
+
+    expect(updateNodeData).not.toHaveBeenCalled();
+  });
+});
--- a/skyvern-frontend/src/routes/workflows/editor/hooks/useResolveDefaultGoogleSheetsCredential.ts
+++ b/skyvern-frontend/src/routes/workflows/editor/hooks/useResolveDefaultGoogleSheetsCredential.ts
@ -0,0 +1,74 @@
+import { useReactFlow } from "@xyflow/react";
+import { useEffect, useMemo } from "react";
+
+import {
+  getDefaultGoogleOAuthCredentialId,
+  useGoogleOAuthCredentials,
+} from "@/hooks/useGoogleOAuthCredentials";
+import { useWorkflowHasChangesStore } from "@/store/WorkflowHasChangesStore";
+
+import type { AppNode } from "../nodes";
+import { isGoogleSheetsReadNode } from "../nodes/GoogleSheetsReadNode/types";
+import { isGoogleSheetsWriteNode } from "../nodes/GoogleSheetsWriteNode/types";
+
+/**
+ * Canvas-level fallback for the per-block GoogleOAuthCredentialSelector, which
+ * only auto-fills the default account while its editor is mounted (build mode +
+ * expanded). Collapsed or never-expanded blocks would otherwise reach save/run
+ * with an empty credential_id and fail with "Google account is required".
+ */
+export function useResolveDefaultGoogleSheetsCredential(
+  nodes: Array<AppNode>,
+  readOnly: boolean = false,
+): void {
+  const { updateNodeData } = useReactFlow<AppNode>();
+  const setHasChanges = useWorkflowHasChangesStore((s) => s.setHasChanges);
+
+  const unconfiguredNodeIds = useMemo(
+    () =>
+      nodes
+        .filter(
+          (node) =>
+            (isGoogleSheetsWriteNode(node) || isGoogleSheetsReadNode(node)) &&
+            node.data.editable &&
+            !node.data.credentialId.trim(),
+        )
+        .map((node) => node.id)
+        .sort(),
+    [nodes],
+  );
+
+  // Only fetch credentials when a block actually needs one, so workflows with no
+  // (unconfigured) Sheets blocks and read-only canvases don't fetch on mount.
+  // Wait out in-flight refetches (isFetching) so an invalidation — e.g. an
+  // account disconnected in another tab — can't get filled from stale cache.
+  const { credentials, isLoading, isFetching } = useGoogleOAuthCredentials({
+    enabled: !readOnly && unconfiguredNodeIds.length > 0,
+  });
+  const defaultCredentialId = getDefaultGoogleOAuthCredentialId(credentials);
+
+  // Stable for a given set of unconfigured blocks regardless of node ordering.
+  const unconfiguredKey = unconfiguredNodeIds.join(",");
+
+  useEffect(() => {
+    if (readOnly || isLoading || isFetching || !defaultCredentialId) {
+      return;
+    }
+    if (unconfiguredNodeIds.length === 0) {
+      return;
+    }
+    for (const id of unconfiguredNodeIds) {
+      updateNodeData(id, { credentialId: defaultCredentialId });
+    }
+    // Filling a credential is a real, persistable edit, but this effect can run
+    // in the same passive-effect flush as Workspace's mount initializer, which
+    // runs later (child effects before parent) and calls setHasChanges(false).
+    // Defer to a microtask so the dirty flag is set after that reset; otherwise
+    // the fill is silently unsaved and a later Run reads the stale credential_id
+    // from the backend.
+    queueMicrotask(() => setHasChanges(true));
+    // unconfiguredNodeIds is derived from unconfiguredKey; depending on the key
+    // keeps the effect from re-firing on unrelated node-array identity changes.
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [readOnly, isLoading, isFetching, defaultCredentialId, unconfiguredKey]);
+}
--- a/skyvern/forge/sdk/copilot/agent.py
+++ b/skyvern/forge/sdk/copilot/agent.py
@ -754,6 +754,37 @@ def _turn_intent_disables_tools(turn_intent: TurnIntent | None) -> bool:
    return not authority.may_update_workflow and not authority.may_run_blocks


+_DRAFT_ONLY_MCP_TOOL_ALLOWLIST = frozenset({"get_block_schema", "validate_block"})
+_DRAFT_ONLY_NATIVE_TOOL_DENYLIST = frozenset(
+    {"discover_workflow_entrypoint", "inspect_page_for_composition", "fill_credential_field"}
+)
+
+
+def _request_policy_disables_browser_scout_tools(request_policy: RequestPolicy | None) -> bool:
+    return (
+        isinstance(request_policy, RequestPolicy)
+        and request_policy.allow_update_workflow
+        and not request_policy.allow_run_blocks
+        and (request_policy.testing_intent == "skip_test" or request_policy.allow_missing_credentials_in_draft)
+    )
+
+
+def _mcp_tool_surface_for_turn(
+    alias_map: dict[str, str],
+    overlays: dict[str, Any],
+    turn_intent: TurnIntent | None,
+    request_policy: RequestPolicy | None = None,
+) -> tuple[dict[str, str], dict[str, Any]]:
+    if _turn_intent_disables_tools(turn_intent):
+        return {}, {}
+    if _request_policy_disables_browser_scout_tools(request_policy):
+        return (
+            {name: target for name, target in alias_map.items() if name in _DRAFT_ONLY_MCP_TOOL_ALLOWLIST},
+            {name: overlay for name, overlay in overlays.items() if name in _DRAFT_ONLY_MCP_TOOL_ALLOWLIST},
+        )
+    return alias_map, overlays
+
+
 def _native_tools_for_turn(
    native_tools: list[Any],
    turn_intent: TurnIntent | None,
@ -763,6 +794,8 @@ def _native_tools_for_turn(
    # use them. The tool implementations enforce TurnIntent/RequestPolicy
    # authority and return structured blockers; removing a tool lets the model
    # hit an SDK-level ModelBehaviorError if static prompt text still names it.
+    if _request_policy_disables_browser_scout_tools(request_policy):
+        return [tool for tool in native_tools if getattr(tool, "name", None) not in _DRAFT_ONLY_NATIVE_TOOL_DENYLIST]
    return list(native_tools)


@ -3366,9 +3399,7 @@ async def _run_copilot_turn_impl(

    alias_map = get_skyvern_mcp_alias_map()
    overlays = _build_skyvern_mcp_overlays(copilot_config.block_authoring_policy)
-    if _turn_intent_disables_tools(ctx.turn_intent):
-        alias_map = {}
-        overlays = {}
+    alias_map, overlays = _mcp_tool_surface_for_turn(alias_map, overlays, ctx.turn_intent, ctx.request_policy)

    native_tools = _native_tools_for_turn(list(NATIVE_TOOLS), ctx.turn_intent, ctx.request_policy)
    tool_info: list[tuple[str, str]] = [(tool.name, tool.description or "") for tool in native_tools]
--- a/skyvern/forge/sdk/copilot/code_block_synthesis.py
+++ b/skyvern/forge/sdk/copilot/code_block_synthesis.py
@ -50,9 +50,10 @@ _DOWNLOAD_OUTPUT_VAR_BASE = "downloaded_files"
 CREDENTIAL_FILL_TOOL_NAME = "fill_credential_field"
 _CREDENTIAL_FIELDS = frozenset({"username", "password", "totp"})

-# Shape of a synthesized credential fill, ``.fill(<param>.<field>)`` — distinguishes a login
-# fill from a plain ``.fill(str(<key>))`` text input.
-CREDENTIAL_FILL_CODE_PATTERN = re.compile(r"\.fill\([A-Za-z_]\w*\.\w+\)")
+# Shape of a synthesized credential fill, ``.fill(<param>.<field>)`` or the runtime OTP
+# accessor ``.fill(await <param>.otp())`` — distinguishes a login fill from a plain
+# ``.fill(str(<key>))`` text input.
+CREDENTIAL_FILL_CODE_PATTERN = re.compile(r"\.fill\(\s*(?:[A-Za-z_]\w*\.\w+|await\s+[A-Za-z_]\w*\.otp\(\))\s*\)")
 _ENTRY_TARGET_TOOLS = frozenset({"click", "type_text", CREDENTIAL_FILL_TOOL_NAME, "select_option", "press_key"})
 _DURABLE_FALLBACK_ENTRY_TARGET_TOOLS = frozenset({"type_text", CREDENTIAL_FILL_TOOL_NAME, "select_option"})
 _OPTIONAL_DISMISSAL_NAME_PATTERN = re.compile(
@ -905,7 +906,10 @@ def synthesize_code_block(
                credential_param_key = _credential_param_key(interaction, used_param_keys)
                credential_param_keys[credential_id] = credential_param_key
                parameters.append({"key": credential_param_key, "credential_id": credential_id})
-            lines.append(f"{action_indent}await {locator}.fill({credential_param_key}.{credential_field})")
+            if credential_field == "totp":
+                lines.append(f"{action_indent}await {locator}.fill(await {credential_param_key}.otp())")
+            else:
+                lines.append(f"{action_indent}await {locator}.fill({credential_param_key}.{credential_field})")
        elif tool_name == "select_option":
            value = str(interaction.get("value") or "").strip()
            if not value:
@ -1113,8 +1117,9 @@ def render_synthesized_offer_text(
            + bindings
            + ". Bind each as a workflow parameter with `workflow_parameter_type: credential_id` and the "
            "credential ID in `default_value`; at runtime the key resolves to a credential object whose "
-            "`.username` / `.password` / `.totp` attributes the snippet reads (`.totp` is a fresh one-time "
-            "code generated when the block starts). Never replace these attribute reads with literal values."
+            "`.username` / `.password` attributes and `.otp()` method the snippet reads (`.otp()` resolves a "
+            "fresh authenticator, email, or SMS one-time code during the run). Never replace these reads with "
+            "literal values."
        )
    if "page.expect_download()" in synthesized.code:
        parts.append(
--- a/skyvern/forge/sdk/copilot/output_policy.py
+++ b/skyvern/forge/sdk/copilot/output_policy.py
@ -34,16 +34,22 @@ _CREDENTIAL_ID_RE = re.compile(r"\bcred_[A-Za-z0-9][A-Za-z0-9_-]*\b")
 _PLACEHOLDER_MARKERS = ("{{", "{%", "[REDACTED_SECRET]")
 # RHS of a secret-keyword assignment that references a bound value instead of carrying one:
 # a `parameters`-rooted lookup (quoted-key subscript / .get / attribute), or an attribute
-# chain ending in a credential field (`cred.password`), optionally wrapped in str(...).
+# chain ending in a credential field (`cred.password` / `await cred.otp()`), optionally wrapped in str(...).
+# `totp` remains allowed for backward compatibility with old synthesized code;
+# new Code-block OTP flows should use `await cred.otp()`.
 # Fully anchored — only closing punctuation may follow, so a literal appended to a
 # reference (`cred.password+"hunter2"`) or a dotted literal (a JWT) never passes.
 _SANCTIONED_SECRET_REFERENCE_RE = re.compile(
    r"^(?:str\()?"
    r"(?:parameters(?:\[(?:'[^']*'|\"[^\"]*\")\]|\.get\((?:'[^']*'|\"[^\"]*\")\)|(?:\.[A-Za-z_][A-Za-z0-9_]*)+)"
-    r"|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*\.(?:username|password|totp))"
+    r"|[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*\.(?:username|password|totp)"
+    r"|(?:await\s+)?[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*\.otp\(\))"
    r"[)\]\},;.'\"]*$"
 )
-_SECRET_ASSIGNMENT_RHS_RE = re.compile(r"[:=]\s*(\S+)\s*$")
+# The RHS can be a multi-token expression such as `await login_credentials.otp()`.
+# Callers pass the single line containing the match, so this is not expected to
+# consume across embedded newlines.
+_SECRET_ASSIGNMENT_RHS_RE = re.compile(r"[:=]\s*(.+)\s*$")
 _UNVALIDATED_PROPOSAL_AFFORDANCE_RE = re.compile(
    r"\baccept\b(?=[\s\S]{0,120}\bsav(?:e|ed|ing)\b)(?=[\s\S]{0,160}\b(?:reject|discard)\b)",
    re.IGNORECASE,
@ -432,8 +438,8 @@ def format_output_policy_tool_error(verdict: OutputPolicyVerdict) -> str:
    if OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes:
        message += (
            " For saved credentials, bind a credential_id workflow parameter and reference fields as "
-            "`<key>.username`, `<key>.password`, or `<key>.totp`; do not split, concatenate, or obfuscate "
-            "literal secrets in workflow code or YAML."
+            "`<key>.username`, `<key>.password`, or `await <key>.otp()` for one-time codes; do not split, "
+            "concatenate, or obfuscate literal secrets in workflow code or YAML."
        )
    return message

@ -447,12 +453,23 @@ def _contains_raw_secret(value: Any) -> bool:
                matched = match.group(0)
                if any(marker in matched for marker in _PLACEHOLDER_MARKERS):
                    continue
-                if pattern is SECRET_KEYWORD_ASSIGNMENT_PATTERN and _is_sanctioned_secret_reference(matched):
+                if pattern is SECRET_KEYWORD_ASSIGNMENT_PATTERN and (
+                    _is_sanctioned_secret_reference(matched)
+                    or _is_sanctioned_secret_reference(_line_containing_match(text, match))
+                ):
                    continue
                return True
    return False


+def _line_containing_match(text: str, match: re.Match[str]) -> str:
+    line_start = text.rfind("\n", 0, match.start()) + 1
+    line_end = text.find("\n", match.end())
+    if line_end == -1:
+        line_end = len(text)
+    return text[line_start:line_end]
+
+
 def _is_sanctioned_secret_reference(matched: str) -> bool:
    rhs_match = _SECRET_ASSIGNMENT_RHS_RE.search(matched)
    if rhs_match is None:
--- a/skyvern/forge/sdk/copilot/request_policy.py
+++ b/skyvern/forge/sdk/copilot/request_policy.py
@ -137,6 +137,9 @@ _NAMED_CREDENTIAL_TOKEN_RE = re.compile(
    r"\b(?:saved\s+credential|credential)\s+(?:named|called)\s+([A-Za-z0-9_.@:-]{2,100})\b",
    re.I,
 )
+_CODE_BLOCK_AUTHORING_MARKERS = ("code block", "code-block", "codeblock")
+_LOGIN_BLOCK_BAN_MARKERS = ("do not create a login block", "don't create a login block", "no login block")
+_CREDENTIAL_CODE_MARKERS = ("saved credential", "login_credentials", ".otp()", "one-time-code")


 _MAX_COMPLETION_CRITERIA = 8
@ -589,6 +592,33 @@ def _classifier_fallback_policy(
    )


+def _explicit_code_block_credential_draft_requested(user_message: str) -> bool:
+    normalized = " ".join((user_message or "").lower().split())
+    if not normalized:
+        return False
+    has_code_marker = any(marker in normalized for marker in _CODE_BLOCK_AUTHORING_MARKERS)
+    if not has_code_marker:
+        return False
+    blocks_login = any(marker in normalized for marker in _LOGIN_BLOCK_BAN_MARKERS)
+    mentions_credential_code = any(marker in normalized for marker in _CREDENTIAL_CODE_MARKERS)
+    return blocks_login or mentions_credential_code
+
+
+def _apply_explicit_code_block_credential_draft_policy(policy: RequestPolicy, user_message: str) -> None:
+    if policy.raw_secret_detected:
+        return
+    if not _explicit_code_block_credential_draft_requested(user_message):
+        return
+    policy.testing_intent = "skip_test"
+    policy.allow_update_workflow = True
+    policy.allow_run_blocks = False
+    policy.allow_missing_credentials_in_draft = True
+    policy.requires_user_clarification = False
+    policy.user_response_policy = "proceed"
+    policy.clarification_reason = "none"
+    policy.clarification_question = None
+
+
 async def _run_request_policy_classifier(handler: Any, prompt: str) -> tuple[Any | None, str, int]:
    retry_count = 0
    last_failure_kind = "none"
@ -1216,6 +1246,9 @@ async def build_request_policy(
            organization_id=organization_id,
            exc_info=True,
        )
+    # This narrows classifier output only after the classifier has identified
+    # credential intent; running it earlier would be overwritten by the model verdict.
+    _apply_explicit_code_block_credential_draft_policy(policy, user_message)
    try:
        await _seed_discovered_credentials(
            policy,
--- a/skyvern/forge/sdk/copilot/tools/init.py
+++ b/skyvern/forge/sdk/copilot/tools/init.py
@ -854,20 +854,23 @@ async def fill_credential_field_tool(
    The secret value is resolved server-side from the stored credential and never
    enters the conversation; the result reports only `typed_length`. Use this
    instead of `type_text` whenever a login form field should receive a saved
-    credential's username, password, or one-time TOTP code — `type_text` cannot
-    type secrets and you never have the values.
+    credential's username, password, or authenticator-app one-time code. Email/SMS
+    OTP credentials are not filled during scouting because scouting has no
+    workflow run/task context for safe polling.

    `selector` must be a CSS selector for the exact input field (no comma-union
    fallbacks — inspect the page first and target the proven field).
    `credential_id` must be a credential from the request policy's
-    `resolved_credentials`. `field` is one of `username`, `password`, `totp`
-    (`totp` generates a fresh code at call time).
+    `resolved_credentials`. `field` is one of `username`, `password`, `totp`.

    This tool only fills; it never clicks or submits. Each successful fill is
    recorded as a scouted interaction, so the SYNTHESIZED CODE BLOCK will bind
-    the credential as a `credential_id` workflow parameter and reference it as
-    `<parameter_key>.username` / `.password` / `.totp` — keep that attribute
-    form when persisting code blocks.
+    the credential as a `credential_id` workflow parameter and reference
+    username/password as `<parameter_key>.username` / `.password`.
+
+    In synthesized code blocks, one-time codes must use
+    `await <parameter_key>.otp()` so authenticator, email, and SMS OTP sources
+    all resolve at runtime.
    """
    result = await _fill_credential_field_impl(ctx.context, selector, credential_id, field)
    return json.dumps(scrub_secrets_from_structure(ctx.context, result))
--- a/skyvern/forge/sdk/copilot/tools/banned_blocks.py
+++ b/skyvern/forge/sdk/copilot/tools/banned_blocks.py
@ -225,7 +225,7 @@ def _code_only_browser_schema_guidance() -> list[str]:
        "Use concrete selectors and text anchors found during exploration. If only intent targeting is available, inspect the page again before mutating.",
        _code_only_browser_validation_guidance(),
        "Keep block outputs JSON-safe and include visible evidence text when extracting records, products, totals, confirmations, or identifiers.",
-        "For saved credentials: bind the credential as a workflow parameter with workflow_parameter_type credential_id and the credential ID in default_value. At runtime the parameter key resolves to a credential object — read <key>.username, <key>.password, and <key>.totp (a fresh one-time code generated when the block starts). Never put literal secret values in code; scout credential fields with fill_credential_field.",
+        "For saved credentials: bind the credential as a workflow parameter with workflow_parameter_type credential_id and the credential ID in default_value. At runtime the parameter key resolves to a credential object — read <key>.username and <key>.password, and use await <key>.otp() for authenticator, email, or SMS one-time codes. Never put literal secret values in code; scout credential fields with fill_credential_field.",
    ]


@ -254,7 +254,7 @@ Runtime facts:
 - For browser reads, prefer visible anchors, locator text, block outputs, and
  MCP/scout evidence gathered before authoring.
 - A `credential_id` workflow parameter resolves to a credential object with
-  `<key>.username`, `<key>.password`, and fresh `<key>.totp`; scout fields with
+  `<key>.username`, `<key>.password`, and `await <key>.otp()` for one-time codes; scout fields with
  `fill_credential_field`, never embed literal secrets.
 - Credentialed login code must be idempotent. After `goto`, wait for either the
  login form or an already-authenticated page anchor; only fill username/password
--- a/skyvern/forge/sdk/copilot/tools/credential_fill.py
+++ b/skyvern/forge/sdk/copilot/tools/credential_fill.py
@ -16,7 +16,7 @@ from skyvern.forge.sdk.copilot.secret_scrub import (
    register_secret_scrub_value,
    scrub_secrets_from_text,
 )
-from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential
+from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential, TotpType
 from skyvern.forge.sdk.services.credentials import parse_totp_secret

 from .banned_blocks import _copilot_block_authoring_policy
@ -41,6 +41,15 @@ _CREDENTIAL_FILL_FIELDS = frozenset({"username", "password", "totp"})
 _CREDENTIAL_FILL_TIMEOUT_MS = 15000


+def _runtime_otp_steering_error(credential_id: str) -> str:
+    return (
+        f"Credential `{credential_id}` receives one-time codes by email/SMS, so `fill_credential_field` cannot "
+        "safely retrieve the code during scouting without a workflow run/task context to anchor polling. "
+        "Persist the OTP step in a code block as `await <credential_parameter>.otp()` after the action that "
+        "triggers delivery; the runtime will poll for the fresh code during the workflow run without exposing it."
+    )
+
+
 def _scrub_secret_from_text(text: str, secret_value: str) -> str:
    if not secret_value:
        return text
@ -120,6 +129,10 @@ async def _resolve_credential_fill_value(
        register_secret_scrub_value(copilot_ctx, value)
    else:
        if not credential.totp:
+            # A saved OTP identifier means the code is delivered out-of-band;
+            # only runtime polling has the run/task context needed to resolve it.
+            if credential.totp_identifier or credential.totp_type in {TotpType.EMAIL, TotpType.TEXT}:
+                return None, "", _runtime_otp_steering_error(credential_id)
            return None, "", f"Credential `{credential_id}` has no TOTP secret configured."
        try:
            value = pyotp.TOTP(parse_totp_secret(credential.totp)).now()
--- a/skyvern/forge/sdk/copilot/tools/guardrails.py
+++ b/skyvern/forge/sdk/copilot/tools/guardrails.py
@ -177,6 +177,19 @@ def _has_reached_download_target(ctx: Any) -> bool:
    return target.already_registered or bool(target.selector.strip())


+def _request_policy_allows_untested_code_block_draft(ctx: Any) -> bool:
+    policy = getattr(ctx, "request_policy", None)
+    return (
+        getattr(ctx, "allow_untested_workflow_draft", False) is True
+        and isinstance(policy, RequestPolicy)
+        and policy.allow_update_workflow
+        and not policy.allow_run_blocks
+        and policy.testing_intent == "skip_test"
+        and policy.credential_input_kind == "credential_name"
+        and policy.allow_missing_credentials_in_draft
+    )
+
+
 def _download_scout_required_error(copilot_ctx: Any, workflow_yaml: str | None) -> str | None:
    """Reject authoring a download-intent code block until the affordance has been scout-acted
    this turn, so the skyvern_evaluate post-hook can populate the reached-download target and the
@ -188,6 +201,14 @@ def _download_scout_required_error(copilot_ctx: Any, workflow_yaml: str | None)
    download_labels = _download_intent_block_labels(workflow_yaml)
    if not download_labels:
        return None
+    if _request_policy_allows_untested_code_block_draft(copilot_ctx):
+        LOG.info(
+            "copilot download scout-act gate skipped for untested credential draft",
+            workflow_permanent_id=getattr(copilot_ctx, "workflow_permanent_id", None),
+            download_intent_block_labels=download_labels,
+            surface="tool_pre_side_effect",
+        )
+        return None
    if _has_reached_download_target(copilot_ctx):
        return None
    if turn_has_scout_interaction(copilot_ctx):
--- a/skyvern/forge/sdk/copilot/tools/workflow_update.py
+++ b/skyvern/forge/sdk/copilot/tools/workflow_update.py
@ -98,7 +98,12 @@ from .frontier import (
    _stale_block_metadata_message,
    _workflow_requires_canonical_persist,
 )
-from .guardrails import _authority_tool_error, _download_binding_required_error, _download_scout_required_error
+from .guardrails import (
+    _authority_tool_error,
+    _download_binding_required_error,
+    _download_scout_required_error,
+    _request_policy_allows_untested_code_block_draft,
+)

 LOG = structlog.get_logger()

@ -266,11 +271,43 @@ _CODE_ARTIFACT_REQUIRED_LIST_FIELDS = (
    "completion_criteria",
    "terminal_verifier_expectations",
 )
-_CREDENTIAL_FIELD_ACCESS_RE = re.compile(r"\b([A-Za-z_][A-Za-z0-9_]*)\.(username|password|totp)\b")
+_CREDENTIAL_FIELD_ACCESS_RE = re.compile(
+    r"\b(?P<parameter>[A-Za-z_][A-Za-z0-9_]*)\.(?:(?P<field>username|password|totp)\b|(?P<otp_method>otp)\s*\()"
+)
 _CODE_SUBMIT_ACTION_RE = re.compile(r"\.(?:click|press)\s*\(")
 _SCOUT_SUBMIT_TOOL_NAMES = frozenset({"click", "press_key"})


+class CredentialFieldAccess(NamedTuple):
+    parameter_key: str
+    field: str
+    requires_live_scout: bool
+
+
+def _credential_field_accesses(code: str) -> list[CredentialFieldAccess]:
+    accesses: list[CredentialFieldAccess] = []
+    for match in _CREDENTIAL_FIELD_ACCESS_RE.finditer(code):
+        field = match.group("field")
+        if field:
+            accesses.append(
+                CredentialFieldAccess(
+                    parameter_key=match.group("parameter"),
+                    field=field,
+                    requires_live_scout=True,
+                )
+            )
+            continue
+        if match.group("otp_method"):
+            accesses.append(
+                CredentialFieldAccess(
+                    parameter_key=match.group("parameter"),
+                    field="totp",
+                    requires_live_scout=False,
+                )
+            )
+    return accesses
+
+
 def _code_artifact_metadata_as_tool_argument(
    metadata: list[CodeArtifactMetadata] | None,
 ) -> list[dict[str, Any]]:
@ -2703,10 +2740,12 @@ def _credentialed_code_block_scout_gate_errors(
        if not code.strip():
            continue
        required_fields_by_credential: dict[str, set[str]] = {}
-        for parameter_key, field in _CREDENTIAL_FIELD_ACCESS_RE.findall(code):
-            credential_id = credential_params_by_key.get(parameter_key)
+        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(field)
+                required_fields_by_credential.setdefault(credential_id, set()).add(access.field)
        if not required_fields_by_credential:
            continue

@ -2858,7 +2897,11 @@ async def _update_workflow(
        ctx.code_artifact_metadata = merged_metadata
        ctx.workflow_verification_evidence.code_artifact_metadata = merged_metadata
        params["code_artifact_metadata"] = merged_metadata
-    credential_scout_errors = _credentialed_code_block_scout_gate_errors(workflow_yaml, ctx)
+    credential_scout_errors = (
+        []
+        if _request_policy_allows_untested_code_block_draft(ctx)
+        else _credentialed_code_block_scout_gate_errors(workflow_yaml, ctx)
+    )
    if credential_scout_errors and code_safety_errors and code_artifact_metadata_error is None:
        return {
            "ok": False,
--- a/skyvern/forge/sdk/db/repositories/otp.py
+++ b/skyvern/forge/sdk/db/repositories/otp.py
@ -2,7 +2,7 @@ from __future__ import annotations

 from datetime import datetime, timedelta, timezone

-from sqlalchemy import and_, asc, select
+from sqlalchemy import and_, asc, or_, select

 from skyvern.config import settings
 from skyvern.forge.sdk.db._error_handling import db_operation
@ -22,6 +22,7 @@ class OTPRepository(BaseRepository):
        valid_lifespan_minutes: int = settings.TOTP_LIFESPAN_MINUTES,
        otp_type: OTPType | None = None,
        workflow_run_id: str | None = None,
+        include_unscoped_workflow_run: bool = False,
        created_after: datetime | None = None,
        limit: int | None = None,
    ) -> list[TOTPCode]:
@ -29,7 +30,7 @@ class OTPRepository(BaseRepository):
        1. filter by:
        - organization_id
        - totp_identifier
-        - workflow_run_id (optional)
+        - workflow_run_id (optional); include unscoped rows too when requested
        - created_after (optional): only codes created at/after this instant
        2. make sure created_at is within the valid lifespan
        3. sort by task_id/workflow_id/workflow_run_id nullslast and created_at desc
@ -51,7 +52,11 @@ class OTPRepository(BaseRepository):
            )
            if otp_type:
                query = query.filter(TOTPCodeModel.otp_type == otp_type)
-            if workflow_run_id is not None:
+            if workflow_run_id is not None and include_unscoped_workflow_run:
+                query = query.filter(
+                    or_(TOTPCodeModel.workflow_run_id == workflow_run_id, TOTPCodeModel.workflow_run_id.is_(None))
+                )
+            elif workflow_run_id is not None:
                query = query.filter(TOTPCodeModel.workflow_run_id == workflow_run_id)
            if created_after is not None:
                query = query.filter(TOTPCodeModel.created_at >= created_after)
--- a/skyvern/forge/sdk/db/repositories/workflow_parameters.py
+++ b/skyvern/forge/sdk/db/repositories/workflow_parameters.py
@ -63,6 +63,7 @@ from skyvern.forge.sdk.workflow.models.parameter import (
    WorkflowParameter,
    WorkflowParameterType,
 )
+from skyvern.utils.action_redaction import redact_action_for_log
 from skyvern.webeye.actions.actions import Action

 LOG = structlog.get_logger()
@ -725,6 +726,8 @@ class WorkflowParametersRepository(BaseRepository):
    @db_operation("create_action")
    async def create_action(self, action: Action) -> Action:
        async with self.Session() as session:
+            raw_action_payload = action.model_dump()
+            action_log_payload = redact_action_for_log(action)
            new_action = ActionModel(
                action_type=action.action_type,
                source_action_id=action.source_action_id,
@ -737,12 +740,12 @@ class WorkflowParametersRepository(BaseRepository):
                status=action.status,
                reasoning=action.reasoning,
                intention=action.intention,
-                response=action.response,
+                response=action_log_payload.get("response"),
                element_id=action.element_id,
                skyvern_element_hash=action.skyvern_element_hash,
                skyvern_element_data=action.skyvern_element_data,
                screenshot_artifact_id=action.screenshot_artifact_id,
-                action_json=action.model_dump(),
+                action_json=raw_action_payload,
                confidence_float=action.confidence_float,
                created_by=action.created_by,
            )
--- a/skyvern/forge/sdk/routes/credentials.py
+++ b/skyvern/forge/sdk/routes/credentials.py
@ -119,7 +119,7 @@ from skyvern.schemas.workflows import (
    WorkflowParameterYAML,
    WorkflowStatus,
 )
-from skyvern.services.otp_service import OTPValue, parse_otp_login
+from skyvern.services.otp_service import OTPValue, parse_otp_login, redact_otp_identifier_for_log
 from skyvern.services.run_service import cancel_workflow_run
 from skyvern.utils.url_validators import validate_url

@ -188,10 +188,11 @@ async def send_totp_code(
    data: TOTPCodeCreate,
    curr_org: Organization = Depends(org_auth_service.get_current_org),
 ) -> TOTPCode:
+    redacted_totp_identifier = redact_otp_identifier_for_log(data.totp_identifier)
    LOG.info(
        "Saving OTP code",
        organization_id=curr_org.organization_id,
-        totp_identifier=data.totp_identifier,
+        totp_identifier=redacted_totp_identifier,
        task_id=data.task_id,
        workflow_id=data.workflow_id,
        workflow_run_id=data.workflow_run_id,
@ -211,18 +212,35 @@ async def send_totp_code(
            raise HTTPException(status_code=400, detail=f"Invalid workflow run id: {data.workflow_run_id}")
    content = data.content.strip()
    otp_value: OTPValue | None = OTPValue(value=content, type=data.type or OTPType.TOTP)
+    parse_exception_type_name: str | None = None
    # We assume the user is sending the code directly when the length of code is less than or equal to 10
    if len(content) > 10:
-        otp_value = await parse_otp_login(content, curr_org.organization_id, enforced_otp_type=data.type)
+        try:
+            otp_value = await parse_otp_login(content, curr_org.organization_id, enforced_otp_type=data.type)
+        except Exception as e:
+            otp_value = None
+            parse_exception_type_name = type(e).__name__
+
+    if parse_exception_type_name:
+        LOG.error(
+            "Failed to parse otp login",
+            totp_identifier=redacted_totp_identifier,
+            task_id=data.task_id,
+            workflow_id=data.workflow_id,
+            workflow_run_id=data.workflow_run_id,
+            content_length=len(data.content),
+            exception_type=parse_exception_type_name,
+        )
+        raise HTTPException(status_code=400, detail="Failed to parse otp login")

    if not otp_value:
        LOG.error(
            "Failed to parse otp login",
-            totp_identifier=data.totp_identifier,
+            totp_identifier=redacted_totp_identifier,
            task_id=data.task_id,
            workflow_id=data.workflow_id,
            workflow_run_id=data.workflow_run_id,
-            content=data.content,
+            content_length=len(data.content),
        )
        raise HTTPException(status_code=400, detail="Failed to parse otp login")

--- a/skyvern/forge/sdk/schemas/sdk_actions.py
+++ b/skyvern/forge/sdk/schemas/sdk_actions.py
@ -4,6 +4,7 @@ from typing import Annotated, Any, Literal, Union
 from pydantic import BaseModel, Field

 from skyvern.config import settings
+from skyvern.utils.action_redaction import redact_input_text_payload_for_log


 class SdkActionType(str, Enum):
@ -62,6 +63,14 @@ class InputTextAction(SdkActionBase):
    totp_url: str | None = Field(None, description="TOTP URL for input_text actions")
    timeout: float = Field(default=settings.BROWSER_ACTION_TIMEOUT_MS, description="Timeout in milliseconds")

+    def __repr__(self) -> str:
+        payload = redact_input_text_payload_for_log(self.model_dump(), value_key="value")
+        fields = ", ".join(f"{key}={value!r}" for key, value in payload.items())
+        return f"{self.__class__.__name__}({fields})"
+
+    def __str__(self) -> str:
+        return self.__repr__()
+
    def get_navigation_goal(self) -> str | None:
        return self.intention

--- a/skyvern/schemas/steps.py
+++ b/skyvern/schemas/steps.py
@ -1,8 +1,11 @@
 from __future__ import annotations

-from pydantic import BaseModel
+from typing import Any, cast
+
+from pydantic import BaseModel, SerializerFunctionWrapHandler, model_serializer

 from skyvern.errors.errors import UserDefinedError
+from skyvern.utils.action_redaction import redact_action_for_log
 from skyvern.webeye.actions.actions import Action
 from skyvern.webeye.actions.responses import ActionResult

@ -26,6 +29,19 @@ class AgentStepOutput(BaseModel):
    browser_metadata: BrowserMetadata | None = None
    step_exception: str | None = None

+    @model_serializer(mode="wrap")
+    def serialize_model(self, handler: SerializerFunctionWrapHandler) -> dict[str, Any]:
+        payload = cast(dict[str, Any], handler(self))
+        if self.actions_and_results is None:
+            return payload
+
+        serialized_pairs = payload.get("actions_and_results") or []
+        redacted_pairs = []
+        for (action, _), (_, serialized_results) in zip(self.actions_and_results, serialized_pairs, strict=True):
+            redacted_pairs.append((redact_action_for_log(action), serialized_results))
+        payload["actions_and_results"] = redacted_pairs
+        return payload
+
    def __repr__(self) -> str:
        return f"AgentStepOutput({self.model_dump()})"

--- a/skyvern/services/otp_service.py
+++ b/skyvern/services/otp_service.py
@ -32,9 +32,14 @@ _NON_ALNUM_PATTERN = re.compile(r"[^a-z0-9]")
 _EXPECTED_TOTP_WEBHOOK_RESPONSE_SHAPE = '{"verification_code":"123456"}'
 # Recovers the verification_code value when the surrounding JSON is malformed
 # (e.g. unescaped quotes inside a relayed email). Assumes verification_code is
-# the final field, which is the common shape.
+# the final field, which is the common shape; the closing brace anchor is a
+# best-effort recovery boundary, not a strict JSON parser.
 _VERIFICATION_CODE_FIELD_PATTERN = re.compile(r'"verification_code"\s*:\s*"(?P<value>.*)"\s*}\s*\Z', re.DOTALL)
-_TOTP_WEBHOOK_BODY_PREVIEW_LIMIT = 200
+_REDACTED_OTP_BODY_PLACEHOLDER = "[REDACTED_OTP_BODY]"
+_REDACTED_OTP_IDENTIFIER_PLACEHOLDER = "[REDACTED_OTP_IDENTIFIER]"
+_TOTP_WEBHOOK_NON_JSON_RESPONSE_REASON = "totp_webhook_non_json_response"
+_TOTP_WEBHOOK_REQUEST_FAILED_REASON = "totp_webhook_request_failed"
+_SAFE_TOTP_ERROR_REASON_PREFIXES = (_TOTP_WEBHOOK_NON_JSON_RESPONSE_REASON, _TOTP_WEBHOOK_REQUEST_FAILED_REASON)
 _TOTP_WEBHOOK_REQUEST_MAX_ATTEMPTS = 3
 _TOTP_WEBHOOK_REQUEST_RETRY_TIMEOUT_SECONDS = 5

@ -154,27 +159,40 @@ def _get_header_value(headers: dict[str, str], header_name: str) -> str | None:


 def _format_content_type_for_error(content_type: str | None) -> str:
-    return content_type if content_type is not None else "<absent>"
+    if content_type is None:
+        return "<absent>"
+    media_type = content_type.split(";", maxsplit=1)[0].strip().lower()
+    if not media_type or "/" not in media_type or len(media_type) > 100:
+        return "<present>"
+    return media_type


 def _response_body_preview(response_body: Any) -> str:
    body = response_body if isinstance(response_body, str) else str(response_body)
-    if len(body) <= _TOTP_WEBHOOK_BODY_PREVIEW_LIMIT:
-        return body
-    return f"{body[:_TOTP_WEBHOOK_BODY_PREVIEW_LIMIT]}... (truncated)"
+    body_start = body.lstrip()[:1]
+    json_like = body_start in {"{", "["}
+    return f"{_REDACTED_OTP_BODY_PLACEHOLDER}(length={len(body)},json_like={str(json_like).lower()})"
+
+
+def _schema_only_otp_error_reason(reason: str | None) -> str:
+    if reason and reason.startswith(_SAFE_TOTP_ERROR_REASON_PREFIXES):
+        return reason
+    return _TOTP_WEBHOOK_REQUEST_FAILED_REASON
+
+
+def redact_otp_identifier_for_log(totp_identifier: str | None) -> str | None:
+    return _REDACTED_OTP_IDENTIFIER_PLACEHOLDER if totp_identifier else None


 def _totp_webhook_contract_error_reason(
    *,
-    url: str,
    status_code: int,
    content_type: str | None,
    response_body: Any,
 ) -> str:
    return (
-        "TOTP webhook returned HTTP 200 but the response was not JSON. "
-        f"endpoint_url={url} "
-        f"HTTP status={status_code} "
+        f"{_TOTP_WEBHOOK_NON_JSON_RESPONSE_REASON} "
+        f"http_status={status_code} "
        f"content_type={_format_content_type_for_error(content_type)} "
        f"body_preview={_response_body_preview(response_body)!r} "
        f"expected_response_shape={_EXPECTED_TOTP_WEBHOOK_RESPONSE_SHAPE}"
@ -235,17 +253,18 @@ async def _post_totp_verification_url(
                return response.status_code, response.headers, response.body, False
            parsed, is_json = _coerce_totp_response_body(response.body)
            return response.status_code, response.headers, parsed, is_json
-        except Exception:
+        except Exception as e:
+            # Avoid exc_info here because network exceptions can include the
+            # webhook URL or response details; keep retry logs diagnostic but sanitized.
            LOG.debug(
                "TOTP webhook request attempt failed",
-                endpoint_url=url,
                attempt=attempt + 1,
                max_attempts=max_attempts,
-                exc_info=True,
+                exception_type=type(e).__name__,
            )
            if attempt < max_attempts - 1 and retry_timeout > 0:
                await asyncio.sleep(retry_timeout)
-    raise _TOTPWebhookRequestError(f"Failed post request url={url}")
+    raise _TOTPWebhookRequestError("Failed post request to TOTP verification URL")


 def try_generate_totp_for_credential(
@ -413,8 +432,6 @@ async def poll_otp_value(
        task_id=task_id,
        workflow_run_id=workflow_run_id,
        workflow_permanent_id=workflow_permanent_id,
-        totp_verification_url=totp_verification_url,
-        totp_identifier=totp_identifier,
    )
    consecutive_failures = 0
    last_error_reason: str | None = None
@ -431,8 +448,6 @@ async def poll_otp_value(
                    task_id=task_id,
                    workflow_run_id=workflow_run_id,
                    workflow_id=workflow_id or workflow_permanent_id,
-                    totp_verification_url=totp_verification_url,
-                    totp_identifier=totp_identifier,
                    reason=last_error_reason,
                )
            LOG.warning("Polling otp value timed out")
@ -440,8 +455,6 @@ async def poll_otp_value(
                task_id=task_id,
                workflow_run_id=workflow_run_id,
                workflow_id=workflow_id or workflow_permanent_id,
-                totp_verification_url=totp_verification_url,
-                totp_identifier=totp_identifier,
            )
        otp_value: OTPValue | None = None
        try:
@ -465,15 +478,13 @@ async def poll_otp_value(
                )
        except FailedToGetTOTPVerificationCode as e:
            consecutive_failures += 1
-            last_error_reason = e.reason
+            last_error_reason = _schema_only_otp_error_reason(e.reason)
            LOG.warning(
                "OTP fetch failed, will retry until wall-clock timeout",
                consecutive_failures=consecutive_failures,
-                last_error_reason=e.reason,
+                last_error_reason=last_error_reason,
                task_id=task_id,
                workflow_run_id=workflow_run_id,
-                totp_verification_url=totp_verification_url,
-                totp_identifier=totp_identifier,
            )
            continue
        consecutive_failures = 0
@ -484,7 +495,6 @@ async def poll_otp_value(
                task_id=task_id,
                workflow_run_id=workflow_run_id,
                workflow_permanent_id=workflow_permanent_id,
-                totp_identifier=totp_identifier,
                otp_type=otp_value.get_otp_type().value,
                otp_length=len(otp_value.value),
            )
@ -518,19 +528,20 @@ async def _get_otp_value_from_url(
            organization_id=organization_id,
        )
    except Exception as e:
-        LOG.error("Failed to get otp value from url", totp_verification_url=url, exc_info=True)
+        LOG.error(
+            "Failed to get otp value from url",
+            exception_type=type(e).__name__,
+        )
        raise FailedToGetTOTPVerificationCode(
            task_id=task_id,
            workflow_run_id=workflow_run_id,
            workflow_id=workflow_permanent_id,
-            totp_verification_url=url,
-            reason=str(e),
+            reason=f"{_TOTP_WEBHOOK_REQUEST_FAILED_REASON} exception_type={type(e).__name__}",
        )
    content_type = _get_header_value(response_headers, "Content-Type")
    if status_code != 200:
        LOG.warning(
            "TOTP webhook returned non-200 response",
-            endpoint_url=url,
            http_status=status_code,
            content_type=content_type,
            body_preview=_response_body_preview(response_body),
@ -539,14 +550,12 @@ async def _get_otp_value_from_url(

    if not is_json_response:
        reason = _totp_webhook_contract_error_reason(
-            url=url,
            status_code=status_code,
            content_type=content_type,
            response_body=response_body,
        )
        LOG.error(
            "TOTP webhook returned non-JSON response",
-            endpoint_url=url,
            http_status=status_code,
            content_type=content_type,
            body_preview=_response_body_preview(response_body),
@ -556,14 +565,12 @@ async def _get_otp_value_from_url(
            task_id=task_id,
            workflow_run_id=workflow_run_id,
            workflow_id=workflow_permanent_id,
-            totp_verification_url=url,
            reason=reason,
        )

    if not isinstance(response_body, dict):
        LOG.warning(
            "TOTP webhook response body is not a JSON object",
-            endpoint_url=url,
            http_status=status_code,
            content_type=content_type,
            response_json_type=type(response_body).__name__,
@ -575,7 +582,6 @@ async def _get_otp_value_from_url(
    if not content:
        LOG.warning(
            "No verification_code found in TOTP webhook response",
-            endpoint_url=url,
            http_status=status_code,
            content_type=content_type,
            response_keys=list(response_body.keys()),
@ -587,8 +593,13 @@ async def _get_otp_value_from_url(
    if isinstance(content, str) and len(content) > 10:
        try:
            otp_value = await parse_otp_login(content, organization_id)
-        except Exception:
-            LOG.warning("faile to parse content by LLM call", exc_info=True)
+        except Exception as e:
+            otp_value = None
+            LOG.warning(
+                "Failed to parse OTP content by LLM call",
+                exception_type=type(e).__name__,
+                content_length=len(content),
+            )

    if not otp_value:
        LOG.warning(
@ -608,8 +619,14 @@ async def _get_otp_value_from_db(
    workflow_run_id: str | None = None,
    created_after: datetime | None = None,
 ) -> OTPValue | None:
+    # Email/SMS deliveries can arrive through /v1/credentials/totp without run
+    # scope, so include both exact run matches and unscoped rows in SQL.
    totp_codes = await app.DATABASE.otp.get_otp_codes(
-        organization_id=organization_id, totp_identifier=totp_identifier, created_after=created_after
+        organization_id=organization_id,
+        totp_identifier=totp_identifier,
+        workflow_run_id=workflow_run_id,
+        include_unscoped_workflow_run=workflow_run_id is not None,
+        created_after=created_after,
    )
    for totp_code in totp_codes:
        if totp_code.workflow_run_id and workflow_run_id and totp_code.workflow_run_id != workflow_run_id:
--- a/skyvern/utils/action_redaction.py
+++ b/skyvern/utils/action_redaction.py
@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from pydantic import BaseModel
+
+REDACTED_OTP_VALUE = "<redacted otp value>"
+REDACTED_OTP_IDENTIFIER = "<redacted otp identifier>"
+REDACTED_OTP_URL = "<redacted otp url>"
+REDACTED_OTP_SECRET = "<redacted otp secret>"
+SDK_INPUT_TEXT_ACTION_TYPE = "ai_input_text"  # Mirrors SdkActionType.AI_INPUT_TEXT without importing sdk_actions.
+
+
+def redact_action_for_log(action: BaseModel) -> dict[str, Any]:
+    return redact_action_payload_for_log(action.model_dump())
+
+
+def redact_action_payload_for_log(action_payload: Mapping[str, Any]) -> dict[str, Any]:
+    redacted_payload = dict(action_payload)
+    action_type = redacted_payload.get("action_type")
+    sdk_action_type = redacted_payload.get("type")
+
+    if action_type == "input_text":
+        return redact_input_text_payload_for_log(redacted_payload, value_key="text")
+
+    if sdk_action_type == SDK_INPUT_TEXT_ACTION_TYPE:
+        return redact_input_text_payload_for_log(redacted_payload, value_key="value")
+
+    return redacted_payload
+
+
+def redact_input_text_payload_for_log(action_payload: Mapping[str, Any], *, value_key: str) -> dict[str, Any]:
+    redacted_payload = dict(action_payload)
+    if _is_otp_input_payload(redacted_payload):
+        if redacted_payload.get(value_key):
+            redacted_payload[value_key] = REDACTED_OTP_VALUE
+        if redacted_payload.get("response"):
+            redacted_payload["response"] = REDACTED_OTP_VALUE
+        if redacted_payload.get("totp_identifier"):
+            redacted_payload["totp_identifier"] = REDACTED_OTP_IDENTIFIER
+        if redacted_payload.get("totp_url"):
+            redacted_payload["totp_url"] = REDACTED_OTP_URL
+        if isinstance(redacted_payload.get("totp_timing_info"), Mapping):
+            timing_info = dict(redacted_payload["totp_timing_info"])
+            if timing_info.get("totp_secret"):
+                timing_info["totp_secret"] = REDACTED_OTP_SECRET
+            redacted_payload["totp_timing_info"] = timing_info
+    return redacted_payload
+
+
+def _is_otp_input_payload(action_payload: Mapping[str, Any]) -> bool:
+    # Keep this in sync with OTP marker fields on SDK/web InputTextAction models.
+    if action_payload.get("totp_identifier") or action_payload.get("totp_url"):
+        return True
+    if action_payload.get("totp_code_required"):
+        return True
+    if isinstance(action_payload.get("totp_timing_info"), Mapping):
+        return True
+
+    return False
--- a/skyvern/webeye/actions/actions.py
+++ b/skyvern/webeye/actions/actions.py
@ -6,6 +6,7 @@ import structlog
 from pydantic import BaseModel, ConfigDict, Field

 from skyvern.errors.errors import UserDefinedError
+from skyvern.utils.action_redaction import redact_input_text_payload_for_log
 from skyvern.webeye.actions.action_types import ActionType

 LOG = structlog.get_logger()
@ -285,9 +286,28 @@ class InputTextAction(WebAction):
    action_type: ActionType = ActionType.INPUT_TEXT
    text: str
    totp_code_required: bool = False
+    totp_identifier: str | None = None
+    totp_url: str | None = None

    def __repr__(self) -> str:
-        return f"InputTextAction(element_id={self.element_id}, text={self.text}, context={self.input_or_select_context}, tool_call_id={self.tool_call_id})"
+        payload = redact_input_text_payload_for_log(
+            {
+                "action_type": self.action_type,
+                "text": self.text,
+                "totp_code_required": self.totp_code_required,
+                "totp_timing_info": self.totp_timing_info,
+                "totp_identifier": self.totp_identifier,
+                "totp_url": self.totp_url,
+            },
+            value_key="text",
+        )
+        return (
+            f"InputTextAction(element_id={self.element_id}, text={payload['text']}, "
+            f"context={self.input_or_select_context}, tool_call_id={self.tool_call_id})"
+        )
+
+    def __str__(self) -> str:
+        return self.__repr__()


 class UploadFileAction(WebAction):
--- a/skyvern/webeye/actions/models.py
+++ b/skyvern/webeye/actions/models.py
@ -8,6 +8,7 @@ from pydantic import BaseModel, ConfigDict
 from skyvern.config import settings
 from skyvern.errors.errors import UserDefinedError
 from skyvern.schemas.steps import AgentStepOutput, BrowserMetadata
+from skyvern.utils.action_redaction import redact_action_for_log
 from skyvern.webeye.actions.actions import Action, DecisiveAction
 from skyvern.webeye.actions.responses import ActionResult
 from skyvern.webeye.scraper.scraped_page import ScrapedPage
@ -31,13 +32,25 @@ class DetailedAgentStepOutput(BaseModel):

    def __repr__(self) -> str:
        if settings.DEBUG_MODE:
-            return f"DetailedAgentStepOutput({self.model_dump()})"
+            return f"DetailedAgentStepOutput({self._redacted_model_dump()})"
        else:
            return f"AgentStepOutput({self.to_agent_step_output().model_dump()})"

    def __str__(self) -> str:
        return self.__repr__()

+    def _redacted_model_dump(self) -> dict[str, Any]:
+        payload = self.model_dump()
+        if self.actions is not None:
+            payload["actions"] = [redact_action_for_log(action) for action in self.actions]
+        if self.actions_and_results is not None:
+            serialized_pairs = payload.get("actions_and_results") or []
+            redacted_pairs = []
+            for (action, _), (_, serialized_results) in zip(self.actions_and_results, serialized_pairs, strict=True):
+                redacted_pairs.append((redact_action_for_log(action), serialized_results))
+            payload["actions_and_results"] = redacted_pairs
+        return payload
+
    def extract_errors(self) -> list[UserDefinedError]:
        errors = []
        if self.actions_and_results:
--- a/tests/unit/forge/sdk/db/test_repositories.py
+++ b/tests/unit/forge/sdk/db/test_repositories.py
@ -1,7 +1,10 @@
 """Tests for all OSS repository instantiations + dependency injection."""

+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch

+import pytest
+

 def test_credential_repository_instantiation():
    from skyvern.forge.sdk.db.repositories.credentials import CredentialRepository
@ -44,6 +47,39 @@ def test_otp_repository_instantiation():
    assert hasattr(repo, "create_otp_code")


+@pytest.mark.asyncio
+async def test_otp_repository_can_include_unscoped_workflow_run_rows_in_sql():
+    from skyvern.forge.sdk.db.repositories.otp import OTPRepository
+
+    class CapturingSession:
+        query = None
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *_args):
+            return False
+
+        async def scalars(self, query):
+            self.query = query
+            return SimpleNamespace(all=lambda: [])
+
+    session = CapturingSession()
+    repo = OTPRepository(session_factory=lambda: session, debug_enabled=False)
+
+    await repo.get_otp_codes(
+        organization_id="o_test",
+        totp_identifier="otp@example.test",
+        workflow_run_id="wr_test",
+        include_unscoped_workflow_run=True,
+    )
+
+    sql = str(session.query)
+    assert "totp_codes.workflow_run_id = :workflow_run_id_1" in sql
+    assert "totp_codes.workflow_run_id IS NULL" in sql
+    assert " OR " in sql
+
+
 def test_debug_repository_instantiation():
    from skyvern.forge.sdk.db.repositories.debug import DebugRepository

--- a/tests/unit/test_actions.py
+++ b/tests/unit/test_actions.py
@ -3,9 +3,24 @@ from unittest.mock import MagicMock
 import pytest
 from pydantic import ValidationError

+from skyvern.forge.sdk.db.repositories.workflow_parameters import WorkflowParametersRepository
+from skyvern.forge.sdk.db.utils import hydrate_action
+from skyvern.forge.sdk.schemas.sdk_actions import InputTextAction as SdkInputTextAction
+from skyvern.forge.sdk.schemas.sdk_actions import SdkActionType
+from skyvern.schemas.steps import AgentStepOutput
+from skyvern.utils.action_redaction import (
+    REDACTED_OTP_IDENTIFIER,
+    REDACTED_OTP_SECRET,
+    REDACTED_OTP_URL,
+    REDACTED_OTP_VALUE,
+    SDK_INPUT_TEXT_ACTION_TYPE,
+    redact_action_for_log,
+)
+from skyvern.webeye.actions.action_types import ActionType
 from skyvern.webeye.actions.actions import (
    Action,
    ClickAction,
+    InputTextAction,
    KeypressAction,
    NewTabAction,
    NullAction,
@ -13,6 +28,7 @@ from skyvern.webeye.actions.actions import (
    SwitchTabAction,
    WebAction,
 )
+from skyvern.webeye.actions.models import DetailedAgentStepOutput
 from skyvern.webeye.actions.parse_actions import parse_action


@ -23,6 +39,10 @@ def _mock_scraped_page() -> MagicMock:
    return page


+def test_sdk_input_text_action_type_constant_matches_sdk_enum() -> None:
+    assert SDK_INPUT_TEXT_ACTION_TYPE == SdkActionType.AI_INPUT_TEXT.value
+
+
 def test_action_parse__no_element_id() -> None:
    action_no_element_id = {
        "action_type": "click",
@ -50,6 +70,245 @@ def test_action_parse__with_element_id() -> None:
    assert action.element_id == "1"


+def test_sdk_input_text_action_repr_redacts_otp_fields() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    secret_identifier = "OTP_IDENTIFIER_SHOULD_NOT_APPEAR"
+    secret_url = "OTP_URL_SHOULD_NOT_APPEAR"
+
+    action = SdkInputTextAction(
+        selector="#otp-field",
+        value=secret_value,
+        intention="Enter one-time code",
+        totp_identifier=secret_identifier,
+        totp_url=secret_url,
+    )
+
+    rendered = repr(action)
+    rendered_str = str(action)
+    raw_payload = action.model_dump()
+    log_payload = redact_action_for_log(action)
+
+    assert secret_value not in rendered
+    assert secret_identifier not in rendered
+    assert secret_url not in rendered
+    assert secret_value not in rendered_str
+    assert secret_identifier not in rendered_str
+    assert secret_url not in rendered_str
+    assert raw_payload["value"] == secret_value
+    assert raw_payload["totp_identifier"] == secret_identifier
+    assert raw_payload["totp_url"] == secret_url
+    assert secret_value not in str(log_payload)
+    assert secret_identifier not in str(log_payload)
+    assert secret_url not in str(log_payload)
+    assert REDACTED_OTP_VALUE in rendered
+    assert REDACTED_OTP_VALUE in rendered_str
+    assert REDACTED_OTP_IDENTIFIER in rendered_str
+    assert REDACTED_OTP_URL in rendered_str
+    assert log_payload["value"] == REDACTED_OTP_VALUE
+    assert "#otp-field" in rendered
+    assert "Enter one-time code" in rendered
+
+
+def test_web_input_text_action_repr_redacts_otp_text() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter verification code",
+        response=secret_value,
+        totp_code_required=True,
+    )
+
+    rendered = repr(action)
+    rendered_str = str(action)
+    raw_payload = action.model_dump()
+    log_payload = redact_action_for_log(action)
+
+    assert secret_value not in rendered
+    assert secret_value not in rendered_str
+    assert raw_payload["text"] == secret_value
+    assert raw_payload["response"] == secret_value
+    assert secret_value not in str(log_payload)
+    assert REDACTED_OTP_VALUE in rendered
+    assert REDACTED_OTP_VALUE in rendered_str
+    assert log_payload["text"] == REDACTED_OTP_VALUE
+    assert log_payload["response"] == REDACTED_OTP_VALUE
+    assert "otp-field" in rendered
+
+
+def test_web_input_text_action_repr_redacts_otp_text_marked_by_identifier() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    secret_identifier = "OTP_IDENTIFIER_SHOULD_NOT_APPEAR"
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter code",
+        response=secret_value,
+        totp_identifier=secret_identifier,
+    )
+
+    rendered = repr(action)
+    rendered_str = str(action)
+    raw_payload = action.model_dump()
+    log_payload = redact_action_for_log(action)
+
+    assert secret_value not in rendered
+    assert secret_value not in rendered_str
+    assert secret_identifier not in rendered_str
+    assert raw_payload["text"] == secret_value
+    assert raw_payload["totp_identifier"] == secret_identifier
+    assert log_payload["text"] == REDACTED_OTP_VALUE
+    assert log_payload["response"] == REDACTED_OTP_VALUE
+    assert log_payload["totp_identifier"] == REDACTED_OTP_IDENTIFIER
+    assert secret_value not in str(log_payload)
+    assert secret_identifier not in str(log_payload)
+    assert REDACTED_OTP_VALUE in rendered_str
+
+
+def test_step_output_serialization_redacts_otp_input_action() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter verification code",
+        response=secret_value,
+        totp_code_required=True,
+    )
+
+    payload = AgentStepOutput(actions_and_results=[(action, [])]).model_dump()
+
+    assert secret_value not in str(payload)
+    assert payload["actions_and_results"][0][0]["text"] == REDACTED_OTP_VALUE
+    assert payload["actions_and_results"][0][0]["response"] == REDACTED_OTP_VALUE
+
+
+def test_detailed_step_output_debug_repr_redacts_otp_input_action(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr("skyvern.config.settings.DEBUG_MODE", True)
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    secret_identifier = "OTP_IDENTIFIER_SHOULD_NOT_APPEAR"
+    secret_url = "OTP_URL_SHOULD_NOT_APPEAR"
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter verification code",
+        response=secret_value,
+        totp_identifier=secret_identifier,
+        totp_url=secret_url,
+    )
+
+    rendered = repr(
+        DetailedAgentStepOutput(
+            scraped_page=None,
+            extract_action_prompt=None,
+            llm_response=None,
+            actions=[action],
+            action_results=None,
+            actions_and_results=[(action, [])],
+        )
+    )
+
+    assert secret_value not in rendered
+    assert secret_identifier not in rendered
+    assert secret_url not in rendered
+    assert REDACTED_OTP_VALUE in rendered
+    assert REDACTED_OTP_IDENTIFIER in rendered
+    assert REDACTED_OTP_URL in rendered
+
+
+def test_action_log_payload_redacts_otp_text_response_and_timing_secret() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    timing_secret = "OTP_TIMING_SECRET_SHOULD_NOT_APPEAR"
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter passcode",
+        response=secret_value,
+        totp_timing_info={"is_totp_sequence": True, "totp_secret": timing_secret, "action_index": 0},
+    )
+
+    payload = redact_action_for_log(action)
+
+    assert payload["text"] == REDACTED_OTP_VALUE
+    assert payload["response"] == REDACTED_OTP_VALUE
+    assert payload["totp_timing_info"]["totp_secret"] == REDACTED_OTP_SECRET
+    assert secret_value not in str(payload)
+    assert timing_secret not in str(payload)
+    assert payload["element_id"] == "otp-field"
+    assert payload["action_type"] == ActionType.INPUT_TEXT
+
+
+def test_action_log_payload_keeps_non_otp_input_debuggable() -> None:
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        element_id="account-field",
+        text="SAFE_ACCOUNT_REFERENCE",
+        intention="Enter account reference",
+        response="SAFE_ACCOUNT_REFERENCE",
+    )
+
+    payload = redact_action_for_log(action)
+
+    assert payload["text"] == "SAFE_ACCOUNT_REFERENCE"
+    assert payload["response"] == "SAFE_ACCOUNT_REFERENCE"
+    assert payload["element_id"] == "account-field"
+    assert payload["intention"] == "Enter account reference"
+
+
+@pytest.mark.asyncio
+async def test_create_action_redacts_response_but_preserves_action_json_for_hydration() -> None:
+    secret_value = "OTP_SECRET_VALUE_SHOULD_NOT_APPEAR"
+    captured_models = []
+
+    class FakeSession:
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        def add(self, model) -> None:
+            captured_models.append(model)
+
+        async def commit(self) -> None:
+            pass
+
+        async def refresh(self, model) -> None:
+            pass
+
+    repo = WorkflowParametersRepository(lambda: FakeSession())
+    action = InputTextAction(
+        action_type=ActionType.INPUT_TEXT,
+        organization_id="o_test",
+        workflow_run_id="wr_test",
+        task_id="tsk_test",
+        step_id="stp_test",
+        step_order=0,
+        action_order=0,
+        element_id="otp-field",
+        text=secret_value,
+        intention="Enter verification code",
+        response=secret_value,
+        totp_code_required=True,
+    )
+
+    await repo.create_action(action)
+
+    persisted_model = captured_models[0]
+    assert persisted_model.response == REDACTED_OTP_VALUE
+    assert persisted_model.action_json["text"] == secret_value
+    assert persisted_model.action_json["response"] == secret_value
+
+    hydrated_action = hydrate_action(persisted_model)
+    assert isinstance(hydrated_action, InputTextAction)
+    assert hydrated_action.text == secret_value
+    assert hydrated_action.response == secret_value
+
+
 def test_web_action_parse__no_element_id() -> None:
    action_no_element_id = {
        "action_type": "click",
--- a/tests/unit/test_copilot_code_block_persist_seam.py
+++ b/tests/unit/test_copilot_code_block_persist_seam.py
@ -16,6 +16,7 @@ from skyvern.forge.sdk.copilot.blocker_signal import assert_clean_user_facing_te
 from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
 from skyvern.forge.sdk.copilot.context import CopilotContext
 from skyvern.forge.sdk.copilot.reached_download_target import ReachedDownloadTarget
+from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
 from skyvern.forge.sdk.copilot.tools import (
    _code_block_safety_errors,
    _detect_stale_block_metadata,
@ -229,6 +230,28 @@ def _standard_ctx() -> CopilotContext:
    return ctx


+def _draft_only_credential_ctx() -> CopilotContext:
+    ctx = _code_only_ctx()
+    ctx.scout_trajectory = []
+    ctx.allow_untested_workflow_draft = True
+    ctx.request_policy = RequestPolicy(
+        testing_intent="skip_test",
+        credential_input_kind="credential_name",
+        credential_refs=["Saved portal credential"],
+        allow_update_workflow=True,
+        allow_run_blocks=False,
+        allow_missing_credentials_in_draft=True,
+        resolved_credentials=[
+            SimpleNamespace(
+                credential_id="cred_missing",
+                name="Saved portal credential",
+                tested_url="https://example.com/login",
+            )
+        ],
+    )
+    return ctx
+
+
 def _enable_imposition(ctx: CopilotContext) -> None:
    ctx.impose_synthesized_code_block = True

@ -2013,6 +2036,43 @@ class TestCredentialScoutPersistGate:
        await page.locator("input[type='submit']").click()
        """
    )
+    _RUNTIME_OTP_CODE_YAML = _credential_code_yaml(
+        code="""
+        await page.locator("#email").fill(login_credential.username)
+        await page.locator("input[type='password']").fill(login_credential.password)
+        await page.locator("#totpmfa").fill(await login_credential.otp())
+        await page.locator("input[type='submit']").click()
+        await page.wait_for_load_state("load")
+        """
+    )
+    _DRAFT_DOWNLOAD_CODE_YAML = _credential_code_yaml(
+        code="""
+        await page.goto("https://example.com/login")
+        await page.locator("#email").fill(login_credential.username)
+        await page.locator("input[type='password']").fill(login_credential.password)
+        await page.locator("#totpmfa").fill(await login_credential.otp())
+        await page.locator("input[type='submit']").click()
+        await page.wait_for_load_state("load")
+        async with page.expect_download() as download_info:
+            await page.locator("a[href='/invoices/monthly.pdf']").click()
+        download = await download_info.value
+        print(download.suggested_filename)
+        """
+    )
+    _DOWNLOAD_CODE_YAML = _yaml(
+        """
+        title: Download invoice
+        workflow_definition:
+          blocks:
+          - block_type: code
+            label: download_monthly_invoice_pdf
+            code: |
+              async with page.expect_download() as download_info:
+                  await page.locator("a[href='/invoices/monthly.pdf']").click()
+              download = await download_info.value
+              print(download.suggested_filename)
+        """
+    )

    @pytest.mark.asyncio
    async def test_rejects_credential_submit_code_without_matching_fill_scouts(self) -> None:
@ -2095,6 +2155,48 @@ class TestCredentialScoutPersistGate:
        assert "click the submit control or press Enter" not in error_text
        assert "saved-credential login flow" not in error_text

+    @pytest.mark.asyncio
+    async def test_runtime_otp_method_does_not_require_impossible_live_otp_fill(self) -> None:
+        ctx = _code_only_ctx()
+        ctx.scout_trajectory = [
+            _credential_fill_interaction("username"),
+            _credential_fill_interaction("password"),
+            _submit_interaction(),
+        ]
+
+        result = await _update_workflow({"workflow_yaml": self._RUNTIME_OTP_CODE_YAML}, ctx)
+
+        assert result["ok"] is False
+        error_text = str(result.get("error") or "")
+        assert "was not found in this organization" in error_text
+        assert "successful `fill_credential_field` scouting for `totp`" not in error_text
+        assert "saved-credential login flow" not in error_text
+
+    @pytest.mark.asyncio
+    async def test_draft_only_credential_code_download_persists_without_scouts(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        _stub_successful_update(monkeypatch)
+        ctx = _draft_only_credential_ctx()
+
+        result = await _update_workflow({"workflow_yaml": self._DRAFT_DOWNLOAD_CODE_YAML}, ctx)
+
+        assert result["ok"] is True
+        assert "login_with_saved_credential" in ctx.workflow_yaml
+        assert "expect_download" in ctx.workflow_yaml
+        assert "login_credential.username" in ctx.workflow_yaml
+
+    @pytest.mark.asyncio
+    async def test_download_scout_gate_still_blocks_normal_code_only_authoring(self) -> None:
+        ctx = _code_only_ctx()
+        ctx.scout_trajectory = []
+
+        result = await _update_workflow({"workflow_yaml": self._DOWNLOAD_CODE_YAML}, ctx)
+
+        assert result["ok"] is False
+        assert "Scout it first" in result["error"]
+        assert "download affordance" in result["error"]
+
    @pytest.mark.asyncio
    async def test_standard_mode_behavior_is_unchanged(self) -> None:
        ctx = _standard_ctx()
--- a/tests/unit/test_copilot_code_block_synthesis.py
+++ b/tests/unit/test_copilot_code_block_synthesis.py
@ -20,6 +20,7 @@ from skyvern.forge.sdk.copilot.code_block_synthesis import (
    _SYNTHESIZED_BLOCK_LABEL,
    CREDENTIAL_FILL_TOOL_NAME,
    build_synthesized_artifact_metadata,
+    code_contains_credential_fill,
    is_optional_dismissal_only_trajectory,
    render_synthesized_offer_text,
    synthesize_code_block,
@ -1465,12 +1466,15 @@ class TestCredentialFillSynthesis:
        assert 'await page.locator("#passwordInput").fill(authtest_simple.password)' in result.code
        assert result.parameters == [{"key": "authtest_simple", "credential_id": "cred_123"}]

-    def test_totp_field_reads_totp_attribute(self) -> None:
+    def test_totp_field_reads_runtime_otp_method(self) -> None:
        result = synthesize_code_block(
            [self._credential_fill(selector="#totpCode", credential_field="totp", typed_length=6)]
        )
        assert result is not None
-        assert 'await page.locator("#totpCode").fill(authtest_simple.totp)' in result.code
+        assert 'await page.locator("#totpCode").fill(await authtest_simple.otp())' in result.code
+
+    def test_runtime_otp_fill_is_detected_as_credential_fill_code(self) -> None:
+        assert code_contains_credential_fill('await page.locator("#otp").fill(await login_credential.otp())')

    def test_missing_credential_reference_is_dropped_with_note(self) -> None:
        result = synthesize_code_block(
@ -1528,7 +1532,7 @@ class TestCredentialFillSynthesis:
        assert "`authtest_simple` -> `cred_123`" in text
        assert "workflow_parameter_type: credential_id" in text
        assert "default_value" in text
-        assert ".username` / `.password` / `.totp`" in text
+        assert ".username` / `.password` attributes and `.otp()`" in text
        assert "authtest_simple" not in [p.get("key") for p in synthesized.parameters if "credential_id" not in p]

    def test_credential_parameters_excluded_from_plain_bind_line(self) -> None:
--- a/tests/unit/test_copilot_fill_credential_field.py
+++ b/tests/unit/test_copilot_fill_credential_field.py
@ -22,7 +22,7 @@ from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy
 from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
 from skyvern.forge.sdk.copilot.tools import credential_fill as credential_fill_module
 from skyvern.forge.sdk.copilot.tools import scouting as scouting_module
-from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential
+from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordCredential, TotpType

 _FAKE_PASSWORD = "fake-test-password-7x9"
 _FAKE_USERNAME = "qa.user@example.test"
@ -175,6 +175,25 @@ class TestResolveCredentialFillValue:
        assert error is not None
        assert "TOTP" in error

+    @pytest.mark.asyncio
+    async def test_email_otp_credential_returns_runtime_otp_steer(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._wire_vault(
+            monkeypatch,
+            PasswordCredential(
+                username=_FAKE_USERNAME,
+                password=_FAKE_PASSWORD,
+                totp=None,
+                totp_type=TotpType.EMAIL,
+                totp_identifier="otp@example.test",
+            ),
+        )
+        value, _, error = await tools_module._resolve_credential_fill_value(_ctx(), "cred_123", "totp")
+        assert value is None
+        assert error is not None
+        assert "await <credential_parameter>.otp()" in error
+        assert "workflow run" in error
+        assert "otp@example.test" not in error
+
    @pytest.mark.asyncio
    async def test_missing_credential_errors(self, monkeypatch: pytest.MonkeyPatch) -> None:
        monkeypatch.setattr(
--- a/tests/unit/test_copilot_output_policy.py
+++ b/tests/unit/test_copilot_output_policy.py
@ -348,6 +348,43 @@ class TestSanctionedSecretReferenceIdiom:

        assert verdict.allowed

+    def test_allows_awaited_credential_otp_assignment_in_workflow_yaml(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            workflow_yaml=(
+                'code: |\n  passcode = await login_credentials.otp()\n  await page.locator("#code").fill(passcode)\n'
+            ),
+        )
+
+        assert verdict.allowed
+
+    def test_allows_keyword_secret_reference_to_awaited_credential_otp_in_tool_arguments(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "code: |\n  token = await login_credential.otp()"},
+        )
+
+        assert verdict.allowed
+
+    def test_allows_secret_keyword_match_when_full_line_is_sanctioned_otp_reference(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "code: |\n  passcode = await login_credential.otp()"},
+        )
+
+        assert verdict.allowed
+
+    def test_rejects_literal_secret_even_when_same_line_has_credential_reference(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={
+                "workflow_yaml": "code: |\n  password = 'hunter2'; password_ref = login_credential.password"
+            },
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
    def test_allows_str_wrapped_parameter_reference(self) -> None:
        verdict = evaluate_output_policy(
            request_policy=_policy(),
@ -370,6 +407,15 @@ class TestSanctionedSecretReferenceIdiom:
            assert not verdict.allowed, rhs
            assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes

+    def test_still_rejects_awaited_non_credential_secret_source(self) -> None:
+        verdict = evaluate_output_policy(
+            request_policy=_policy(),
+            tool_arguments={"workflow_yaml": "code: |\n  passcode = await page.locator('#code').inner_text()"},
+        )
+
+        assert not verdict.allowed
+        assert OutputPolicyReason.RAW_SECRET_LEAK in verdict.reason_codes
+
    def test_allows_keyword_argument_reference_with_closing_paren(self) -> None:
        verdict = evaluate_output_policy(
            request_policy=_policy(),
--- a/tests/unit/test_copilot_request_policy_workflow_credentials.py
+++ b/tests/unit/test_copilot_request_policy_workflow_credentials.py
@ -284,3 +284,55 @@ async def test_no_discovered_credentials_leaves_approved_set_empty() -> None:

    assert policy.discovered_credentials == []
    get_by_ids.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_fallback_code_block_credential_request_saves_draft_without_running() -> None:
+    credential = SimpleNamespace(
+        credential_id="cred_email_otp",
+        name="mock-portal-login-email-otp",
+        tested_url="http://localhost:8900/telco_billing/northwind/?mfa=email",
+    )
+    with patch(
+        "skyvern.forge.app.DATABASE.credentials.get_credentials",
+        new=AsyncMock(return_value=[credential]),
+    ):
+        policy = await build_request_policy(
+            user_message=(
+                "Build this as a Code block credential test using the saved credential named "
+                "mock-portal-login-email-otp. Do not create a Login block for sign-in or MFA, "
+                "and use await login_credentials.otp() for the email one-time-code."
+            ),
+            workflow_yaml="",
+            chat_history=[],
+            global_llm_context="",
+            organization_id="o_test",
+            handler=None,
+        )
+
+    assert policy.classifier_status == "fallback"
+    assert policy.credential_input_kind == "credential_name"
+    assert policy.credential_refs == ["mock-portal-login-email-otp"]
+    assert policy.resolved_credentials == [credential]
+    assert policy.allow_update_workflow is True
+    assert policy.allow_run_blocks is False
+    assert policy.allow_missing_credentials_in_draft is True
+    assert policy.testing_intent == "skip_test"
+    assert policy.requires_user_clarification is False
+
+
+@pytest.mark.asyncio
+async def test_fallback_code_block_generic_one_time_code_does_not_skip_run() -> None:
+    policy = await build_request_policy(
+        user_message="Build a code block that handles a one time code after sign in.",
+        workflow_yaml="",
+        chat_history=[],
+        global_llm_context="",
+        organization_id="o_test",
+        handler=None,
+    )
+
+    assert policy.classifier_status == "fallback"
+    assert policy.allow_run_blocks is True
+    assert policy.allow_missing_credentials_in_draft is False
+    assert policy.testing_intent == "unspecified"
--- a/tests/unit/test_copilot_turn_intent_tool_gate.py
+++ b/tests/unit/test_copilot_turn_intent_tool_gate.py
@ -5,16 +5,18 @@ from unittest.mock import AsyncMock, patch

 import pytest

-from skyvern.forge.sdk.copilot.agent import _native_tools_for_turn
+from skyvern.forge.sdk.copilot.agent import _mcp_tool_surface_for_turn, _native_tools_for_turn
 from skyvern.forge.sdk.copilot.blocker_signal import CopilotToolBlockerSignal
 from skyvern.forge.sdk.copilot.context import CopilotContext
 from skyvern.forge.sdk.copilot.request_policy import RequestPolicy
 from skyvern.forge.sdk.copilot.tools import (
    NATIVE_TOOLS,
    _authority_tool_error,
+    _build_skyvern_mcp_overlays,
    _get_run_results,
    _turn_intent_tool_error,
    _update_workflow,
+    get_skyvern_mcp_alias_map,
 )
 from skyvern.forge.sdk.copilot.turn_intent import (
    UNRESOLVED_BLOCK_REF_TARGET_ENTITY,
@ -116,6 +118,38 @@ def test_turn_intent_gate_allows_draft_update_without_run_authority() -> None:
    assert _turn_intent_tool_error(_ctx(intent), "update_workflow") is None


+def test_draft_only_credential_code_policy_prunes_browser_tool_surface() -> None:
+    intent = TurnIntent(
+        mode=TurnIntentMode.DRAFT_ONLY,
+        authority=TurnIntentAuthority(may_update_workflow=True, may_run_blocks=False),
+    )
+    policy = RequestPolicy(
+        testing_intent="skip_test",
+        allow_update_workflow=True,
+        allow_run_blocks=False,
+        allow_missing_credentials_in_draft=True,
+    )
+
+    alias_map, overlays = _mcp_tool_surface_for_turn(
+        get_skyvern_mcp_alias_map(),
+        _build_skyvern_mcp_overlays(),
+        intent,
+        policy,
+    )
+    native_names = {getattr(tool, "name", None) for tool in _native_tools_for_turn(list(NATIVE_TOOLS), intent, policy)}
+
+    assert set(alias_map) == {"get_block_schema", "validate_block"}
+    assert set(overlays) == {"get_block_schema", "validate_block"}
+    for browser_tool in {"navigate_browser", "evaluate", "click", "type_text"}:
+        assert browser_tool not in alias_map
+        assert browser_tool not in overlays
+    assert "update_workflow" in native_names
+    assert "list_credentials" in native_names
+    assert "fill_credential_field" not in native_names
+    assert "discover_workflow_entrypoint" not in native_names
+    assert "inspect_page_for_composition" not in native_names
+
+
 def test_turn_intent_gate_allows_build_page_inspection_with_update_authority() -> None:
    intent = TurnIntent(
        mode=TurnIntentMode.BUILD,
--- a/tests/unit/test_credentials_totp_route_logging.py
+++ b/tests/unit/test_credentials_totp_route_logging.py
@ -0,0 +1,120 @@
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import structlog.testing
+from fastapi import HTTPException
+
+from skyvern.forge.sdk.routes import credentials
+from skyvern.forge.sdk.schemas.totp_codes import TOTPCodeCreate
+
+
+def _database_with_otp_create(create_otp_code: AsyncMock) -> SimpleNamespace:
+    return SimpleNamespace(otp=SimpleNamespace(create_otp_code=create_otp_code))
+
+
+def _assert_raw_values_not_logged(logs: list[dict[str, object]], *raw_values: str) -> None:
+    for record in logs:
+        record_repr = repr(record)
+        for raw_value in raw_values:
+            assert raw_value not in record_repr
+            for value in record.values():
+                assert raw_value not in str(value)
+
+
+@pytest.mark.asyncio
+async def test_send_totp_code_save_log_redacts_identifier_but_stores_raw_values(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    raw_identifier = "qa-email-otp@example.test"
+    raw_content = "135790"
+    create_otp_code = AsyncMock(return_value=SimpleNamespace(totp_code_id="otp_1"))
+    monkeypatch.setattr(credentials.app, "DATABASE", _database_with_otp_create(create_otp_code))
+
+    with structlog.testing.capture_logs() as logs:
+        result = await credentials.send_totp_code(
+            TOTPCodeCreate(totp_identifier=raw_identifier, content=raw_content),
+            curr_org=SimpleNamespace(organization_id="o_test"),
+        )
+
+    assert result.totp_code_id == "otp_1"
+    save_log = next((r for r in logs if r.get("event") == "Saving OTP code"), None)
+    assert save_log is not None
+    assert save_log["totp_identifier"] == "[REDACTED_OTP_IDENTIFIER]"
+    _assert_raw_values_not_logged(logs, raw_identifier, raw_content)
+
+    create_otp_code.assert_awaited_once()
+    storage_kwargs = create_otp_code.await_args.kwargs
+    assert storage_kwargs["totp_identifier"] == raw_identifier
+    assert storage_kwargs["content"] == raw_content
+    assert storage_kwargs["code"] == raw_content
+
+
+@pytest.mark.asyncio
+async def test_send_totp_code_parse_failure_log_redacts_identifier_and_content(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    raw_identifier = "qa-email-otp@example.test"
+    raw_content = "Use 135790 to finish sign in for qa-email-otp@example.test."
+    create_otp_code = AsyncMock()
+    monkeypatch.setattr(credentials.app, "DATABASE", _database_with_otp_create(create_otp_code))
+    monkeypatch.setattr(credentials, "parse_otp_login", AsyncMock(return_value=None))
+
+    with structlog.testing.capture_logs() as logs:
+        with pytest.raises(HTTPException) as exc_info:
+            await credentials.send_totp_code(
+                TOTPCodeCreate(totp_identifier=raw_identifier, content=raw_content),
+                curr_org=SimpleNamespace(organization_id="o_test"),
+            )
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.detail == "Failed to parse otp login"
+    assert exc_info.value.__cause__ is None
+    assert exc_info.value.__context__ is None
+    create_otp_code.assert_not_awaited()
+
+    error_log = next((r for r in logs if r.get("event") == "Failed to parse otp login"), None)
+    assert error_log is not None
+    assert error_log["totp_identifier"] == "[REDACTED_OTP_IDENTIFIER]"
+    assert error_log["content_length"] == len(raw_content)
+    assert "content" not in error_log
+    _assert_raw_values_not_logged(logs, raw_identifier, raw_content, "135790")
+
+
+@pytest.mark.asyncio
+async def test_send_totp_code_parser_exception_log_redacts_raw_exception_context(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    raw_identifier = "qa-email-otp@example.test"
+    raw_content = "Email body says OTP 246810 for qa-email-otp@example.test."
+    create_otp_code = AsyncMock()
+    monkeypatch.setattr(credentials.app, "DATABASE", _database_with_otp_create(create_otp_code))
+
+    async def parse_otp_login_raises(*_args: object, **_kwargs: object) -> None:
+        raise RuntimeError(f"parser saw {raw_content} for {raw_identifier}")
+
+    monkeypatch.setattr(credentials, "parse_otp_login", parse_otp_login_raises)
+
+    with structlog.testing.capture_logs() as logs:
+        with pytest.raises(HTTPException) as exc_info:
+            await credentials.send_totp_code(
+                TOTPCodeCreate(totp_identifier=raw_identifier, content=raw_content),
+                curr_org=SimpleNamespace(organization_id="o_test"),
+            )
+
+    assert exc_info.value.status_code == 400
+    assert exc_info.value.detail == "Failed to parse otp login"
+    assert exc_info.value.__cause__ is None
+    assert exc_info.value.__context__ is None
+    assert raw_identifier not in str(exc_info.value)
+    assert raw_content not in str(exc_info.value)
+    assert "246810" not in str(exc_info.value)
+    create_otp_code.assert_not_awaited()
+
+    error_log = next((r for r in logs if r.get("event") == "Failed to parse otp login"), None)
+    assert error_log is not None
+    assert error_log["totp_identifier"] == "[REDACTED_OTP_IDENTIFIER]"
+    assert error_log["content_length"] == len(raw_content)
+    assert error_log["exception_type"] == "RuntimeError"
+    assert "content" not in error_log
+    _assert_raw_values_not_logged(logs, raw_identifier, raw_content, "246810")
--- a/tests/unit/test_otp_service.py
+++ b/tests/unit/test_otp_service.py
@ -177,10 +177,13 @@ class TestGetOtpValueFromUrl:
            )

        reason = exc_info.value.reason or ""
-        assert "https://example.com/totp" in reason
-        assert "HTTP status=200" in reason
-        assert "content_type=text/plain; charset=utf-8" in reason
-        assert "body_preview='147258'" in reason
+        assert "https://example.com/totp" not in reason
+        assert "totp_webhook_non_json_response" in reason
+        assert "http_status=200" in reason
+        assert "content_type=text/plain" in reason
+        assert "charset" not in reason
+        assert "body_preview='[REDACTED_OTP_BODY](length=6,json_like=false)'" in reason
+        assert "147258" not in reason
        assert '{"verification_code":"123456"}' in reason
        assert "api-key" not in reason

@ -204,7 +207,7 @@ class TestGetOtpValueFromUrl:
            )

        reason = exc_info.value.reason or ""
-        assert f"body_preview='{long_body[:200]}... (truncated)'" in reason
+        assert "body_preview='[REDACTED_OTP_BODY](length=250,json_like=false)'" in reason
        assert long_body not in reason

    @pytest.mark.asyncio
@ -247,8 +250,9 @@ class TestGetOtpValueFromUrl:
            )

        reason = exc_info.value.reason or ""
-        assert "body_preview='line1\\nline2'" in reason
-        assert "body_preview='line1\\\\nline2'" not in reason
+        assert "body_preview='[REDACTED_OTP_BODY](length=11,json_like=false)'" in reason
+        assert "line1" not in reason
+        assert "line2" not in reason

    @pytest.mark.asyncio
    async def test_json_missing_verification_code_returns_none_and_logs(self, monkeypatch: pytest.MonkeyPatch) -> None:
@ -271,7 +275,13 @@ class TestGetOtpValueFromUrl:
            )

        assert result is None
-        assert any("No verification_code found in TOTP webhook response" in log.get("event", "") for log in logs)
+        missing_code_log = next(
+            (log for log in logs if "No verification_code found in TOTP webhook response" in log.get("event", "")),
+            None,
+        )
+        assert missing_code_log is not None
+        assert "endpoint_url" not in missing_code_log
+        assert "https://example.com/totp" not in repr(logs)

    @pytest.mark.asyncio
    async def test_json_non_object_response_returns_none_and_logs_distinct_event(
@ -296,7 +306,13 @@ class TestGetOtpValueFromUrl:
            )

        assert result is None
-        assert any("TOTP webhook response body is not a JSON object" in log.get("event", "") for log in logs)
+        non_object_log = next(
+            (log for log in logs if "TOTP webhook response body is not a JSON object" in log.get("event", "")),
+            None,
+        )
+        assert non_object_log is not None
+        assert "endpoint_url" not in non_object_log
+        assert "https://example.com/totp" not in repr(logs)
        assert not any("No verification_code found in TOTP webhook response" in log.get("event", "") for log in logs)

    @pytest.mark.asyncio
@ -320,7 +336,12 @@ class TestGetOtpValueFromUrl:
            )

        assert result is None
-        assert any("TOTP webhook returned non-200 response" in log.get("event", "") for log in logs)
+        non_200_log = next(
+            (log for log in logs if "TOTP webhook returned non-200 response" in log.get("event", "")), None
+        )
+        assert non_200_log is not None
+        assert "endpoint_url" not in non_200_log
+        assert "https://example.com/totp" not in repr(logs)

    @pytest.mark.asyncio
    async def test_relayed_email_in_malformed_json_extracts_code_instead_of_raising(
@ -534,7 +555,7 @@ class TestPollOtpValueRetry:

    @pytest.mark.asyncio
    async def test_raises_failed_with_reason_when_timeout_during_failure_streak(self) -> None:
-        """When the wall-clock timeout fires while the webhook is still failing, surface the underlying reason."""
+        """When timeout fires while webhook still fails, surface only sanitized failure context."""
        base = datetime(2026, 1, 1, 12, 0, 0)
        utcnow_returns = [
            base,  # start_datetime
@ -553,15 +574,44 @@ class TestPollOtpValueRetry:
            mock_datetime.utcnow.side_effect = utcnow_returns
            mock_settings.VERIFICATION_CODE_POLLING_TIMEOUT_MINS = 15
            mock_app.DATABASE.organizations.get_valid_org_auth_token = AsyncMock(return_value=_mock_org_token())
-            mock_fetch.side_effect = FailedToGetTOTPVerificationCode(reason="connection refused")
+            raw_identifier = "otp@example.com"
+            raw_url = "https://example.com/mfa?token=secret"
+            raw_code = "135790"
+            mock_fetch.side_effect = FailedToGetTOTPVerificationCode(
+                reason=f"connection refused for {raw_identifier} at {raw_url} after code {raw_code}"
+            )

-            with pytest.raises(FailedToGetTOTPVerificationCode) as exc_info:
-                await poll_otp_value(
-                    organization_id="o_test",
-                    task_id="tsk_test",
-                    totp_verification_url="https://example.com/mfa",
-                )
-            assert exc_info.value.reason == "connection refused"
+            with structlog.testing.capture_logs() as logs:
+                with pytest.raises(FailedToGetTOTPVerificationCode) as exc_info:
+                    await poll_otp_value(
+                        organization_id="o_test",
+                        task_id="tsk_test",
+                        totp_identifier=raw_identifier,
+                        totp_verification_url=raw_url,
+                    )
+
+            assert exc_info.value.reason == "totp_webhook_request_failed"
+            assert raw_identifier not in str(exc_info.value)
+            assert raw_url not in str(exc_info.value)
+            assert raw_code not in str(exc_info.value)
+
+            retry_log = next(
+                (r for r in logs if r.get("event") == "OTP fetch failed, will retry until wall-clock timeout"), None
+            )
+            assert retry_log is not None
+            assert "totp_identifier" not in retry_log
+            assert "totp_verification_url" not in retry_log
+
+            timeout_log = next(
+                (r for r in logs if r.get("event") == "Polling otp value timed out while webhook was still failing"),
+                None,
+            )
+            assert timeout_log is not None
+            assert timeout_log["last_error_reason"] == exc_info.value.reason
+            for record in logs:
+                assert raw_identifier not in repr(record)
+                assert raw_url not in repr(record)
+                assert raw_code not in repr(record)

    @pytest.mark.asyncio
    async def test_raises_no_otp_at_timeout_when_polls_were_clean(self) -> None:
@ -597,13 +647,15 @@ class TestPollOtpValueRetry:
    @patch("skyvern.services.otp_service._get_otp_value_from_url", new_callable=AsyncMock)
    @patch("skyvern.services.otp_service.app")
    @patch("skyvern.services.otp_service.settings")
-    async def test_success_log_omits_raw_otp_code_but_keeps_ids_and_type(
+    async def test_success_log_omits_raw_otp_code_and_otp_locator_fields(
        self, mock_settings: MagicMock, mock_app: MagicMock, mock_fetch: AsyncMock, mock_sleep: AsyncMock
    ) -> None:
-        """SKY-11011: the success log must not emit the raw OTP code, only ids + OTP type."""
+        """The success and polling logs must not emit OTP codes, identifiers, or verification URLs."""
        import structlog.testing

        raw = "135790"
+        raw_identifier = "otp@example.com"
+        raw_url = "https://example.com/mfa"
        mock_settings.VERIFICATION_CODE_POLLING_TIMEOUT_MINS = 15
        mock_app.DATABASE.organizations.get_valid_org_auth_token = AsyncMock(return_value=_mock_org_token())
        mock_fetch.return_value = OTPValue(value=raw, type=OTPType.TOTP)
@ -614,8 +666,8 @@ class TestPollOtpValueRetry:
                task_id="tsk_test",
                workflow_run_id="wr_test",
                workflow_permanent_id="wpid_test",
-                totp_identifier="otp@example.com",
-                totp_verification_url="https://example.com/mfa",
+                totp_identifier=raw_identifier,
+                totp_verification_url=raw_url,
            )

        assert result is not None
@ -623,9 +675,13 @@ class TestPollOtpValueRetry:

        for record in logs:
            assert raw not in repr(record)
+            assert raw_identifier not in repr(record)
+            assert raw_url not in repr(record)
            assert raw not in str(record.values())
            for value in record.values():
                assert raw not in str(value)
+                assert raw_identifier not in str(value)
+                assert raw_url not in str(value)
            assert "otp_value" not in record
            assert record.get("value") != raw

@ -634,7 +690,8 @@ class TestPollOtpValueRetry:
        assert success["task_id"] == "tsk_test"
        assert success["workflow_run_id"] == "wr_test"
        assert success["workflow_permanent_id"] == "wpid_test"
-        assert success["totp_identifier"] == "otp@example.com"
+        assert "totp_identifier" not in success
+        assert "totp_verification_url" not in success
        assert success["otp_type"] == "totp"
        assert success["otp_length"] == 6
        assert isinstance(success["otp_length"], int)
@ -642,7 +699,8 @@ class TestPollOtpValueRetry:

        polling = next((r for r in logs if r.get("event") == "Polling otp value"), None)
        assert polling is not None
-        assert polling["totp_identifier"] == "otp@example.com"
+        assert "totp_identifier" not in polling
+        assert "totp_verification_url" not in polling
        assert raw not in str(polling.values())


@ -1082,3 +1140,108 @@ async def test_get_otp_value_from_db_forwards_created_after(monkeypatch: pytest.
    assert result is None
    get_otp_codes.assert_awaited_once()
    assert get_otp_codes.await_args.kwargs["created_after"] == started_at
+    assert get_otp_codes.await_args.kwargs["workflow_run_id"] == "wr_test"
+    assert get_otp_codes.await_args.kwargs["include_unscoped_workflow_run"] is True
+
+
+@pytest.mark.asyncio
+async def test_get_otp_value_from_db_scopes_query_to_workflow_run_when_provided() -> None:
+    """Run-scoped polling prefers exact rows but keeps unscoped email/SMS pushes eligible."""
+    unscoped = SimpleNamespace(
+        code="111111",
+        otp_type=OTPType.TOTP,
+        workflow_run_id=None,
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+    other_run = SimpleNamespace(
+        code="333333",
+        otp_type=OTPType.TOTP,
+        workflow_run_id="wr_other",
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+    scoped = SimpleNamespace(
+        code="222222",
+        otp_type=OTPType.TOTP,
+        workflow_run_id="wr_test",
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+
+    async def get_otp_codes(**kwargs: object) -> list[SimpleNamespace]:
+        assert kwargs["workflow_run_id"] == "wr_test"
+        assert kwargs["include_unscoped_workflow_run"] is True
+        return [other_run, scoped, unscoped]
+
+    with patch("skyvern.services.otp_service.app") as mock_app:
+        mock_app.DATABASE.otp.get_otp_codes = AsyncMock(side_effect=get_otp_codes)
+        result = await _get_otp_value_from_db(
+            "o_test",
+            "otp@example.com",
+            workflow_run_id="wr_test",
+            created_after=datetime(2026, 6, 8, 20, 3, 0),
+        )
+
+    assert result == OTPValue(value="222222", type=OTPType.TOTP)
+    mock_app.DATABASE.otp.get_otp_codes.assert_awaited_once()
+    assert mock_app.DATABASE.otp.get_otp_codes.await_args.kwargs["workflow_run_id"] == "wr_test"
+    assert mock_app.DATABASE.otp.get_otp_codes.await_args.kwargs["include_unscoped_workflow_run"] is True
+
+
+@pytest.mark.asyncio
+async def test_get_otp_value_from_db_allows_unscoped_code_for_run_scoped_poll() -> None:
+    unscoped = SimpleNamespace(
+        code="111111",
+        otp_type=OTPType.TOTP,
+        workflow_run_id=None,
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+    other_run = SimpleNamespace(
+        code="333333",
+        otp_type=OTPType.TOTP,
+        workflow_run_id="wr_other",
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+    get_otp_codes = AsyncMock(return_value=[other_run, unscoped])
+
+    with patch("skyvern.services.otp_service.app") as mock_app:
+        mock_app.DATABASE.otp.get_otp_codes = get_otp_codes
+        result = await _get_otp_value_from_db(
+            "o_test",
+            "otp@example.com",
+            workflow_run_id="wr_test",
+            created_after=datetime(2026, 6, 8, 20, 3, 0),
+        )
+
+    assert result == OTPValue(value="111111", type=OTPType.TOTP)
+    assert get_otp_codes.await_args.kwargs["workflow_run_id"] == "wr_test"
+    assert get_otp_codes.await_args.kwargs["include_unscoped_workflow_run"] is True
+
+
+@pytest.mark.asyncio
+async def test_get_otp_value_from_db_preserves_unscoped_lookup_without_workflow_run() -> None:
+    unscoped = SimpleNamespace(
+        code="111111",
+        otp_type=OTPType.TOTP,
+        workflow_run_id=None,
+        workflow_id=None,
+        task_id=None,
+        expired_at=None,
+    )
+    get_otp_codes = AsyncMock(return_value=[unscoped])
+
+    with patch("skyvern.services.otp_service.app") as mock_app:
+        mock_app.DATABASE.otp.get_otp_codes = get_otp_codes
+        result = await _get_otp_value_from_db("o_test", "otp@example.com")
+
+    assert result == OTPValue(value="111111", type=OTPType.TOTP)
+    assert get_otp_codes.await_args.kwargs["workflow_run_id"] is None
+    assert get_otp_codes.await_args.kwargs["include_unscoped_workflow_run"] is False
--- a/tests/unit/test_otp_service_redaction.py
+++ b/tests/unit/test_otp_service_redaction.py
@ -0,0 +1,31 @@
+"""Redaction coverage for OTP webhook/email handling."""
+
+from __future__ import annotations
+
+from skyvern.services.otp_service import _response_body_preview, _totp_webhook_contract_error_reason
+
+
+def test_response_body_preview_redacts_raw_verification_code_and_email_body() -> None:
+    body = '{"verification_code":"Your sign-in code is 246810. Do not share this code. Reply-To: otp@example.test"}'
+
+    preview = _response_body_preview(body)
+
+    assert "246810" not in preview
+    assert "otp@example.test" not in preview
+    assert "[REDACTED_OTP_BODY]" in preview
+    assert "json_like=true" in preview
+
+
+def test_contract_error_reason_redacts_body_preview() -> None:
+    reason = _totp_webhook_contract_error_reason(
+        status_code=200,
+        content_type="text/plain; charset=utf-8",
+        response_body="verification_code=135790 email text",
+    )
+
+    assert "https://otp.example.test/hook" not in reason
+    assert "135790" not in reason
+    assert "verification_code=135790" not in reason
+    assert "content_type=text/plain" in reason
+    assert "charset" not in reason
+    assert "[REDACTED_OTP_BODY]" in reason