diff --git a/src/strands_tools/workflow.py b/src/strands_tools/workflow.py
index c9dd0555..dea2262f 100644
--- a/src/strands_tools/workflow.py
+++ b/src/strands_tools/workflow.py
@@ -608,6 +608,26 @@ def start_workflow(self, workflow_id: str) -> Dict:
                 # Get all ready tasks
                 ready_tasks = self.get_ready_tasks(workflow)
 
+                if not active_futures and not ready_tasks:
+                    skipped_at = datetime.now(timezone.utc).isoformat()
+                    for task in workflow["tasks"]:
+                        task_id = task["task_id"]
+                        if workflow["task_results"][task_id]["status"] != "pending":
+                            continue
+                        blocked_by = [
+                            dep_id
+                            for dep_id in task.get("dependencies", [])
+                            if workflow["task_results"][dep_id]["status"] != "completed"
+                        ]
+                        workflow["task_results"][task_id] = {
+                            **workflow["task_results"][task_id],
+                            "status": "skipped",
+                            "result": [{"text": f"Task skipped due to failed dependencies: {', '.join(blocked_by)}"}],
+                            "completed_at": skipped_at,
+                        }
+                        completed_tasks.add(task_id)
+                        logger.warning(f"⏭️ Task '{task_id}' skipped due to failed dependencies: {blocked_by}")
+
                 # Prepare tasks for parallel submission with batching
                 tasks_to_submit = []
                 max_concurrent = self.task_executor.max_workers
@@ -694,15 +714,21 @@ def start_workflow(self, workflow_id: str) -> Dict:
 
             # Calculate success rate
             completed_count = sum(1 for result in workflow["task_results"].values() if result["status"] == "completed")
+            failed_count = sum(1 for result in workflow["task_results"].values() if result["status"] == "error")
+            skipped_count = sum(1 for result in workflow["task_results"].values() if result["status"] == "skipped")
             success_rate = (completed_count / total_tasks) * 100 if total_tasks > 0 else 0
+            status_text = (
+                "completed with partial success" if failed_count or skipped_count else "completed successfully"
+            )
 
             return {
                 "status": "success",
                 "content": [
                     {
                         "text": (
-                            f"🎉 Workflow '{workflow_id}' completed successfully! "
-                            f"({completed_count}/{total_tasks} tasks succeeded - {success_rate:.1f}%)"
+                            f"🎉 Workflow '{workflow_id}' {status_text}! "
+                            f"({completed_count}/{total_tasks} tasks succeeded, "
+                            f"{failed_count} failed, {skipped_count} skipped - {success_rate:.1f}%)"
                         )
                     }
                 ],
@@ -805,7 +831,7 @@ def get_workflow_status(self, workflow_id: str) -> Dict:
             table.add_column("⏱️ Duration", justify="right")
 
             # Count statuses
-            status_counts = {"pending": 0, "completed": 0, "error": 0, "running": 0}
+            status_counts = {"pending": 0, "completed": 0, "error": 0, "running": 0, "skipped": 0}
             total_tasks = len(workflow["tasks"])
 
             for task in workflow["tasks"]:
@@ -837,6 +863,8 @@ def get_workflow_status(self, workflow_id: str) -> Dict:
                     status_display = "[red]❌[/red]"
                 elif status == "running":
                     status_display = "[yellow]🔄[/yellow]"
+                elif status == "skipped":
+                    status_display = "[yellow]⏭️[/yellow]"
                 else:
                     status_display = "[blue]⏳[/blue]"
 
@@ -861,6 +889,7 @@ def get_workflow_status(self, workflow_id: str) -> Dict:
                     f"✅ **Completed:** {status_counts['completed']}",
                     f"⏳ **Pending:** {status_counts['pending']}",
                     f"❌ **Failed:** {status_counts['error']}",
+                    f"⏭️ **Skipped:** {status_counts['skipped']}",
                     f"🔄 **Active Workers:** {self.task_executor.active_workers}/{self.task_executor.max_workers}",
                 ]
             )
diff --git a/tests/test_workflow.py b/tests/test_workflow.py
index bfc35b3f..adce1139 100644
--- a/tests/test_workflow.py
+++ b/tests/test_workflow.py
@@ -4,6 +4,7 @@
 
 import json
 import tempfile
+from concurrent.futures import Future
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -572,6 +573,52 @@ def test_get_ready_tasks_with_completed_dependencies(self, mock_parent_agent):
         assert len(ready_tasks) == 1
         assert ready_tasks[0]["task_id"] == "task2"
 
+    def test_start_workflow_skips_tasks_blocked_by_failed_dependencies(self, mock_parent_agent, mock_workflow_dir):
+        """Test workflow exits when failed dependencies block pending tasks."""
+        with patch.object(workflow_module, "WORKFLOW_DIR", Path(mock_workflow_dir)):
+            manager = workflow_module.WorkflowManager(mock_parent_agent)
+            workflow_id = "deadlock_test"
+            tasks = [
+                {"task_id": "research", "description": "Research"},
+                {"task_id": "analysis1", "description": "Analyze", "dependencies": ["research"]},
+                {"task_id": "analysis2", "description": "Analyze", "dependencies": ["research"]},
+                {"task_id": "report", "description": "Report", "dependencies": ["analysis1", "analysis2"]},
+            ]
+            manager.create_workflow(workflow_id, tasks)
+
+            def submit_tasks(tasks_to_submit):
+                futures = {}
+                for namespaced_task_id, _task_func, args, _kwargs in tasks_to_submit:
+                    task = args[0]
+                    future = Future()
+                    if task["task_id"] == "analysis2":
+                        future.set_result({"status": "error", "content": [{"text": "failed"}]})
+                    else:
+                        future.set_result({"status": "success", "content": [{"text": "done"}]})
+                    futures[namespaced_task_id] = future
+                return futures
+
+            sleep_calls = 0
+
+            def fail_if_loop_spins(_seconds):
+                nonlocal sleep_calls
+                sleep_calls += 1
+                if sleep_calls > 10:
+                    raise AssertionError("workflow loop did not terminate")
+
+            with (
+                patch.object(manager.task_executor, "submit_tasks", side_effect=submit_tasks),
+                patch("strands_tools.workflow.time.sleep", side_effect=fail_if_loop_spins),
+            ):
+                result = manager.start_workflow(workflow_id)
+
+            workflow = manager.get_workflow(workflow_id)
+            assert result["status"] == "success"
+            assert "partial success" in result["content"][0]["text"]
+            assert workflow["task_results"]["analysis2"]["status"] == "error"
+            assert workflow["task_results"]["report"]["status"] == "skipped"
+            assert "failed dependencies" in workflow["task_results"]["report"]["result"][0]["text"]
+
 
 class TestWorkflowEdgeCases:
     """Test edge cases and error conditions."""