feat: show credits consumed per run on workflow run pages (#SKY-8320) (#6133)

skyvern-frontend/src/routes/workflows/debugger/DebuggerRunTimeline.tsx

@@ -59,13 +59,23 @@ function DebuggerRunTimeline({
         "animate-pulse": firstActionOrThoughtIsPending,
       })}
     >
-      <div className="grid w-full grid-cols-2 gap-2">
+      <div className="grid w-full grid-cols-3 gap-2">
         <div className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs">
           Actions: {numberOfActions}
         </div>
         <div className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs">
           Steps: {workflowRun.total_steps ?? 0}
         </div>
+        <div
+          className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs"
+          title="Credits consumed by this run (live + cached)"
+        >
+          Credits:{" "}
+          {(
+            (workflowRun.credits_used ?? 0) +
+            (workflowRun.cached_credits_used ?? 0)
+          ).toLocaleString()}
+        </div>
       </div>
       <ScrollArea>
         <ScrollAreaViewport className="h-full w-full">

skyvern-frontend/src/routes/workflows/workflowRun/WorkflowRunTimeline.tsx

@@ -82,13 +82,23 @@ function WorkflowRunTimeline({
 
   return (
     <div className="min-w-0 space-y-4 overflow-hidden rounded bg-slate-elevation1 p-4">
-      <div className="grid grid-cols-2 gap-2">
+      <div className="grid grid-cols-3 gap-2">
         <div className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs">
           Actions: {numberOfActions}
         </div>
         <div className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs">
           Steps: {workflowRun.total_steps ?? 0}
         </div>
+        <div
+          className="flex items-center justify-center rounded bg-slate-elevation3 px-4 py-3 text-xs"
+          title="Credits consumed by this run (live + cached)"
+        >
+          Credits:{" "}
+          {(
+            (workflowRun.credits_used ?? 0) +
+            (workflowRun.cached_credits_used ?? 0)
+          ).toLocaleString()}
+        </div>
       </div>
       <ScrollArea>
         <ScrollAreaViewport className="h-[37rem] max-h-[37rem] [&>div]:!block [&>div]:!overflow-x-hidden">

skyvern/constants.py

@@ -40,6 +40,7 @@ DROPDOWN_MENU_MAX_DISTANCE = 100
 BROWSER_DOWNLOADING_SUFFIX = ".crdownload"
 MAX_UPLOAD_FILE_COUNT = 50
 AZURE_BLOB_STORAGE_MAX_UPLOAD_FILE_COUNT = 50
+CUSTOMER_STORAGE_UPLOAD_MAX_BYTES = 1 * 1024 * 1024 * 1024  # 1 GB per file (FileUploadBlock)
 DEFAULT_MAX_SCREENSHOT_SCROLLS = 3
 
 # Default navigation_goal for LoginBlocks. Instructs the LLM how to find the login

skyvern/exceptions.py

@@ -567,6 +567,16 @@ class DownloadFileMaxSizeExceeded(SkyvernException):
         super().__init__(f"Download file size exceeded the maximum allowed size of {max_size} MB.")
 
 
+class UploadFileMaxSizeExceeded(SkyvernException):
+    def __init__(self, file_size_bytes: int, max_size_bytes: int) -> None:
+        self.file_size_bytes = file_size_bytes
+        self.max_size_bytes = max_size_bytes
+        super().__init__(
+            f"Upload file size {file_size_bytes / 1024 / 1024:.1f} MB exceeded the maximum "
+            f"allowed size of {max_size_bytes / 1024 / 1024:.0f} MB."
+        )
+
+
 class DownloadFileMaxWaitingTime(SkyvernException):
     def __init__(self, downloading_files: list[str]) -> None:
         self.downloading_files = downloading_files

skyvern/forge/agent_functions.py

@@ -1,6 +1,7 @@
 import asyncio
 import copy
 import hashlib
+import os
 from datetime import timedelta
 from typing import TYPE_CHECKING, Any, Dict, List
 
@@ -9,11 +10,18 @@ import structlog
 from playwright.async_api import Frame, Page
 
 from skyvern.config import settings
-from skyvern.constants import SKYVERN_ID_ATTR
-from skyvern.exceptions import DisabledBlockExecutionError, StepUnableToExecuteError, TaskAlreadyTimeout
+from skyvern.constants import CUSTOMER_STORAGE_UPLOAD_MAX_BYTES, SKYVERN_ID_ATTR
+from skyvern.exceptions import (
+    AzureConfigurationError,
+    DisabledBlockExecutionError,
+    StepUnableToExecuteError,
+    TaskAlreadyTimeout,
+    UploadFileMaxSizeExceeded,
+)
 from skyvern.forge import app
 from skyvern.forge.async_operations import AsyncOperation
 from skyvern.forge.prompts import prompt_engine
+from skyvern.forge.sdk.api.aws import AsyncAWSClient
 from skyvern.forge.sdk.api.azure import AzureClientFactory
 from skyvern.forge.sdk.api.llm.exceptions import LLMProviderError
 from skyvern.forge.sdk.copilot.config import CopilotConfig
@@ -25,6 +33,7 @@ from skyvern.forge.sdk.schemas.tasks import Task, TaskStatus
 from skyvern.forge.sdk.services import google_oauth_service
 from skyvern.forge.sdk.trace import traced
 from skyvern.forge.sdk.workflow.models.block import BlockTypeVar
+from skyvern.schemas.workflows import FileStorageType, FileUploadDestination
 from skyvern.webeye.actions.actions import Action
 from skyvern.webeye.browser_state import BrowserState
 from skyvern.webeye.scraper.scraped_page import ELEMENT_NODE_ATTRIBUTES, CleanupElementTreeFunc, json_to_html
@@ -1070,6 +1079,66 @@ class AgentFunction:
                 timeout=httpx.Timeout(timeout_seconds),
             )
 
+    async def upload_file_to_customer_storage(
+        self,
+        file_path: str,
+        destination: FileUploadDestination,
+        organization_id: str | None = None,
+        run_id: str | None = None,
+    ) -> str:
+        """Upload a single file to customer-specified S3 or Azure storage.
+
+        Returns the customer-facing URI (``destination.customer_uri``).  The
+        cloud override routes NAT-org traffic through the egress proxy so it
+        egresses from a static IP; the OSS base path uploads directly via the
+        AWS / Azure SDK.
+
+        Enforces ``CUSTOMER_STORAGE_UPLOAD_MAX_BYTES`` (1 GB) regardless of
+        route so the proxy pod and the customer's quota are both protected
+        from runaway uploads.
+        """
+        await self._enforce_upload_size_cap(file_path)
+
+        if destination.storage_type == FileStorageType.S3:
+            aws_client = AsyncAWSClient(
+                aws_access_key_id=destination.aws_access_key_id,
+                aws_secret_access_key=destination.aws_secret_access_key,
+                region_name=destination.aws_region_name,
+            )
+            await aws_client.upload_file_from_path(
+                uri=destination.sdk_uri,
+                file_path=file_path,
+                raise_exception=True,
+            )
+            return destination.customer_uri
+
+        if destination.storage_type == FileStorageType.AZURE:
+            if not destination.azure_storage_account_name or not destination.azure_storage_account_key:
+                raise AzureConfigurationError("Azure Storage is not configured")
+            azure_client = app.AZURE_CLIENT_FACTORY.create_storage_client(
+                storage_account_name=destination.azure_storage_account_name,
+                storage_account_key=destination.azure_storage_account_key,
+            )
+            await azure_client.upload_file_from_path(destination.sdk_uri, file_path)
+            return destination.customer_uri
+
+        raise ValueError(f"Unsupported storage type: {destination.storage_type}")
+
+    @staticmethod
+    async def _enforce_upload_size_cap(file_path: str) -> None:
+        """Reject files larger than CUSTOMER_STORAGE_UPLOAD_MAX_BYTES.
+
+        Centralized so direct and proxied paths share the same cap. Stat the
+        file (fast, no IO of contents) and raise a typed exception so callers
+        translate it into a clean block failure.
+        """
+        try:
+            size = await asyncio.to_thread(os.path.getsize, file_path)
+        except FileNotFoundError:
+            raise
+        if size > CUSTOMER_STORAGE_UPLOAD_MAX_BYTES:
+            raise UploadFileMaxSizeExceeded(file_size_bytes=size, max_size_bytes=CUSTOMER_STORAGE_UPLOAD_MAX_BYTES)
+
     def get_copilot_security_rules(self) -> str:
         """Return security guardrails for the workflow copilot system prompt.
 

skyvern/forge/sdk/artifact/manager.py

@@ -53,6 +53,16 @@ def _safe_file_size_from_path(path: str | None) -> int | None:
         return None
 
 
+def _bundling_enabled() -> bool:
+    """Bundling and Skyvern-origin URLs are coupled: enabled together when HMAC signing is configured.
+
+    When unset, bundling is skipped at flush time so every artifact has its
+    own storage URI and can be served via a presigned URL — no Skyvern-origin
+    proxying, no HMAC, no API-key requirement.
+    """
+    return bool(settings.ARTIFACT_CONTENT_HMAC_KEYRING)
+
+
 @dataclass
 class ArtifactBatchData:
     """
@@ -1096,37 +1106,50 @@ class ArtifactManager:
             extra["artifact_type"] = artifact_type
         return f"{path}?{urlencode(extra)}" if extra else path
 
-    async def get_share_link(self, artifact: Artifact) -> str | None:
-        """Return a Skyvern-origin signed ``/v1/artifacts/{id}/content`` URL.
+    async def resolve_share_url(
+        self,
+        artifact: Artifact,
+        expiry_seconds: int | None = None,
+    ) -> str | None:
+        """Return the customer-facing URL for an artifact.
 
-        SKY-8861: every customer-visible artifact URL goes through the content
-        endpoint — short, our origin, per-org TTL, no S3/Azure presigned URLs
-        leaking into webhooks or API responses. Bundled and non-bundled
-        artifacts share the same path; the content endpoint already serves
-        either (``retrieve_artifact`` handles bundle extraction transparently).
+        Skyvern-origin signed URL when HMAC signing is configured *or* when the
+        artifact is a bundled member (its URI points at the parent ZIP, so a
+        storage presigned URL would download the wrong bytes — fall through to
+        the endpoint, which knows how to extract the member). Otherwise the
+        storage backend's presigned URL (S3 / Azure SAS / local URI).
+        """
+        if _bundling_enabled() or artifact.bundle_key:
+            return self._bundle_content_url(
+                artifact.artifact_id,
+                artifact_name=artifact.bundle_key,
+                artifact_type=artifact.artifact_type,
+                expiry_seconds=expiry_seconds,
+            )
+        return await app.STORAGE.get_share_link(artifact)
+
+    async def get_share_link(self, artifact: Artifact) -> str | None:
+        """Return a customer-facing URL for one artifact.
+
+        HMAC keyring set: Skyvern signed ``/v1/artifacts/{id}/content`` URL.
+        HMAC keyring unset: storage backend's presigned URL, except for
+        legacy bundled rows which still route through the endpoint.
         """
         expiry_seconds = await self.resolve_artifact_url_expiry_seconds(artifact.organization_id)
-        return self._bundle_content_url(
-            artifact.artifact_id,
-            artifact_name=artifact.bundle_key,
-            artifact_type=artifact.artifact_type,
-            expiry_seconds=expiry_seconds,
-        )
+        return await self.resolve_share_url(artifact, expiry_seconds=expiry_seconds)
 
     async def get_share_links(self, artifacts: list[Artifact]) -> list[str | None]:
-        """Return signed content URLs for a batch of artifacts."""
+        """Return URLs for a batch of artifacts."""
         return await self.get_share_links_with_bundle_support(artifacts)
 
     async def get_share_links_with_bundle_support(self, artifacts: list[Artifact]) -> list[str | None]:
-        """Mint a signed ``/v1/artifacts/{id}/content`` URL for every artifact.
+        """Mint a customer-facing URL for every artifact in the batch.
 
-        Bundled vs non-bundled used to branch here — bundled went through the
-        content endpoint, non-bundled fell through to ``STORAGE.get_share_links``
-        and leaked S3 presigned / Azure SAS URLs into webhooks and customer
-        API responses. SKY-8861: unify on the signed origin URL for both.
-
-        The per-org TTL is resolved once per batch (callers look up by
-        run/workflow scope so all artifacts share an org).
+        HMAC keyring set: every artifact → Skyvern signed URL.
+        HMAC keyring unset: non-bundled artifacts → one batched
+        ``STORAGE.get_share_links`` call (single backend round-trip). Bundled
+        legacy rows → individual Skyvern unsigned URL each (still requires
+        API-key auth on fetch; see safety-net rationale in `resolve_share_url`).
         """
         if not artifacts:
             return []
@@ -1134,23 +1157,46 @@ class ArtifactManager:
         organization_id = artifacts[0].organization_id
         expiry_seconds = await self.resolve_artifact_url_expiry_seconds(organization_id)
 
-        result: list[str | None] = [
-            self._bundle_content_url(
-                artifact.artifact_id,
-                artifact_name=artifact.bundle_key,
-                artifact_type=artifact.artifact_type,
+        if _bundling_enabled():
+            return [
+                self._bundle_content_url(
+                    artifact.artifact_id,
+                    artifact_name=artifact.bundle_key,
+                    artifact_type=artifact.artifact_type,
+                    expiry_seconds=expiry_seconds,
+                )
+                for artifact in artifacts
+            ]
+
+        bundled_indices = [i for i, a in enumerate(artifacts) if a.bundle_key]
+        non_bundled_indices = [i for i, a in enumerate(artifacts) if not a.bundle_key]
+        non_bundled = [artifacts[i] for i in non_bundled_indices]
+
+        presigned: list[str] | None = await app.STORAGE.get_share_links(non_bundled) if non_bundled else []
+
+        result: list[str | None] = [None] * len(artifacts)
+        if presigned is None:
+            for idx in non_bundled_indices:
+                result[idx] = None
+        else:
+            for idx, presigned_url in zip(non_bundled_indices, presigned, strict=True):
+                result[idx] = presigned_url
+        for idx in bundled_indices:
+            a = artifacts[idx]
+            result[idx] = self._bundle_content_url(
+                a.artifact_id,
+                artifact_name=a.bundle_key,
+                artifact_type=a.artifact_type,
                 expiry_seconds=expiry_seconds,
             )
-            for artifact in artifacts
-        ]
 
         LOG.debug(
             "get_share_links_with_bundle_support",
             total=len(artifacts),
-            bundled=sum(1 for a in artifacts if a.bundle_key),
-            non_bundled=sum(1 for a in artifacts if not a.bundle_key),
+            bundled=len(bundled_indices),
+            non_bundled=len(non_bundled_indices),
+            keyring_set=_bundling_enabled(),
         )
-
         return result
 
     async def mark_archived_artifacts(self, artifacts: list[Artifact]) -> None:
@@ -1378,24 +1424,52 @@ class ArtifactManager:
         return buf.getvalue()
 
     async def flush_step_archive(self, step_id: str) -> None:
-        """Build the ZIP, upload as one S3 object, and create DB rows for all member artifacts.
+        """Persist the step's accumulated artifacts.
 
-        Call this as soon as a step finishes executing (before moving on to the next step) to
-        release the in-memory buffer immediately rather than waiting until the end of the task.
-        Safe to call multiple times for the same step_id — subsequent calls are no-ops because
-        the accumulator is popped on the first flush.
+        Bundled mode (HMAC keyring set): one ZIP storage object, one parent
+        STEP_ARCHIVE row, N member rows with ``bundle_key``.
+        Unbundled mode (HMAC keyring unset): one storage object per member,
+        one ArtifactModel per member with no ``bundle_key``, no parent row.
+
+        Call this as soon as a step finishes executing so the in-memory buffer
+        is released. Safe to call multiple times — subsequent calls are
+        no-ops because the accumulator is popped on the first flush.
         """
         accumulator = self._step_archives.pop(step_id, None)
         if not accumulator or not accumulator.entries or not accumulator.member_types:
             return
 
-        step = accumulator.step
         LOG.debug(
             "Flushing step archive",
             step_id=step_id,
             artifact_types=[t.value for t, _, _ in accumulator.member_types],
             entry_count=len(accumulator.entries),
+            bundled=_bundling_enabled(),
         )
+
+        if _bundling_enabled():
+            await self._flush_step_archive_bundled(accumulator)
+        else:
+            await self._flush_step_archive_unbundled(accumulator)
+
+        for organization_id, action_id, artifact_id in accumulator.pending_action_screenshot_updates:
+            try:
+                await app.DATABASE.artifacts.update_action_screenshot_artifact_id(
+                    organization_id=organization_id,
+                    action_id=action_id,
+                    screenshot_artifact_id=artifact_id,
+                )
+            except Exception:
+                LOG.warning(
+                    "Failed to update action with screenshot artifact id after archive flush",
+                    action_id=action_id,
+                    artifact_id=artifact_id,
+                    exc_info=True,
+                )
+
+    async def _flush_step_archive_bundled(self, accumulator: StepArchiveAccumulator) -> None:
+        """Bundle all accumulated entries into a single ZIP, one storage PUT, parent + member rows."""
+        step = accumulator.step
         archive_artifact_id = generate_artifact_id()
         archive_uri = app.STORAGE.build_uri(
             organization_id=step.organization_id,
@@ -1452,21 +1526,52 @@ class ArtifactManager:
         ]
         await app.DATABASE.artifacts.bulk_create_artifacts([parent_model, *member_models])
 
-        # Apply deferred action.screenshot_artifact_id updates now that artifact rows exist.
-        for organization_id, action_id, artifact_id in accumulator.pending_action_screenshot_updates:
-            try:
-                await app.DATABASE.artifacts.update_action_screenshot_artifact_id(
-                    organization_id=organization_id,
-                    action_id=action_id,
-                    screenshot_artifact_id=artifact_id,
-                )
-            except Exception:
-                LOG.warning(
-                    "Failed to update action with screenshot artifact id after archive flush",
-                    action_id=action_id,
+    async def _flush_step_archive_unbundled(self, accumulator: StepArchiveAccumulator) -> None:
+        """Persist each accumulated entry as its own storage object + ArtifactModel row.
+
+        No STEP_ARCHIVE parent — the ZIP doesn't exist in this mode. Storage
+        PUTs are sequenced; the member count per step is small (typically < 20).
+        """
+        step = accumulator.step
+        now = datetime.now(UTC)
+        member_models: list[ArtifactModel] = []
+        for artifact_type, filename, artifact_id in accumulator.member_types:
+            data = accumulator.entries[filename]
+            uri = app.STORAGE.build_uri(
+                organization_id=step.organization_id,
+                artifact_id=artifact_id,
+                step=step,
+                artifact_type=artifact_type,
+            )
+            artifact = Artifact(
+                artifact_id=artifact_id,
+                artifact_type=artifact_type,
+                uri=uri,
+                organization_id=step.organization_id,
+                step_id=step.step_id,
+                task_id=step.task_id,
+                workflow_run_id=accumulator.workflow_run_id,
+                workflow_run_block_id=accumulator.workflow_run_block_id,
+                run_id=accumulator.run_id,
+                created_at=now,
+                modified_at=now,
+            )
+            await app.STORAGE.store_artifact(artifact, data)
+            member_models.append(
+                self._build_artifact_model(
                     artifact_id=artifact_id,
-                    exc_info=True,
+                    artifact_type=artifact_type,
+                    uri=uri,
+                    file_size=len(data),
+                    organization_id=step.organization_id,
+                    step_id=step.step_id,
+                    task_id=step.task_id,
+                    workflow_run_id=accumulator.workflow_run_id,
+                    workflow_run_block_id=accumulator.workflow_run_block_id,
+                    run_id=accumulator.run_id,
                 )
+            )
+        await app.DATABASE.artifacts.bulk_create_artifacts(member_models)
 
     async def create_task_archive(
         self,

skyvern/forge/sdk/artifact/storage/base.py

@@ -30,15 +30,13 @@ async def _file_infos_from_artifacts(artifacts: list[Artifact], *, artifact_type
         return []
     organization_id = artifacts[0].organization_id
     expiry_seconds = await app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds(organization_id)
+    _ = artifact_type  # kept for call-site compatibility; URL hint now sourced from the artifact row.
     infos: list[FileInfo] = []
     for artifact in artifacts:
         filename = artifact.uri.rsplit("/", 1)[-1] if artifact.uri else ""
-        url = app.ARTIFACT_MANAGER.build_signed_content_url(
-            artifact_id=artifact.artifact_id,
-            artifact_name=filename,
-            artifact_type=artifact_type.value,
-            expiry_seconds=expiry_seconds,
-        )
+        url = await app.ARTIFACT_MANAGER.resolve_share_url(artifact, expiry_seconds=expiry_seconds)
+        if url is None:
+            continue
         infos.append(
             FileInfo(
                 url=url,

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

@@ -123,7 +123,14 @@ from skyvern.forge.sdk.workflow.models.parameter import (
     WorkflowParameter,
 )
 from skyvern.schemas.runs import RunEngine
-from skyvern.schemas.workflows import BlockResult, BlockStatus, BlockType, FileStorageType, FileType
+from skyvern.schemas.workflows import (
+    BlockResult,
+    BlockStatus,
+    BlockType,
+    FileStorageType,
+    FileType,
+    FileUploadDestination,
+)
 from skyvern.services.error_detection_service import detect_user_defined_errors_for_task
 from skyvern.utils.strings import generate_random_string
 from skyvern.utils.templating import get_missing_variables
@@ -4187,6 +4194,49 @@ class FileUploadBlock(Block):
     def _get_azure_blob_uri(self, workflow_run_id: str, blob_name: str) -> str:
         return f"https://{self.azure_storage_account_name}.blob.core.windows.net/{self.azure_blob_container_name}/{blob_name}"
 
+    def _build_s3_destination(
+        self,
+        workflow_run_id: str,
+        file_path: str,
+        aws_access_key_id: str | None,
+        aws_secret_access_key: str | None,
+    ) -> FileUploadDestination:
+        s3_uri = self._get_s3_uri(workflow_run_id, file_path)
+        # ``_get_s3_uri`` returns ``s3://{bucket}/{key}`` — split it back out for
+        # the destination so the cloud override can compute a presigned URL.
+        without_scheme = s3_uri[len("s3://") :]
+        bucket, _, key = without_scheme.partition("/")
+        return FileUploadDestination(
+            storage_type=FileStorageType.S3,
+            customer_uri=s3_uri,
+            sdk_uri=s3_uri,
+            s3_bucket=bucket,
+            s3_key=key,
+            aws_access_key_id=aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            aws_region_name=self.region_name,
+        )
+
+    def _build_azure_destination(
+        self,
+        workflow_run_id: str,
+        file_path: str,
+        azure_storage_account_name: str,
+        azure_storage_account_key: str,
+    ) -> FileUploadDestination:
+        blob_name = self._get_azure_blob_name(workflow_run_id, file_path)
+        customer_uri = self._get_azure_blob_uri(workflow_run_id, blob_name)
+        sdk_uri = f"azure://{self.azure_blob_container_name or ''}/{blob_name}"
+        return FileUploadDestination(
+            storage_type=FileStorageType.AZURE,
+            customer_uri=customer_uri,
+            sdk_uri=sdk_uri,
+            azure_storage_account_name=azure_storage_account_name,
+            azure_storage_account_key=azure_storage_account_key,
+            azure_blob_container_name=self.azure_blob_container_name,
+            azure_blob_name=blob_name,
+        )
+
     async def execute(
         self,
         workflow_run_id: str,
@@ -4283,15 +4333,20 @@ class FileUploadBlock(Block):
                     workflow_run_context.get_original_secret_value_or_none(self.aws_secret_access_key)
                     or self.aws_secret_access_key
                 )
-                aws_client = AsyncAWSClient(
-                    aws_access_key_id=actual_aws_access_key_id,
-                    aws_secret_access_key=actual_aws_secret_access_key,
-                    region_name=self.region_name,
-                )
                 for file_path in files_to_upload:
-                    s3_uri = self._get_s3_uri(workflow_run_id, file_path)
-                    uploaded_uris.append(s3_uri)
-                    await aws_client.upload_file_from_path(uri=s3_uri, file_path=file_path, raise_exception=True)
+                    destination = self._build_s3_destination(
+                        workflow_run_id=workflow_run_id,
+                        file_path=file_path,
+                        aws_access_key_id=actual_aws_access_key_id,
+                        aws_secret_access_key=actual_aws_secret_access_key,
+                    )
+                    customer_uri = await app.AGENT_FUNCTION.upload_file_to_customer_storage(
+                        file_path=file_path,
+                        destination=destination,
+                        organization_id=organization_id,
+                        run_id=workflow_run_id,
+                    )
+                    uploaded_uris.append(customer_uri)
                 LOG.info("FileUploadBlock File(s) uploaded to S3", file_path=self.path)
             elif self.storage_type == FileStorageType.AZURE:
                 actual_azure_storage_account_name = (
@@ -4305,17 +4360,21 @@ class FileUploadBlock(Block):
                 if actual_azure_storage_account_name is None or actual_azure_storage_account_key is None:
                     raise AzureConfigurationError("Azure Storage is not configured")
 
-                azure_client = app.AZURE_CLIENT_FACTORY.create_storage_client(
-                    storage_account_name=actual_azure_storage_account_name,
-                    storage_account_key=actual_azure_storage_account_key,
-                )
                 for file_path in files_to_upload:
                     LOG.info("FileUploadBlock Uploading file to Azure Blob Storage", file_path=file_path)
-                    blob_name = self._get_azure_blob_name(workflow_run_id, file_path)
-                    azure_uri = self._get_azure_blob_uri(workflow_run_id, blob_name)
-                    uploaded_uris.append(azure_uri)
-                    uri = f"azure://{self.azure_blob_container_name or ''}/{blob_name}"
-                    await azure_client.upload_file_from_path(uri, file_path)
+                    destination = self._build_azure_destination(
+                        workflow_run_id=workflow_run_id,
+                        file_path=file_path,
+                        azure_storage_account_name=actual_azure_storage_account_name,
+                        azure_storage_account_key=actual_azure_storage_account_key,
+                    )
+                    customer_uri = await app.AGENT_FUNCTION.upload_file_to_customer_storage(
+                        file_path=file_path,
+                        destination=destination,
+                        organization_id=organization_id,
+                        run_id=workflow_run_id,
+                    )
+                    uploaded_uris.append(customer_uri)
                 LOG.info("FileUploadBlock File(s) uploaded to Azure Blob Storage", file_path=self.path)
             else:
                 # This case should ideally be caught by the initial validation

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

@@ -341,6 +341,7 @@ class WorkflowRunResponseBase(BaseModel):
     browser_address: str | None = None
     run_with: str = "agent"
     script_run: ScriptRunResponse | None = None
+    script_id: str | None = None
     errors: list[dict[str, Any]] | None = None
 
     @field_validator("run_with", mode="before")

skyvern/forge/sdk/workflow/service.py

@@ -742,22 +742,14 @@ class WorkflowService:
             )
             download_id_set = set(download_ids)
             for artifact in artifacts:
-                url = app.ARTIFACT_MANAGER.build_signed_content_url(
-                    artifact_id=artifact.artifact_id,
-                    artifact_name=artifact.bundle_key,
-                    artifact_type=artifact.artifact_type,
-                    expiry_seconds=expiry_seconds,
-                )
+                url = await app.ARTIFACT_MANAGER.resolve_share_url(artifact, expiry_seconds=expiry_seconds)
+                if url is None:
+                    continue
                 url_map[artifact.artifact_id] = url
                 if artifact.artifact_id in download_id_set:
                     filename = artifact.uri.rsplit("/", 1)[-1] if artifact.uri else ""
                     fileinfo_map[artifact.artifact_id] = FileInfo(
-                        url=app.ARTIFACT_MANAGER.build_signed_content_url(
-                            artifact_id=artifact.artifact_id,
-                            artifact_name=filename,
-                            artifact_type=artifact.artifact_type,
-                            expiry_seconds=expiry_seconds,
-                        ),
+                        url=url,
                         checksum=artifact.checksum,
                         filename=filename,
                         file_size=artifact.file_size,
@@ -5328,6 +5320,7 @@ class WorkflowService:
             browser_address=workflow_run.browser_address,
             run_with=workflow_run.run_with,
             script_run=workflow_run.script_run,
+            script_id=workflow_run.script_run.script_id if workflow_run.script_run else None,
             errors=errors,
         )
 

skyvern/schemas/runs.py

@@ -432,6 +432,10 @@ class WorkflowRunResponse(BaseRunResponse):
         default=None,
         description="Whether to fallback to AI if code run fails.",
     )
+    script_id: str | None = Field(
+        default=None,
+        description="ID of the cached script used for this workflow run, if any.",
+    )
     run_request: WorkflowRunRequest | None = Field(
         default=None, description="The original request parameters used to start this workflow run"
     )

skyvern/schemas/workflows.py

@@ -485,6 +485,30 @@ class FileStorageType(StrEnum):
     AZURE = "azure"
 
 
+class FileUploadDestination(BaseModel):
+    """Customer-storage destination for a single file upload.
+
+    Used by ``AgentFunction.upload_file_to_customer_storage``. The cloud
+    override inspects this to decide whether to compute a presigned/SAS URL
+    and route through the NAT egress proxy or upload directly via the SDK.
+    """
+
+    storage_type: FileStorageType
+    customer_uri: str
+    sdk_uri: str
+
+    s3_bucket: str | None = None
+    s3_key: str | None = None
+    aws_access_key_id: str | None = None
+    aws_secret_access_key: str | None = None
+    aws_region_name: str | None = None
+
+    azure_storage_account_name: str | None = None
+    azure_storage_account_key: str | None = None
+    azure_blob_container_name: str | None = None
+    azure_blob_name: str | None = None
+
+
 class ParameterYAML(BaseModel, abc.ABC):
     parameter_type: ParameterType
     key: str

skyvern/services/workflow_service.py

@@ -157,6 +157,7 @@ async def get_workflow_run_response(
         browser_profile_id=workflow_run.browser_profile_id,
         max_screenshot_scrolls=workflow_run.max_screenshot_scrolls,
         script_run=workflow_run.script_run,
+        script_id=workflow_run.script_run.script_id if workflow_run.script_run else None,
         run_request=WorkflowRunRequest(
             workflow_id=workflow_run.workflow_permanent_id,
             title=workflow_run_resp.workflow_title,

tests/unit/forge/test_agent_function_upload_file.py

@@ -0,0 +1,137 @@
+"""Cover the OSS ``AgentFunction.upload_file_to_customer_storage`` base path.
+
+The cloud override (NAT proxy routing) is tested in
+``tests/cloud/test_nat_egress_proxy_uploads.py``. This file pins down the
+direct (SDK) path: S3 via aioboto3 and Azure via the AZURE_CLIENT_FACTORY,
+plus the shared 1 GB size cap.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from skyvern.constants import CUSTOMER_STORAGE_UPLOAD_MAX_BYTES
+from skyvern.exceptions import AzureConfigurationError, UploadFileMaxSizeExceeded
+from skyvern.forge.agent_functions import AgentFunction
+from skyvern.schemas.workflows import FileStorageType, FileUploadDestination
+
+
+def _s3_destination(bucket: str = "customer-bucket", key: str = "k.bin") -> FileUploadDestination:
+    return FileUploadDestination(
+        storage_type=FileStorageType.S3,
+        customer_uri=f"s3://{bucket}/{key}",
+        sdk_uri=f"s3://{bucket}/{key}",
+        s3_bucket=bucket,
+        s3_key=key,
+        aws_access_key_id="AKIA-test",
+        aws_secret_access_key="secret-test",
+    )
+
+
+def _azure_destination() -> FileUploadDestination:
+    return FileUploadDestination(
+        storage_type=FileStorageType.AZURE,
+        customer_uri="https://acc.blob.core.windows.net/c/blob",
+        sdk_uri="azure://c/blob",
+        azure_storage_account_name="acc",
+        azure_storage_account_key="key",
+        azure_blob_container_name="c",
+        azure_blob_name="blob",
+    )
+
+
+@pytest.fixture
+def small_file(tmp_path: Path) -> Path:
+    fp = tmp_path / "f.bin"
+    fp.write_bytes(b"abc")
+    return fp
+
+
+@pytest.mark.asyncio
+async def test_s3_direct_path_calls_async_aws_client(small_file: Path) -> None:
+    destination = _s3_destination()
+    fake_aws = AsyncMock()
+    fake_aws.upload_file_from_path = AsyncMock()
+
+    with patch("skyvern.forge.agent_functions.AsyncAWSClient", return_value=fake_aws) as MockClient:
+        result = await AgentFunction().upload_file_to_customer_storage(
+            file_path=str(small_file),
+            destination=destination,
+            organization_id="o_1",
+        )
+
+    assert result == destination.customer_uri
+    MockClient.assert_called_once_with(
+        aws_access_key_id="AKIA-test",
+        aws_secret_access_key="secret-test",
+        region_name=None,
+    )
+    fake_aws.upload_file_from_path.assert_awaited_once_with(
+        uri=destination.sdk_uri,
+        file_path=str(small_file),
+        raise_exception=True,
+    )
+
+
+@pytest.mark.asyncio
+async def test_azure_direct_path_calls_factory(small_file: Path) -> None:
+    destination = _azure_destination()
+    fake_azure = AsyncMock()
+    fake_azure.upload_file_from_path = AsyncMock()
+
+    fake_factory = MagicMock()
+    fake_factory.create_storage_client = MagicMock(return_value=fake_azure)
+
+    with patch("skyvern.forge.agent_functions.app") as mock_app:
+        mock_app.AZURE_CLIENT_FACTORY = fake_factory
+
+        result = await AgentFunction().upload_file_to_customer_storage(
+            file_path=str(small_file),
+            destination=destination,
+        )
+
+    assert result == destination.customer_uri
+    fake_factory.create_storage_client.assert_called_once_with(
+        storage_account_name="acc",
+        storage_account_key="key",
+    )
+    fake_azure.upload_file_from_path.assert_awaited_once_with(destination.sdk_uri, str(small_file))
+
+
+@pytest.mark.asyncio
+async def test_azure_missing_creds_raises(small_file: Path) -> None:
+    destination = FileUploadDestination(
+        storage_type=FileStorageType.AZURE,
+        customer_uri="https://acc.blob.core.windows.net/c/blob",
+        sdk_uri="azure://c/blob",
+        azure_storage_account_name=None,
+        azure_storage_account_key=None,
+        azure_blob_container_name="c",
+        azure_blob_name="blob",
+    )
+    with pytest.raises(AzureConfigurationError):
+        await AgentFunction().upload_file_to_customer_storage(
+            file_path=str(small_file),
+            destination=destination,
+        )
+
+
+@pytest.mark.asyncio
+async def test_size_cap_enforced_on_direct_path(small_file: Path) -> None:
+    destination = _s3_destination()
+
+    with patch(
+        "skyvern.forge.agent_functions.os.path.getsize",
+        return_value=CUSTOMER_STORAGE_UPLOAD_MAX_BYTES + 1,
+    ):
+        fake_aws = AsyncMock()
+        with patch("skyvern.forge.agent_functions.AsyncAWSClient", return_value=fake_aws):
+            with pytest.raises(UploadFileMaxSizeExceeded):
+                await AgentFunction().upload_file_to_customer_storage(
+                    file_path=str(small_file),
+                    destination=destination,
+                )
+        fake_aws.upload_file_from_path.assert_not_awaited()

tests/unit/services/test_get_workflow_run_response_passthrough.py

@@ -65,6 +65,7 @@ async def test_get_workflow_run_response_passes_through_all_fields() -> None:
 
     assert resp is not None
     assert resp.script_run == script_run
+    assert resp.script_id == "s_abc"
     assert resp.ai_fallback is True
     assert resp.browser_session_id == "pbs_123"
     assert resp.max_screenshot_scrolls == 5

tests/unit/test_artifact_url_expiry_org_config.py

@@ -12,6 +12,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from skyvern.config import settings
 from skyvern.forge.sdk.artifact.manager import ArtifactManager
 from skyvern.forge.sdk.artifact.signing import (
     ARTIFACT_URL_EXPIRY_SECONDS,
@@ -21,6 +22,8 @@ from skyvern.forge.sdk.artifact.signing import (
 from skyvern.forge.sdk.routes.agent_protocol import _artifact_content_response_headers
 from skyvern.forge.sdk.schemas.organizations import Organization
 
+_DUMMY_KEYRING_JSON = '{"current_kid":"k1","keys":{"k1":{"secret":"deadbeef"}}}'
+
 
 def _make_org(artifact_url_expiry_seconds: int | None) -> Organization:
     now = datetime.now(timezone.utc)
@@ -306,6 +309,7 @@ class TestGetShareLinkAlwaysUsesSignedContentUrl:
 
         resolve = AsyncMock(return_value=12 * 3600)
         with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
             patch.object(manager, "resolve_artifact_url_expiry_seconds", resolve),
             patch.object(
                 manager, "_bundle_content_url", return_value="https://api/v1/artifacts/a_42/content?sig=x"
@@ -329,6 +333,7 @@ class TestGetShareLinkAlwaysUsesSignedContentUrl:
 
         resolve = AsyncMock(return_value=12 * 3600)
         with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
             patch.object(manager, "resolve_artifact_url_expiry_seconds", resolve),
             patch.object(
                 manager, "_bundle_content_url", return_value="https://api/v1/artifacts/a_b/content?sig=x"
@@ -360,6 +365,7 @@ class TestGetShareLinksWithBundleSupportAlwaysUsesSignedContentUrl:
 
         resolve = AsyncMock(return_value=2 * 3600)
         with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
             patch.object(manager, "resolve_artifact_url_expiry_seconds", resolve),
             patch.object(
                 manager,
@@ -390,6 +396,7 @@ class TestGetShareLinksWithBundleSupportAlwaysUsesSignedContentUrl:
 
         resolve = AsyncMock(return_value=3 * 3600)
         with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
             patch.object(manager, "resolve_artifact_url_expiry_seconds", resolve),
             patch.object(manager, "_bundle_content_url", return_value="https://x") as bundle,
             patch("skyvern.forge.sdk.artifact.manager.app") as app,

tests/unit/test_block_output_download_refresh.py

@@ -13,7 +13,7 @@ IDs on every workflow-run-status response.
 
 from __future__ import annotations
 
-from unittest.mock import AsyncMock, MagicMock, Mock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -139,8 +139,8 @@ async def test_refresh_rebuilds_downloaded_files_from_artifact_ids():
             new=AsyncMock(return_value=3600),
         ),
         patch(
-            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.build_signed_content_url",
-            new=Mock(return_value=fresh_url),
+            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.resolve_share_url",
+            new=AsyncMock(return_value=fresh_url),
         ),
     ):
         service = WorkflowService()
@@ -211,9 +211,9 @@ async def test_refresh_issues_one_batch_db_call_for_n_blocks():
             new=AsyncMock(return_value=3600),
         ),
         patch(
-            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.build_signed_content_url",
-            new=Mock(
-                side_effect=lambda artifact_id, **_: f"https://api.skyvern.com/v1/artifacts/{artifact_id}/content"
+            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.resolve_share_url",
+            new=AsyncMock(
+                side_effect=lambda artifact, **_: f"https://api.skyvern.com/v1/artifacts/{artifact.artifact_id}/content"
             ),
         ),
     ):
@@ -254,9 +254,9 @@ async def test_refresh_substitutes_screenshot_urls_from_map():
             new=AsyncMock(return_value=3600),
         ),
         patch(
-            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.build_signed_content_url",
-            new=Mock(
-                side_effect=lambda artifact_id, **_: f"https://api.skyvern.com/v1/artifacts/{artifact_id}/content"
+            "skyvern.forge.sdk.workflow.service.app.ARTIFACT_MANAGER.resolve_share_url",
+            new=AsyncMock(
+                side_effect=lambda artifact, **_: f"https://api.skyvern.com/v1/artifacts/{artifact.artifact_id}/content"
             ),
         ),
     ):

tests/unit/test_browser_session_download_artifacts.py

@@ -308,12 +308,12 @@ async def test_get_shared_downloaded_files_in_browser_session_uses_artifact_urls
         file_size=2048,
     )
     mock_list = AsyncMock(return_value=[artifact])
-    build_url = MagicMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
+    resolve_url = AsyncMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
 
     with patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app:
         with patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app:
             s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-            base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+            base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
             base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
             result = await storage.get_shared_downloaded_files_in_browser_session(
                 organization_id="o_1", browser_session_id="pbs_1"
@@ -346,12 +346,12 @@ async def test_get_shared_downloaded_files_in_browser_session_falls_back_to_pres
     )
 
     mock_list = AsyncMock(return_value=[])
-    build_url = MagicMock()  # must NOT be called
+    resolve_url = AsyncMock()  # must NOT be called
 
     with patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app:
         with patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app:
             s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-            base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+            base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
             result = await storage.get_shared_downloaded_files_in_browser_session(
                 organization_id="o_1", browser_session_id="pbs_old"
             )
@@ -359,7 +359,7 @@ async def test_get_shared_downloaded_files_in_browser_session_falls_back_to_pres
     assert len(result) == 1
     assert _is_amazonaws_s3_url(result[0].url)
     assert result[0].file_size == 1024
-    build_url.assert_not_called()
+    resolve_url.assert_not_awaited()
 
 
 @pytest.mark.asyncio
@@ -380,14 +380,14 @@ async def test_get_shared_downloaded_files_in_browser_session_filters_partial_ar
         "s3://skyvern-artifacts/v1/local/o_1/browser_sessions/pbs_1/downloads/inflight.pdf.crdownload",
     )
     mock_list = AsyncMock(return_value=[partial, completed])
-    build_url = MagicMock(return_value="https://api.skyvern.com/v1/artifacts/a_done/content?expiry=x&kid=y&sig=z")
+    resolve_url = AsyncMock(return_value="https://api.skyvern.com/v1/artifacts/a_done/content?expiry=x&kid=y&sig=z")
 
     with (
         patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app,
         patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app,
     ):
         s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-        base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+        base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
         base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
         result = await storage.get_shared_downloaded_files_in_browser_session(
             organization_id="o_1", browser_session_id="pbs_1"

tests/unit/test_browser_session_recording_artifacts.py

@@ -236,14 +236,14 @@ async def test_get_shared_recordings_returns_short_signed_urls(keyring_configure
         file_size=16384,
     )
     mock_list = AsyncMock(return_value=[artifact])
-    build_url = MagicMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
+    resolve_url = AsyncMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
 
     with (
         patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app,
         patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app,
     ):
         s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-        base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+        base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
         base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
         result = await storage.get_shared_recordings_in_browser_session(
             organization_id="o_1", browser_session_id="pbs_1"
@@ -267,14 +267,14 @@ async def test_get_shared_recordings_filters_unsupported_extensions(keyring_conf
     good = _make_recording_artifact("a_good", "s3://b/v1/.../videos/2026-04-26/ok.webm")
     bad = _make_recording_artifact("a_bad", "s3://b/v1/.../videos/2026-04-26/sneaky.exe")
     mock_list = AsyncMock(return_value=[good, bad])
-    build_url = MagicMock(return_value="https://api.skyvern.com/v1/artifacts/a_good/content")
+    resolve_url = AsyncMock(return_value="https://api.skyvern.com/v1/artifacts/a_good/content")
 
     with (
         patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app,
         patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app,
     ):
         s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-        base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+        base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
         base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
         result = await storage.get_shared_recordings_in_browser_session(
             organization_id="o_1", browser_session_id="pbs_1"
@@ -296,8 +296,8 @@ async def test_get_shared_recordings_sorts_newest_first(keyring_configured):
         "a_new", "s3://b/v1/.../videos/2026-04-26/newer.webm", created_at="2026-04-26T00:00:00Z"
     )
     mock_list = AsyncMock(return_value=[older, newer])
-    build_url = MagicMock(
-        side_effect=lambda artifact_id, **_: f"https://api.skyvern.com/v1/artifacts/{artifact_id}/content"
+    resolve_url = AsyncMock(
+        side_effect=lambda artifact, **_: f"https://api.skyvern.com/v1/artifacts/{artifact.artifact_id}/content"
     )
 
     with (
@@ -305,7 +305,7 @@ async def test_get_shared_recordings_sorts_newest_first(keyring_configured):
         patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app,
     ):
         s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-        base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+        base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
         base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
         result = await storage.get_shared_recordings_in_browser_session(
             organization_id="o_1", browser_session_id="pbs_1"
@@ -334,14 +334,14 @@ async def test_get_shared_recordings_falls_back_to_presigned_for_legacy_session(
     )
 
     mock_list = AsyncMock(return_value=[])  # no rows
-    build_url = MagicMock()  # must NOT be called
+    resolve_url = AsyncMock()  # must NOT be called
 
     with (
         patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app,
         patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app,
     ):
         s3_app.DATABASE.artifacts.list_artifacts_for_browser_session_by_type = mock_list
-        base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+        base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
         base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
         result = await storage.get_shared_recordings_in_browser_session(
             organization_id="o_1", browser_session_id="pbs_old"
@@ -350,7 +350,7 @@ async def test_get_shared_recordings_falls_back_to_presigned_for_legacy_session(
     assert len(result) == 1
     assert _is_amazonaws_s3_url(result[0].url)
     assert result[0].file_size == 1024
-    build_url.assert_not_called()
+    resolve_url.assert_not_awaited()
 
 
 @pytest.mark.asyncio

tests/unit/test_cdp_connect_headers_migration.py

@@ -1,199 +0,0 @@
-import importlib.util
-from pathlib import Path
-
-import pytest
-from sqlalchemy.exc import DBAPIError
-
-MIGRATION_PATH = (
-    Path(__file__).resolve().parents[2] / "alembic/versions/2026_05_22_2059-0b0bd1875c6e_add_cdp_connect_headers.py"
-)
-
-
-class _OrigLockError(Exception):
-    sqlstate = "55P03"
-
-
-class _OrigPgcodeLockError(Exception):
-    pgcode = "55P03"
-
-
-class _FakeOp:
-    def __init__(
-        self, lock_failures: int = 0, fail_rollback: bool = False, lock_error: Exception | None = None
-    ) -> None:
-        self.lock_failures = lock_failures
-        self.fail_rollback = fail_rollback
-        self.lock_error = lock_error or _OrigLockError()
-        self.statements: list[str] = []
-
-    def execute(self, statement: str) -> None:
-        self.statements.append(statement)
-        if statement == "ROLLBACK" and self.fail_rollback:
-            raise RuntimeError("rollback failed")
-        if statement.startswith("LOCK TABLE") and self.lock_failures:
-            self.lock_failures -= 1
-            raise DBAPIError(statement, None, self.lock_error)
-
-
-def _load_migration_module():
-    spec = importlib.util.spec_from_file_location("cdp_connect_headers_migration", MIGRATION_PATH)
-    assert spec is not None
-    assert spec.loader is not None
-    module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
-
-
-def test_retry_uses_bounded_lock_wait_and_rolls_back_failed_attempts(monkeypatch: pytest.MonkeyPatch) -> None:
-    migration = _load_migration_module()
-    fake_op = _FakeOp(lock_failures=2)
-    sleeps: list[int] = []
-
-    monkeypatch.setattr(migration, "op", fake_op)
-    monkeypatch.setattr(migration.time, "sleep", sleeps.append)
-    monkeypatch.setattr(migration.time, "monotonic", lambda: 0.0)
-    monkeypatch.setattr(migration.random, "random", lambda: 0.5)
-
-    migration._execute_with_retry(
-        "tasks",
-        'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-        deadline=100.0,
-    )
-
-    assert sleeps == [0.625, 0.625]
-    assert fake_op.statements == [
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        "ROLLBACK",
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        "ROLLBACK",
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-        "COMMIT",
-    ]
-
-
-def test_retry_stops_when_deadline_is_exhausted(monkeypatch: pytest.MonkeyPatch) -> None:
-    migration = _load_migration_module()
-    fake_op = _FakeOp(lock_failures=1)
-    sleeps: list[int] = []
-
-    monkeypatch.setattr(migration, "op", fake_op)
-    monkeypatch.setattr(migration.time, "sleep", sleeps.append)
-    monkeypatch.setattr(migration.time, "monotonic", lambda: 101.0)
-    monkeypatch.setattr(migration.random, "random", lambda: 0.5)
-
-    with pytest.raises(DBAPIError):
-        migration._execute_with_retry(
-            "observer_cruises",
-            'ALTER TABLE "observer_cruises" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-            deadline=100.0,
-        )
-
-    assert sleeps == []
-    assert fake_op.statements == [
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "observer_cruises" IN ACCESS EXCLUSIVE MODE',
-        "ROLLBACK",
-    ]
-
-
-def test_retry_recognizes_pgcode_lock_errors(monkeypatch: pytest.MonkeyPatch) -> None:
-    migration = _load_migration_module()
-    fake_op = _FakeOp(lock_failures=1, lock_error=_OrigPgcodeLockError())
-    sleeps: list[int] = []
-
-    monkeypatch.setattr(migration, "op", fake_op)
-    monkeypatch.setattr(migration.time, "sleep", sleeps.append)
-    monkeypatch.setattr(migration.time, "monotonic", lambda: 0.0)
-    monkeypatch.setattr(migration.random, "random", lambda: 1.0)
-
-    migration._execute_with_retry(
-        "tasks",
-        'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-        deadline=100.0,
-    )
-
-    assert sleeps == [1.0]
-    assert fake_op.statements == [
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        "ROLLBACK",
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-        "COMMIT",
-    ]
-
-
-def test_retry_aborts_when_rollback_fails(monkeypatch: pytest.MonkeyPatch) -> None:
-    migration = _load_migration_module()
-    fake_op = _FakeOp(lock_failures=1, fail_rollback=True)
-    sleeps: list[int] = []
-
-    monkeypatch.setattr(migration, "op", fake_op)
-    monkeypatch.setattr(migration.time, "sleep", sleeps.append)
-    monkeypatch.setattr(migration.time, "monotonic", lambda: 0.0)
-    monkeypatch.setattr(migration.random, "random", lambda: 0.5)
-
-    with pytest.raises(RuntimeError, match="rollback failed"):
-        migration._execute_with_retry(
-            "tasks",
-            'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-            deadline=100.0,
-        )
-
-    assert sleeps == []
-    assert fake_op.statements == [
-        "BEGIN",
-        "SET LOCAL statement_timeout = '5s'",
-        "SET LOCAL lock_timeout = '250ms'",
-        'LOCK TABLE "tasks" IN ACCESS EXCLUSIVE MODE',
-        "ROLLBACK",
-    ]
-
-
-def test_retry_budget_is_twenty_minutes() -> None:
-    migration = _load_migration_module()
-
-    assert migration._MIGRATION_RETRY_SECONDS == 20 * 60
-
-
-def test_lock_wait_budget_is_bounded_to_250ms() -> None:
-    migration = _load_migration_module()
-
-    assert migration._LOCK_TIMEOUT == "250ms"
-
-
-def test_retry_backoff_samples_sub_second_jitter(monkeypatch: pytest.MonkeyPatch) -> None:
-    migration = _load_migration_module()
-    fake_op = _FakeOp(lock_failures=2)
-    sleeps: list[float] = []
-    draws = iter([0.0, 1.0])
-
-    monkeypatch.setattr(migration, "op", fake_op)
-    monkeypatch.setattr(migration.time, "sleep", sleeps.append)
-    monkeypatch.setattr(migration.time, "monotonic", lambda: 0.0)
-    monkeypatch.setattr(migration.random, "random", lambda: next(draws))
-
-    migration._execute_with_retry(
-        "tasks",
-        'ALTER TABLE "tasks" ADD COLUMN IF NOT EXISTS cdp_connect_headers JSON',
-        deadline=100.0,
-    )
-
-    assert sleeps == [0.25, 1.0]

tests/unit/test_downloaded_files_artifact_urls.py

@@ -212,12 +212,12 @@ async def test_get_downloaded_files_uses_artifact_urls_when_rows_exist(keyring_c
         file_size=4096,
     )
     mock_list = AsyncMock(return_value=[artifact])
-    build_url = MagicMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
+    resolve_url = AsyncMock(return_value="https://api.skyvern.com/v1/artifacts/a_42/content?expiry=x&kid=y&sig=z")
 
     with patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app:
         with patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app:
             s3_app.DATABASE.artifacts.list_artifacts_for_run_by_type = mock_list
-            base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+            base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
             base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
             result = await storage.get_downloaded_files(organization_id="o_1", run_id="wr_1")
 
@@ -251,14 +251,14 @@ async def test_get_downloaded_files_preserves_artifact_row_order(keyring_configu
         created_at="2026-04-23T00:01:00Z",
     )
     mock_list = AsyncMock(return_value=[first, second])
-    build_url = MagicMock(
-        side_effect=lambda artifact_id, **_: f"https://api.skyvern.com/v1/artifacts/{artifact_id}/content"
+    resolve_url = AsyncMock(
+        side_effect=lambda artifact, **_: f"https://api.skyvern.com/v1/artifacts/{artifact.artifact_id}/content"
     )
 
     with patch("skyvern.forge.sdk.artifact.storage.base.app") as base_app:
         with patch("skyvern.forge.sdk.artifact.storage.s3.app") as s3_app:
             s3_app.DATABASE.artifacts.list_artifacts_for_run_by_type = mock_list
-            base_app.ARTIFACT_MANAGER.build_signed_content_url = build_url
+            base_app.ARTIFACT_MANAGER.resolve_share_url = resolve_url
             base_app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=12 * 60 * 60)
             result = await storage.get_downloaded_files(organization_id="o_1", run_id="wr_1")
 

tests/unit/test_self_host_artifact_url_fallback.py

@@ -0,0 +1,346 @@
+"""Self-host artifact URL fallback.
+
+When ``ARTIFACT_CONTENT_HMAC_KEYRING`` is unset:
+1. Bundling is skipped at step-archive flush — each artifact gets its own URI.
+2. URL minting falls back to ``STORAGE.get_share_link[s]`` (presigned).
+
+When it is set: today's cloud behavior (bundling on, Skyvern signed URLs).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from skyvern.config import settings
+from skyvern.forge.sdk.artifact.manager import ArtifactManager, _bundling_enabled
+from skyvern.forge.sdk.artifact.models import Artifact, ArtifactType
+from skyvern.forge.sdk.artifact.storage.test_helpers import create_fake_step
+
+_DUMMY_KEYRING_JSON = '{"current_kid":"k1","keys":{"k1":{"secret":"deadbeef"}}}'
+
+
+def _artifact(artifact_id: str, *, bundle_key: str | None = None, uri: str | None = None) -> Artifact:
+    now = datetime.now(timezone.utc)
+    return Artifact(
+        artifact_id=artifact_id,
+        artifact_type=ArtifactType.SCREENSHOT_ACTION,
+        uri=uri or f"s3://bucket/{artifact_id}.png",
+        bundle_key=bundle_key,
+        organization_id="o_1",
+        created_at=now,
+        modified_at=now,
+    )
+
+
+class TestBundlingEnabledPredicate:
+    def test_disabled_when_keyring_is_none(self) -> None:
+        with patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None):
+            assert _bundling_enabled() is False
+
+    def test_disabled_when_keyring_is_empty_string(self) -> None:
+        with patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", ""):
+            assert _bundling_enabled() is False
+
+    def test_enabled_when_keyring_is_set(self) -> None:
+        with patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON):
+            assert _bundling_enabled() is True
+
+
+class TestResolveShareUrl:
+    @pytest.mark.asyncio
+    async def test_keyring_set_non_bundled_returns_signed_url(self) -> None:
+        manager = ArtifactManager()
+        artifact = _artifact("a_1")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
+            patch.object(
+                manager, "_bundle_content_url", return_value="https://api/v1/artifacts/a_1/content?sig=x"
+            ) as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_link = AsyncMock()
+            url = await manager.resolve_share_url(artifact, expiry_seconds=3600)
+        assert url == "https://api/v1/artifacts/a_1/content?sig=x"
+        bundle.assert_called_once()
+        app.STORAGE.get_share_link.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_keyring_set_bundled_returns_signed_url(self) -> None:
+        manager = ArtifactManager()
+        artifact = _artifact("a_b", bundle_key="screenshot_action_0.png")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
+            patch.object(
+                manager, "_bundle_content_url", return_value="https://api/v1/artifacts/a_b/content?sig=x"
+            ) as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_link = AsyncMock()
+            url = await manager.resolve_share_url(artifact, expiry_seconds=3600)
+        assert url == "https://api/v1/artifacts/a_b/content?sig=x"
+        bundle.assert_called_once()
+        assert bundle.call_args.kwargs["artifact_name"] == "screenshot_action_0.png"
+        app.STORAGE.get_share_link.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_non_bundled_returns_storage_presigned(self) -> None:
+        manager = ArtifactManager()
+        artifact = _artifact("a_2")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(manager, "_bundle_content_url") as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_link = AsyncMock(
+                return_value="https://bucket.s3.amazonaws.com/...?X-Amz-Signature=abc"
+            )
+            url = await manager.resolve_share_url(artifact, expiry_seconds=3600)
+        assert url == "https://bucket.s3.amazonaws.com/...?X-Amz-Signature=abc"
+        app.STORAGE.get_share_link.assert_awaited_once_with(artifact)
+        bundle.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_bundled_legacy_row_routes_through_signed_url(self) -> None:
+        """Safety net: legacy rows with bundle_key set must NOT be presigned —
+        their uri points at the ZIP, not the member. Route through the Skyvern
+        endpoint (which 403s in webhooks but at least doesn't silently return
+        the wrong bytes)."""
+        manager = ArtifactManager()
+        artifact = _artifact("a_legacy", bundle_key="screenshot_action_0.png")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(
+                manager, "_bundle_content_url", return_value="https://api/v1/artifacts/a_legacy/content"
+            ) as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_link = AsyncMock()
+            url = await manager.resolve_share_url(artifact, expiry_seconds=3600)
+        assert url == "https://api/v1/artifacts/a_legacy/content"
+        bundle.assert_called_once()
+        app.STORAGE.get_share_link.assert_not_awaited()
+
+
+class TestGetShareLinksBatchedFallback:
+    @pytest.mark.asyncio
+    async def test_keyring_unset_batches_through_storage_get_share_links(self) -> None:
+        manager = ArtifactManager()
+        artifacts = [_artifact(f"a_{i}") for i in range(3)]
+        resolve = AsyncMock(return_value=12 * 3600)
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(manager, "resolve_artifact_url_expiry_seconds", resolve),
+            patch.object(manager, "_bundle_content_url") as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_links = AsyncMock(
+                return_value=[
+                    "https://bucket.s3.amazonaws.com/a_0?sig=p0",
+                    "https://bucket.s3.amazonaws.com/a_1?sig=p1",
+                    "https://bucket.s3.amazonaws.com/a_2?sig=p2",
+                ]
+            )
+            result = await manager.get_share_links_with_bundle_support(artifacts)
+        assert result == [
+            "https://bucket.s3.amazonaws.com/a_0?sig=p0",
+            "https://bucket.s3.amazonaws.com/a_1?sig=p1",
+            "https://bucket.s3.amazonaws.com/a_2?sig=p2",
+        ]
+        app.STORAGE.get_share_links.assert_awaited_once_with(artifacts)
+        bundle.assert_not_called()
+        resolve.assert_awaited_once_with("o_1")
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_storage_returns_none_yields_all_none(self) -> None:
+        manager = ArtifactManager()
+        artifacts = [_artifact("a_0"), _artifact("a_1")]
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(manager, "resolve_artifact_url_expiry_seconds", AsyncMock(return_value=3600)),
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_links = AsyncMock(return_value=None)
+            result = await manager.get_share_links_with_bundle_support(artifacts)
+        assert result == [None, None]
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_empty_input_returns_empty(self) -> None:
+        manager = ArtifactManager()
+        with patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None):
+            result = await manager.get_share_links_with_bundle_support([])
+        assert result == []
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_get_share_link_single_uses_storage(self) -> None:
+        manager = ArtifactManager()
+        artifact = _artifact("a_solo")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(manager, "resolve_artifact_url_expiry_seconds", AsyncMock(return_value=3600)),
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_link = AsyncMock(return_value="https://bucket/a_solo?sig=p")
+            url = await manager.get_share_link(artifact)
+        assert url == "https://bucket/a_solo?sig=p"
+        app.STORAGE.get_share_link.assert_awaited_once_with(artifact)
+
+    @pytest.mark.asyncio
+    async def test_keyring_unset_mixed_batch_legacy_bundled_routes_to_signed(self) -> None:
+        manager = ArtifactManager()
+        artifacts = [
+            _artifact("a_plain"),
+            _artifact("a_legacy_bundle", bundle_key="screenshot_action_0.png"),
+            _artifact("a_plain2"),
+        ]
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch.object(manager, "resolve_artifact_url_expiry_seconds", AsyncMock(return_value=3600)),
+            patch.object(
+                manager,
+                "_bundle_content_url",
+                side_effect=lambda artifact_id, **_: f"https://api/v1/artifacts/{artifact_id}/content",
+            ) as bundle,
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.STORAGE.get_share_links = AsyncMock(
+                return_value=[
+                    "https://bucket/a_plain?sig=1",
+                    "https://bucket/a_plain2?sig=2",
+                ]
+            )
+            result = await manager.get_share_links_with_bundle_support(artifacts)
+
+        assert result == [
+            "https://bucket/a_plain?sig=1",
+            "https://api/v1/artifacts/a_legacy_bundle/content",
+            "https://bucket/a_plain2?sig=2",
+        ]
+        bundle.assert_called_once()
+        # Non-bundled list passed verbatim, preserving input order.
+        app.STORAGE.get_share_links.assert_awaited_once()
+        passed = app.STORAGE.get_share_links.await_args.args[0]
+        assert [a.artifact_id for a in passed] == ["a_plain", "a_plain2"]
+
+
+class TestFileInfosFromArtifactsRespectsKeyring:
+    @pytest.mark.asyncio
+    async def test_keyring_unset_yields_storage_presigned_url(self) -> None:
+        from skyvern.forge.sdk.artifact.storage.base import _file_infos_from_artifacts
+
+        artifact = _artifact("a_dl", uri="azure://container/o_1/wr_1/invoice.pdf")
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch("skyvern.forge.sdk.artifact.storage.base.app") as app,
+        ):
+            app.ARTIFACT_MANAGER.resolve_artifact_url_expiry_seconds = AsyncMock(return_value=3600)
+            app.ARTIFACT_MANAGER.resolve_share_url = AsyncMock(
+                return_value="https://account.blob.core.windows.net/container/invoice.pdf?sas=abc"
+            )
+            infos = await _file_infos_from_artifacts([artifact], artifact_type=ArtifactType.DOWNLOAD)
+        assert len(infos) == 1
+        assert infos[0].url == "https://account.blob.core.windows.net/container/invoice.pdf?sas=abc"
+        app.ARTIFACT_MANAGER.resolve_share_url.assert_awaited_once()
+
+
+class TestFlushStepArchiveUnbundled:
+    @pytest.mark.asyncio
+    async def test_unbundled_flush_writes_one_artifact_per_member(self) -> None:
+        """Keyring unset → no ZIP, no STEP_ARCHIVE parent, no bundle_key on members."""
+        manager = ArtifactManager()
+        step = create_fake_step("step_unbundled_1")
+        manager.accumulate_screenshot_to_step_archive(
+            step=step, screenshots=[b"png0", b"png1"], artifact_type=ArtifactType.SCREENSHOT_ACTION
+        )
+        manager.accumulate_scrape_to_archive(
+            step=step,
+            html=b"<html/>",
+            id_css_map=b"{}",
+            id_frame_map=b"{}",
+            element_tree=b"{}",
+            element_tree_trimmed=b"{}",
+            element_tree_in_prompt=b"",
+        )
+
+        bulk_create = AsyncMock()
+        store = AsyncMock()
+        build_uri = MagicMock(side_effect=lambda **kw: f"s3://bucket/{kw['artifact_id']}.bin")
+
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.DATABASE.artifacts.bulk_create_artifacts = bulk_create
+            app.STORAGE.store_artifact = store
+            app.STORAGE.build_uri = build_uri
+            await manager.flush_step_archive("step_unbundled_1")
+
+        # Eight members: 2 screenshots + 6 scrape entries.
+        assert store.await_count == 8
+        bulk_create.assert_awaited_once()
+        models = bulk_create.await_args.args[0]
+        assert len(models) == 8  # No parent STEP_ARCHIVE row.
+        assert all(m.bundle_key is None for m in models)
+        assert all(m.artifact_type != ArtifactType.STEP_ARCHIVE for m in models)
+        called_types = [call.kwargs["artifact_type"] for call in build_uri.call_args_list]
+        assert ArtifactType.SCREENSHOT_ACTION in called_types
+        assert ArtifactType.HTML_SCRAPE in called_types
+
+    @pytest.mark.asyncio
+    async def test_bundled_flush_unchanged_when_keyring_set(self) -> None:
+        manager = ArtifactManager()
+        step = create_fake_step("step_bundled_1")
+        manager.accumulate_screenshot_to_step_archive(
+            step=step, screenshots=[b"a"], artifact_type=ArtifactType.SCREENSHOT_ACTION
+        )
+
+        bulk_create = AsyncMock()
+        store = AsyncMock()
+        build_uri = MagicMock(return_value="s3://bucket/parent.zip")
+
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON),
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.DATABASE.artifacts.bulk_create_artifacts = bulk_create
+            app.STORAGE.store_artifact = store
+            app.STORAGE.build_uri = build_uri
+            await manager.flush_step_archive("step_bundled_1")
+
+        store.assert_awaited_once()
+        bulk_create.assert_awaited_once()
+        models = bulk_create.await_args.args[0]
+        assert len(models) == 2
+        assert models[0].artifact_type == ArtifactType.STEP_ARCHIVE
+        assert models[0].bundle_key is None
+        assert models[1].bundle_key == "screenshot_action_0.png"
+
+    @pytest.mark.asyncio
+    async def test_unbundled_flush_applies_pending_screenshot_fk_updates(self) -> None:
+        """Deferred action.screenshot_artifact_id writes must still fire in the unbundled path."""
+        manager = ArtifactManager()
+        step = create_fake_step("step_unbundled_fk")
+        ids = manager.accumulate_screenshot_to_step_archive(
+            step=step, screenshots=[b"png"], artifact_type=ArtifactType.SCREENSHOT_ACTION
+        )
+        acc = manager._step_archives["step_unbundled_fk"]
+        acc.pending_action_screenshot_updates.append((step.organization_id, "act_1", ids[0]))
+
+        update_fk = AsyncMock()
+        with (
+            patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", None),
+            patch("skyvern.forge.sdk.artifact.manager.app") as app,
+        ):
+            app.DATABASE.artifacts.bulk_create_artifacts = AsyncMock()
+            app.DATABASE.artifacts.update_action_screenshot_artifact_id = update_fk
+            app.STORAGE.store_artifact = AsyncMock()
+            app.STORAGE.build_uri = MagicMock(return_value="s3://bucket/x.bin")
+            await manager.flush_step_archive("step_unbundled_fk")
+
+        update_fk.assert_awaited_once_with(
+            organization_id=step.organization_id,
+            action_id="act_1",
+            screenshot_artifact_id=ids[0],
+        )

tests/unit/test_step_archive_accumulator.py

@@ -10,12 +10,14 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from skyvern.config import settings
 from skyvern.forge.sdk.artifact.manager import ArtifactManager, StepArchiveAccumulator
 from skyvern.forge.sdk.artifact.models import ArtifactType
 from skyvern.forge.sdk.artifact.storage.test_helpers import create_fake_step
 
 TEST_STEP_ID = "step_archive_test_001"
 TEST_STEP_ID_2 = "step_archive_test_002"
+_DUMMY_KEYRING_JSON = '{"current_kid":"k1","keys":{"k1":{"secret":"deadbeef"}}}'
 
 
 # ---------------------------------------------------------------------------
@@ -533,7 +535,16 @@ def _make_app_mocks() -> tuple[MagicMock, MagicMock]:
 
 
 class TestFlushStepArchive:
-    """Tests for ArtifactManager.flush_step_archive (per-step early flush)."""
+    """Tests for ArtifactManager.flush_step_archive (per-step early flush).
+
+    These cover the **bundled** path (HMAC keyring set, cloud behavior). The
+    unbundled (self-host) path is covered in test_self_host_artifact_url_fallback.
+    """
+
+    @pytest.fixture(autouse=True)
+    def _force_bundled(self):
+        with patch.object(settings, "ARTIFACT_CONTENT_HMAC_KEYRING", _DUMMY_KEYRING_JSON):
+            yield
 
     @pytest.mark.asyncio
     async def test_flush_uploads_zip_and_creates_db_rows(self) -> None: