feat(SKY-11345): default new copilot data-write blocks to continue_on_failure=false (#6779)

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,

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

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`

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,

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