fix(SKY-9178): prefer credential TOTP before webhook polling for OTP (#6451)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

skyvern/forge/agent.py

@@ -119,11 +119,7 @@ from skyvern.schemas.steps import AgentStepOutput
 from skyvern.services import run_service, service_utils
 from skyvern.services.action_service import get_action_history
 from skyvern.services.error_detection_service import detect_user_defined_errors_for_task
-from skyvern.services.otp_service import (
-    extract_totp_from_navigation_inputs,
-    poll_otp_value,
-    try_generate_totp_from_credential,
-)
+from skyvern.services.otp_service import poll_otp_value, resolve_otp_value
 from skyvern.services.webhook_delivery import WEBHOOK_DELIVERY_MAX_ATTEMPTS, deliver_webhook_with_retries
 from skyvern.utils.image_resizer import Resolution
 from skyvern.utils.prompt_engine import (
@@ -5895,25 +5891,7 @@ class ForgeAgent:
             return json_response
 
         LOG.info("Need verification code")
-        otp_value = extract_totp_from_navigation_inputs(task.navigation_payload)
-        if not otp_value and (task.totp_verification_url or task.totp_identifier) and task.organization_id:
-            workflow_id = workflow_permanent_id = None
-            if task.workflow_run_id:
-                workflow_run = await app.DATABASE.workflow_runs.get_workflow_run(task.workflow_run_id)
-                if workflow_run:
-                    workflow_id = workflow_run.workflow_id
-                    workflow_permanent_id = workflow_run.workflow_permanent_id
-            otp_value = await poll_otp_value(
-                organization_id=task.organization_id,
-                task_id=task.task_id,
-                workflow_id=workflow_id,
-                workflow_run_id=task.workflow_run_id,
-                workflow_permanent_id=workflow_permanent_id,
-                totp_verification_url=task.totp_verification_url,
-                totp_identifier=task.totp_identifier,
-            )
-        if not otp_value:
-            otp_value = try_generate_totp_from_credential(task.workflow_run_id)
+        otp_value = await resolve_otp_value(task)
 
         if not otp_value or otp_value.get_otp_type() != OTPType.TOTP:
             return json_response

skyvern/services/otp_service.py

@@ -9,6 +9,7 @@ import structlog
 from pydantic import BaseModel, Field
 
 if TYPE_CHECKING:
+    from skyvern.forge.sdk.schemas.tasks import Task
     from skyvern.forge.sdk.workflow.context_manager import WorkflowRunContext
 
 from skyvern.config import settings
@@ -274,6 +275,35 @@ def _try_generate_totp_for_credential(
         return None
 
 
+def has_credential_totp_candidate(workflow_run_id: str | None) -> bool:
+    """Return True when try_generate_totp_from_credential would have a credential to consult.
+
+    Mirrors try_generate_totp_from_credential's selection: active-with-TOTP if an
+    active credential is recorded, else exactly one TOTP-bearing candidate.
+    Used to drive prompt gating and classifier branches without actually
+    generating a code.
+    """
+    if not workflow_run_id:
+        return False
+
+    workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
+    if not workflow_run_context:
+        return False
+
+    current_context = skyvern_context.current()
+    active_credential_key = current_context.active_credential_parameter_key if current_context else None
+    if active_credential_key:
+        value = workflow_run_context.values.get(active_credential_key)
+        return isinstance(value, dict) and isinstance(value.get("totp"), str)
+
+    candidate_keys = [
+        key
+        for key, value in workflow_run_context.values.items()
+        if isinstance(value, dict) and isinstance(value.get("totp"), str)
+    ]
+    return len(candidate_keys) == 1
+
+
 def try_generate_totp_from_credential(workflow_run_id: str | None) -> OTPValue | None:
     """Generate a TOTP only for the credential the agent is currently typing into.
 
@@ -308,6 +338,45 @@ def try_generate_totp_from_credential(workflow_run_id: str | None) -> OTPValue |
     return _try_generate_totp_for_credential(workflow_run_context, candidate_keys[0], workflow_run_id)
 
 
+async def resolve_otp_value(task: "Task") -> OTPValue | None:
+    """Resolve the OTP value to use for a verification step.
+
+    Priority is payload -> credential-backed TOTP -> webhook polling. The
+    workflow-run metadata lookup needed by polling is performed lazily so
+    payload/credential resolutions do not touch the database. Polling raises
+    NoTOTPVerificationCodeFound or FailedToGetTOTPVerificationCode on timeout;
+    those propagate so callers can build the right terminate action. Returns
+    None when no source is configured.
+    """
+    otp_value = extract_totp_from_navigation_inputs(task.navigation_payload)
+    if otp_value:
+        return otp_value
+
+    otp_value = try_generate_totp_from_credential(task.workflow_run_id)
+    if otp_value:
+        return otp_value
+
+    if (task.totp_verification_url or task.totp_identifier) and task.organization_id:
+        workflow_id: str | None = None
+        workflow_permanent_id: str | None = None
+        if task.workflow_run_id:
+            workflow_run = await app.DATABASE.workflow_runs.get_workflow_run(task.workflow_run_id)
+            if workflow_run:
+                workflow_id = workflow_run.workflow_id
+                workflow_permanent_id = workflow_run.workflow_permanent_id
+        return await poll_otp_value(
+            organization_id=task.organization_id,
+            task_id=task.task_id,
+            workflow_id=workflow_id,
+            workflow_run_id=task.workflow_run_id,
+            workflow_permanent_id=workflow_permanent_id,
+            totp_verification_url=task.totp_verification_url,
+            totp_identifier=task.totp_identifier,
+        )
+
+    return None
+
+
 async def poll_otp_value(
     organization_id: str,
     task_id: str | None = None,

skyvern/webeye/actions/parse_actions.py

@@ -18,9 +18,9 @@ from skyvern.forge.sdk.schemas.tasks import Task
 from skyvern.forge.sdk.schemas.totp_codes import OTPType
 from skyvern.forge.sdk.trace import traced
 from skyvern.services.otp_service import (
-    extract_totp_from_navigation_inputs,
+    has_credential_totp_candidate,
     poll_otp_value,
-    try_generate_totp_from_credential,
+    resolve_otp_value,
 )
 from skyvern.utils.image_resizer import Resolution, scale_coordinates
 from skyvern.utils.url_validators import strip_query_params
@@ -57,29 +57,6 @@ from skyvern.webeye.scraper.scraped_page import ScrapedPage
 LOG = structlog.get_logger()
 
 
-def _has_credential_totp_candidate(workflow_run_id: str | None) -> bool:
-    # Mirrors try_generate_totp_from_credential selection: active-with-TOTP, or
-    # exactly one TOTP-bearing candidate when no active is set (multi-no-active -> False).
-    if not workflow_run_id:
-        return False
-    workflow_run_context = app.WORKFLOW_CONTEXT_MANAGER.get_workflow_run_context(workflow_run_id)
-    if not workflow_run_context:
-        return False
-
-    current_context = skyvern_context.current()
-    active_credential_key = current_context.active_credential_parameter_key if current_context else None
-    if active_credential_key:
-        value = workflow_run_context.values.get(active_credential_key)
-        return isinstance(value, dict) and isinstance(value.get("totp"), str)
-
-    candidate_keys = [
-        key
-        for key, value in workflow_run_context.values.items()
-        if isinstance(value, dict) and isinstance(value.get("totp"), str)
-    ]
-    return len(candidate_keys) == 1
-
-
 def parse_action(
     action: Dict[str, Any],
     scraped_page: ScrapedPage,
@@ -976,37 +953,16 @@ async def generate_cua_fallback_actions(
             )
 
     elif skyvern_action_type == "get_verification_code":
-        # 1. Check navigation payload first for inline OTP.
-        otp_value = extract_totp_from_navigation_inputs(task.navigation_payload)
-        # 2. Then try to generate TOTP from credential if payload has no OTP.
-        if not otp_value:
-            otp_value = try_generate_totp_from_credential(task.workflow_run_id)
-        # 3. Lastly, poll for OTP if organization has config and no OTP was found yet.
-        if not otp_value and (task.totp_verification_url or task.totp_identifier) and task.organization_id:
-            LOG.info(
-                "Getting verification code for CUA",
-                task_id=task.task_id,
-                organization_id=task.organization_id,
-                workflow_run_id=task.workflow_run_id,
-                totp_verification_url=strip_query_params(task.totp_verification_url)
-                if task.totp_verification_url
-                else None,
-                totp_identifier=task.totp_identifier,
-            )
-            try:
-                otp_value = await poll_otp_value(
-                    organization_id=task.organization_id,
-                    task_id=task.task_id,
-                    workflow_run_id=task.workflow_run_id,
-                    totp_verification_url=task.totp_verification_url,
-                    totp_identifier=task.totp_identifier,
-                )
-            except NoTOTPVerificationCodeFound:
-                reasoning_suffix = "No verification code found"
-                reasoning = f"{reasoning}. {reasoning_suffix}" if reasoning else reasoning_suffix
-            except FailedToGetTOTPVerificationCode as e:
-                reasoning_suffix = f"Failed to get verification code. Reason: {e.reason}"
-                reasoning = f"{reasoning}. {reasoning_suffix}" if reasoning else reasoning_suffix
+        try:
+            otp_value = await resolve_otp_value(task)
+        except NoTOTPVerificationCodeFound:
+            otp_value = None
+            reasoning_suffix = "No verification code found"
+            reasoning = f"{reasoning}. {reasoning_suffix}" if reasoning else reasoning_suffix
+        except FailedToGetTOTPVerificationCode as e:
+            otp_value = None
+            reasoning_suffix = f"Failed to get verification code. Reason: {e.reason}"
+            reasoning = f"{reasoning}. {reasoning_suffix}" if reasoning else reasoning_suffix
 
         # OTP value obtained - use it as verification code
         if otp_value and otp_value.get_otp_type() == OTPType.TOTP:
@@ -1021,7 +977,7 @@ async def generate_cua_fallback_actions(
             # Three-way classification: missing config, configured-but-empty (any of
             # URL / identifier / credential), or wrong-type otp_value. Each gets a
             # distinct customer-visible signal so webhook consumers can branch.
-            has_credential = _has_credential_totp_candidate(task.workflow_run_id)
+            has_credential = has_credential_totp_candidate(task.workflow_run_id)
             no_source_configured = (
                 otp_value is None and not task.totp_verification_url and not task.totp_identifier and not has_credential
             )

tests/unit/test_agent_otp_routing.py

@@ -0,0 +1,116 @@
+"""Regression tests for normal Agent OTP routing.
+
+Covers:
+- handle_potential_verification_code delegates to resolve_otp_value without a
+  pre-resolver DB roundtrip.
+- With a webhook-configured task, the resolver returns a credential-backed TOTP
+  before polling is attempted (the actual SKY-9178 customer scenario).
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from skyvern.forge.agent import ForgeAgent
+from skyvern.forge.sdk.core import skyvern_context
+from skyvern.forge.sdk.core.skyvern_context import SkyvernContext
+from skyvern.forge.sdk.schemas.totp_codes import OTPType
+from skyvern.services.otp_service import OTPValue
+
+
+def _make_task(
+    *,
+    totp_verification_url: str | None = "https://example.com/webhook",
+    totp_identifier: str | None = "user@example.com",
+    navigation_payload: object = None,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        task_id="tsk_test",
+        organization_id="o_test",
+        workflow_run_id="wr_test",
+        workflow_permanent_id="wpid_test",
+        totp_verification_url=totp_verification_url,
+        totp_identifier=totp_identifier,
+        navigation_payload=navigation_payload,
+        url="https://example.com",
+        navigation_goal="log in",
+        llm_key=None,
+        workflow_system_prompt=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_potential_verification_code_uses_resolver_without_db_lookup(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    task = _make_task(navigation_payload={"otp_code": "654321"})
+    step = MagicMock()
+    scraped_page = MagicMock()
+    browser_state = MagicMock()
+    json_response = {
+        "place_to_enter_verification_code": True,
+        "should_enter_verification_code": True,
+    }
+
+    resolver = AsyncMock(return_value=None)
+    db_get = AsyncMock()
+    monkeypatch.setattr("skyvern.forge.agent.resolve_otp_value", resolver)
+    monkeypatch.setattr("skyvern.forge.agent.app.DATABASE.workflow_runs.get_workflow_run", db_get)
+
+    agent = ForgeAgent.__new__(ForgeAgent)
+    await agent.handle_potential_verification_code(task, step, scraped_page, browser_state, json_response)
+
+    resolver.assert_awaited_once_with(task)
+    db_get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_handle_potential_verification_code_skips_polling_when_credential_returns_first(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """SKY-9178 regression: when both webhook URL and credential TOTP are configured,
+    the resolver yields the credential code first and webhook polling is never invoked."""
+    task = _make_task()
+    step = MagicMock()
+    scraped_page = MagicMock()
+    browser_state = MagicMock()
+    json_response = {
+        "place_to_enter_verification_code": True,
+        "should_enter_verification_code": True,
+    }
+
+    credential_code = OTPValue(value="123456", type=OTPType.TOTP)
+    resolver = AsyncMock(return_value=credential_code)
+    poll = AsyncMock()
+    db_get = AsyncMock()
+    monkeypatch.setattr("skyvern.forge.agent.resolve_otp_value", resolver)
+    monkeypatch.setattr("skyvern.forge.agent.poll_otp_value", poll)
+    monkeypatch.setattr("skyvern.forge.agent.app.DATABASE.workflow_runs.get_workflow_run", db_get)
+
+    rebuilt = AsyncMock(return_value=("prompt", False, "prompt_name"))
+    monkeypatch.setattr(ForgeAgent, "_build_extract_action_prompt", rebuilt)
+    monkeypatch.setattr("skyvern.forge.agent.service_utils.is_cua_task", AsyncMock(return_value=False))
+
+    rescrape = AsyncMock(return_value={"actions": []})
+    monkeypatch.setattr(
+        "skyvern.forge.agent.LLMAPIHandlerFactory.get_override_llm_api_handler",
+        lambda *args, **kwargs: rescrape,
+    )
+
+    agent = ForgeAgent.__new__(ForgeAgent)
+    agent.async_operation_pool = MagicMock()
+
+    skyvern_context.set(SkyvernContext(task_id=task.task_id))
+    try:
+        result = await agent.handle_potential_verification_code(task, step, scraped_page, browser_state, json_response)
+    finally:
+        skyvern_context.reset()
+
+    resolver.assert_awaited_once_with(task)
+    poll.assert_not_awaited()
+    db_get.assert_not_awaited()
+    rescrape.assert_awaited_once()
+    assert result == {"actions": []}

tests/unit/test_missing_totp_source.py

@@ -50,11 +50,11 @@ def _patch_cua_fallback_common(monkeypatch: pytest.MonkeyPatch, action_type: str
 
     monkeypatch.setattr("skyvern.webeye.actions.parse_actions.app.LLM_API_HANDLER", AsyncMock(side_effect=_fake_llm))
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions.extract_totp_from_navigation_inputs",
+        "skyvern.services.otp_service.extract_totp_from_navigation_inputs",
         lambda *_: None,
     )
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions.try_generate_totp_from_credential",
+        "skyvern.services.otp_service.try_generate_totp_from_credential",
         lambda *_: None,
     )
 
@@ -68,7 +68,7 @@ async def test_get_verification_code_with_no_source_emits_missing_totp_source(mo
     step = _make_step()
     _patch_cua_fallback_common(monkeypatch, "get_verification_code")
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions._has_credential_totp_candidate",
+        "skyvern.webeye.actions.parse_actions.has_credential_totp_candidate",
         lambda *_: False,
     )
 
@@ -93,14 +93,14 @@ async def test_get_verification_code_with_url_configured_emits_configured_but_em
     step = _make_step()
     _patch_cua_fallback_common(monkeypatch, "get_verification_code")
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions._has_credential_totp_candidate",
+        "skyvern.webeye.actions.parse_actions.has_credential_totp_candidate",
         lambda *_: False,
     )
 
     async def _no_poll(*args, **kwargs):  # type: ignore[no-untyped-def]
         return None
 
-    monkeypatch.setattr("skyvern.webeye.actions.parse_actions.poll_otp_value", _no_poll)
+    monkeypatch.setattr("skyvern.services.otp_service.poll_otp_value", _no_poll)
 
     actions = await generate_cua_fallback_actions(task, step, assistant_message=None, reasoning=None)
 
@@ -127,7 +127,7 @@ async def test_get_verification_code_with_only_credential_configured_but_failing
     step = _make_step()
     _patch_cua_fallback_common(monkeypatch, "get_verification_code")
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions._has_credential_totp_candidate",
+        "skyvern.webeye.actions.parse_actions.has_credential_totp_candidate",
         lambda *_: True,
     )
 
@@ -155,7 +155,7 @@ async def test_get_verification_code_with_credential_totp_returns_verification(
     fake_otp_value.get_otp_type.return_value = OTPType.TOTP
 
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions.try_generate_totp_from_credential",
+        "skyvern.services.otp_service.try_generate_totp_from_credential",
         lambda *_: fake_otp_value,
     )
 
@@ -179,7 +179,7 @@ async def test_get_verification_code_with_multiple_unselected_credentials_emits_
     step = _make_step()
     _patch_cua_fallback_common(monkeypatch, "get_verification_code")
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions._has_credential_totp_candidate",
+        "skyvern.webeye.actions.parse_actions.has_credential_totp_candidate",
         lambda *_: False,  # mirroring multi-no-active selection result
     )
 
@@ -223,7 +223,7 @@ async def test_get_magic_link_with_credential_candidate_still_emits_missing_sour
     # Even if _has_credential_totp_candidate would return True, the magic-link
     # branch must NOT consult it. Patch to True to prove the branch ignores it.
     monkeypatch.setattr(
-        "skyvern.webeye.actions.parse_actions._has_credential_totp_candidate",
+        "skyvern.webeye.actions.parse_actions.has_credential_totp_candidate",
         lambda *_: True,
     )
 

tests/unit/test_resolve_otp_value.py

@@ -0,0 +1,286 @@
+"""Tests for skyvern.services.otp_service.resolve_otp_value.
+
+Verifies the unified OTP source priority used by both the normal Agent flow
+and the CUA flow: navigation payload -> credential-backed TOTP -> webhook
+polling. Polling exceptions surface unchanged so callers can react.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from skyvern.exceptions import FailedToGetTOTPVerificationCode, NoTOTPVerificationCodeFound
+from skyvern.forge.sdk.schemas.totp_codes import OTPType
+from skyvern.services.otp_service import OTPValue, resolve_otp_value
+
+
+def _make_task(
+    *,
+    task_id: str = "tsk_test",
+    workflow_run_id: str | None = "wr_test",
+    organization_id: str | None = "o_test",
+    totp_verification_url: str | None = None,
+    totp_identifier: str | None = None,
+    navigation_payload: object = None,
+) -> SimpleNamespace:
+    return SimpleNamespace(
+        task_id=task_id,
+        workflow_run_id=workflow_run_id,
+        organization_id=organization_id,
+        totp_verification_url=totp_verification_url,
+        totp_identifier=totp_identifier,
+        navigation_payload=navigation_payload,
+    )
+
+
+def _otp_value(code: str = "123456") -> OTPValue:
+    return OTPValue(value=code, type=OTPType.TOTP)
+
+
+@pytest.mark.asyncio
+async def test_payload_otp_returns_immediately_skipping_credential_and_poll() -> None:
+    payload_value = _otp_value("999000")
+    task = _make_task(
+        navigation_payload={"otp_code": "999000"},
+        totp_verification_url="https://example.com/webhook",
+    )
+    with (
+        patch(
+            "skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=payload_value
+        ) as payload,
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential") as credential,
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is payload_value
+    payload.assert_called_once_with(task.navigation_payload)
+    credential.assert_not_called()
+    poll.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_credential_returns_value_skipping_poll_even_when_url_configured() -> None:
+    """Key SKY-9178 fix: credential TOTP wins over webhook polling when both are configured."""
+    credential_value = _otp_value("424242")
+    task = _make_task(
+        totp_verification_url="https://example.com/webhook",
+        totp_identifier="user@example.com",
+    )
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch(
+            "skyvern.services.otp_service.try_generate_totp_from_credential",
+            return_value=credential_value,
+        ) as credential,
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is credential_value
+    credential.assert_called_once_with(task.workflow_run_id)
+    poll.assert_not_called()
+
+
+def _stub_workflow_run_lookup(
+    monkeypatch_target, *, workflow_id: str | None, workflow_permanent_id: str | None
+) -> MagicMock:
+    """Patch app.DATABASE.workflow_runs.get_workflow_run with a recording mock."""
+    workflow_run = SimpleNamespace(workflow_id=workflow_id, workflow_permanent_id=workflow_permanent_id)
+    get = AsyncMock(return_value=workflow_run)
+    monkeypatch_target.setattr(
+        "skyvern.services.otp_service.app.DATABASE.workflow_runs.get_workflow_run",
+        get,
+    )
+    return get
+
+
+@pytest.mark.asyncio
+async def test_falls_through_to_poll_when_no_payload_or_credential(monkeypatch: pytest.MonkeyPatch) -> None:
+    poll_value = _otp_value("303030")
+    task = _make_task(
+        totp_verification_url="https://example.com/webhook",
+        totp_identifier="user@example.com",
+    )
+    db_get = _stub_workflow_run_lookup(monkeypatch, workflow_id="w_test", workflow_permanent_id="wpid_test")
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock(return_value=poll_value)) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is poll_value
+    db_get.assert_awaited_once_with("wr_test")
+    poll.assert_awaited_once()
+    kwargs = poll.await_args.kwargs
+    assert kwargs["organization_id"] == "o_test"
+    assert kwargs["task_id"] == "tsk_test"
+    assert kwargs["workflow_id"] == "w_test"
+    assert kwargs["workflow_run_id"] == "wr_test"
+    assert kwargs["workflow_permanent_id"] == "wpid_test"
+    assert kwargs["totp_verification_url"] == "https://example.com/webhook"
+    assert kwargs["totp_identifier"] == "user@example.com"
+
+
+@pytest.mark.asyncio
+async def test_workflow_run_lookup_skipped_when_payload_resolves(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Codex must-fix: payload-resolved OTP must not pay a DB roundtrip."""
+    payload_value = _otp_value("111111")
+    task = _make_task(
+        navigation_payload={"otp_code": "111111"},
+        totp_verification_url="https://example.com/webhook",
+    )
+    db_get = _stub_workflow_run_lookup(monkeypatch, workflow_id="w_test", workflow_permanent_id="wpid_test")
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=payload_value),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential") as credential,
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is payload_value
+    db_get.assert_not_awaited()
+    credential.assert_not_called()
+    poll.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_workflow_run_lookup_skipped_when_credential_resolves(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Codex must-fix: credential-resolved OTP must not pay a DB roundtrip."""
+    credential_value = _otp_value("222222")
+    task = _make_task(
+        totp_verification_url="https://example.com/webhook",
+        totp_identifier="user@example.com",
+    )
+    db_get = _stub_workflow_run_lookup(monkeypatch, workflow_id="w_test", workflow_permanent_id="wpid_test")
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=credential_value),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is credential_value
+    db_get.assert_not_awaited()
+    poll.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_polling_handles_missing_workflow_run_lookup(monkeypatch: pytest.MonkeyPatch) -> None:
+    """If DB returns None for the workflow_run, polling still proceeds with None metadata."""
+    poll_value = _otp_value("999")
+    task = _make_task(totp_verification_url="https://example.com/webhook")
+    get = AsyncMock(return_value=None)
+    monkeypatch.setattr("skyvern.services.otp_service.app.DATABASE.workflow_runs.get_workflow_run", get)
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock(return_value=poll_value)) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is poll_value
+    kwargs = poll.await_args.kwargs
+    assert kwargs["workflow_id"] is None
+    assert kwargs["workflow_permanent_id"] is None
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_no_source_configured_at_all() -> None:
+    task = _make_task()
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is None
+    poll.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_polling_skipped_when_organization_id_missing() -> None:
+    task = _make_task(organization_id=None, totp_verification_url="https://example.com/webhook")
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock()) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is None
+    poll.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_no_totp_found_exception_propagates_from_poll(monkeypatch: pytest.MonkeyPatch) -> None:
+    task = _make_task(totp_verification_url="https://example.com/webhook")
+    _stub_workflow_run_lookup(monkeypatch, workflow_id=None, workflow_permanent_id=None)
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch(
+            "skyvern.services.otp_service.poll_otp_value",
+            new=AsyncMock(side_effect=NoTOTPVerificationCodeFound(task_id=task.task_id)),
+        ),
+    ):
+        with pytest.raises(NoTOTPVerificationCodeFound):
+            await resolve_otp_value(task)
+
+
+@pytest.mark.asyncio
+async def test_failed_to_get_totp_exception_propagates_from_poll(monkeypatch: pytest.MonkeyPatch) -> None:
+    task = _make_task(totp_verification_url="https://example.com/webhook")
+    _stub_workflow_run_lookup(monkeypatch, workflow_id=None, workflow_permanent_id=None)
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch(
+            "skyvern.services.otp_service.poll_otp_value",
+            new=AsyncMock(side_effect=FailedToGetTOTPVerificationCode(reason="bad body")),
+        ),
+    ):
+        with pytest.raises(FailedToGetTOTPVerificationCode):
+            await resolve_otp_value(task)
+
+
+@pytest.mark.asyncio
+async def test_identifier_only_routes_through_poll(monkeypatch: pytest.MonkeyPatch) -> None:
+    poll_value = _otp_value("777")
+    task = _make_task(totp_identifier="user@example.com")
+    _stub_workflow_run_lookup(monkeypatch, workflow_id=None, workflow_permanent_id=None)
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None),
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock(return_value=poll_value)) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is poll_value
+    kwargs = poll.await_args.kwargs
+    assert kwargs["totp_verification_url"] is None
+    assert kwargs["totp_identifier"] == "user@example.com"
+
+
+@pytest.mark.asyncio
+async def test_credential_check_runs_even_when_workflow_run_id_missing(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Helper still attempts credential lookup; DB lookup is skipped when no workflow_run_id."""
+    task = _make_task(workflow_run_id=None, totp_verification_url="https://example.com/webhook")
+    db_get = AsyncMock()
+    monkeypatch.setattr("skyvern.services.otp_service.app.DATABASE.workflow_runs.get_workflow_run", db_get)
+    with (
+        patch("skyvern.services.otp_service.extract_totp_from_navigation_inputs", return_value=None),
+        patch("skyvern.services.otp_service.try_generate_totp_from_credential", return_value=None) as credential,
+        patch("skyvern.services.otp_service.poll_otp_value", new=AsyncMock(return_value=None)) as poll,
+    ):
+        result = await resolve_otp_value(task)
+
+    assert result is None
+    credential.assert_called_once_with(None)
+    db_get.assert_not_awaited()
+    poll.assert_awaited_once()