vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-07-09 16:09:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

feat: execute workflow runs against the version stamped at creation (#6942)

Browse source
This commit is contained in:
Shuchang Zheng 2026-06-30 19:35:18 -07:00 • committed by GitHub
parent 9266ba3539
commit 17d4201376
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 132 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

40
skyvern/forge/sdk/workflow/service.py
View file

@ -1035,6 +1035,7 @@ class WorkflowService:
workflow_schedule_id: str | None = None,
ignore_inherited_workflow_system_prompt: bool = False,
copilot_session_id: str | None = None,
resolved_workflow_id: str | None = None,
) -> WorkflowRun:
"""
Create a workflow run and its parameters. Validate the workflow and the organization. If there are missing
@ -1044,15 +1045,24 @@ class WorkflowService:
:param workflow_id: The workflow id to run.
:param organization_id: The organization id for the workflow.
:param max_steps_override: The max steps override for the workflow run, if any.
:param resolved_workflow_id: Pin the exact workflow version row to run against, resolved by
workflow_id. Used when the (permanent_id, version) index is non-unique and a version=
lookup could resolve the wrong row. When None, resolve by permanent id + version.
:return: The created workflow run.
"""
async with app.DATABASE.workflow_runs.Session() as outer_session:
# Validate the workflow and the organization
workflow = await self.get_workflow_by_permanent_id(
workflow_permanent_id=workflow_permanent_id,
organization_id=None if is_template_workflow else organization.organization_id,
version=version,
)
if resolved_workflow_id is not None:
workflow = await app.DATABASE.workflows.get_workflow(
workflow_id=resolved_workflow_id,
organization_id=None if is_template_workflow else organization.organization_id,
)
else:
workflow = await self.get_workflow_by_permanent_id(
workflow_permanent_id=workflow_permanent_id,
organization_id=None if is_template_workflow else organization.organization_id,
version=version,
)
if workflow is None:
LOG.error(f"Workflow {workflow_permanent_id} not found", workflow_version=version)
raise WorkflowNotFound(workflow_permanent_id=workflow_permanent_id, version=version)
@ -1709,15 +1719,11 @@ class WorkflowService:
block_outputs=block_outputs,
)
workflow_run = await self.get_workflow_run(workflow_run_id=workflow_run_id, organization_id=organization_id)
workflow = workflow_override or await self.get_workflow_by_permanent_id(
workflow_permanent_id=workflow_run.workflow_permanent_id
)
has_conditionals = workflow_script_service.workflow_has_conditionals(workflow)
browser_profile_id = workflow_run.browser_profile_id
close_browser_on_completion = browser_session_id is None and not workflow_run.browser_address
# Guard: if the run was canceled while queued (before Temporal picked it up),
# don't overwrite the canceled status with running.
# Guard: if the run was canceled while queued (before Temporal picked it up), don't
# overwrite the canceled status with running. Checked BEFORE workflow resolution so a run
# whose stamped workflow version was deleted after cancellation does not raise
# WorkflowNotFound on a late worker pickup.
if workflow_run.status == WorkflowRunStatus.canceled:
LOG.info(
"Workflow run was canceled before execution started, skipping",
@ -1726,6 +1732,14 @@ class WorkflowService:
)
return workflow_run
# Resolve the exact version stamped on the run (run.workflow_id is always set), not the
# latest published version, so a publish between run creation and execution does not change
# what executes.
workflow = workflow_override or await self.get_workflow(workflow_id=workflow_run.workflow_id)
has_conditionals = workflow_script_service.workflow_has_conditionals(workflow)
browser_profile_id = workflow_run.browser_profile_id
close_browser_on_completion = browser_session_id is None and not workflow_run.browser_address
enterprise_gated_features = _collect_enterprise_gated_workflow_features(workflow, block_labels=block_labels)
if enterprise_gated_features:
try:

21
skyvern/services/workflow_service.py
View file

