fix(SKY-11881): persist typed code-block recorder actions (#7147)

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

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

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