From 298ab5550ced9f40810edfb9897100e16f84d8a4 Mon Sep 17 00:00:00 2001 From: Aaron Perez Date: Mon, 6 Jul 2026 21:27:07 -0500 Subject: [PATCH] fix(SKY-11881): persist typed code-block recorder actions (#7147) --- .../workflow/models/code_block_recorder.py | 161 +++++++++++++++++- .../workflow/models/code_block_recording.py | 4 +- .../unit/workflow/test_code_block_recorder.py | 44 ++++- 3 files changed, 197 insertions(+), 12 deletions(-) diff --git a/skyvern/forge/sdk/workflow/models/code_block_recorder.py b/skyvern/forge/sdk/workflow/models/code_block_recorder.py index 3ff448655..19bba57ad 100644 --- a/skyvern/forge/sdk/workflow/models/code_block_recorder.py +++ b/skyvern/forge/sdk/workflow/models/code_block_recorder.py @@ -2,13 +2,15 @@ from __future__ import annotations import sys import time +from os import PathLike, fspath from types import FrameType from typing import Any, Awaitable, Callable +import pydantic import structlog from skyvern.webeye.actions.action_types import ActionType -from skyvern.webeye.actions.actions import Action, ActionStatus +from skyvern.webeye.actions.actions import Action, ActionStatus, SelectOption LOG = structlog.get_logger() @@ -114,6 +116,143 @@ def _factory_selector(name: str, args: tuple[Any, ...]) -> str: return f"{name}({arg})" if arg is not None else name +def _string_value(value: Any) -> str | None: + if isinstance(value, (str, int, float)): + return str(value) + if isinstance(value, PathLike): + return fspath(value) + return None + + +def _arg(args: tuple[Any, ...], index: int) -> Any: + return args[index] if len(args) > index else None + + +def _page_value_index(name: str, target: str | None) -> int: + # Direct Playwright page web actions take selector first, value second. + return 1 if target is None and name.startswith("page.") else 0 + + +def _element_id(name: str, target: str | None, args: tuple[Any, ...]) -> str: + return target or (_string_value(_arg(args, 0)) if name.startswith("page.") else None) or "" + + +def _select_option(value: Any, kwargs: dict[str, Any]) -> SelectOption | None: + if value is None: + if not {"label", "value", "index"} & kwargs.keys(): + return None + value = {key: kwargs.get(key) for key in ("label", "value", "index")} + if isinstance(value, str): + return SelectOption(value=value) + if isinstance(value, int): + return SelectOption(index=value) + if isinstance(value, dict): + return SelectOption( + label=_string_value(value.get("label")), + value=_string_value(value.get("value")), + index=value.get("index") if isinstance(value.get("index"), int) else None, + ) + return None + + +def _recorded_action_fields( + action_type: ActionType, + name: str, + target: str | None, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> dict[str, Any]: + fields: dict[str, Any] = {} + if action_type.is_web_action(): + fields["element_id"] = _element_id(name, target, args) + + value_index = _page_value_index(name, target) + if action_type == ActionType.GOTO_URL: + fields["url"] = _string_value(kwargs.get("url", _arg(args, 0))) + elif action_type == ActionType.INPUT_TEXT: + # Preserve the existing recorder privacy boundary: input values may be credentials, + # so the typed action carries the required field without retaining the raw value. + fields["text"] = "" + elif action_type == ActionType.UPLOAD_FILE: + fields["file_url"] = _string_value(kwargs.get("file_url", _arg(args, value_index))) + elif action_type == ActionType.DOWNLOAD_FILE: + fields["file_name"] = _string_value(kwargs.get("file_name", _arg(args, 0))) or "download_file" + download_url = _string_value(kwargs.get("download_url", _arg(args, 1))) + if download_url is not None: + fields["download_url"] = download_url + elif action_type == ActionType.SELECT_OPTION: + option = _select_option(kwargs.get("value", _arg(args, value_index)), kwargs) + if option is not None: + fields["option"] = option + elif action_type == ActionType.CHECKBOX: + fields["is_checked"] = not name.endswith(".uncheck") + elif action_type == ActionType.EXTRACT: + prompt = kwargs.get("prompt", _arg(args, 0)) + if isinstance(prompt, str): + fields["data_extraction_goal"] = prompt + schema = kwargs.get("schema", _arg(args, 1)) + if schema is not None: + fields["data_extraction_schema"] = schema + elif action_type == ActionType.EXECUTE_JS: + fields["js_code"] = _string_value(kwargs.get("expression", _arg(args, 0))) + elif action_type == ActionType.KEYPRESS: + keys = kwargs.get("keys", _arg(args, 0)) + fields["keys"] = ( + [str(key) for key in keys] if isinstance(keys, list) else [str(keys)] if keys is not None else [] + ) + fields["hold"] = bool(kwargs.get("hold", False)) + if "duration" in kwargs: + fields["duration"] = int(kwargs["duration"]) + elif action_type == ActionType.SCROLL: + fields["scroll_x"] = kwargs.get("scroll_x", _arg(args, 0)) + fields["scroll_y"] = kwargs.get("scroll_y", _arg(args, 1)) + elif action_type == ActionType.MOVE: + fields["x"] = kwargs.get("x", _arg(args, 0)) + fields["y"] = kwargs.get("y", _arg(args, 1)) + elif action_type == ActionType.DRAG: + fields["start_x"] = kwargs.get("start_x", _arg(args, 0)) + fields["start_y"] = kwargs.get("start_y", _arg(args, 1)) + fields["path"] = kwargs.get("path", _arg(args, 2)) + elif action_type == ActionType.LEFT_MOUSE: + fields["x"] = kwargs.get("x", _arg(args, 0)) + fields["y"] = kwargs.get("y", _arg(args, 1)) + fields["direction"] = kwargs.get("direction", _arg(args, 2)) + return {key: value for key, value in fields.items() if value is not None} + + +def _action_from_fields( + action_type: ActionType, + fields: dict[str, Any], + *, + warning: str, +) -> Action: + # Import lazily: db.utils imports workflow models through schema conversion helpers. + from skyvern.forge.sdk.db.utils import ACTION_TYPE_TO_CLASS + + action_class = ACTION_TYPE_TO_CLASS.get(action_type, Action) + if action_class is Action: + return Action(**fields) + try: + return action_class(**fields) + except pydantic.ValidationError as exc: + LOG.warning( + warning, + action_type=action_type, + subclass=action_class.__name__, + errors=exc.errors(), + ) + return Action(**fields) + + +def recorded_action_from_payload(raw: dict[str, Any]) -> Action: + action_type = ActionType(raw["action_type"]) + return _action_from_fields( + action_type, + raw, + warning="Failed to instantiate masked recorded action subclass, falling back to base Action", + ) + + class _Recorder: def __init__(self, on_action: OnAction | None = None) -> None: self.actions: list[Action] = [] @@ -127,12 +266,13 @@ class _Recorder: target: str | None, call: Callable[[], Awaitable[Any]], args: tuple[Any, ...], + kwargs: dict[str, Any], description: str | None = None, ) -> Any: started = time.monotonic() # Input values may be credentials (incl. derived TOTP codes); never describe them. describe_args = () if action_type == ActionType.INPUT_TEXT else args - action = Action( + common_fields = dict( action_type=action_type, status=ActionStatus.completed, action_order=len(self.actions), @@ -142,6 +282,11 @@ class _Recorder: description=description if description is not None else _describe(name, target, describe_args), output={"code_line": _frame_user_line()}, ) + action = _action_from_fields( + action_type, + {**common_fields, **_recorded_action_fields(action_type, name, target, args, kwargs)}, + warning="Failed to instantiate recorded action subclass, falling back to base Action", + ) try: result = await call() except BaseException as exc: @@ -192,7 +337,7 @@ class RecordingLocator: async def recorded(*args: Any, **kwargs: Any) -> Any: return await self.__recorder.record( - action_type, f"locator.{name}", self.__selector, lambda: attr(*args, **kwargs), args + action_type, f"locator.{name}", self.__selector, lambda: attr(*args, **kwargs), args, kwargs ) return recorded @@ -210,7 +355,7 @@ class RecordingKeyboard: async def recorded(*args: Any, **kwargs: Any) -> Any: return await self.__recorder.record( - ActionType.KEYPRESS, "keyboard.press", None, lambda: attr(*args, **kwargs), args + ActionType.KEYPRESS, "keyboard.press", None, lambda: attr(*args, **kwargs), args, kwargs ) return recorded @@ -263,7 +408,13 @@ class RecordingPage: if isinstance(prompt, str) and prompt.strip(): description = " ".join(prompt.split())[:200] return await self.__recorder.record( - action_type, f"page.{name}", None, lambda: attr(*args, **kwargs), args, description=description + action_type, + f"page.{name}", + None, + lambda: attr(*args, **kwargs), + args, + kwargs, + description=description, ) return recorded diff --git a/skyvern/forge/sdk/workflow/models/code_block_recording.py b/skyvern/forge/sdk/workflow/models/code_block_recording.py index 667299b95..0c7e783f4 100644 --- a/skyvern/forge/sdk/workflow/models/code_block_recording.py +++ b/skyvern/forge/sdk/workflow/models/code_block_recording.py @@ -12,7 +12,7 @@ from skyvern.forge import app from skyvern.forge.sdk.artifact.models import ArtifactType from skyvern.forge.sdk.models import StepStatus from skyvern.forge.sdk.schemas.tasks import TaskStatus -from skyvern.forge.sdk.workflow.models.code_block_recorder import RecordingPage +from skyvern.forge.sdk.workflow.models.code_block_recorder import RecordingPage, recorded_action_from_payload from skyvern.schemas.steps import AgentStepOutput from skyvern.webeye.actions.actions import Action @@ -152,7 +152,7 @@ class CodeBlockActionRecording: try: masked = self._workflow_run_context.mask_secrets_in_data([a.model_dump(mode="json") for a in recorded]) for raw in masked: - action = Action.model_validate(raw) + action = recorded_action_from_payload(raw) action.task_id = self._task.task_id action.step_id = self._step.step_id action.step_order = self._step.order diff --git a/tests/unit/workflow/test_code_block_recorder.py b/tests/unit/workflow/test_code_block_recorder.py index c8fb29672..3f195c619 100644 --- a/tests/unit/workflow/test_code_block_recorder.py +++ b/tests/unit/workflow/test_code_block_recorder.py @@ -14,6 +14,8 @@ from skyvern.core.script_generations.skyvern_page import SkyvernPage from skyvern.forge import app from skyvern.forge.agent import ForgeAgent from skyvern.forge.sdk.copilot.code_block_steps import _METHOD_ACTION_TYPES +from skyvern.forge.sdk.db.models import ActionModel +from skyvern.forge.sdk.db.utils import hydrate_action from skyvern.forge.sdk.models import StepStatus from skyvern.forge.sdk.schemas.tasks import TaskStatus from skyvern.forge.sdk.workflow.context_manager import WorkflowRunContext @@ -30,7 +32,7 @@ from skyvern.forge.sdk.workflow.models.code_block_recorder import ( from skyvern.forge.sdk.workflow.models.parameter import OutputParameter, ParameterType from skyvern.schemas.workflows import BlockStatus from skyvern.webeye.actions.action_types import ActionType -from skyvern.webeye.actions.actions import Action, ActionStatus +from skyvern.webeye.actions.actions import Action, ActionStatus, ClickAction, GotoUrlAction, InputTextAction from skyvern.webeye.browser_artifacts import BrowserArtifacts @@ -127,6 +129,13 @@ async def test_records_goto_click_fill_with_types_and_order() -> None: assert [a.action_order for a in recorded] == [0, 1, 2] assert all(a.status == ActionStatus.completed for a in recorded) assert recorded[0].description == "page.goto https://example.com" + assert isinstance(recorded[0], GotoUrlAction) + assert recorded[0].url == "https://example.com" + assert isinstance(recorded[1], InputTextAction) + assert recorded[1].element_id == "#q" + assert recorded[1].text == "" + assert isinstance(recorded[2], ClickAction) + assert recorded[2].element_id == "#go" @pytest.mark.asyncio @@ -651,17 +660,42 @@ async def test_goalless_code_block_skips_screenshots(monkeypatch: pytest.MonkeyP async def test_recorded_calls_persist_as_actions_on_the_step(monkeypatch: pytest.MonkeyPatch) -> None: """Each recorded playwright call becomes a real Action row tied to the task/step.""" page = FakePage() - context = FakeWorkflowRunContext() + context = FakeWorkflowRunContext(secrets={"pw": "secret-password"}) mocks = _patch_execute_environment(monkeypatch, page, context) - block = _make_code_block("await page.goto('https://example.com')\nawait page.locator('#go').click()", goal="go") + block = _make_code_block( + "await page.goto('https://example.com')\n" + "await page.locator('#pw').fill('secret-password')\n" + "await page.locator('#go').click()", + goal="go", + ) result = await block.execute(workflow_run_id="wr_test", workflow_run_block_id="wrb_test", organization_id="o_test") assert result.success is True actions = _created_actions(mocks) - assert [a.action_type for a in actions] == [ActionType.GOTO_URL, ActionType.CLICK] + assert [a.action_type for a in actions] == [ActionType.GOTO_URL, ActionType.INPUT_TEXT, ActionType.CLICK] assert all(a.task_id == "tsk_code" and a.step_id == "stp_code" and a.step_order == 0 for a in actions) - assert [a.action_order for a in actions] == [0, 1] + assert [a.action_order for a in actions] == [0, 1, 2] + assert isinstance(actions[0], GotoUrlAction) + assert actions[0].url == "https://example.com" + assert isinstance(actions[1], InputTextAction) + assert actions[1].element_id == "#pw" + assert actions[1].text == "" + assert isinstance(actions[2], ClickAction) + assert actions[2].element_id == "#go" + dumped = json.dumps([a.model_dump(mode="json") for a in actions]) + assert "secret-password" not in dumped + hydrated = [ + hydrate_action( + ActionModel( + action_type=action.action_type, + status=action.status, + action_json=action.model_dump(mode="json"), + ) + ) + for action in actions + ] + assert [type(action) for action in hydrated] == [GotoUrlAction, InputTextAction, ClickAction] @pytest.mark.asyncio