@ -31,9 +31,13 @@ async def prepare_workflow(
workflow_run_id: str | None = None,
ignore_inherited_workflow_system_prompt: bool = False,
copilot_session_id: str | None = None,
resolved_workflow_id: str | None = None,
) -> WorkflowRun:
"""
Prepare a workflow to be run.
``resolved_workflow_id`` pins the exact workflow version row; when None, resolve by
permanent id + version.
"""
if template:
if workflow_id not in await app.STORAGE.retrieve_global_workflows():
@ -55,13 +59,20 @@ async def prepare_workflow(
workflow_schedule_id=workflow_schedule_id,
ignore_inherited_workflow_system_prompt=ignore_inherited_workflow_system_prompt,
copilot_session_id=copilot_session_id,
resolved_workflow_id=resolved_workflow_id,
)
workflow = await app.WORKFLOW_SERVICE.get_workflow_by_permanent_id(
workflow_permanent_id=workflow_id,
organization_id=None if template else organization.organization_id,
version=version,
)
if resolved_workflow_id is not None:
workflow = await app.WORKFLOW_SERVICE.get_workflow(
workflow_id=resolved_workflow_id,
organization_id=None if template else organization.organization_id,
)
else:
workflow = await app.WORKFLOW_SERVICE.get_workflow_by_permanent_id(
workflow_permanent_id=workflow_id,
organization_id=None if template else organization.organization_id,
version=version,
)
await app.DATABASE.tasks.create_task_run(
task_run_type=RunType.workflow_run,

79
tests/unit/test_execute_workflow_resolution.py Normal file
View file

@ -0,0 +1,79 @@
"""Unit tests for the global execute_workflow resolve-by-run.workflow_id change.
execute_workflow resolves the exact workflow version stamped on the run (get_workflow by
workflow_id) instead of latest-by-permanent-id, and short-circuits a canceled run before that
resolution.
"""
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import AsyncMock
import pytest
from skyvern.forge.sdk.workflow.models.workflow import WorkflowRunStatus
from skyvern.forge.sdk.workflow.service import WorkflowService
class _StopForTest(Exception):
"""Sentinel to abort execute_workflow right after workflow resolution."""
@pytest.mark.asyncio
async def test_execute_workflow_resolves_by_run_workflow_id(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, Any] = {}
async def _capture_resolution(**kwargs: Any) -> Any:
captured.update(kwargs)
raise _StopForTest
workflow_run = SimpleNamespace(
workflow_permanent_id="wpid_1",
workflow_id="w_v7",
status=WorkflowRunStatus.queued,
)
service = WorkflowService()
monkeypatch.setattr(service, "get_workflow_run", AsyncMock(return_value=workflow_run))
# Latest-by-permanent-id must NOT be used for execution resolution anymore.
monkeypatch.setattr(
service,
"get_workflow_by_permanent_id",
AsyncMock(side_effect=AssertionError("execution must resolve by run.workflow_id")),
)
monkeypatch.setattr(service, "get_workflow", _capture_resolution)
organization = SimpleNamespace(organization_id="o_1")
with pytest.raises(_StopForTest):
await service.execute_workflow(
workflow_run_id="wr_1",
api_key="k",
organization=cast(Any, organization),
)
# The exact version stamped on the run executes, not latest-by-permanent-id.
assert captured["workflow_id"] == "w_v7"
@pytest.mark.asyncio
async def test_execute_workflow_canceled_run_skips_resolution(monkeypatch: pytest.MonkeyPatch) -> None:
"""A run canceled while queued short-circuits BEFORE workflow resolution, so a run whose
stamped version was deleted after cancellation does not raise WorkflowNotFound."""
workflow_run = SimpleNamespace(
workflow_permanent_id="wpid_1",
workflow_id="w_deleted",
status=WorkflowRunStatus.canceled,
)
service = WorkflowService()
monkeypatch.setattr(service, "get_workflow_run", AsyncMock(return_value=workflow_run))
get_workflow = AsyncMock(side_effect=AssertionError("must not resolve a canceled run's workflow"))
monkeypatch.setattr(service, "get_workflow", get_workflow)
organization = SimpleNamespace(organization_id="o_1")
result = await service.execute_workflow(
workflow_run_id="wr_1",
api_key="k",
organization=cast(Any, organization),
)
assert result is workflow_run
get_workflow.assert_not_awaited()

16
tests/unit/test_workflow_elapsed_timeout.py
View file

@ -222,7 +222,7 @@ async def test_execute_workflow_returns_after_elapsed_timeout_without_finally(mo
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -300,7 +300,7 @@ async def test_execute_workflow_times_out_slow_pre_block_script_lookup(monkeypat
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -384,7 +384,7 @@ async def test_execute_workflow_preserves_completed_status_after_post_run_timeou
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -473,7 +473,7 @@ async def test_execute_workflow_preserves_timed_out_status_after_non_terminal_po
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -566,7 +566,7 @@ async def test_execute_workflow_marks_timed_out_when_post_run_budget_is_exhauste
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -741,7 +741,7 @@ async def test_execute_workflow_refreshes_terminal_status_after_immediate_post_r
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -830,7 +830,7 @@ async def test_execute_workflow_returns_finalized_status_after_post_run_timeout(
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))
@ -917,7 +917,7 @@ async def test_execute_workflow_runs_finally_for_existing_timed_out_status(
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=created_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "mark_workflow_run_as_running", AsyncMock(return_value=running_run))
monkeypatch.setattr(svc, "get_workflow_run_parameter_tuples", AsyncMock(return_value=[]))
monkeypatch.setattr(svc, "get_workflow_output_parameters", AsyncMock(return_value=[]))

3
tests/unit/workflow/test_enterprise_feature_gating.py
View file

@ -187,6 +187,7 @@ async def test_execute_workflow_cleans_up_after_enterprise_gate_failure(monkeypa
workflow = _workflow([_navigation_block("openai", engine=RunEngine.openai_cua)])
workflow_run = SimpleNamespace(
workflow_run_id="wr_1",
workflow_id=workflow.workflow_id,
workflow_permanent_id=workflow.workflow_permanent_id,
browser_profile_id=None,
browser_address=None,
@ -208,7 +209,7 @@ async def test_execute_workflow_cleans_up_after_enterprise_gate_failure(monkeypa
svc = WorkflowService()
monkeypatch.setattr(svc, "get_workflow_run", AsyncMock(return_value=workflow_run))
monkeypatch.setattr(svc, "get_workflow_by_permanent_id", AsyncMock(return_value=workflow))
monkeypatch.setattr(svc, "get_workflow", AsyncMock(return_value=workflow))
mark_workflow_run_as_failed = AsyncMock(return_value=failed_workflow_run)
clean_up_workflow = AsyncMock()
monkeypatch.setattr(svc, "mark_workflow_run_as_failed", mark_workflow_run_as_failed)