diff --git a/skyvern/forge/sdk/executor/async_executor.py b/skyvern/forge/sdk/executor/async_executor.py
index 84b51af8d..5ca1407df 100644
--- a/skyvern/forge/sdk/executor/async_executor.py
+++ b/skyvern/forge/sdk/executor/async_executor.py
@@ -36,6 +36,7 @@ async def execute_workflow(
         workflow_run_id: str,
         max_steps_override: int | None,
         api_key: str | None,
+        browser_session_id: str | None,
         **kwargs: dict,
     ) -> None:
         pass
@@ -93,6 +94,7 @@ async def execute_workflow(
         workflow_run_id: str,
         max_steps_override: int | None,
         api_key: str | None,
+        browser_session_id: str | None,
         **kwargs: dict,
     ) -> None:
         LOG.info(
@@ -109,4 +111,5 @@ async def execute_workflow(
             workflow_run_id=workflow_run_id,
             api_key=api_key,
             organization=organization,
+            browser_session_id=browser_session_id,
         )
diff --git a/skyvern/forge/sdk/routes/agent_protocol.py b/skyvern/forge/sdk/routes/agent_protocol.py
index b2a4e8b1e..9b619cd3e 100644
--- a/skyvern/forge/sdk/routes/agent_protocol.py
+++ b/skyvern/forge/sdk/routes/agent_protocol.py
@@ -567,6 +567,7 @@ async def execute_workflow(
         organization_id=current_org.organization_id,
         workflow_id=workflow_run.workflow_id,
         workflow_run_id=workflow_run.workflow_run_id,
+        browser_session_id=workflow_request.browser_session_id,
         max_steps_override=x_max_steps_override,
         api_key=x_api_key,
     )
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index c61e80572..d0d638501 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -199,12 +199,15 @@ async def execute_workflow(
         workflow_run_id: str,
         api_key: str,
         organization: Organization,
+        browser_session_id: str | None,
     ) -> WorkflowRun:
         """Execute a workflow."""
         organization_id = organization.organization_id
         workflow_run = await self.get_workflow_run(workflow_run_id=workflow_run_id)
         workflow = await self.get_workflow(workflow_id=workflow_run.workflow_id, organization_id=organization_id)
 
+        uses_persistent_browser_state = browser_session_id is not None
+
         # Set workflow run status to running, create workflow run parameters
         await self.mark_workflow_run_as_running(workflow_run_id=workflow_run.workflow_run_id)
 
@@ -245,6 +248,7 @@ async def execute_workflow(
                         workflow_run=workflow_run,
                         api_key=api_key,
                         need_call_webhook=True,
+                        close_browser_on_completion=not uses_persistent_browser_state,
                     )
                     return workflow_run
                 parameters = block.get_all_parameters(workflow_run_id)
@@ -277,6 +281,7 @@ async def execute_workflow(
                         workflow_run=workflow_run,
                         api_key=api_key,
                         need_call_webhook=False,
+                        close_browser_on_completion=not uses_persistent_browser_state,
                     )
                     return workflow_run
                 elif block_result.status == BlockStatus.failed:
@@ -298,6 +303,7 @@ async def execute_workflow(
                             workflow=workflow,
                             workflow_run=workflow_run,
                             api_key=api_key,
+                            close_browser_on_completion=not uses_persistent_browser_state,
                         )
                         return workflow_run
 
@@ -332,6 +338,7 @@ async def execute_workflow(
                             workflow=workflow,
                             workflow_run=workflow_run,
                             api_key=api_key,
+                            close_browser_on_completion=not uses_persistent_browser_state,
                         )
                         return workflow_run
 
@@ -363,7 +370,12 @@ async def execute_workflow(
                 await self.mark_workflow_run_as_failed(
                     workflow_run_id=workflow_run.workflow_run_id, failure_reason=failure_reason
                 )
-                await self.clean_up_workflow(workflow=workflow, workflow_run=workflow_run, api_key=api_key)
+                await self.clean_up_workflow(
+                    workflow=workflow,
+                    workflow_run=workflow_run,
+                    api_key=api_key,
+                    close_browser_on_completion=not uses_persistent_browser_state,
+                )
                 return workflow_run
 
         refreshed_workflow_run = await app.DATABASE.get_workflow_run(workflow_run_id=workflow_run.workflow_run_id)
@@ -379,7 +391,12 @@ async def execute_workflow(
                 workflow_run_id=workflow_run.workflow_run_id,
                 workflow_run_status=refreshed_workflow_run.status if refreshed_workflow_run else None,
             )
-        await self.clean_up_workflow(workflow=workflow, workflow_run=workflow_run, api_key=api_key)
+        await self.clean_up_workflow(
+            workflow=workflow,
+            workflow_run=workflow_run,
+            api_key=api_key,
+            close_browser_on_completion=not uses_persistent_browser_state,
+        )
         return workflow_run
 
     async def create_workflow(