|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Test nested Workflow blocks in for_each with pause/resume. |
| 3 | +
|
| 4 | +This test reproduces a bug where: |
| 5 | +1. Parent workflow has for_each: sequential with Workflow blocks |
| 6 | +2. Child workflow pauses (Prompt block) |
| 7 | +3. User provides feedback/response |
| 8 | +4. Response is incorrectly routed to parent instead of child |
| 9 | +
|
| 10 | +The bug causes: |
| 11 | +- Child workflow never completes its approval flow |
| 12 | +- Remaining for_each iterations are skipped |
| 13 | +- Sub-workflows are never saved |
| 14 | +
|
| 15 | +Test Scenario: |
| 16 | +1. Parent workflow with for_each: sequential calling child workflows |
| 17 | +2. Child workflows have Prompt blocks that pause for approval |
| 18 | +3. Resume first child with "yes" |
| 19 | +4. Verify first child completes with approval |
| 20 | +5. Second child should pause for its approval |
| 21 | +6. Resume second child with "yes" |
| 22 | +7. Verify both children completed successfully |
| 23 | +""" |
| 24 | + |
| 25 | +import json |
| 26 | +from typing import Any |
| 27 | + |
| 28 | +import pytest |
| 29 | +from mcp.types import TextContent |
| 30 | +from test_mcp_client import get_mcp_client |
| 31 | + |
| 32 | + |
| 33 | +class TestNestedWorkflowPauseResume: |
| 34 | + """Test pause/resume for nested Workflow blocks in for_each loops.""" |
| 35 | + |
| 36 | + @pytest.mark.asyncio |
| 37 | + async def test_nested_workflow_in_foreach_sequential_pause_resume(self) -> None: |
| 38 | + """Test that nested workflow pause/resume works correctly in for_each sequential. |
| 39 | +
|
| 40 | + This is the core bug reproduction test: |
| 41 | + 1. Start parent workflow with 2 items |
| 42 | + 2. First child pauses for approval |
| 43 | + 3. Resume with "yes" -> first child should complete |
| 44 | + 4. Second child should pause for approval (not parent!) |
| 45 | + 5. Resume with "yes" -> second child should complete |
| 46 | + 6. Parent workflow should complete with all items processed |
| 47 | + """ |
| 48 | + async with get_mcp_client() as client: |
| 49 | + # Step 1: Start parent workflow |
| 50 | + exec_result = await client.call_tool( |
| 51 | + "execute_workflow", |
| 52 | + arguments={ |
| 53 | + "workflow": "nested-workflow-in-foreach-parent", |
| 54 | + "inputs": { |
| 55 | + "work_items": [ |
| 56 | + {"name": "item1", "value": "value1"}, |
| 57 | + {"name": "item2", "value": "value2"}, |
| 58 | + ] |
| 59 | + }, |
| 60 | + "debug": True, |
| 61 | + }, |
| 62 | + ) |
| 63 | + |
| 64 | + exec_content = exec_result.content[0] |
| 65 | + assert isinstance(exec_content, TextContent) |
| 66 | + exec_response: dict[str, Any] = json.loads(exec_content.text) |
| 67 | + |
| 68 | + # Verify workflow paused at first child |
| 69 | + assert exec_response["status"] == "paused", ( |
| 70 | + f"Expected workflow to pause, got status: {exec_response.get('status')}" |
| 71 | + ) |
| 72 | + assert "job_id" in exec_response, "Expected job_id in paused response" |
| 73 | + job_id = exec_response["job_id"] |
| 74 | + |
| 75 | + # The prompt should be from the CHILD workflow (item1) |
| 76 | + prompt = exec_response.get("prompt", "") |
| 77 | + assert "item1" in prompt, ( |
| 78 | + f"Expected prompt to mention 'item1' (first child), got: {prompt}" |
| 79 | + ) |
| 80 | + |
| 81 | + # Step 2: Resume first child with "yes" |
| 82 | + resume1_result = await client.call_tool( |
| 83 | + "resume_workflow", |
| 84 | + arguments={ |
| 85 | + "job_id": job_id, |
| 86 | + "response": "yes", |
| 87 | + "debug": True, |
| 88 | + }, |
| 89 | + ) |
| 90 | + |
| 91 | + resume1_content = resume1_result.content[0] |
| 92 | + assert isinstance(resume1_content, TextContent) |
| 93 | + resume1_response: dict[str, Any] = json.loads(resume1_content.text) |
| 94 | + |
| 95 | + # Step 3: Should pause again for SECOND child (item2) |
| 96 | + # THIS IS WHERE THE BUG MANIFESTS: |
| 97 | + # - Bug behavior: status == "success" (parent completes, skipping item2) |
| 98 | + # - Correct behavior: status == "paused" with prompt for item2 |
| 99 | + assert resume1_response["status"] == "paused", ( |
| 100 | + f"Expected workflow to pause for item2, got: {resume1_response.get('status')}. " |
| 101 | + f"Resume response was incorrectly routed to parent workflow. " |
| 102 | + f"Full response: {resume1_response}" |
| 103 | + ) |
| 104 | + |
| 105 | + # Verify prompt is for second child (item2) |
| 106 | + prompt2 = resume1_response.get("prompt", "") |
| 107 | + assert "item2" in prompt2, ( |
| 108 | + f"Expected prompt to mention 'item2' (second child), got: {prompt2}. " |
| 109 | + f"This indicates for_each iteration was not correctly resumed." |
| 110 | + ) |
| 111 | + |
| 112 | + job_id_2 = resume1_response.get("job_id", job_id) |
| 113 | + |
| 114 | + # Step 4: Resume second child with "yes" |
| 115 | + resume2_result = await client.call_tool( |
| 116 | + "resume_workflow", |
| 117 | + arguments={ |
| 118 | + "job_id": job_id_2, |
| 119 | + "response": "yes", |
| 120 | + "debug": True, |
| 121 | + }, |
| 122 | + ) |
| 123 | + |
| 124 | + resume2_content = resume2_result.content[0] |
| 125 | + assert isinstance(resume2_content, TextContent) |
| 126 | + resume2_response: dict[str, Any] = json.loads(resume2_content.text) |
| 127 | + |
| 128 | + # Step 5: Now workflow should complete successfully |
| 129 | + assert resume2_response["status"] == "success", ( |
| 130 | + f"Expected workflow to complete after both children approved, " |
| 131 | + f"got status: {resume2_response.get('status')}. " |
| 132 | + f"Error: {resume2_response.get('error')}" |
| 133 | + ) |
| 134 | + |
| 135 | + # Verify outputs |
| 136 | + outputs = resume2_response.get("outputs", {}) |
| 137 | + assert outputs.get("setup_completed") is True |
| 138 | + assert outputs.get("all_items_processed") is True |
| 139 | + assert outputs.get("finalize_completed") is True |
| 140 | + |
| 141 | + @pytest.mark.asyncio |
| 142 | + async def test_nested_workflow_single_item_pause_resume(self) -> None: |
| 143 | + """Test simpler case: single item for_each with nested workflow pause. |
| 144 | +
|
| 145 | + This isolates the core pause/resume logic without multiple iterations. |
| 146 | + """ |
| 147 | + async with get_mcp_client() as client: |
| 148 | + # Start with single item |
| 149 | + exec_result = await client.call_tool( |
| 150 | + "execute_workflow", |
| 151 | + arguments={ |
| 152 | + "workflow": "nested-workflow-in-foreach-parent", |
| 153 | + "inputs": { |
| 154 | + "work_items": [ |
| 155 | + {"name": "single_item", "value": "single_value"}, |
| 156 | + ] |
| 157 | + }, |
| 158 | + "debug": True, |
| 159 | + }, |
| 160 | + ) |
| 161 | + |
| 162 | + exec_content = exec_result.content[0] |
| 163 | + assert isinstance(exec_content, TextContent) |
| 164 | + exec_response: dict[str, Any] = json.loads(exec_content.text) |
| 165 | + |
| 166 | + # Should pause for the single child |
| 167 | + assert exec_response["status"] == "paused" |
| 168 | + job_id = exec_response["job_id"] |
| 169 | + |
| 170 | + prompt = exec_response.get("prompt", "") |
| 171 | + assert "single_item" in prompt |
| 172 | + |
| 173 | + # Resume with approval |
| 174 | + resume_result = await client.call_tool( |
| 175 | + "resume_workflow", |
| 176 | + arguments={ |
| 177 | + "job_id": job_id, |
| 178 | + "response": "yes", |
| 179 | + "debug": True, |
| 180 | + }, |
| 181 | + ) |
| 182 | + |
| 183 | + resume_content = resume_result.content[0] |
| 184 | + assert isinstance(resume_content, TextContent) |
| 185 | + resume_response: dict[str, Any] = json.loads(resume_content.text) |
| 186 | + |
| 187 | + # Should complete (only one item) |
| 188 | + assert resume_response["status"] == "success", ( |
| 189 | + f"Expected success after single item approval, got: {resume_response}" |
| 190 | + ) |
| 191 | + |
| 192 | + outputs = resume_response.get("outputs", {}) |
| 193 | + assert outputs.get("all_items_processed") is True |
| 194 | + |
| 195 | + @pytest.mark.asyncio |
| 196 | + async def test_nested_workflow_feedback_iteration(self) -> None: |
| 197 | + """Test that feedback (non-approval response) triggers re-design iteration. |
| 198 | +
|
| 199 | + This simulates the workflow-creator flow where: |
| 200 | + 1. First response is feedback ("add more details") |
| 201 | + 2. Second response is approval ("approve") |
| 202 | + """ |
| 203 | + async with get_mcp_client() as client: |
| 204 | + # Start workflow |
| 205 | + exec_result = await client.call_tool( |
| 206 | + "execute_workflow", |
| 207 | + arguments={ |
| 208 | + "workflow": "nested-workflow-in-foreach-parent", |
| 209 | + "inputs": { |
| 210 | + "work_items": [ |
| 211 | + {"name": "test_item", "value": "test_value"}, |
| 212 | + ] |
| 213 | + }, |
| 214 | + "debug": True, |
| 215 | + }, |
| 216 | + ) |
| 217 | + |
| 218 | + exec_content = exec_result.content[0] |
| 219 | + assert isinstance(exec_content, TextContent) |
| 220 | + exec_response: dict[str, Any] = json.loads(exec_content.text) |
| 221 | + |
| 222 | + assert exec_response["status"] == "paused" |
| 223 | + job_id = exec_response["job_id"] |
| 224 | + |
| 225 | + # Send denial (should complete with denied branch) |
| 226 | + resume_result = await client.call_tool( |
| 227 | + "resume_workflow", |
| 228 | + arguments={ |
| 229 | + "job_id": job_id, |
| 230 | + "response": "no", |
| 231 | + "debug": True, |
| 232 | + }, |
| 233 | + ) |
| 234 | + |
| 235 | + resume_content = resume_result.content[0] |
| 236 | + assert isinstance(resume_content, TextContent) |
| 237 | + resume_response: dict[str, Any] = json.loads(resume_content.text) |
| 238 | + |
| 239 | + # Should complete (denial is a valid response that completes the workflow) |
| 240 | + assert resume_response["status"] == "success", ( |
| 241 | + f"Expected success after denial, got: {resume_response}" |
| 242 | + ) |
| 243 | + |
| 244 | + |
| 245 | +if __name__ == "__main__": |
| 246 | + pytest.main([__file__, "-v"]) |
0 commit comments