Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 58 additions & 0 deletions workflows/tests/test_schema_recording_load.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
"""Load Chrome recorder JSON without requiring a trailing extract step."""

from workflow_use.schema.views import ClickStep, WorkflowDefinitionSchema


def _chrome_recording_payload(*, last_step_type: str = 'click') -> dict:
last_step: dict
if last_step_type == 'click':
last_step = {
'type': 'click',
'targetText': 'Order with Careem',
'url': 'https://www.ubereats.com/feed',
}
else:
last_step = {'type': 'extract', 'extractionGoal': 'Get confirmation'}

return {
'name': 'Recorded Workflow (Semantic)',
'description': 'Recorded on 8/29/2026, 5:50:27 PM',
'version': '1.0',
'input_schema': [],
'steps': [
{'type': 'navigation', 'url': 'https://www.ubereats.com/feed'},
last_step,
],
}


def test_chrome_recording_ending_in_click_loads():
workflow = WorkflowDefinitionSchema.model_validate(_chrome_recording_payload())

assert len(workflow.steps) == 2
assert workflow.steps[-1].type == 'click'
assert isinstance(workflow.steps[-1], ClickStep)
assert workflow.steps[-1].target_text == 'Order with Careem'


def test_snake_case_target_text_still_loads():
workflow = WorkflowDefinitionSchema.model_validate(
{
'name': 'Semantic workflow',
'description': 'Hand-written',
'version': '1.0',
'input_schema': [],
'steps': [
{'type': 'navigation', 'url': 'https://example.com'},
{'type': 'click', 'target_text': 'Submit'},
],
}
)

assert workflow.steps[-1].target_text == 'Submit'


def test_extract_ending_workflow_still_loads():
workflow = WorkflowDefinitionSchema.model_validate(_chrome_recording_payload(last_step_type='extract'))

assert workflow.steps[-1].type == 'extract'
33 changes: 11 additions & 22 deletions workflows/workflow_use/schema/views.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from typing import Any, Dict, List, Literal, Optional, Union

from pydantic import BaseModel, Field, validator
from pydantic import AliasChoices, BaseModel, Field


# --- Base Step Model ---
Expand Down Expand Up @@ -28,6 +28,8 @@ class BaseWorkflowStep(BaseModel):

# --- Steps that require interaction with a DOM element ---
class SelectorWorkflowSteps(BaseWorkflowStep):
model_config = {'extra': 'allow', 'populate_by_name': True}

# Legacy fields - kept for backward compatibility but discouraged
cssSelector: Optional[str] = Field(
None, description='[LEGACY] CSS selector - avoid in new workflows, use target_text instead.'
Expand All @@ -43,8 +45,10 @@ class SelectorWorkflowSteps(BaseWorkflowStep):
)

# PRIMARY: Text-based semantic targeting (non-brittle)
# Chrome recorder exports `targetText`; semantic replay reads `target_text`.
target_text: Optional[str] = Field(
None,
validation_alias=AliasChoices('target_text', 'targetText'),
description='Visible or accessible text to identify the element. Use hierarchical context for disambiguation (e.g., "Submit (in Personal Information)", "Edit (item 2 of 3)"). If None, relies on selectorStrategies fallback.',
)

Expand Down Expand Up @@ -214,7 +218,12 @@ class WorkflowInputSchemaDefinition(BaseModel):


class WorkflowDefinitionSchema(BaseModel):
"""Pydantic model representing the structure of the workflow JSON file."""
"""Pydantic model representing the structure of the workflow JSON file.

A trailing extract step is recommended for LLM generation (`run-as-tool`) but is
not required. Chrome recorder exports often end on click/input and must still load
for `run-workflow-no-ai`.
"""

workflow_analysis: Optional[str] = Field(
None,
Expand All @@ -237,26 +246,6 @@ class WorkflowDefinitionSchema(BaseModel):
description='List of input schema definitions.',
)

@validator('steps')
def validate_ends_with_extract(cls, steps: List[WorkflowStep]) -> List[WorkflowStep]:
"""Validate that the workflow ends with an extract step."""
if not steps:
raise ValueError('Workflow must have at least one step')

last_step = steps[-1]
# Check if last step is an extract step
# We need to check the 'type' attribute from the step dict/model
step_type = getattr(last_step, 'type', None)

if step_type not in ['extract', 'extract_page_content']:
raise ValueError(
f'Workflow must end with an extract step (extract or extract_page_content). '
f'Current last step type: {step_type}. '
f'AI processing is always needed at the end of a workflow.'
)

return steps

# Add loader from json file
@classmethod
def load_from_json(cls, json_path: str):
Expand Down