vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-10-03 04:06:21 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Remove task v1 cache replay (SKY-17334) (#8720)

Browse source
This commit is contained in:
Shuchang Zheng 2026-09-29 18:12:01 -04:00 • committed by GitHub
parent 28be5d4822
commit f81f9f44e2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 0 additions and 400 deletions
Download patch file Download diff file Expand all files Collapse all files

5
skyvern/exceptions.py
View file

@ -1507,11 +1507,6 @@ class IllegitComplete(SkyvernException):
super().__init__(f"Illegit complete{data_str}")
class CachedActionPlanError(SkyvernException):
def __init__(self, message: str) -> None:
super().__init__(message)
class InvalidUrl(SkyvernHTTPException):
def __init__(
self, url: str, *, field_name: str = "url", reason: Literal["malformed", "unsupported scheme"] = "malformed"

31
skyvern/forge/sdk/db/repositories/tasks.py
View file

@ -1297,37 +1297,6 @@ class TasksRepository(BaseRepository):
task_run.captcha_cost = captcha_cost
await session.commit()
@db_operation("cache_task_run")
async def cache_task_run(self, run_id: str, organization_id: str | None = None) -> Run:
async with self.Session() as session:
task_run = (
await session.scalars(
select(TaskRunModel).filter_by(organization_id=organization_id).filter_by(run_id=run_id)
)
).first()
if task_run:
task_run.cached = True
await session.commit()
await session.refresh(task_run)
return Run.model_validate(task_run)
raise NotFoundError(f"Run {run_id} not found")
@db_operation("get_cached_task_run")
async def get_cached_task_run(
self, task_run_type: RunType, url_hash: str | None = None, organization_id: str | None = None
) -> Run | None:
async with self.Session() as session:
query = select(TaskRunModel)
if task_run_type:
query = query.filter_by(task_run_type=task_run_type)
if url_hash:
query = query.filter_by(url_hash=url_hash)
if organization_id:
query = query.filter_by(organization_id=organization_id)
query = query.filter_by(cached=True).order_by(TaskRunModel.created_at.desc())
task_run = (await session.scalars(query)).first()
return Run.model_validate(task_run) if task_run else None
@db_operation("get_run")
async def get_run(
self,

27
skyvern/forge/sdk/db/repositories/workflow_parameters.py
View file

@ -46,7 +46,6 @@ from skyvern.forge.sdk.db.models import (
OnePasswordCredentialParameterModel,
OutputParameterModel,
TaskGenerationModel,
TaskModel,
WorkflowCopilotChatMessageModel,
WorkflowCopilotChatModel,
WorkflowCopilotCompletionCriteriaSetModel,
@ -65,7 +64,6 @@ from skyvern.forge.sdk.db.utils import (
from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
from skyvern.forge.sdk.schemas.copilot_turn_outcome import TurnOutcome
from skyvern.forge.sdk.schemas.task_generations import TaskGeneration
from skyvern.forge.sdk.schemas.tasks import Task, TaskStatus
from skyvern.forge.sdk.schemas.workflow_copilot import (
COPILOT_PROPOSAL_METADATA_KEY,
CopilotAttachedFile,
@ -2023,28 +2021,3 @@ class WorkflowParametersRepository(BaseRepository):
await session.refresh(action)
return Action.model_validate(action)
raise NotFoundError(f"Action {action_id}")
@db_operation("retrieve_action_plan")
async def retrieve_action_plan(self, task: Task) -> list[Action]:
async with self.Session() as session:
subquery = (
select(TaskModel.task_id)
.filter(TaskModel.url == task.url)
.filter(TaskModel.navigation_goal == task.navigation_goal)
.filter(TaskModel.status == TaskStatus.completed)
.order_by(TaskModel.created_at.desc())
.limit(1)
.subquery()
)
query = (
select(ActionModel)
.filter(ActionModel.task_id == subquery.c.task_id)
.order_by(ActionModel.step_order, ActionModel.action_order, ActionModel.created_at)
)
actions = (await session.scalars(query)).all()
# hydrate_action, not Action.model_validate: the base model has no action_json merge, so
# validating the row directly drops every subclass field a cached action was recorded
# with. Matches every other retrieval site.
return [hydrate_action(action) for action in actions]

273
skyvern/webeye/actions/caching.py
View file

@ -1,273 +0,0 @@
import structlog
from skyvern.exceptions import CachedActionPlanError
from skyvern.forge import app
from skyvern.forge.prompts import prompt_engine
from skyvern.forge.sdk.api.llm.api_handler_factory import get_org_aware_secondary_llm_api_handler
from skyvern.forge.sdk.models import Step
from skyvern.forge.sdk.schemas.tasks import Task
from skyvern.webeye.actions.action_types import ActionType
from skyvern.webeye.actions.actions import Action, ActionStatus, SelectOption
from skyvern.webeye.scraper.scraped_page import ScrapedPage
LOG = structlog.get_logger()
async def retrieve_action_plan(task: Task, step: Step, scraped_page: ScrapedPage) -> list[Action]:
try:
return await _retrieve_action_plan(task, step, scraped_page)
except Exception as e:
LOG.exception("Failed to retrieve action plan", exception=e)
return []
async def _retrieve_action_plan(task: Task, step: Step, scraped_page: ScrapedPage) -> list[Action]:
# V0: use the previous action plan if there is a completed task with the same url and navigation goal
# get completed task with the same url and navigation goal
# TODO(kerem): don't use step_order, get all the previous actions instead
cached_actions = await app.DATABASE.workflow_params.retrieve_action_plan(task=task)
if not cached_actions:
LOG.info("No cached actions found for the task, fallback to no-cache mode")
return []
# Get the existing actions for this task from the database. Then find the actions that are already executed by looking at
# the source_action_id field for this task's actions.
previous_actions = await app.DATABASE.tasks.get_previous_actions_for_task(task_id=task.task_id)
executed_cached_actions = []
remaining_cached_actions = []
action_matching_complete = False
if previous_actions:
for idx, cached_action in enumerate(cached_actions):
if not action_matching_complete:
should_be_matching_action = previous_actions[idx]
if not should_be_matching_action.source_action_id:
# If there is an action without a source_action_id, it means we already went back to no-cache mode
# and we should not try to reuse the previous action plan since it's not possible to determine which
# action we should execute next
return []
action_id_to_match = (
cached_action.source_action_id if cached_action.source_action_id else cached_action.action_id
)
if should_be_matching_action.source_action_id == action_id_to_match:
executed_cached_actions.append(cached_action)
if idx == len(previous_actions) - 1:
# If we've reached the end of the previous actions, we've completed matching.
action_matching_complete = True
else:
# If we've reached an action that doesn't match the source_action_id of the previous actions,
# we've completed matching.
action_matching_complete = True
remaining_cached_actions.append(cached_action)
else:
remaining_cached_actions.append(cached_action)
else:
remaining_cached_actions = cached_actions
action_matching_complete = True
# For any remaining cached action,
# check if the element hash exists in the current scraped page. Add them to a list until we can't find a match. Always keep the
# actions without an element hash.
cached_actions_to_execute: list[Action] = []
found_element_with_no_hash = False
for cached_action in remaining_cached_actions:
# The actions without an element hash: TerminateAction CompleteAction NullAction SolveCaptchaAction WaitAction
# For these actions, we can't check if the element hash exists in the current scraped page.
# For that reason, we're going to make sure they're executed always as the first action in each step.
if not cached_action.skyvern_element_hash:
if not found_element_with_no_hash and len(cached_actions_to_execute) > 0:
# If we've already added actions with element hashes to the list before we encounter an action without an element hash,
# we need to execute the actions we already added first. We want the actions without an element hash
# to be executed as the first actions in each step. We're ok with executing multiple actions without an element hash
# in a row, but we want them to be executed in a new step after we wait & scrape the page again.
break
cached_actions_to_execute.append(cached_action)
found_element_with_no_hash = True
continue
matching_element_ids = scraped_page.hash_to_element_ids.get(cached_action.skyvern_element_hash)
if matching_element_ids and len(matching_element_ids) == 1:
cached_actions_to_execute.append(cached_action)
continue
# After this point, we can't continue adding actions to the plan, so we break and continue with what we have.
# Because this action has either no hash-match or multiple hash-matches, we can't continue.
elif matching_element_ids and len(matching_element_ids) > 1:
LOG.warning(
"Found multiple elements with the same hash, stop matching",
element_hash=cached_action.skyvern_element_hash,
element_ids=matching_element_ids,
)
break
else:
LOG.info("No element found with the hash", element_hash=cached_action.skyvern_element_hash, sampling=True)
break
# If there are no items in the list we just built, we need to revert back to no-cache mode. Return empty list.
if not cached_actions_to_execute:
return []
LOG.info("Found cached actions to execute", actions=cached_actions_to_execute)
actions: list[Action] = []
for idx, cached_action in enumerate(cached_actions_to_execute):
updated_action = cached_action.model_copy()
updated_action.status = ActionStatus.pending
updated_action.source_action_id = (
cached_action.source_action_id if cached_action.source_action_id else cached_action.action_id
)
updated_action.workflow_run_id = task.workflow_run_id
updated_action.task_id = task.task_id
updated_action.step_id = step.step_id
updated_action.step_order = step.order
updated_action.action_order = idx
# Reset the action response to None so we don't use the previous answers
updated_action.response = None
# The code-block recorder's line stamp belongs to the run that produced it. Carrying it into
# a reused action would point the model at unrelated code, and readers treat the stamp as
# proof the row came from the recorder. model_copy is shallow, so rebuild rather than mutate.
if isinstance(updated_action.output, dict) and "code_line" in updated_action.output:
updated_action.output = {k: v for k, v in updated_action.output.items() if k != "code_line"}
# Update the element id with the element id from the current scraped page, matched by element hash
if cached_action.skyvern_element_hash:
matching_element_ids = scraped_page.hash_to_element_ids.get(cached_action.skyvern_element_hash)
if matching_element_ids and len(matching_element_ids) == 1:
matching_element_id = matching_element_ids[0]
updated_action.element_id = matching_element_id
updated_action.skyvern_element_data = scraped_page.id_to_element_dict.get(matching_element_id)
else:
raise CachedActionPlanError(
"All elements with either no hash or multiple hashes should have been already filtered out"
)
actions.append(updated_action)
# Check for unsupported actions before personalizing the actions
# Classify the supported actions into two groups:
# 1. Actions that can be cached with a query
# 2. Actions that can be cached without a query
# We'll use this classification to determine if we should continue with caching or fallback to no-cache mode
check_for_unsupported_actions(actions)
personalized_actions = await personalize_actions(task=task, step=step, scraped_page=scraped_page, actions=actions)
LOG.info("Personalized cached actions are ready", actions=personalized_actions)
return personalized_actions
async def personalize_actions(
task: Task,
step: Step,
actions: list[Action],
scraped_page: ScrapedPage,
) -> list[Action]:
queries_and_answers: dict[str, str | None] = {action.intention: None for action in actions if action.intention}
answered_queries: dict[str, str] = {}
if queries_and_answers:
# Call LLM to get answers for the queries only if there are queries to answer
answered_queries = await get_user_detail_answers(
task=task, step=step, scraped_page=scraped_page, queries_and_answers=queries_and_answers
)
personalized_actions = []
for action in actions:
query = action.intention
if query and (personalized_answer := answered_queries.get(query)):
current_personized_actions = await personalize_action(
action, query, personalized_answer, task, step, scraped_page
)
personalized_actions.extend(current_personized_actions)
else:
personalized_actions.append(action)
return personalized_actions
async def get_user_detail_answers(
task: Task, step: Step, scraped_page: ScrapedPage, queries_and_answers: dict[str, str | None]
) -> dict[str, str]:
try:
question_answering_prompt = prompt_engine.load_prompt(
"answer-user-detail-questions",
navigation_goal=task.navigation_goal,
navigation_payload=task.navigation_payload,
queries_and_answers=queries_and_answers,
)
llm_response = await get_org_aware_secondary_llm_api_handler(default=app.SECONDARY_LLM_API_HANDLER)(
prompt=question_answering_prompt, step=step, screenshots=None, prompt_name="answer-user-detail-questions"
)
return llm_response
except Exception as e:
LOG.exception("Failed to get user detail answers", exception=e)
# TODO: custom exception so we can fallback to no-cache mode by catching it
raise e
async def personalize_action(
action: Action,
query: str,
answer: str,
task: Task,
step: Step,
scraped_page: ScrapedPage,
) -> list[Action]:
action.intention = query
action.response = answer
if action.action_type == ActionType.INPUT_TEXT:
action.text = answer
if not answer:
return []
elif action.action_type == ActionType.UPLOAD_FILE:
action.file_url = answer
elif action.action_type == ActionType.CLICK:
# TODO: we only use cached action.intention. send the intention, navigation payload + navigation goal, html
# to small llm and make a decision of which elements to click. Not clicking anything is also an option here
# FIXME: what about the cached action is to use click action to upload a file with file_url?
return [action]
elif action.action_type == ActionType.SELECT_OPTION:
# TODO: send the selection action with the original/previous option value. Our current selection agent
# is already able to handle it. But we need to pass the empty option. otherwise, there will be an unexpected error about the null value
action.option = SelectOption()
return [action]
elif action.action_type in [
ActionType.COMPLETE,
ActionType.WAIT,
ActionType.SOLVE_CAPTCHA,
ActionType.NULL_ACTION,
ActionType.DOWNLOAD_FILE,
]:
return [action]
elif action.action_type == ActionType.TERMINATE:
return []
else:
raise CachedActionPlanError(
f"Unsupported action type for personalization, fallback to no-cache mode: {action.action_type}"
)
return [action]
def check_for_unsupported_actions(actions: list[Action]) -> None:
supported_actions = [
ActionType.INPUT_TEXT,
ActionType.WAIT,
ActionType.CLICK,
ActionType.COMPLETE,
ActionType.DOWNLOAD_FILE,
]
supported_actions_with_query = [ActionType.INPUT_TEXT]
for action in actions:
query = action.intention
if action.action_type not in supported_actions:
raise CachedActionPlanError(
f"This action type does not support caching: {action.action_type}, fallback to no-cache mode"
)
if query and action.action_type not in supported_actions_with_query:
raise CachedActionPlanError(
f"This action type does not support caching with a query: {action.action_type}, fallback to no-cache mode"
)

64
tests/unit/test_hydrate_action_resilience.py
View file

@ -4,7 +4,6 @@ Regression for SKY-9512: a single bad action row should not crash the timeline
endpoint via ValidationError propagation.
"""
import inspect
from datetime import datetime, timezone
from types import SimpleNamespace
from typing import Any
@ -15,7 +14,6 @@ from fastapi import FastAPI
from fastapi.testclient import TestClient
from skyvern.forge.sdk.db.repositories.tasks import TasksRepository
from skyvern.forge.sdk.db.repositories.workflow_parameters import WorkflowParametersRepository
from skyvern.forge.sdk.db.utils import ACTION_TYPE_TO_CLASS, hydrate_action
from skyvern.forge.sdk.routes.routers import legacy_base_router
from skyvern.forge.sdk.schemas.organizations import Organization
@ -26,18 +24,12 @@ from skyvern.webeye.actions.actions import (
Action,
ActionStatus,
ActionType,
ClickAction,
MoveAction,
UploadFileAction,
)
from tests.unit.helpers import make_action_row as _action_row
from tests.unit.helpers import make_session_factory_yielding as _session_yielding
def _task() -> SimpleNamespace:
return SimpleNamespace(task_id="tsk_test", url="https://example.com", navigation_goal="goal")
def _organization() -> Organization:
now = datetime.now(timezone.utc)
return Organization(
@ -95,62 +87,6 @@ def test_every_action_type_hydrates_as_its_concrete_model() -> None:
assert len(set(ACTION_TYPE_TO_CLASS.values())) == len(ActionType)
@pytest.mark.asyncio
async def test_subclass_fields_survive_the_production_retrieval_path() -> None:
"""SKY-12874 / AC6. Drives the repository method itself, not a hand-rolled stand-in.
``retrieve_action_plan`` used ``Action.model_validate(row)``, which has no action_json merge, so
every cached action came back as a base ``Action`` with its subclass fields gone. A test that
calls ``hydrate_action`` directly proves the helper works and says nothing about retrieval.
"""
row = _action_row(
action_type=ActionType.UPLOAD_FILE,
element_id="7",
action_json={"element_id": "7", "file_url": "https://example.com/a.pdf", "is_upload_file_tag": True},
)
repo = WorkflowParametersRepository.__new__(WorkflowParametersRepository)
with patch.object(WorkflowParametersRepository, "Session", _session_yielding([row]), create=True):
retrieved = await inspect.unwrap(WorkflowParametersRepository.retrieve_action_plan)(repo, task=_task())
assert len(retrieved) == 1
action = retrieved[0]
assert isinstance(action, UploadFileAction)
assert action.file_url == "https://example.com/a.pdf"
assert action.is_upload_file_tag is True
# caching.retrieve_action_plan then copies, retargets by element hash and personalizes.
retargeted = action.model_copy()
retargeted.element_id = "9"
retargeted.file_url = "https://example.com/b.pdf"
assert isinstance(retargeted, UploadFileAction)
assert retargeted.action_type == ActionType.UPLOAD_FILE
assert retargeted.file_url == "https://example.com/b.pdf"
assert retargeted.is_upload_file_tag is True
@pytest.mark.asyncio
async def test_a_cached_download_click_is_retrieved_as_a_download_click() -> None:
"""The behavioural consequence of the fix above, pinned deliberately.
``ActionHandler.handle_action`` selects its download-capturing path with
``isinstance(action, ClickAction) and action.download``. A base ``Action`` failed that isinstance
check, so a cached click recorded as a download silently skipped download capture on replay.
This affects unenrolled runs, which is the population the security ACs give no cover for.
"""
row = _action_row(
action_type=ActionType.CLICK,
element_id="3",
action_json={"element_id": "3", "download": True},
)
repo = WorkflowParametersRepository.__new__(WorkflowParametersRepository)
with patch.object(WorkflowParametersRepository, "Session", _session_yielding([row]), create=True):
retrieved = await inspect.unwrap(WorkflowParametersRepository.retrieve_action_plan)(repo, task=_task())
action = retrieved[0]
assert isinstance(action, ClickAction)
assert action.download is True
def _get_task_actions(*rows: SimpleNamespace, headers: dict[str, str] | None = None) -> Any:
repo = TasksRepository.__new__(TasksRepository)
test_app = FastAPI()