diff --git a/skyvern/exceptions.py b/skyvern/exceptions.py index 90d937070..4e5f2da42 100644 --- a/skyvern/exceptions.py +++ b/skyvern/exceptions.py @@ -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" diff --git a/skyvern/forge/sdk/db/repositories/tasks.py b/skyvern/forge/sdk/db/repositories/tasks.py index 5e0b71671..671f11209 100644 --- a/skyvern/forge/sdk/db/repositories/tasks.py +++ b/skyvern/forge/sdk/db/repositories/tasks.py @@ -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, diff --git a/skyvern/forge/sdk/db/repositories/workflow_parameters.py b/skyvern/forge/sdk/db/repositories/workflow_parameters.py index cfbf5547c..a8c42db72 100644 --- a/skyvern/forge/sdk/db/repositories/workflow_parameters.py +++ b/skyvern/forge/sdk/db/repositories/workflow_parameters.py @@ -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] diff --git a/skyvern/webeye/actions/caching.py b/skyvern/webeye/actions/caching.py deleted file mode 100644 index 255512163..000000000 --- a/skyvern/webeye/actions/caching.py +++ /dev/null @@ -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" - ) diff --git a/tests/unit/test_hydrate_action_resilience.py b/tests/unit/test_hydrate_action_resilience.py index f096a9fc1..13a296aa5 100644 --- a/tests/unit/test_hydrate_action_resilience.py +++ b/tests/unit/test_hydrate_action_resilience.py @@ -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()