fix(SKY-11948): consume output-contract grants at dispatch (#7188)

skyvern/config.py

@@ -205,6 +205,7 @@ class Settings(BaseSettings):
     # Experimental Workflow Copilot v2 branch mode.
     # Off = standard block authoring. On = prefer code blocks for browser work.
     WORKFLOW_COPILOT_CODE_BLOCK_MODE: bool = False
+    WORKFLOW_COPILOT_AUTHOR_TIME_GATE_LOG_ONLY: bool = False
     # Default code_only for MCP block/workflow tools. Off = permissive.
     MCP_CODE_ONLY_MODE: bool = False
     # Default for the bounded code-block self-heal; off by default.

skyvern/forge/sdk/api/files.py

@@ -6,6 +6,7 @@ import re
 import shutil
 import tempfile
 import zipfile
+from http import HTTPStatus
 from pathlib import Path
 from typing import TYPE_CHECKING
 from urllib.parse import parse_qsl, unquote, urlparse
@@ -17,9 +18,20 @@ from yarl import URL
 
 from skyvern.config import settings
 from skyvern.constants import BROWSER_DOWNLOAD_TIMEOUT, BROWSER_DOWNLOADING_SUFFIX, REPO_ROOT_DIR
-from skyvern.exceptions import DownloadFileMaxSizeExceeded, DownloadFileMaxWaitingTime
+from skyvern.exceptions import DownloadFileMaxSizeExceeded, DownloadFileMaxWaitingTime, SkyvernHTTPException
 from skyvern.forge import app
-from skyvern.utils.url_validators import encode_url
+from skyvern.forge.sdk.core.aiohttp_helper import (
+    SSRFGuardedResolver,
+    ssrf_guarded_tcp_connector,
+    strip_cross_origin_redirect_credentials,
+    validate_and_pin_fetch_url,
+    validate_and_pin_redirect_url,
+)
+from skyvern.utils.url_validators import (
+    MAX_SAFE_REDIRECTS,
+    SAFE_REDIRECT_STATUS_CODES,
+    encode_url,
+)
 
 if TYPE_CHECKING:
     from skyvern.forge.sdk.core.skyvern_context import SkyvernContext
@@ -209,70 +221,93 @@ async def download_file(
                 LOG.info("Downloading file from local file system", url=url)
                 return local_path
 
-        async with aiohttp.ClientSession(raise_for_status=True) as session:
+        resolver = SSRFGuardedResolver()
+        current_url = await validate_and_pin_fetch_url(url, resolver)
+        request_headers = dict(headers or {})
+        async with aiohttp.ClientSession(
+            raise_for_status=True, connector=ssrf_guarded_tcp_connector(resolver)
+        ) as session:
             LOG.info("Starting to download file", url=url)
-            encoded_url = encode_url(url)
-            async with session.get(URL(encoded_url, encoded=True), headers=headers) as response:
-                # Check the content length if available
-                if max_size_mb and response.content_length and response.content_length > max_size_mb * 1024 * 1024:
-                    # todo: move to root exception.py
-                    raise DownloadFileMaxSizeExceeded(max_size_mb)
+            for _ in range(MAX_SAFE_REDIRECTS + 1):
+                encoded_url = encode_url(current_url)
+                async with session.get(
+                    URL(encoded_url, encoded=True), headers=request_headers, allow_redirects=False
+                ) as response:
+                    if response.status in SAFE_REDIRECT_STATUS_CODES and response.headers.get("Location"):
+                        next_url = await validate_and_pin_redirect_url(
+                            current_url, response.headers["Location"], resolver
+                        )
+                        request_headers, _ = strip_cross_origin_redirect_credentials(
+                            request_headers, None, current_url, next_url
+                        )
+                        current_url = next_url
+                        continue
 
-                # Get the file name
-                if output_dir:
-                    download_dir_path = Path(output_dir)
-                    download_dir_path.mkdir(parents=True, exist_ok=True)
-                else:
-                    download_dir_path = Path(make_temp_directory(prefix="skyvern_downloads_"))
+                    # Check the content length if available
+                    if max_size_mb and response.content_length and response.content_length > max_size_mb * 1024 * 1024:
+                        # todo: move to root exception.py
+                        raise DownloadFileMaxSizeExceeded(max_size_mb)
 
-                download_dir_resolved = download_dir_path.resolve()
-                # response.headers stays a CIMultiDictProxy: dict() would make header
-                # lookups case-sensitive and miss lowercase wire headers.
-                file_name = _determine_download_filename(filename, response.headers, url)
-                allowed_dir = os.path.realpath(download_dir_resolved)
-                resolved_final_path = os.path.realpath(os.path.join(allowed_dir, file_name))
-                # sanitize_filename strips separators but keeps dots, so a dots-only name can
-                # still resolve outside the download dir; require a direct child. Checked
-                # before streaming so a rejected name downloads zero bytes.
-                if (
-                    resolved_final_path == allowed_dir
-                    or not resolved_final_path.startswith(allowed_dir + os.sep)
-                    or os.path.dirname(resolved_final_path) != allowed_dir
-                ):
-                    raise ValueError(f"Unsafe filename derived from download: {file_name!r}")
-                final_path = Path(resolved_final_path)
+                    # Get the file name
+                    if output_dir:
+                        download_dir_path = Path(output_dir)
+                        download_dir_path.mkdir(parents=True, exist_ok=True)
+                    else:
+                        download_dir_path = Path(make_temp_directory(prefix="skyvern_downloads_"))
 
-                temp_file = tempfile.NamedTemporaryFile(mode="wb", dir=download_dir_resolved, delete=False)
-                file_path = Path(temp_file.name).resolve()
-                if file_path != download_dir_resolved and not file_path.is_relative_to(download_dir_resolved):
-                    temp_file.close()
-                    raise ValueError("Unsafe temporary file path created for download")
+                    download_dir_resolved = download_dir_path.resolve()
+                    # response.headers stays a CIMultiDictProxy: dict() would make header
+                    # lookups case-sensitive and miss lowercase wire headers.
+                    file_name = _determine_download_filename(filename, response.headers, url)
+                    allowed_dir = os.path.realpath(download_dir_resolved)
+                    resolved_final_path = os.path.realpath(os.path.join(allowed_dir, file_name))
+                    # sanitize_filename strips separators but keeps dots, so a dots-only name can
+                    # still resolve outside the download dir; require a direct child. Checked
+                    # before streaming so a rejected name downloads zero bytes.
+                    if (
+                        resolved_final_path == allowed_dir
+                        or not resolved_final_path.startswith(allowed_dir + os.sep)
+                        or os.path.dirname(resolved_final_path) != allowed_dir
+                    ):
+                        raise ValueError(f"Unsafe filename derived from download: {file_name!r}")
+                    final_path = Path(resolved_final_path)
 
-                LOG.info("Downloading file to temporary path", file_path=str(file_path))
-                try:
-                    with temp_file as f:
-                        # Write the content of the request into the file
-                        total_bytes_downloaded = 0
-                        async for chunk in response.content.iter_chunked(1024):
-                            f.write(chunk)
-                            total_bytes_downloaded += len(chunk)
-                            if max_size_mb and total_bytes_downloaded > max_size_mb * 1024 * 1024:
-                                raise DownloadFileMaxSizeExceeded(max_size_mb)
+                    temp_file = tempfile.NamedTemporaryFile(mode="wb", dir=download_dir_resolved, delete=False)
+                    file_path = Path(temp_file.name).resolve()
+                    if file_path != download_dir_resolved and not file_path.is_relative_to(download_dir_resolved):
+                        temp_file.close()
+                        raise ValueError("Unsafe temporary file path created for download")
 
-                    file_path.replace(final_path)
-                except BaseException:
-                    # An orphaned temp file in a run download dir would get synced to storage
-                    # under the tmpXXXX name; drop it on any failure (incl. cancellation).
-                    file_path.unlink(missing_ok=True)
-                    raise
+                    LOG.info("Downloading file to temporary path", file_path=str(file_path))
+                    try:
+                        with temp_file as f:
+                            # Write the content of the request into the file
+                            total_bytes_downloaded = 0
+                            async for chunk in response.content.iter_chunked(1024):
+                                f.write(chunk)
+                                total_bytes_downloaded += len(chunk)
+                                if max_size_mb and total_bytes_downloaded > max_size_mb * 1024 * 1024:
+                                    raise DownloadFileMaxSizeExceeded(max_size_mb)
 
-                LOG.info(f"File downloaded successfully to {final_path}")
-                return str(final_path)
+                        file_path.replace(final_path)
+                    except BaseException:
+                        # An orphaned temp file in a run download dir would get synced to storage
+                        # under the tmpXXXX name; drop it on any failure (incl. cancellation).
+                        file_path.unlink(missing_ok=True)
+                        raise
+
+                    LOG.info(f"File downloaded successfully to {final_path}")
+                    return str(final_path)
+            raise SkyvernHTTPException(
+                message=f"Too many redirects while downloading file: {current_url}",
+                status_code=HTTPStatus.BAD_REQUEST,
+            )
     except aiohttp.ClientResponseError as e:
-        LOG.exception(f"Failed to download file, status code: {e.status}")
+        # Re-raised and handled at the action/block boundary; server rejections are external.
+        LOG.warning(f"Failed to download file, status code: {e.status}", exc_info=True)
         raise
     except DownloadFileMaxSizeExceeded as e:
-        LOG.exception(f"Failed to download file, max size exceeded: {e.max_size}")
+        LOG.warning(f"Failed to download file, max size exceeded: {e.max_size}", exc_info=True)
         raise
     except PermissionError as e:
         LOG.warning(
@@ -282,6 +317,10 @@ async def download_file(
             reason=str(e),
         )
         raise
+    except aiohttp.InvalidURL:
+        # Malformed customer-provided URL - a client-data error, not a platform fault.
+        LOG.warning("Failed to download file, invalid URL", exc_info=True)
+        raise
     except Exception:
         LOG.exception("Failed to download file")
         raise

skyvern/forge/sdk/copilot/build_test_outcome.py

@@ -539,8 +539,7 @@ def _grounding_payload_from_evidence(
     challenge_gated = _challenge_gated_page_evidence(evidence)
     capture_degraded = not has_bounded_page_schema(evidence)
     observed_empty_page = _observed_empty_page_evidence(evidence)
-    if (challenge_gated or capture_degraded or observed_empty_page) and not (isinstance(run_id, str) and run_id):
-        return None
+    # No-run degraded captures are typed grounding evidence; downstream binding remains a separate gate.
     diagnostic_reason: Literal["none", "empty_page", "challenge_gated", "capture_degraded"] = "none"
     if challenge_gated:
         diagnostic_reason = "challenge_gated"

skyvern/forge/sdk/copilot/enforcement.py

@@ -102,6 +102,7 @@ from skyvern.forge.sdk.copilot.run_outcome import (
     RecordedRunOutcome,
     run_outcome_display_reason,
 )
+from skyvern.forge.sdk.copilot.runtime import AuthorTimeGateAblationPayload, record_author_time_gate_ablation_event
 from skyvern.forge.sdk.copilot.screenshot_utils import ScreenshotEntry
 from skyvern.forge.sdk.copilot.terminal_predicates import (
     artifact_health_blocked,
@@ -2384,7 +2385,7 @@ def _uncovered_output_reject_rescout_key(canonical_paths: set[str], structural_f
 def _active_uncovered_output_reject_paths(ctx: AgentContext) -> set[str]:
     canonical = {
         _canonical_output_path(path)
-        for path in author_time_reject_missing_output_paths(ctx.latest_recorded_build_test_outcome)
+        for path in author_time_reject_missing_output_paths(getattr(ctx, "latest_recorded_build_test_outcome", None))
     }
     return canonical & uncovered_requested_output_paths(ctx) if canonical else set()
 
@@ -2425,6 +2426,19 @@ def uncovered_output_reject_scout_steer_signal(ctx: AgentContext, tool_name: str
     key = _uncovered_output_reject_rescout_key(active, latest.structural_failure_identity)
     if ctx.uncovered_output_rescout_steer_key == key:
         return None
+    payload: AuthorTimeGateAblationPayload = {
+        "uncovered_output_paths": sorted(active),
+        "structural_failure_identity": latest.structural_failure_identity,
+    }
+    if record_author_time_gate_ablation_event(
+        ctx,
+        gate_id="uncovered_output_rescout_steer",
+        reason_code=UNCOVERED_OUTPUT_RESCOUT_STEER_REASON_CODE,
+        fingerprint=key,
+        blocked_tool=tool_name,
+        payload=payload,
+    ):
+        return None
     ctx.uncovered_output_rescout_steer_key = key
     named_paths = ", ".join(sorted(active))
     return CopilotToolBlockerSignal(

skyvern/forge/sdk/copilot/runtime.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import asyncio
 import inspect
+from collections.abc import Sequence
 from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, AsyncIterator, Awaitable, NotRequired, TypeAlias, TypedDict, cast
@@ -65,6 +66,16 @@ CodeArtifactMetadataValue: TypeAlias = (
     str | int | float | bool | None | list["CodeArtifactMetadataValue"] | dict[str, "CodeArtifactMetadataValue"]
 )
 CodeArtifactMetadataPayload: TypeAlias = dict[str, CodeArtifactMetadataValue]
+AuthorTimeGateAblationPayloadValue: TypeAlias = (
+    str
+    | int
+    | float
+    | bool
+    | None
+    | Sequence["AuthorTimeGateAblationPayloadValue"]
+    | dict[str, "AuthorTimeGateAblationPayloadValue"]
+)
+AuthorTimeGateAblationPayload: TypeAlias = dict[str, AuthorTimeGateAblationPayloadValue]
 SdkActionWorkflowRunCacheKey: TypeAlias = tuple[str, str]
 
 
@@ -144,6 +155,16 @@ class RegisteredArtifactEvidence:
     workflow_run_id: str
 
 
+@dataclass(frozen=True)
+class AuthorTimeGateAblationEvent:
+    gate_id: str
+    reason_code: str
+    fingerprint: str
+    log_only: bool
+    blocked_tool: str | None = None
+    payload: AuthorTimeGateAblationPayload = field(default_factory=dict)
+
+
 class ScoutedInteraction(TypedDict):
     tool_name: str
     selector: NotRequired[str]
@@ -423,6 +444,43 @@ class AgentContext:
     # extraction_schema declares fields that map to no output the block produces.
     # Surfaced into the persisted TurnOutcome so a later turn can report it.
     latest_schema_incompatibility: SchemaIncompatibility | None = None
+    author_time_gate_ablation_events: list[AuthorTimeGateAblationEvent] = field(default_factory=list)
+
+
+def copilot_author_time_gate_log_only_enabled() -> bool:
+    return not settings.is_cloud_environment() and settings.WORKFLOW_COPILOT_AUTHOR_TIME_GATE_LOG_ONLY
+
+
+def record_author_time_gate_ablation_event(
+    ctx: AgentContext,
+    *,
+    gate_id: str,
+    reason_code: str,
+    fingerprint: str,
+    blocked_tool: str | None = None,
+    payload: AuthorTimeGateAblationPayload | None = None,
+) -> bool:
+    if not copilot_author_time_gate_log_only_enabled():
+        return False
+    event = AuthorTimeGateAblationEvent(
+        gate_id=gate_id,
+        reason_code=reason_code,
+        fingerprint=fingerprint,
+        blocked_tool=blocked_tool,
+        payload=dict(payload or {}),
+        log_only=True,
+    )
+    ctx.author_time_gate_ablation_events.append(event)
+    LOG.info(
+        "copilot_author_time_gate_ablation_event",
+        gate_id=event.gate_id,
+        reason_code=event.reason_code,
+        fingerprint=event.fingerprint,
+        blocked_tool=event.blocked_tool,
+        log_only=event.log_only,
+        payload=event.payload,
+    )
+    return True
 
 
 def output_contract_ladder_unresolved(ctx: AgentContext) -> bool:

skyvern/forge/sdk/copilot/tools/__init__.py

@@ -253,7 +253,7 @@ from .workflow_update import _record_workflow_update_result as _record_workflow_
 from .workflow_update import _scaffold_metadata_contract_for_update as _scaffold_metadata_contract_for_update
 from .workflow_update import _update_workflow as _update_workflow
 from .workflow_update import (
-    consume_output_contract_advisory_grant_for_run as consume_output_contract_advisory_grant_for_run,
+    consume_output_contract_advisory_grant_for_run_result as consume_output_contract_advisory_grant_for_run_result,
 )
 
 LOG = structlog.get_logger()
@@ -468,8 +468,6 @@ async def run_blocks_tool(
         )
         return json.dumps(result)
 
-    consume_output_contract_advisory_grant_for_run(copilot_ctx)
-
     with copilot_span(
         "run_blocks",
         data=_run_blocks_span_data(
@@ -487,6 +485,7 @@ async def run_blocks_tool(
             block_outputs_to_seed=block_outputs_to_seed,
             frontier_start_label=frontier_start_label,
         )
+        consume_output_contract_advisory_grant_for_run_result(copilot_ctx, result)
         completion_verification = await _verify_and_record_run_blocks_result(copilot_ctx, result, handler_start)
         tool_visible_result = _tool_visible_result_after_completion_verification(
             copilot_ctx,
@@ -768,8 +767,6 @@ async def update_and_run_blocks_tool(
         )
         return json.dumps(result)
 
-    consume_output_contract_advisory_grant_for_run(copilot_ctx)
-
     with copilot_span(
         "run_blocks",
         data=_run_blocks_span_data(
@@ -787,6 +784,7 @@ async def update_and_run_blocks_tool(
             block_outputs_to_seed=block_outputs_to_seed,
             frontier_start_label=frontier_start_label,
         )
+        consume_output_contract_advisory_grant_for_run_result(copilot_ctx, run_result)
         completion_verification = await _verify_and_record_run_blocks_result(copilot_ctx, run_result, handler_start)
         tool_visible_result = _tool_visible_result_after_completion_verification(
             copilot_ctx,

skyvern/forge/sdk/copilot/tools/blockers.py

@@ -39,7 +39,12 @@ from skyvern.forge.sdk.copilot.failure_tracking import (
 from skyvern.forge.sdk.copilot.loop_detection import detect_failed_tool_step_loop_for_ctx, detect_tool_loop
 from skyvern.forge.sdk.copilot.reached_download_target import REGISTERED_DOWNLOAD_OUTPUT_KEYS
 from skyvern.forge.sdk.copilot.run_outcome import trusted_terminal_challenge_category_name
-from skyvern.forge.sdk.copilot.runtime import AgentContext, output_contract_ladder_unresolved
+from skyvern.forge.sdk.copilot.runtime import (
+    AgentContext,
+    AuthorTimeGateAblationPayload,
+    output_contract_ladder_unresolved,
+    record_author_time_gate_ablation_event,
+)
 from skyvern.forge.sdk.workflow.models.workflow import WorkflowRun, WorkflowRunStatus
 from skyvern.schemas.workflows import BlockType
 
@@ -1663,6 +1668,26 @@ def _recorded_outcome_grounding_signal(ctx: AgentContext, tool_name: str) -> Cop
         return None
     if not recorded_outcome_grounding_requires_current_page(ctx):
         return None
+    requirement = ctx.recorded_outcome_grounding_requirement
+    if requirement is None:
+        return None
+    payload: AuthorTimeGateAblationPayload = {
+        "phase": requirement.phase,
+        "outcome_reason_code": requirement.reason_code,
+        "workflow_run_id": requirement.workflow_run_id,
+        "block_labels": list(requirement.block_labels),
+        "required_tool": requirement.required_tool,
+        "required_target_url": requirement.required_target_url,
+    }
+    if record_author_time_gate_ablation_event(
+        ctx,
+        gate_id="recorded_outcome_grounding",
+        reason_code="recorded_outcome_grounding_required",
+        fingerprint=requirement.structural_key,
+        blocked_tool=tool_name,
+        payload=payload,
+    ):
+        return None
     return CopilotToolBlockerSignal(
         blocker_kind="missing_required_context",
         agent_steering_text=(

skyvern/forge/sdk/copilot/tools/workflow_update.py

@@ -113,7 +113,14 @@ from skyvern.forge.sdk.copilot.request_policy import (
     RequestedOutputEvidenceSource,
     _coerce_requested_output_evidence_source,
 )
-from skyvern.forge.sdk.copilot.runtime import AgentContext, ScoutedInteraction, output_contract_ladder_unresolved
+from skyvern.forge.sdk.copilot.runtime import (
+    AgentContext,
+    AuthorTimeGateAblationPayload,
+    ScoutedInteraction,
+    copilot_author_time_gate_log_only_enabled,
+    output_contract_ladder_unresolved,
+    record_author_time_gate_ablation_event,
+)
 from skyvern.forge.sdk.copilot.schema_incompatibility import (
     SCHEMA_INCOMPATIBILITY_FAILURE_TYPE,
     SchemaIncompatibility,
@@ -1559,7 +1566,7 @@ def _recorded_outcome_convergence_reject(
     workflow_yaml: str,
     code_artifact_metadata: object,
 ) -> _ConvergenceReject | None:
-    latest = ctx.latest_recorded_build_test_outcome
+    latest = getattr(ctx, "latest_recorded_build_test_outcome", None)
     if not isinstance(latest, RecordedBuildTestOutcome) or not latest.is_authoritative:
         return None
     candidate_signature = authored_structure_signature_from_workflow(workflow_yaml, code_artifact_metadata)
@@ -1775,6 +1782,8 @@ _MAX_OUTPUT_CONTRACT_REJECTS = 4
 _MAX_OUTPUT_CONTRACT_DEFERRALS = 3
 _MAX_OUTPUT_CONTRACT_ACTUATIONS_WITHOUT_RUN = 3
 _OUTPUT_CONTRACT_PAGE_EXTRACT_VAR = "output_contract_page_values"
+_OUTPUT_CONTRACT_ABLATION_GATE_ID = "output_contract_actuation"
+_METADATA_PREFLIGHT_ABLATION_GATE_ID = "metadata_run_preflight_reject"
 
 
 @dataclass(frozen=True)
@@ -1803,6 +1812,37 @@ class _OutputContractEvaluation:
         )
 
 
+def _record_output_contract_ablation_event(
+    ctx: AgentContext,
+    evaluation: _OutputContractEvaluation,
+    *,
+    gate_id: str,
+    blocked_tool: str,
+    fingerprint: str,
+) -> bool:
+    reason_code = str(
+        evaluation.payload.get("reason_code") or evaluation.reason_code or _OUTPUT_CONTRACT_REJECT_REASON_CODE
+    )
+    payload: AuthorTimeGateAblationPayload = {
+        "block_label": evaluation.block_label,
+        "canonical_output_contract_signature": evaluation.canonical_signature,
+        "canonical_required_child_paths": sorted(evaluation.required_paths),
+        "missing_metadata_paths": list(evaluation.missing_metadata_paths),
+        "missing_schema_paths": list(evaluation.missing_schema_paths),
+        "missing_return_paths": list(evaluation.missing_return_paths),
+        "shape_violations": list(evaluation.shape_violations),
+        "can_attempt_run": evaluation.can_attempt_run,
+    }
+    return record_author_time_gate_ablation_event(
+        ctx,
+        gate_id=gate_id,
+        reason_code=reason_code,
+        fingerprint=fingerprint,
+        blocked_tool=blocked_tool,
+        payload=payload,
+    )
+
+
 @dataclass(frozen=True)
 class _RuntimeOutputRepairContract:
     required_paths: set[str]
@@ -2502,6 +2542,7 @@ def _adjudicate_output_contract_ladder_after_reject(
         required_paths=evaluation.required_paths,
         signature=signature,
         current_fingerprint=current_fingerprint,
+        advisory_run_grantable=blockers == [_OUTPUT_CONTRACT_REJECT_REASON_CODE],
     )
     if actuation.kind == OutputContractActuationKind.BLOCKED_TERMINAL:
         _stash_output_source_unobservable_terminal(
@@ -3154,8 +3195,6 @@ def _observed_required_output_values(ctx: AgentContext, required_paths: set[str]
         return False
     if covered & required_paths:
         return True
-    # Match on ancestor/descendant lineage, not the top-level root: every requested-output path
-    # shares the ``output`` root, so a root-only match would credit any coverage as the required value.
     return any(_output_paths_share_lineage(str(path), required) for path in covered for required in required_paths)
 
 
@@ -3238,7 +3277,9 @@ def _grant_output_contract_advisory_run(ctx: AgentContext, signature: str) -> No
     )
 
 
-def consume_output_contract_advisory_grant_for_run(ctx: AgentContext) -> list[str]:
+def consume_output_contract_advisory_grant_for_run(
+    ctx: AgentContext, *, workflow_run_id: str | None = None
+) -> list[str]:
     consumed: list[str] = []
     for signature, state in list(ctx.output_contract_actuation_by_signature.items()):
         if state == OutputContractAdvisoryState.GRANTED:
@@ -3248,6 +3289,7 @@ def consume_output_contract_advisory_grant_for_run(ctx: AgentContext) -> list[st
         LOG.info(
             "copilot_output_contract_advisory_run_consumed",
             canonical_output_contract_signature=signature,
+            workflow_run_id=workflow_run_id,
         )
     if ctx.output_contract_actuation_count_by_signature:
         ctx.output_contract_actuation_count_by_signature.clear()
@@ -3256,6 +3298,37 @@ def consume_output_contract_advisory_grant_for_run(ctx: AgentContext) -> list[st
     return consumed
 
 
+def consume_output_contract_advisory_grant_for_run_result(
+    ctx: AgentContext, run_result: Mapping[str, object]
+) -> list[str]:
+    workflow_run_id = _workflow_run_id_from_run_result(run_result)
+    if workflow_run_id is None:
+        data = run_result.get("data")
+        if isinstance(data, Mapping):
+            LOG.warning(
+                "copilot_output_contract_advisory_run_result_missing_workflow_run_id",
+                data_keys=sorted(str(key) for key in data),
+            )
+        return []
+    consumed = consume_output_contract_advisory_grant_for_run(ctx, workflow_run_id=workflow_run_id)
+    for signature in consumed:
+        LOG.info(
+            "copilot_output_contract_advisory_run_dispatched_at_seam",
+            canonical_output_contract_signature=signature,
+            workflow_run_id=workflow_run_id,
+        )
+    return consumed
+
+
+def _workflow_run_id_from_run_result(run_result: Mapping[str, object]) -> str | None:
+    data = run_result.get("data")
+    if isinstance(data, Mapping):
+        workflow_run_id = data.get("workflow_run_id")
+        if isinstance(workflow_run_id, str) and workflow_run_id.strip():
+            return workflow_run_id
+    return None
+
+
 def _stash_output_source_unobservable_terminal(
     ctx: AgentContext,
     *,
@@ -3267,6 +3340,23 @@ def _stash_output_source_unobservable_terminal(
 ) -> None:
     if ctx.turn_halt is not None:
         return
+    payload: AuthorTimeGateAblationPayload = {
+        "block_label": block_label,
+        "canonical_output_contract_signature": signature,
+        "canonical_required_child_paths": sorted(required_paths),
+        "spine_split_blockers": list(blockers),
+    }
+    if copilot_author_time_gate_log_only_enabled() and ctx.output_contract_bail_actuated_this_call:
+        return
+    if record_author_time_gate_ablation_event(
+        ctx,
+        gate_id=_OUTPUT_CONTRACT_ABLATION_GATE_ID,
+        reason_code=reason_code,
+        fingerprint=signature,
+        blocked_tool="update_workflow",
+        payload=payload,
+    ):
+        return
     signal = build_output_source_unobservable_blocker_signal(
         reason_code=reason_code,
         required_paths=required_paths,
@@ -3332,7 +3422,7 @@ def _registered_output_paths(result: object) -> set[str]:
         if isinstance(value, dict) and value:
             for key, child in value.items():
                 absorb(f"{prefix}.{key}" if prefix else str(key), child)
-        elif prefix:
+        elif prefix and _runtime_output_value_is_meaningful(value):
             paths.add(prefix)
 
     for block in data.get("blocks") or []:
@@ -3360,9 +3450,8 @@ def record_output_contract_run_output_evidence(ctx: AgentContext, result: object
     observed_paths = {_strip_output_namespace_root(path) for path in _registered_output_paths(result)}
     for signature, paths in list(pending.items()):
         required_paths = {_strip_output_namespace_root(str(path)) for path in paths}
-        bound = any(
-            _output_paths_share_lineage(observed, required)
-            for observed in observed_paths
+        bound = bool(required_paths) and all(
+            any(_output_paths_share_lineage(observed, required) for observed in observed_paths)
             for required in required_paths
         )
         ctx.output_contract_run_output_observed_by_signature[signature] = True
@@ -3404,15 +3493,6 @@ def _reopen_dispatch_lacked_bound_extraction(ctx: AgentContext, signature: str)
     return True
 
 
-def _page_source_extraction_prompt(required_paths: set[str]) -> str:
-    fields = ", ".join(sorted(_metadata_contract_required_paths(required_paths))) or "the requested values"
-    return (
-        "Read the values currently shown on the page and return them keyed exactly to the requested output "
-        f"structure. Required output path(s): {fields}. Return only values visible on the page; if a value is "
-        "not present, return an empty string for it."
-    )
-
-
 def _strip_top_level_returns(code: str) -> str:
     try:
         tree = ast.parse(code)
@@ -3424,6 +3504,14 @@ def _strip_top_level_returns(code: str) -> str:
     return ast.unparse(ast.Module(body=kept, type_ignores=[]))
 
 
+def _page_source_extraction_prompt(required_paths: set[str]) -> str:
+    paths = ", ".join(sorted(_metadata_contract_required_paths(required_paths)))
+    return (
+        "Extract the requested output values visible on the current page. "
+        f"Populate these output paths when visible: {paths}."
+    )
+
+
 def _page_source_extraction_code(target_code: str, required_paths: set[str]) -> str:
     body = _strip_top_level_returns(textwrap.dedent(target_code).strip())
     schema = _schema_template_for_required_paths(required_paths)
@@ -3431,7 +3519,7 @@ def _page_source_extraction_code(target_code: str, required_paths: set[str]) ->
         f"{_OUTPUT_CONTRACT_PAGE_EXTRACT_VAR} = await page.extract(\n"
         f"    prompt={json.dumps(_page_source_extraction_prompt(required_paths))},\n"
         f"    schema={schema!r},\n"
-        f")\n"
+        ")\n"
         f"return {_OUTPUT_CONTRACT_PAGE_EXTRACT_VAR}"
     )
     return f"{body}\n{extract_block}" if body else extract_block
@@ -3482,9 +3570,27 @@ def _actuate_output_contract_bail(
     )
     if actuation.kind == OutputContractActuationKind.ADVISORY_RUN and not _run_authority_permits_dispatch(ctx):
         actuation = OutputContractActuation(OutputContractActuationKind.STRUCTURE_DIRECTIVE, actuation.family)
+    payload: AuthorTimeGateAblationPayload = {
+        "actuation_kind": actuation.kind.value,
+        "family": actuation.family.value,
+        "blockers": list(blockers),
+        "canonical_required_child_paths": sorted(required_paths),
+        "advisory_state": evidence.advisory_state.value,
+        "advisory_run_grantable": evidence.advisory_run_grantable,
+    }
     ctx.output_contract_bail_actuated_this_call = True
+    if record_author_time_gate_ablation_event(
+        ctx,
+        gate_id=_OUTPUT_CONTRACT_ABLATION_GATE_ID,
+        reason_code=actuation.reason_code or actuation.kind.value,
+        fingerprint=current_fingerprint,
+        blocked_tool="update_workflow",
+        payload=payload,
+    ):
+        return actuation
     if actuation.kind == OutputContractActuationKind.ADVISORY_RUN:
         _grant_output_contract_advisory_run(ctx, signature)
+        _arm_pending_run_evidence(ctx, signature, required_paths)
     elif actuation.kind == OutputContractActuationKind.STRUCTURE_DIRECTIVE and not evidence.prior_directive_unconsumed:
         _record_output_contract_actuation_progress(ctx, signature)
     if (
@@ -3616,6 +3722,8 @@ def _impose_output_contract_envelope_after_steering(
                 return workflow_yaml, raw_code_artifact_metadata, False
             if actuation.kind == OutputContractActuationKind.ADVISORY_RUN:
                 return workflow_yaml, raw_code_artifact_metadata, False
+            if copilot_author_time_gate_log_only_enabled():
+                return workflow_yaml, raw_code_artifact_metadata, False
             attempt_key = _output_contract_spine_directive_attempt_key(
                 signature=signature, block_label=label, workflow_yaml=workflow_yaml
             )
@@ -3722,6 +3830,45 @@ def _metadata_contract_run_preflight_reject(
         workflow_yaml,
         raw_code_artifact_metadata,
     )
+    convergence_reject = _recorded_outcome_convergence_reject(
+        ctx,
+        workflow_yaml=workflow_yaml,
+        code_artifact_metadata=raw_code_artifact_metadata,
+    )
+    if convergence_reject is not None:
+        block_labels = sorted(_workflow_yaml_code_blocks_by_label(workflow_yaml))
+        _record_author_time_reject_outcome(
+            ctx,
+            reason_code="unchanged_after_recorded_outcome",
+            summary="The authored code and output structure are unchanged after the last recorded test outcome.",
+            structural_payload={
+                "reason_code": "unchanged_after_recorded_outcome",
+                "authored_structure_signature": convergence_reject.authored_structure_signature,
+                "block_labels": block_labels,
+            },
+            authored_structure_signature=convergence_reject.authored_structure_signature,
+            block_labels=block_labels,
+        )
+        _record_code_authoring_guardrail_reject(
+            ctx, frontier_unchanged=convergence_reject.reason == "frontier_unchanged"
+        )
+        LOG.info(
+            "copilot recorded outcome convergence behavior",
+            convergence_reason=convergence_reject.reason,
+            commit_early_terminal=convergence_reject.commit_early_terminal,
+            block_labels=block_labels,
+        )
+        if convergence_reject.commit_early_terminal:
+            _commit_recorded_outcome_early_terminal(ctx)
+        return {
+            "ok": False,
+            "error": (
+                "Submitted workflow left the frontier the last recorded test outcome named unchanged. "
+                "Revise the code block or output metadata that owns that frontier before testing again."
+            ),
+            "user_facing_summary": _compiled_authoring_user_summary(),
+            "data": _code_repair_progress_data(),
+        }
     evaluation = _evaluate_output_contract_for_code_block(
         ctx,
         workflow_yaml,
@@ -3730,33 +3877,31 @@ def _metadata_contract_run_preflight_reject(
     )
     if evaluation is None or not evaluation.has_deficiencies:
         return None
+    authored_fingerprint = _output_contract_structural_fingerprint(workflow_yaml, evaluation.canonical_signature)
     advisory_granted = _output_contract_advisory_granted(ctx, evaluation.canonical_signature)
     # A granted advisory must arm run-output evidence before the grant is consumed, even when the
     # workflow is otherwise run-attemptable — otherwise the dispatched run records no evidence.
     if evaluation.can_attempt_run and not advisory_granted:
         return None
+    if _record_output_contract_ablation_event(
+        ctx,
+        evaluation,
+        gate_id=_METADATA_PREFLIGHT_ABLATION_GATE_ID,
+        blocked_tool="update_and_run_blocks",
+        fingerprint=authored_fingerprint,
+    ):
+        return None
     payload = _record_output_contract_reject(
         ctx,
         evaluation,
         summary="Submitted workflow does not satisfy the requested output contract before run.",
-        authored_structural_fingerprint=_output_contract_structural_fingerprint(
-            workflow_yaml, evaluation.canonical_signature
-        ),
+        authored_structural_fingerprint=authored_fingerprint,
         workflow_yaml=workflow_yaml,
     )
     if advisory_granted:
         _arm_pending_run_evidence(ctx, evaluation.canonical_signature, set(evaluation.required_paths))
-        consume_output_contract_advisory_grant_for_run(ctx)
-        LOG.info(
-            "copilot_output_contract_advisory_run_dispatched_at_seam",
-            block_label=evaluation.block_label,
-            canonical_output_contract_signature=evaluation.canonical_signature,
-        )
         return None
-    if evaluation.repair_context is not None and (
-        evaluation.repair_context.required_block_structure
-        or evaluation.repair_context.reason_code == OUTPUT_OWNER_AMBIGUITY_REASON_CODE
-    ):
+    if evaluation.repair_context is not None:
         ctx.last_code_authoring_repair_context = evaluation.repair_context
     payload = dict(payload)
     payload["output_contract_reason_code"] = payload.get("reason_code")
@@ -5028,11 +5173,20 @@ def _should_impose_after_update_attempt(ctx: AgentContext) -> bool:
 
 def _author_time_reject_reopens_synthesized_imposition(ctx: AgentContext) -> bool:
     latest = ctx.latest_recorded_build_test_outcome
-    return (
+    if not (
         isinstance(latest, RecordedBuildTestOutcome)
         and latest.is_authoritative
         and latest.phase == "author_time_reject"
-        and latest.reason_code in {"metadata_reject", "synthesized_parameter_binding_ambiguous"}
+    ):
+        return False
+    if latest.reason_code in {"metadata_reject", "synthesized_parameter_binding_ambiguous"}:
+        return True
+    repair_context = ctx.last_code_authoring_repair_context
+    # Ambiguous bare selectors are repaired by the same strict imposition path as metadata gaps.
+    return (
+        latest.reason_code == "code_safety_reject"
+        and isinstance(repair_context, CodeAuthoringRepairContext)
+        and repair_context.reason_code == "ambiguous_bare_selector"
     )
 
 
@@ -8118,27 +8272,35 @@ async def _update_workflow(
         and output_contract_evaluation.has_deficiencies
         and not output_contract_static_advisory_allowed
     ):
-        payload = _record_output_contract_reject(
+        authored_fingerprint = _output_contract_structural_fingerprint(
+            workflow_yaml, output_contract_evaluation.canonical_signature
+        )
+        if not _record_output_contract_ablation_event(
             ctx,
             output_contract_evaluation,
-            summary="Submitted workflow does not satisfy the requested output contract.",
-            authored_structural_fingerprint=_output_contract_structural_fingerprint(
-                workflow_yaml, output_contract_evaluation.canonical_signature
-            ),
-            workflow_yaml=workflow_yaml,
-        )
-        reject_result = _output_contract_reject_result(
-            output_contract_evaluation,
-            payload=payload,
-            tool_name="update_workflow",
-        )
-        return reject(
-            error=str(reject_result["error"]),
-            user_facing_summary=str(reject_result.get("user_facing_summary") or _compiled_authoring_user_summary()),
-            data=reject_result.get("data") if isinstance(reject_result.get("data"), dict) else None,
-            repair_context=output_contract_evaluation.repair_context,
-            record_repair_context_outcome=False,
-        )
+            gate_id=_OUTPUT_CONTRACT_ABLATION_GATE_ID,
+            blocked_tool="update_workflow",
+            fingerprint=authored_fingerprint,
+        ):
+            payload = _record_output_contract_reject(
+                ctx,
+                output_contract_evaluation,
+                summary="Submitted workflow does not satisfy the requested output contract.",
+                authored_structural_fingerprint=authored_fingerprint,
+                workflow_yaml=workflow_yaml,
+            )
+            reject_result = _output_contract_reject_result(
+                output_contract_evaluation,
+                payload=payload,
+                tool_name="update_workflow",
+            )
+            return reject(
+                error=str(reject_result["error"]),
+                user_facing_summary=str(reject_result.get("user_facing_summary") or _compiled_authoring_user_summary()),
+                data=reject_result.get("data") if isinstance(reject_result.get("data"), dict) else None,
+                repair_context=output_contract_evaluation.repair_context,
+                record_repair_context_outcome=False,
+            )
     missing_metadata_error = _missing_code_artifact_metadata_error(
         workflow_yaml,
         ctx,

skyvern/forge/sdk/core/aiohttp_helper.py

@@ -1,15 +1,124 @@
 import asyncio
+import ipaddress
 import os
+import socket
 from typing import Any
+from urllib.parse import urlparse
 
 import aiofiles
 import aiohttp
 import structlog
+from aiohttp.abc import AbstractResolver, ResolveResult
+from aiohttp.resolver import DefaultResolver
 
-from skyvern.exceptions import HttpException
+from skyvern.exceptions import HttpException, InvalidUrl
+from skyvern.utils.url_validators import (
+    MAX_SAFE_REDIRECTS,
+    SAFE_REDIRECT_STATUS_CODES,
+    resolve_fetch_host_ips,
+    validate_fetch_url_with_resolved_ips,
+    validate_redirect_url_with_resolved_ips,
+)
 
 LOG = structlog.get_logger()
 DEFAULT_REQUEST_TIMEOUT = 30
+_REDIRECT_CREDENTIAL_HEADERS = {"authorization", "cookie"}
+
+
+def _url_origin(url: str) -> tuple[str, str, int | None]:
+    parsed = urlparse(url)
+    scheme = parsed.scheme.lower()
+    port = parsed.port
+    if port is None:
+        port = 443 if scheme == "https" else 80 if scheme == "http" else None
+    return scheme, (parsed.hostname or "").lower().rstrip("."), port
+
+
+def strip_cross_origin_redirect_credentials(
+    headers: dict[str, str],
+    cookies: dict[str, str] | None,
+    current_url: str,
+    next_url: str,
+) -> tuple[dict[str, str], dict[str, str] | None]:
+    if _url_origin(current_url) == _url_origin(next_url):
+        return headers, cookies
+    return {key: value for key, value in headers.items() if key.lower() not in _REDIRECT_CREDENTIAL_HEADERS}, None
+
+
+class SSRFGuardedResolver(AbstractResolver):
+    def __init__(self) -> None:
+        self._pinned_host_ips: dict[str, tuple[str, ...]] = {}
+        self._trusted_proxy_hosts: set[str] = set()
+        self._default_resolver = DefaultResolver()
+
+    @staticmethod
+    def _host_key(host: str) -> str:
+        return host.strip().lower().rstrip(".")
+
+    def pin_host_ips(self, host: str, ips: tuple[str, ...]) -> None:
+        if not ips:
+            raise OSError(f"No safe addresses resolved for host: {host}")
+        self._pinned_host_ips[self._host_key(host)] = ips
+
+    def pin_url_ips(self, url: str, ips: tuple[str, ...]) -> None:
+        host = urlparse(url).hostname
+        if not host:
+            raise InvalidUrl(url=url)
+        self.pin_host_ips(host, ips)
+
+    def trust_proxy_url(self, proxy: str) -> None:
+        host = urlparse(proxy).hostname
+        if host:
+            self._trusted_proxy_hosts.add(self._host_key(host))
+
+    async def resolve(
+        self, host: str, port: int = 0, family: socket.AddressFamily = socket.AF_UNSPEC
+    ) -> list[ResolveResult]:
+        results: list[ResolveResult] = []
+        host_key = self._host_key(host)
+        if host_key in self._trusted_proxy_hosts:
+            return await self._default_resolver.resolve(host, port, family)
+
+        ips = self._pinned_host_ips.get(host_key)
+        resolved_ips = await asyncio.to_thread(resolve_fetch_host_ips, host) if ips is None else ips
+        for ip in resolved_ips:
+            ip_family = (
+                socket.AF_INET6 if isinstance(ipaddress.ip_address(ip), ipaddress.IPv6Address) else socket.AF_INET
+            )
+            if family not in (socket.AF_UNSPEC, ip_family):
+                continue
+            results.append(
+                {
+                    "hostname": host,
+                    "host": ip,
+                    "port": port,
+                    "family": ip_family,
+                    "proto": 0,
+                    "flags": 0,
+                }
+            )
+        if not results:
+            raise OSError(f"No safe addresses resolved for host: {host}")
+        return results
+
+    async def close(self) -> None:
+        await self._default_resolver.close()
+
+
+async def validate_and_pin_fetch_url(url: str, resolver: SSRFGuardedResolver) -> str:
+    validated_url, ips = await asyncio.to_thread(validate_fetch_url_with_resolved_ips, url)
+    resolver.pin_url_ips(validated_url, ips)
+    return validated_url
+
+
+async def validate_and_pin_redirect_url(url: str, location: str, resolver: SSRFGuardedResolver) -> str:
+    validated_url, ips = await asyncio.to_thread(validate_redirect_url_with_resolved_ips, url, location)
+    resolver.pin_url_ips(validated_url, ips)
+    return validated_url
+
+
+def ssrf_guarded_tcp_connector(resolver: SSRFGuardedResolver | None = None) -> aiohttp.TCPConnector:
+    return aiohttp.TCPConnector(resolver=resolver or SSRFGuardedResolver(), use_dns_cache=False)
 
 
 async def aiohttp_request(
@@ -44,64 +153,86 @@ async def aiohttp_request(
         Tuple of (status_code, response_headers, response_body)
         where response_body can be dict (for JSON) or str (for text)
     """
-    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=timeout)) as session:
-        # Ensure headers is always a dict for type safety
-        headers_dict: dict[str, str] = headers or {}
-        request_kwargs: dict[str, Any] = {
-            "url": url,
-            "headers": headers_dict,
-            "cookies": cookies,
-            "proxy": proxy,
-            "allow_redirects": follow_redirects,
-        }
+    resolver = SSRFGuardedResolver()
+    if proxy:
+        resolver.trust_proxy_url(proxy)
+    current_url = await validate_and_pin_fetch_url(url, resolver)
+    request_method = method.upper()
+    request_headers = dict(headers or {})
+    request_cookies = cookies
+    strip_body_headers = False
+
+    async with aiohttp.ClientSession(
+        timeout=aiohttp.ClientTimeout(total=timeout), connector=ssrf_guarded_tcp_connector(resolver)
+    ) as session:
+
+        async def build_request_kwargs() -> dict[str, Any]:
+            headers_dict = dict(request_headers)
+            if strip_body_headers:
+                headers_dict.pop("Content-Length", None)
+                headers_dict.pop("content-length", None)
+            request_kwargs: dict[str, Any] = {
+                "headers": headers_dict,
+                "cookies": request_cookies,
+                "proxy": proxy,
+                "allow_redirects": False,
+            }
+
+            if request_method == "GET":
+                return request_kwargs
 
-        # Handle body based on content type and method
-        if method.upper() != "GET":
-            # If files are provided, use multipart/form-data
             if files:
                 form = aiohttp.FormData()
-
-                # Add files to form
                 for field_name, file_path in files.items():
                     if not os.path.exists(file_path):
                         raise FileNotFoundError(f"File not found: {file_path}")
 
                     filename = os.path.basename(file_path)
                     async with aiofiles.open(file_path, "rb") as f:
-                        file_content = await f.read()
-                        form.add_field(field_name, file_content, filename=filename)
+                        form.add_field(field_name, await f.read(), filename=filename)
 
-                # Add data fields to form if provided
                 if data is not None and isinstance(data, dict):
                     for key, value in data.items():
                         form.add_field(key, str(value))
 
                 request_kwargs["data"] = form
-                # Remove Content-Type header if present, let aiohttp set it for multipart
-                # headers_dict is already typed as dict[str, str] from initialization
-                if "Content-Type" in headers_dict:
-                    del headers_dict["Content-Type"]
-                if "content-type" in headers_dict:
-                    del headers_dict["content-type"]
-            # Explicit overrides first
+                headers_dict.pop("Content-Type", None)
+                headers_dict.pop("content-type", None)
             elif json_data is not None:
                 request_kwargs["json"] = json_data
             elif data is not None:
                 content_type = headers_dict.get("Content-Type") or headers_dict.get("content-type") or ""
-                if "application/json" in content_type.lower():
-                    request_kwargs["json"] = data
-                else:
-                    request_kwargs["data"] = data
+                request_kwargs["json" if "application/json" in content_type.lower() else "data"] = data
+            return request_kwargs
 
-        async with session.request(method.upper(), **request_kwargs) as response:
-            response_headers = dict(response.headers)
+        for _ in range(MAX_SAFE_REDIRECTS + 1):
+            request_kwargs = await build_request_kwargs()
+            request_kwargs["url"] = current_url
+            async with session.request(request_method, **request_kwargs) as response:
+                if (
+                    follow_redirects
+                    and response.status in SAFE_REDIRECT_STATUS_CODES
+                    and response.headers.get("Location")
+                ):
+                    next_url = await validate_and_pin_redirect_url(current_url, response.headers["Location"], resolver)
+                    request_headers, request_cookies = strip_cross_origin_redirect_credentials(
+                        request_headers, request_cookies, current_url, next_url
+                    )
+                    current_url = next_url
+                    if response.status == 303 or (response.status in (301, 302) and request_method == "POST"):
+                        request_method = "GET"
+                        strip_body_headers = True
+                    continue
 
-            try:
-                response_body = await response.json()
-            except (aiohttp.ContentTypeError, Exception):
-                response_body = await response.text()
+                response_headers = dict(response.headers)
 
-            return response.status, response_headers, response_body
+                try:
+                    response_body = await response.json()
+                except (aiohttp.ContentTypeError, Exception):
+                    response_body = await response.text()
+
+                return response.status, response_headers, response_body
+        raise HttpException(400, current_url, "Too many redirects while making HTTP request")
 
 
 async def aiohttp_get_json(

skyvern/forge/sdk/routes/credentials.py

@@ -40,6 +40,7 @@ import pyotp
 import structlog
 from fastapi import BackgroundTasks, Body, Depends, Header, HTTPException, Path, Query, Response
 from onepassword.client import Client as OnePasswordClient
+from onepassword.errors import DesktopSessionExpiredException, RateLimitExceededException
 
 from skyvern.config import settings
 from skyvern.exceptions import HttpException as SkyvernHttpException
@@ -115,6 +116,8 @@ from skyvern.forge.sdk.services.credential.credential_vault_service import Crede
 from skyvern.forge.sdk.services.credentials import (
     AuthenticatorTotpErrorCode,
     AuthenticatorTotpParseResult,
+    extract_onepassword_upstream_5xx_status,
+    is_onepassword_credential_error,
     normalize_totp_config,
     parse_totp_config,
 )
@@ -2323,14 +2326,72 @@ async def list_onepassword_items(
                 )
 
         return OnePasswordItemsResponse(configured=True, items=items)
-    except Exception as e:
-        LOG.error(
-            "Failed to list 1Password items",
+    except asyncio.TimeoutError as e:
+        LOG.warning(
+            "Timed out while listing 1Password items",
             organization_id=current_org.organization_id,
-            error=str(e),
             exc_info=True,
         )
-        raise HTTPException(status_code=502, detail="Failed to list 1Password items") from e
+        raise HTTPException(
+            status_code=502,
+            detail="1Password is temporarily unavailable. Please retry in a few minutes.",
+        ) from e
+    except RateLimitExceededException as e:
+        LOG.warning(
+            "1Password rate limit exceeded while listing items",
+            organization_id=current_org.organization_id,
+            error=str(e),
+        )
+        raise HTTPException(
+            status_code=429,
+            detail="1Password rate limit exceeded. Please retry in a few minutes.",
+        ) from e
+    except DesktopSessionExpiredException as e:
+        LOG.warning(
+            "1Password session expired while listing items",
+            organization_id=current_org.organization_id,
+            error=str(e),
+        )
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                "Your 1Password session appears to be expired. Please update your token in Settings and try again."
+            ),
+        ) from e
+    except Exception as e:
+        raw = str(e)
+        # The 1Password SDK exposes these as plain string exceptions; credential
+        # evidence takes precedence over embedded status text.
+        if is_onepassword_credential_error(raw):
+            LOG.warning(
+                "Invalid 1Password service account token while listing items",
+                organization_id=current_org.organization_id,
+                error=raw,
+            )
+            raise HTTPException(
+                status_code=400,
+                detail=(
+                    "Your 1Password service account token appears to be invalid or expired. "
+                    "Please update it in Settings and try again."
+                ),
+            ) from e
+        upstream_status = extract_onepassword_upstream_5xx_status(raw)
+        if upstream_status is not None:
+            LOG.warning(
+                "1Password is temporarily unavailable while listing items",
+                organization_id=current_org.organization_id,
+                upstream_status=upstream_status,
+            )
+            raise HTTPException(
+                status_code=502,
+                detail="1Password is temporarily unavailable. Please retry in a few minutes.",
+            ) from e
+        LOG.error(
+            "Unexpected failure while listing 1Password items",
+            organization_id=current_org.organization_id,
+            exc_info=True,
+        )
+        raise HTTPException(status_code=500, detail="Failed to list 1Password items.") from e
 
 
 @base_router.get(

skyvern/forge/sdk/services/credentials.py

@@ -10,6 +10,39 @@ LOG = structlog.get_logger()
 
 _SECRET_QUERY_PATTERN = re.compile(r"(?i)(?:^|[?&])secret=([^&#]+)")
 
+# 1Password's Python SDK forwards generic 5xx upstream failures as plain Exceptions
+# whose stringified message embeds the HTTP status.
+_ONEPASSWORD_UPSTREAM_5XX_PATTERN = re.compile(
+    r"(?i)"
+    r"(?:\b(?:HTTP|status(?:\s+code)?|code|response)\s*[:=]?\s*(5\d{2})\b)"
+    r"|"
+    r"(?:\b(5\d{2})\s+(?:service\s+unavailable|bad\s+gateway|gateway\s+timeout|internal\s+server\s+error)\b)"
+)
+_ONEPASSWORD_CREDENTIAL_ERROR_PATTERN = re.compile(
+    r"(?i)"
+    r"(?:\b(?:unauthorized|forbidden|not\s+authenticated|authentication\s+failed)\b)"
+    r"|"
+    r"(?:\b(?:invalid|expired|malformed|parse(?:d)?)\b.{0,48}\b(?:token|credential|service\s+account)\b)"
+    r"|"
+    r"(?:\b(?:token|credential|service\s+account)\b.{0,48}\b(?:invalid|expired|malformed|not\s+valid|parse(?:d)?)\b)"
+)
+
+
+def extract_onepassword_upstream_5xx_status(message: str) -> int | None:
+    """Return the upstream 5xx HTTP status embedded in a 1Password SDK error message, if any.
+
+    A match means the failure originated on 1Password's side.
+    """
+    match = _ONEPASSWORD_UPSTREAM_5XX_PATTERN.search(message)
+    if not match:
+        return None
+    status_digits = match.group(1) or match.group(2)
+    return int(status_digits) if status_digits else None
+
+
+def is_onepassword_credential_error(message: str) -> bool:
+    return bool(_ONEPASSWORD_CREDENTIAL_ERROR_PATTERN.search(message))
+
 
 class OnePasswordConstants(StrEnum):
     """Constants for 1Password integration."""

skyvern/forge/sdk/services/org_auth_service.py

@@ -364,7 +364,8 @@ async def resolve_org_from_api_key(
             else:
                 normalized_api_key, normalization_flags = _normalize_api_key_with_flags(x_api_key)
             api_key_debug_fields = _get_api_key_debug_fields(x_api_key, normalized_api_key, normalization_flags)
-            LOG.error(
+            # Malformed client key rejected with 403 — a client error, not a server fault.
+            LOG.warning(
                 "Error decoding JWT",
                 error_type=type(exc).__name__,
                 error_reason=_get_safe_auth_error_reason(exc),

skyvern/forge/sdk/workflow/context_manager.py

@@ -32,7 +32,12 @@ from skyvern.forge.sdk.schemas.credentials import CredentialVaultType, PasswordC
 from skyvern.forge.sdk.schemas.organizations import Organization
 from skyvern.forge.sdk.schemas.tasks import TaskStatus
 from skyvern.forge.sdk.services.bitwarden import BitwardenConstants, BitwardenService
-from skyvern.forge.sdk.services.credentials import AzureVaultConstants, OnePasswordConstants, normalize_totp_config
+from skyvern.forge.sdk.services.credentials import (
+    AzureVaultConstants,
+    OnePasswordConstants,
+    extract_onepassword_upstream_5xx_status,
+    normalize_totp_config,
+)
 from skyvern.forge.sdk.workflow.credential_selection import select_credential_for_run
 from skyvern.forge.sdk.workflow.exceptions import MissingJinjaVariables, OutputParameterKeyCollisionError
 from skyvern.forge.sdk.workflow.models.parameter import (
@@ -76,14 +81,6 @@ _CREDENTIAL_PARAMETER_TYPES: tuple[type, ...] = (
     OnePasswordCredentialParameter,
 )
 
-# 1Password's Python SDK forwards generic 5xx upstream failures as plain Exceptions
-# whose stringified message embeds the HTTP status.
-_ONEPASSWORD_5XX_PATTERN = re.compile(
-    r"(?i)"
-    r"(?:\b(?:HTTP|status(?:\s+code)?|code|response)\s*[:=]?\s*(5\d{2})\b)"
-    r"|"
-    r"(?:\b(5\d{2})\s+(?:service\s+unavailable|bad\s+gateway|gateway\s+timeout|internal\s+server\s+error)\b)"
-)
 _SECRET_FIELD_KEY_PATTERN = re.compile(r"[^A-Za-z0-9_]+")
 
 
@@ -822,11 +819,10 @@ class WorkflowRunContext:
             raise OnePasswordSessionExpiredError(f"{str(e)} {lookup_context}") from e
         except Exception as e:
             raw = str(e)
-            match = _ONEPASSWORD_5XX_PATTERN.search(raw)
-            if match:
-                status_digits = match.group(1) or match.group(2)
+            upstream_status = extract_onepassword_upstream_5xx_status(raw)
+            if upstream_status is not None:
                 raise OnePasswordServiceUnavailableError(
-                    status_code=int(status_digits),
+                    status_code=upstream_status,
                     lookup_context=lookup_context,
                 ) from e
             raise OnePasswordGetItemError(f"{raw} {lookup_context}") from e

skyvern/forge/sdk/workflow/service.py

@@ -4388,7 +4388,8 @@ class WorkflowService:
             workflow_run = await self.mark_workflow_run_as_canceled(workflow_run_id=workflow_run_id)
             return workflow_run, True
         if block_result.status == BlockStatus.failed:
-            LOG.error(
+            # Run-level outcome, recorded as the run's failure_reason below; not a platform fault.
+            LOG.warning(
                 f"Block with type {block.block_type} at index {block_idx}/{blocks_cnt - 1} failed for workflow run {workflow_run_id}",
                 block_type=block.block_type,
                 workflow_run_id=workflow_run_id,

skyvern/services/task_v2_service.py

@@ -1372,7 +1372,8 @@ async def handle_block_result(
         )
 
     elif block_result.status == BlockStatus.failed:
-        LOG.error(
+        # Run-level outcome, surfaced via the run's failure handling below; not a platform fault.
+        LOG.warning(
             f"Block with type {block.block_type} failed for workflow run {workflow_run_id}",
             block_type=block.block_type,
             workflow_run_id=workflow_run.workflow_run_id,

skyvern/utils/url_validators.py

@@ -1,12 +1,19 @@
 import ipaddress
+import socket
 from http import HTTPStatus
-from urllib.parse import quote, urlparse, urlsplit, urlunsplit
+from urllib.parse import quote, urljoin, urlparse, urlsplit, urlunsplit
 
 from pydantic import HttpUrl, ValidationError
 
 from skyvern.config import settings
 from skyvern.exceptions import BlockedHost, InvalidUrl, SkyvernHTTPException
 
+SAFE_REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308}
+MAX_SAFE_REDIRECTS = 10
+
+_BLOCKED_INTERNAL_HOSTNAMES = frozenset({"localhost", "metadata.google.internal", "kubernetes.default.svc"})
+_BLOCKED_INTERNAL_SUFFIXES = (".local", ".localhost", ".internal", ".cluster.local")
+
 
 def strip_query_params(url: str) -> str:
     """Return scheme://host/path with query string, fragment, and userinfo removed.
@@ -65,32 +72,146 @@ def prepend_scheme_and_validate_url(url: str) -> str:
     return url
 
 
-def is_blocked_host(host: str) -> bool:
+def _normalize_host(host: str) -> str:
     # RFC 3986 wraps IPv6 literals in [...]; ip_address() only accepts the bare form.
-    bare = host[1:-1] if host.startswith("[") and host.endswith("]") else host
+    return (host[1:-1] if host.startswith("[") and host.endswith("]") else host).strip().lower().rstrip(".")
+
+
+def _normalize_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
+        return ip.ipv4_mapped
+    return ip
+
+
+def _is_blocked_ip(ip: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
+    ip = _normalize_ip(ip)
+    return bool(
+        ip.is_private or ip.is_link_local or ip.is_loopback or ip.is_reserved or ip.is_multicast or ip.is_unspecified
+    )
+
+
+def _is_allowed_host(host: str) -> bool:
+    normalized = _normalize_host(host)
     ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None
     try:
-        ip = ipaddress.ip_address(bare)
-        if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
-            ip = ip.ipv4_mapped
+        ip = _normalize_ip(ipaddress.ip_address(normalized))
     except ValueError:
         ip = None
     except Exception:
         return False
 
-    candidate_forms = {host.lower(), bare.lower()}
+    candidate_forms = {host.lower(), normalized}
     if ip is not None:
         candidate_forms.add(str(ip).lower())
 
     allowed = {h.lower() for h in settings.ALLOWED_HOSTS}
-    if candidate_forms & allowed:
+    return bool(candidate_forms & allowed)
+
+
+def _is_internal_hostname(host: str) -> bool:
+    normalized = _normalize_host(host)
+    if normalized in _BLOCKED_INTERNAL_HOSTNAMES:
+        return True
+    if normalized.endswith(_BLOCKED_INTERNAL_SUFFIXES):
+        return True
+    return normalized.endswith(".svc")
+
+
+def is_blocked_host(host: str, *, resolve_dns: bool = False) -> bool:
+    normalized = _normalize_host(host)
+    if not normalized:
+        return True
+
+    if _is_allowed_host(host):
         return False
 
-    if ip is not None:
-        return ip.is_private or ip.is_link_local or ip.is_loopback or ip.is_reserved
+    blocked = {b.lower().rstrip(".") for b in settings.BLOCKED_HOSTS}
+    if normalized in blocked or _is_internal_hostname(normalized):
+        return True
 
-    blocked = {b.lower() for b in settings.BLOCKED_HOSTS}
-    return host.lower() in blocked
+    ip: ipaddress.IPv4Address | ipaddress.IPv6Address | None
+    try:
+        ip = ipaddress.ip_address(normalized)
+    except ValueError:
+        ip = None
+    except Exception:
+        return True
+
+    if ip is not None:
+        return _is_blocked_ip(ip)
+
+    if not resolve_dns:
+        return False
+
+    try:
+        infos = socket.getaddrinfo(normalized, None, type=socket.SOCK_STREAM)
+    except OSError:
+        return True
+
+    resolved_any = False
+    for info in infos:
+        sockaddr = info[4]
+        ip_str = sockaddr[0] if sockaddr else None
+        if not ip_str:
+            continue
+        try:
+            resolved_ip = ipaddress.ip_address(ip_str)
+        except ValueError:
+            continue
+        resolved_any = True
+        if _is_blocked_ip(resolved_ip):
+            return True
+    return not resolved_any
+
+
+def resolve_fetch_host_ips(host: str) -> tuple[str, ...]:
+    normalized = _normalize_host(host)
+    if not normalized:
+        raise BlockedHost(host=host)
+
+    allowed = _is_allowed_host(host)
+    if not allowed and (normalized in {b.lower().rstrip(".") for b in settings.BLOCKED_HOSTS}):
+        raise BlockedHost(host=host)
+    if not allowed and _is_internal_hostname(normalized):
+        raise BlockedHost(host=host)
+
+    try:
+        ip = ipaddress.ip_address(normalized)
+    except ValueError:
+        ip = None
+    except Exception:
+        raise BlockedHost(host=host)
+
+    if ip is not None:
+        normalized_ip = _normalize_ip(ip)
+        if not allowed and _is_blocked_ip(normalized_ip):
+            raise BlockedHost(host=host)
+        return (str(normalized_ip),)
+
+    try:
+        infos = socket.getaddrinfo(normalized, None, type=socket.SOCK_STREAM)
+    except OSError:
+        raise BlockedHost(host=host)
+
+    resolved_ips: list[str] = []
+    for info in infos:
+        sockaddr = info[4]
+        ip_str = sockaddr[0] if sockaddr else None
+        if not ip_str:
+            continue
+        try:
+            resolved_ip = _normalize_ip(ipaddress.ip_address(ip_str))
+        except ValueError:
+            continue
+        if not allowed and _is_blocked_ip(resolved_ip):
+            raise BlockedHost(host=host)
+        resolved_ip_str = str(resolved_ip)
+        if resolved_ip_str not in resolved_ips:
+            resolved_ips.append(resolved_ip_str)
+
+    if not resolved_ips:
+        raise BlockedHost(host=host)
+    return tuple(resolved_ips)
 
 
 def validate_url(url: str) -> str | None:
@@ -109,6 +230,30 @@ def validate_url(url: str) -> str | None:
     return str(v)
 
 
+def validate_fetch_url_with_resolved_ips(url: str) -> tuple[str, tuple[str, ...]]:
+    try:
+        url = prepend_scheme_and_validate_url(url=url)
+        v = HttpUrl(url=url)
+    except Exception as e:
+        raise SkyvernHTTPException(message=str(e), status_code=HTTPStatus.BAD_REQUEST)
+
+    if not v.host:
+        raise InvalidUrl(url=url)
+    return str(v), resolve_fetch_host_ips(v.host)
+
+
+def validate_fetch_url(url: str) -> str:
+    return validate_fetch_url_with_resolved_ips(url)[0]
+
+
+def validate_redirect_url_with_resolved_ips(url: str, location: str) -> tuple[str, tuple[str, ...]]:
+    return validate_fetch_url_with_resolved_ips(urljoin(url, location))
+
+
+def validate_redirect_url(url: str, location: str) -> str:
+    return validate_redirect_url_with_resolved_ips(url, location)[0]
+
+
 def encode_url(url: str) -> str:
     parts = list(urlsplit(url))
     # Encode the path while preserving "/" and "%"

skyvern/webeye/actions/handler.py

@@ -61,6 +61,7 @@ from skyvern.exceptions import (
     NoSuitableAutoCompleteOption,
     OptionIndexOutOfBound,
     PhoneNumberInputMismatch,
+    SkyvernException,
 )
 from skyvern.experimentation.wait_utils import get_or_create_wait_config, get_wait_time
 from skyvern.forge import app
@@ -3748,6 +3749,12 @@ async def handle_select_option_action(
             return results
         suggested_value = result.value
 
+    except SkyvernException as e:
+        # Expected selection outcomes on non-standard dropdowns (no matching option,
+        # no incremental elements); recorded as ActionFailure like any other miss.
+        LOG.warning("Custom select error", exc_info=True)
+        results.append(ActionFailure(exception=e))
+        return results
     except Exception as e:
         LOG.exception("Custom select error")
         results.append(ActionFailure(exception=e))

skyvern/webeye/actions/handler_utils.py

@@ -27,10 +27,12 @@ async def download_file(
     try:
         return await download_file_api(file_url, organization_id=organization_id)
     except Exception:
-        LOG.exception(
+        # Fully self-recovering: the action proceeds without the file.
+        LOG.warning(
             "Failed to download file, continuing without it",
             action=action,
             file_url=file_url,
+            exc_info=True,
         )
         return []