diff --git a/skyvern/forge/sdk/copilot/agent.py b/skyvern/forge/sdk/copilot/agent.py index ad6ffc760..7a3d1a18c 100644 --- a/skyvern/forge/sdk/copilot/agent.py +++ b/skyvern/forge/sdk/copilot/agent.py @@ -73,6 +73,7 @@ from skyvern.forge.sdk.copilot.context import ( TurnNarrativePayload, finalize_discovery_counter_in_global_llm_context, ) +from skyvern.forge.sdk.copilot.data_write_defaults import default_data_write_continue_on_failure from skyvern.forge.sdk.copilot.enforcement import ( artifact_health_blocked, outcome_fully_verified, @@ -2433,6 +2434,10 @@ async def _translate_to_agent_result( ctx.last_test_ok = None workflow_yaml = "" if workflow_yaml: + # Inline REPLACE_WORKFLOW bypasses the update_workflow tool, so apply the same default here. + workflow_yaml = default_data_write_continue_on_failure( + workflow_yaml, ctx.last_workflow_yaml or ctx.workflow_yaml + ) try: last_workflow = _process_workflow_yaml( workflow_id=chat_request.workflow_id, diff --git a/skyvern/forge/sdk/copilot/data_write_defaults.py b/skyvern/forge/sdk/copilot/data_write_defaults.py new file mode 100644 index 000000000..a02357800 --- /dev/null +++ b/skyvern/forge/sdk/copilot/data_write_defaults.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from typing import Any + +import yaml + +from skyvern.forge.sdk.copilot.workflow_credential_utils import parse_workflow_yaml, workflow_blocks +from skyvern.schemas.workflows import BlockType + +# Block types that persist a payload to an external store, where a swallowed +# failure reads back as a completed run with nothing written. +DATA_WRITE_BLOCK_TYPES: frozenset[str] = frozenset( + { + BlockType.GOOGLE_SHEETS_WRITE.value, + BlockType.FILE_UPLOAD.value, + BlockType.UPLOAD_TO_S3.value, + BlockType.DOWNLOAD_TO_S3.value, + } +) + + +def _block_type_name(block: dict[str, Any]) -> str: + return str(block.get("block_type") or "").strip().lower() + + +def _data_write_labels_with_truthy_continue_on_failure(parsed: Any) -> set[str]: + if not isinstance(parsed, dict): + return set() + return { + label + for block in workflow_blocks(parsed) + if _block_type_name(block) in DATA_WRITE_BLOCK_TYPES + and block.get("continue_on_failure") + and isinstance((label := block.get("label")), str) + } + + +def default_data_write_continue_on_failure(new_yaml: str, prior_yaml: str | None) -> str: + """Force continue_on_failure=false on data-write blocks unless the prior workflow + already set it truthy on a block with the same label. Keying on the prior flag value + (not just the label) means a block first added earlier in the same turn, already + defaulted to false in the staged draft, is re-defaulted rather than treated as a + pre-existing explicit value. Pure; recurses into loop_blocks/branches and returns the + input unchanged when nothing needs overriding.""" + parsed = parse_workflow_yaml(new_yaml) + if not isinstance(parsed, dict): + return new_yaml + prior_truthy_labels = _data_write_labels_with_truthy_continue_on_failure( + parse_workflow_yaml(prior_yaml) if prior_yaml else None + ) + changed = False + for block in workflow_blocks(parsed): + if _block_type_name(block) not in DATA_WRITE_BLOCK_TYPES: + continue + if not block.get("continue_on_failure"): + continue + label = block.get("label") + if isinstance(label, str) and label in prior_truthy_labels: + continue + block["continue_on_failure"] = False + changed = True + return yaml.safe_dump(parsed, sort_keys=False) if changed else new_yaml diff --git a/skyvern/forge/sdk/copilot/tools/workflow_update.py b/skyvern/forge/sdk/copilot/tools/workflow_update.py index 3cea0daef..dd4583ae3 100644 --- a/skyvern/forge/sdk/copilot/tools/workflow_update.py +++ b/skyvern/forge/sdk/copilot/tools/workflow_update.py @@ -51,6 +51,7 @@ from skyvern.forge.sdk.copilot.composition_evidence import ( ) from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy from skyvern.forge.sdk.copilot.context import CopilotContext +from skyvern.forge.sdk.copilot.data_write_defaults import default_data_write_continue_on_failure from skyvern.forge.sdk.copilot.enforcement import ( MAX_CODE_AUTHORING_GUARDRAIL_REJECTS, MAX_CREDENTIAL_PRIORITY_AUTHORING_REJECTS, @@ -3377,6 +3378,9 @@ async def _update_workflow( ) return {"ok": False, "error": composition_evidence_error} + # New data-write blocks default to surfacing failures rather than swallowing them. + workflow_yaml = default_data_write_continue_on_failure(workflow_yaml, ctx.workflow_yaml) + try: # A code block renders code-first (goal + plain step timeline) only when it # carries a `prompt`; the model authors the goal as artifact `declared_goal` diff --git a/skyvern/forge/sdk/routes/workflow_copilot.py b/skyvern/forge/sdk/routes/workflow_copilot.py index 643f980da..3755fa805 100644 --- a/skyvern/forge/sdk/routes/workflow_copilot.py +++ b/skyvern/forge/sdk/routes/workflow_copilot.py @@ -38,6 +38,7 @@ from skyvern.forge.sdk.copilot.completion_criteria_store import ( ) from skyvern.forge.sdk.copilot.config import BlockAuthoringPolicy, CopilotConfig, normalize_block_authoring_policy from skyvern.forge.sdk.copilot.context import AgentResult, ProposalDisposition, TurnNarrativePayload +from skyvern.forge.sdk.copilot.data_write_defaults import default_data_write_continue_on_failure from skyvern.forge.sdk.copilot.llm_config import resolve_main_copilot_handler from skyvern.forge.sdk.copilot.output_utils import truncate_output from skyvern.forge.sdk.copilot.recoverable_failure import ( @@ -1174,7 +1175,9 @@ async def copilot_call_llm( global_llm_context = str(global_llm_context) if action_type == "REPLACE_WORKFLOW": - llm_workflow_yaml = action_data.get("workflow_yaml", "") + llm_workflow_yaml = default_data_write_continue_on_failure( + action_data.get("workflow_yaml", ""), chat_request.workflow_yaml + ) applied_workflow_yaml = llm_workflow_yaml try: updated_workflow = _process_workflow_yaml( @@ -1191,15 +1194,18 @@ async def copilot_call_llm( timestamp=datetime.now(timezone.utc), ) ) - corrected_workflow_yaml = await _auto_correct_workflow_yaml( - llm_api_handler=llm_api_handler, - organization_id=organization_id, - user_response=user_response, - workflow_yaml=llm_workflow_yaml, - chat_history=chat_history, - global_llm_context=global_llm_context, - debug_run_info_text=debug_run_info_text, - error=e, + corrected_workflow_yaml = default_data_write_continue_on_failure( + await _auto_correct_workflow_yaml( + llm_api_handler=llm_api_handler, + organization_id=organization_id, + user_response=user_response, + workflow_yaml=llm_workflow_yaml, + chat_history=chat_history, + global_llm_context=global_llm_context, + debug_run_info_text=debug_run_info_text, + error=e, + ), + chat_request.workflow_yaml, ) updated_workflow = _process_workflow_yaml( workflow_id=chat_request.workflow_id, diff --git a/tests/unit/copilot/test_data_write_continue_on_failure.py b/tests/unit/copilot/test_data_write_continue_on_failure.py new file mode 100644 index 000000000..163f9436a --- /dev/null +++ b/tests/unit/copilot/test_data_write_continue_on_failure.py @@ -0,0 +1,186 @@ +from __future__ import annotations + +import textwrap +from typing import Any + +from skyvern.forge.sdk.copilot.data_write_defaults import ( + DATA_WRITE_BLOCK_TYPES, + default_data_write_continue_on_failure, +) +from skyvern.forge.sdk.copilot.workflow_credential_utils import parse_workflow_yaml, workflow_blocks + + +def _continue_on_failure_by_label(workflow_yaml: str) -> dict[str, Any]: + parsed = parse_workflow_yaml(workflow_yaml) + assert isinstance(parsed, dict) + return {block.get("label"): block.get("continue_on_failure") for block in workflow_blocks(parsed)} + + +def _wf(blocks_yaml: str) -> str: + return "workflow_definition:\n blocks:\n" + textwrap.indent(textwrap.dedent(blocks_yaml).strip() + "\n", " ") + + +def test_new_data_write_block_with_true_is_forced_false() -> None: + new_yaml = _wf( + """ + - block_type: google_sheets_write + label: write_rows + continue_on_failure: true + """ + ) + result = default_data_write_continue_on_failure(new_yaml, None) + assert _continue_on_failure_by_label(result)["write_rows"] is False + + +def test_existing_explicit_true_is_preserved() -> None: + saved = _wf( + """ + - block_type: google_sheets_write + label: write_rows + continue_on_failure: true + """ + ) + # Same label re-submitted (e.g. an unrelated edit elsewhere) keeps the user's value. + result = default_data_write_continue_on_failure(saved, saved) + assert _continue_on_failure_by_label(result)["write_rows"] is True + assert result == saved # untouched -> no re-dump + + +def test_non_write_block_is_untouched() -> None: + new_yaml = _wf( + """ + - block_type: google_sheets_read + label: read_rows + continue_on_failure: true + """ + ) + result = default_data_write_continue_on_failure(new_yaml, None) + assert result == new_yaml + assert _continue_on_failure_by_label(result)["read_rows"] is True + + +def test_nested_loop_blocks_and_branches_are_recursed() -> None: + new_yaml = _wf( + """ + - block_type: for_loop + label: loop + loop_blocks: + - block_type: file_upload + label: upload_in_loop + continue_on_failure: true + - block_type: conditional + label: cond + branches: + - blocks: + - block_type: google_sheets_write + label: write_in_branch + continue_on_failure: true + """ + ) + by_label = _continue_on_failure_by_label(default_data_write_continue_on_failure(new_yaml, None)) + assert by_label["upload_in_loop"] is False + assert by_label["write_in_branch"] is False + + +def test_no_prior_forces_all_new_write_blocks_false() -> None: + new_yaml = _wf( + """ + - block_type: upload_to_s3 + label: up + continue_on_failure: true + - block_type: download_to_s3 + label: down + continue_on_failure: true + """ + ) + by_label = _continue_on_failure_by_label(default_data_write_continue_on_failure(new_yaml, None)) + assert by_label["up"] is False + assert by_label["down"] is False + + +def test_absent_continue_on_failure_is_passthrough_no_redump() -> None: + new_yaml = _wf( + """ + - block_type: google_sheets_write + label: write_rows + """ + ) + # Already-safe default (absent) must not trigger a YAML round-trip. + assert default_data_write_continue_on_failure(new_yaml, None) == new_yaml + + +def test_new_write_block_added_to_existing_workflow_is_defaulted() -> None: + prior = _wf( + """ + - block_type: google_sheets_write + label: existing_write + continue_on_failure: true + """ + ) + new_yaml = _wf( + """ + - block_type: google_sheets_write + label: existing_write + continue_on_failure: true + - block_type: file_upload + label: brand_new_upload + continue_on_failure: true + """ + ) + by_label = _continue_on_failure_by_label(default_data_write_continue_on_failure(new_yaml, prior)) + assert by_label["existing_write"] is True # untouched + assert by_label["brand_new_upload"] is False # new -> defaulted + + +def test_same_turn_recreated_block_is_redefaulted_not_preserved() -> None: + # The prior is the staged draft after the copilot already defaulted this block to + # false earlier in the turn; a re-emitted true must be re-defaulted, not preserved. + prior = _wf( + """ + - block_type: google_sheets_write + label: write_rows + continue_on_failure: false + """ + ) + new_yaml = _wf( + """ + - block_type: google_sheets_write + label: write_rows + continue_on_failure: true + """ + ) + result = _continue_on_failure_by_label(default_data_write_continue_on_failure(new_yaml, prior)) + assert result["write_rows"] is False + + +def test_malformed_yaml_returns_input_unchanged() -> None: + assert default_data_write_continue_on_failure("not: [valid", None) == "not: [valid" + assert default_data_write_continue_on_failure("", None) == "" + + +def test_coexisting_code_block_content_survives_redump() -> None: + new_yaml = _wf( + """ + - block_type: code + label: compute + code: | + x = 1 + return {"x": x} + - block_type: google_sheets_write + label: write_rows + continue_on_failure: true + """ + ) + result = default_data_write_continue_on_failure(new_yaml, None) + parsed = parse_workflow_yaml(result) + assert isinstance(parsed, dict) + by_label = {block.get("label"): block for block in workflow_blocks(parsed)} + assert by_label["write_rows"].get("continue_on_failure") is False + assert by_label["compute"].get("code") == 'x = 1\nreturn {"x": x}\n' + + +def test_data_write_set_excludes_reads_and_notifications() -> None: + assert "google_sheets_read" not in DATA_WRITE_BLOCK_TYPES + assert "send_email" not in DATA_WRITE_BLOCK_TYPES + assert "http_request" not in DATA_WRITE_BLOCK_TYPES + assert "google_sheets_write" in DATA_WRITE_BLOCK_TYPES