diff --git a/src/graph/nodes.py b/src/graph/nodes.py
index 990864b..e97fd1e 100644
--- a/src/graph/nodes.py
+++ b/src/graph/nodes.py
@@ -100,6 +100,26 @@ def validate_and_fix_plan(plan: dict, enforce_web_search: bool = False) -> dict:
 
     steps = plan.get("steps", [])
 
+    # ============================================================
+    # SECTION 1: Repair missing step_type fields (Issue #650 fix)
+    # ============================================================
+    for idx, step in enumerate(steps):
+        if not isinstance(step, dict):
+            continue
+        
+        # Check if step_type is missing or empty
+        if "step_type" not in step or not step.get("step_type"):
+            # Infer step_type based on need_search value
+            inferred_type = "research" if step.get("need_search", False) else "processing"
+            step["step_type"] = inferred_type
+            logger.info(
+                f"Repaired missing step_type for step {idx} ({step.get('title', 'Untitled')}): "
+                f"inferred as '{inferred_type}' based on need_search={step.get('need_search', False)}"
+            )
+
+    # ============================================================
+    # SECTION 2: Enforce web search requirements
+    # ============================================================
     if enforce_web_search:
         # Check if any step has need_search=true
         has_search_step = any(step.get("need_search", False) for step in steps)
diff --git a/src/prompts/planner.md b/src/prompts/planner.md
index 12be23b..8a71fa8 100644
--- a/src/prompts/planner.md
+++ b/src/prompts/planner.md
@@ -164,6 +164,30 @@ When planning information gathering, consider these key aspects and ensure COMPR
 - Do not include steps for summarizing or consolidating the gathered information.
 - **CRITICAL**: Verify that your plan includes at least one step with `need_search: true` before finalizing
 
+## CRITICAL REQUIREMENT: step_type Field
+
+**⚠️ IMPORTANT: You MUST include the `step_type` field for EVERY step in your plan. This is mandatory and cannot be omitted.**
+
+For each step you create, you MUST explicitly set ONE of these values:
+- `"research"` - For steps that gather information via web search or retrieval (when `need_search: true`)
+- `"processing"` - For steps that analyze, compute, or process data without web search (when `need_search: false`)
+
+**Validation Checklist - For EVERY Step, Verify ALL 4 Fields Are Present:**
+- [ ] `need_search`: Must be either `true` or `false`
+- [ ] `title`: Must describe what the step does
+- [ ] `description`: Must specify exactly what data to collect
+- [ ] `step_type`: Must be either `"research"` or `"processing"`
+
+**Common Mistake to Avoid:**
+- ❌ WRONG: `{"need_search": true, "title": "...", "description": "..."}`  (missing `step_type`)
+- ✅ CORRECT: `{"need_search": true, "title": "...", "description": "...", "step_type": "research"}`
+
+**Step Type Assignment Rules:**
+- If `need_search` is `true` → use `step_type: "research"`
+- If `need_search` is `false` → use `step_type: "processing"`
+
+Failure to include `step_type` for any step will cause validation errors and prevent the research plan from executing.
+
 # Output Format
 
 **CRITICAL: You MUST output a valid JSON object that exactly matches the Plan interface below. Do not include any text before or after the JSON. Do not use markdown code blocks. Output ONLY the raw JSON.**
@@ -189,24 +213,38 @@ interface Plan {
 }
 ```
 
-**Example Output:**
+**Example Output (with BOTH research and processing steps):**
 ```json
 {
   "locale": "en-US",
   "has_enough_context": false,
-  "thought": "To understand the current market trends in AI, we need to gather comprehensive information about recent developments, key players, and market dynamics.",
+  "thought": "To understand the current market trends in AI, we need to gather comprehensive information about recent developments, key players, and market dynamics, then analyze and synthesize this data.",
   "title": "AI Market Research Plan",
   "steps": [
     {
       "need_search": true,
       "title": "Current AI Market Analysis",
-      "description": "Collect data on market size, growth rates, major players, and investment trends in AI sector.",
+      "description": "Collect data on market size, growth rates, major players, investment trends, recent product launches, and technological breakthroughs in the AI sector from reliable sources.",
       "step_type": "research"
+    },
+    {
+      "need_search": true,
+      "title": "Emerging Trends and Future Outlook",
+      "description": "Research emerging trends, expert forecasts, and future predictions for the AI market including expected growth, new market segments, and regulatory changes.",
+      "step_type": "research"
+    },
+    {
+      "need_search": false,
+      "title": "Synthesize and Analyze Market Data",
+      "description": "Analyze and synthesize all collected data to identify patterns, calculate market growth projections, compare competitor positions, and create data visualizations.",
+      "step_type": "processing"
     }
   ]
 }
 ```
 
+**NOTE:** Every step must have a `step_type` field set to either `"research"` or `"processing"`. Research steps (with `need_search: true`) gather data. Processing steps (with `need_search: false`) analyze the gathered data.
+
 # Notes
 
 - Focus on information gathering in research steps - delegate all calculations to processing steps
diff --git a/src/prompts/planner.zh_CN.md b/src/prompts/planner.zh_CN.md
index 89e5f24..3c19773 100644
--- a/src/prompts/planner.zh_CN.md
+++ b/src/prompts/planner.zh_CN.md
@@ -164,6 +164,30 @@ CURRENT_TIME: {{ CURRENT_TIME }}
 - 不要包括总结或整合收集信息的步骤。
 - **关键**：在最终确定之前验证你的计划包括至少一个带有`need_search: true`的步骤
 
+## 关键要求：step_type字段
+
+**⚠️ 重要：你必须为计划中的每一个步骤包含`step_type`字段。这是强制性的，不能省略。**
+
+对于你创建的每个步骤，你必须显式设置以下值之一：
+- `"research"` - 用于通过网络搜索或检索来收集信息的步骤（当`need_search: true`时）
+- `"processing"` - 用于分析、计算或处理数据而不进行网络搜索的步骤（当`need_search: false`时）
+
+**验证清单 - 对于每一个步骤，验证所有4个字段都存在：**
+- [ ] `need_search`：必须是`true`或`false`
+- [ ] `title`：必须描述步骤的作用
+- [ ] `description`：必须指定要收集的确切数据
+- [ ] `step_type`：必须是`"research"`或`"processing"`
+
+**常见错误避免：**
+- ❌ 错误：`{"need_search": true, "title": "...", "description": "..."}` （缺少`step_type`）
+- ✅ 正确：`{"need_search": true, "title": "...", "description": "...", "step_type": "research"}`
+
+**步骤类型分配规则：**
+- 如果`need_search`是`true` → 使用`step_type: "research"`
+- 如果`need_search`是`false` → 使用`step_type: "processing"`
+
+任何步骤缺少`step_type`都将导致验证错误，阻止研究计划执行。
+
 # 输出格式
 
 **关键：你必须输出与下面的Plan接口完全匹配的有效JSON对象。不包括JSON之前或之后的任何文本。不使用markdown代码块。仅输出原始JSON。**
@@ -189,24 +213,38 @@ interface Plan {
 }
 ```
 
-**示例输出：**
+**示例输出（包含研究步骤和处理步骤）：**
 ```json
 {
   "locale": "zh-CN",
   "has_enough_context": false,
-  "thought": "要理解AI中当前的市场趋势，我们需要收集关于最近发展、主要参与者和市场动态的全面信息。",
+  "thought": "要理解AI中当前的市场趋势，我们需要收集关于最近发展、主要参与者和市场动态的全面信息，然后分析和综合这些数据。",
   "title": "AI市场研究计划",
   "steps": [
     {
       "need_search": true,
       "title": "当前AI市场分析",
-      "description": "收集关于AI部门市场规模、增长率、主要参与者和投资趋势的数据。",
+      "description": "从可靠来源收集关于市场规模、增长率、主要参与者、投资趋势、最近的产品发布和AI部门技术突破的数据。",
       "step_type": "research"
+    },
+    {
+      "need_search": true,
+      "title": "新兴趋势和未来前景",
+      "description": "研究新兴趋势、专家预测和AI市场的未来预测，包括预期增长、新的市场细分和监管变化。",
+      "step_type": "research"
+    },
+    {
+      "need_search": false,
+      "title": "综合和分析市场数据",
+      "description": "分析和综合所有收集的数据，以识别模式、计算市场增长预测、比较竞争对手位置并创建数据可视化。",
+      "step_type": "processing"
     }
   ]
 }
 ```
 
+**注意：** 每个步骤必须有一个`step_type`字段，设置为`"research"`或`"processing"`。研究步骤（带有`need_search: true`）收集数据。处理步骤（带有`need_search: false`）分析收集的数据。
+
 # 注意
 
 - 在研究步骤中关注信息收集——将所有计算委托给处理步骤
diff --git a/tests/integration/test_nodes.py b/tests/integration/test_nodes.py
index f7de25b..074862e 100644
--- a/tests/integration/test_nodes.py
+++ b/tests/integration/test_nodes.py
@@ -1864,6 +1864,214 @@ def test_clarification_no_history_defaults_to_topic():
     assert result.update["clarified_research_topic"] == "What is quantum computing?"
 
 
+# ============================================================================
+# Issue #650: Pydantic validation errors (missing step_type field)
+# ============================================================================
+
+
+def test_planner_node_issue_650_missing_step_type_basic():
+    """Test planner_node with missing step_type fields (Issue #650)."""
+    from src.graph.nodes import validate_and_fix_plan
+
+    # Simulate LLM response with missing step_type (Issue #650 scenario)
+    llm_response = {
+        "locale": "en-US",
+        "has_enough_context": False,
+        "thought": "Need to gather data",
+        "title": "Test Plan",
+        "steps": [
+            {
+                "need_search": True,
+                "title": "Research Step",
+                "description": "Gather info",
+                # step_type MISSING - this is the issue
+            },
+            {
+                "need_search": False,
+                "title": "Processing Step",
+                "description": "Analyze",
+                # step_type MISSING
+            },
+        ],
+    }
+
+    # Apply the fix
+    fixed_plan = validate_and_fix_plan(llm_response)
+
+    # Verify all steps have step_type after fix
+    assert isinstance(fixed_plan, dict)
+    assert fixed_plan["steps"][0]["step_type"] == "research"
+    assert fixed_plan["steps"][1]["step_type"] == "processing"
+    assert all("step_type" in step for step in fixed_plan["steps"])
+
+
+def test_planner_node_issue_650_water_footprint_scenario():
+    """Test the exact water footprint query scenario from Issue #650."""
+    from src.graph.nodes import validate_and_fix_plan
+
+    # Approximate the exact plan structure that caused Issue #650
+    # "How many liters of water are required to produce 1 kg of beef?"
+    llm_response = {
+        "locale": "en-US",
+        "has_enough_context": False,
+        "thought": "You asked about water footprint of beef - need comprehensive data gathering",
+        "title": "Research Plan — Water Footprint of 1 kg of Beef",
+        "steps": [
+            {
+                "need_search": True,
+                "title": "Authoritative global estimates",
+                "description": "Collect peer-reviewed estimates",
+                # MISSING step_type
+            },
+            {
+                "need_search": True,
+                "title": "System-specific data",
+                "description": "Gather system-level variation data",
+                # MISSING step_type
+            },
+            {
+                "need_search": False,
+                "title": "Synthesize estimates",
+                "description": "Calculate scenario-based estimates",
+                # MISSING step_type
+            },
+        ],
+    }
+
+    # Apply the fix
+    fixed_plan = validate_and_fix_plan(llm_response)
+
+    # Verify structure - all steps should have step_type filled in
+    assert len(fixed_plan["steps"]) == 3
+    assert fixed_plan["steps"][0]["step_type"] == "research"
+    assert fixed_plan["steps"][1]["step_type"] == "research"
+    assert fixed_plan["steps"][2]["step_type"] == "processing"
+    assert all("step_type" in step for step in fixed_plan["steps"])
+
+
+def test_planner_node_issue_650_validation_error_fixed():
+    """Test that the validation error from Issue #650 is now prevented."""
+    from src.graph.nodes import validate_and_fix_plan
+
+    # This is the exact type of response that caused the error in Issue #650
+    malformed_response = {
+        "locale": "en-US",
+        "has_enough_context": False,
+        "title": "Test",
+        "thought": "Test",
+        "steps": [
+            {
+                "need_search": True,
+                "title": "Step 1",
+                "description": "Test description",
+                # Missing step_type - caused "Field required" error
+            },
+        ],
+    }
+
+    # Before fix would raise:
+    # ValidationError: 1 validation error for Plan
+    # steps.0.step_type Field required [type=missing, ...]
+
+    # After fix should succeed without raising exception
+    fixed = validate_and_fix_plan(malformed_response)
+
+    # Verify the fix was applied
+    assert fixed["steps"][0]["step_type"] in ["research", "processing"]
+    assert "step_type" in fixed["steps"][0]
+
+
+def test_human_feedback_node_issue_650_plan_parsing():
+    """Test human_feedback_node with Issue #650 plan that has missing step_type."""
+    from src.graph.nodes import human_feedback_node
+
+    # Plan with missing step_type fields
+    state = {
+        "current_plan": json.dumps(
+            {
+                "locale": "en-US",
+                "has_enough_context": False,
+                "title": "Test Plan",
+                "thought": "Test",
+                "steps": [
+                    {
+                        "need_search": True,
+                        "title": "Step 1",
+                        "description": "Gather",
+                        # MISSING step_type
+                    },
+                ],
+            }
+        ),
+        "plan_iterations": 0,
+        "auto_accepted_plan": True,
+    }
+
+    config = MagicMock()
+    with patch(
+        "src.graph.nodes.Configuration.from_runnable_config",
+        return_value=MagicMock(enforce_web_search=False),
+    ):
+        with patch("src.graph.nodes.Plan.model_validate", side_effect=lambda x: x):
+            with patch("src.graph.nodes.repair_json_output", side_effect=lambda x: x):
+                result = human_feedback_node(state, config)
+
+                # Should succeed without validation error
+                assert isinstance(result, Command)
+                assert result.goto == "research_team"
+
+
+def test_plan_validation_with_all_issue_650_error_scenarios():
+    """Test all variations of Issue #650 error scenarios."""
+    from src.graph.nodes import validate_and_fix_plan
+
+    test_scenarios = [
+        # Missing step_type with need_search=true
+        {
+            "steps": [
+                {"need_search": True, "title": "R", "description": "D"},
+            ]
+        },
+        # Missing step_type with need_search=false
+        {
+            "steps": [
+                {"need_search": False, "title": "P", "description": "D"},
+            ]
+        },
+        # Multiple missing step_types
+        {
+            "steps": [
+                {"need_search": True, "title": "R1", "description": "D"},
+                {"need_search": True, "title": "R2", "description": "D"},
+                {"need_search": False, "title": "P", "description": "D"},
+            ]
+        },
+        # Mix of missing and present step_type
+        {
+            "steps": [
+                {"need_search": True, "title": "R", "description": "D", "step_type": "research"},
+                {"need_search": False, "title": "P", "description": "D"},
+            ]
+        },
+    ]
+
+    for scenario in test_scenarios:
+        plan = {
+            "locale": "en-US",
+            "has_enough_context": False,
+            "title": "Test",
+            "thought": "Test",
+            **scenario,
+        }
+
+        # Should not raise exception
+        fixed = validate_and_fix_plan(plan)
+
+        # All steps should have step_type after fix
+        for step in fixed["steps"]:
+            assert "step_type" in step
+            assert step["step_type"] in ["research", "processing"]
+
 def test_clarification_skips_specific_topics():
     """Coordinator should skip clarification for already specific topics."""
     from langchain_core.messages import AIMessage
diff --git a/tests/unit/graph/test_plan_validation.py b/tests/unit/graph/test_plan_validation.py
new file mode 100644
index 0000000..64d6ac7
--- /dev/null
+++ b/tests/unit/graph/test_plan_validation.py
@@ -0,0 +1,484 @@
+# Copyright (c) 2025 Bytedance Ltd. and/or its affiliates
+# SPDX-License-Identifier: MIT
+
+import pytest
+from unittest.mock import patch, MagicMock
+
+from src.graph.nodes import validate_and_fix_plan
+
+
+class TestValidateAndFixPlanStepTypeRepair:
+    """Test step_type field repair logic (Issue #650 fix)."""
+
+    def test_repair_missing_step_type_with_need_search_true(self):
+        """Test that missing step_type is inferred as 'research' when need_search=true."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research Step",
+                    "description": "Gather data",
+                    # step_type is MISSING
+                }
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "research"
+
+    def test_repair_missing_step_type_with_need_search_false(self):
+        """Test that missing step_type is inferred as 'processing' when need_search=false."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze data",
+                    # step_type is MISSING
+                }
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "processing"
+
+    def test_repair_missing_step_type_default_to_processing(self):
+        """Test that missing step_type defaults to 'processing' when need_search is not specified."""
+        plan = {
+            "steps": [
+                {
+                    "title": "Unknown Step",
+                    "description": "Do something",
+                    # need_search is MISSING, step_type is MISSING
+                }
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "processing"
+
+    def test_repair_empty_step_type_field(self):
+        """Test that empty step_type field is repaired."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research Step",
+                    "description": "Gather data",
+                    "step_type": "",  # Empty string
+                }
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "research"
+
+    def test_repair_null_step_type_field(self):
+        """Test that null step_type field is repaired."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze data",
+                    "step_type": None,
+                }
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "processing"
+
+    def test_multiple_steps_with_mixed_missing_step_types(self):
+        """Test repair of multiple steps with different missing step_type scenarios."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research 1",
+                    "description": "Gather",
+                    # MISSING step_type
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing 1",
+                    "description": "Analyze",
+                    "step_type": "processing",  # Already has step_type
+                },
+                {
+                    "need_search": True,
+                    "title": "Research 2",
+                    "description": "More gathering",
+                    # MISSING step_type
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][1]["step_type"] == "processing"  # Should remain unchanged
+        assert result["steps"][2]["step_type"] == "research"
+
+    def test_preserve_explicit_step_type(self):
+        """Test that explicitly provided step_type values are preserved."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research Step",
+                    "description": "Gather",
+                    "step_type": "research",
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze",
+                    "step_type": "processing",
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        # Should remain unchanged
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][1]["step_type"] == "processing"
+
+    def test_repair_logs_warning(self):
+        """Test that repair operations are logged."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Missing Type Step",
+                    "description": "Gather",
+                }
+            ]
+        }
+
+        with patch("src.graph.nodes.logger") as mock_logger:
+            validate_and_fix_plan(plan)
+            # Should log repair operation
+            mock_logger.info.assert_called()
+            call_args = str(mock_logger.info.call_args)
+            assert "Repaired missing step_type" in call_args
+
+    def test_non_dict_plan_returns_unchanged(self):
+        """Test that non-dict plans are returned unchanged."""
+        plan = "not a dict"
+        result = validate_and_fix_plan(plan)
+        assert result == plan
+
+    def test_plan_with_non_dict_step_skipped(self):
+        """Test that non-dict step items are skipped without error."""
+        plan = {
+            "steps": [
+                "not a dict step",  # This should be skipped
+                {
+                    "need_search": True,
+                    "title": "Valid Step",
+                    "description": "Gather",
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        # Non-dict step should be unchanged, valid step should be fixed
+        assert result["steps"][0] == "not a dict step"
+        assert result["steps"][1]["step_type"] == "research"
+
+    def test_empty_steps_list(self):
+        """Test that plan with empty steps list is handled gracefully."""
+        plan = {"steps": []}
+        result = validate_and_fix_plan(plan)
+        assert result["steps"] == []
+
+    def test_missing_steps_key(self):
+        """Test that plan without steps key is handled gracefully."""
+        plan = {"locale": "en-US", "title": "Test"}
+        result = validate_and_fix_plan(plan)
+        assert "steps" not in result
+
+
+class TestValidateAndFixPlanWebSearchEnforcement:
+    """Test web search enforcement logic."""
+
+    def test_enforce_web_search_sets_first_research_step(self):
+        """Test that enforce_web_search=True sets need_search on first research step."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": False,
+                    "title": "Research Step",
+                    "description": "Gather",
+                    "step_type": "research",
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze",
+                    "step_type": "processing",
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        # First research step should have web search enabled
+        assert result["steps"][0]["need_search"] is True
+        assert result["steps"][1]["need_search"] is False
+
+    def test_enforce_web_search_converts_first_step(self):
+        """Test that enforce_web_search converts first step to research if needed."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": False,
+                    "title": "First Step",
+                    "description": "Do something",
+                    "step_type": "processing",
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        # First step should be converted to research with web search
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][0]["need_search"] is True
+
+    def test_enforce_web_search_with_existing_search_step(self):
+        """Test that enforce_web_search doesn't modify if search step already exists."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research Step",
+                    "description": "Gather",
+                    "step_type": "research",
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze",
+                    "step_type": "processing",
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        # Steps should remain unchanged
+        assert result["steps"][0]["need_search"] is True
+        assert result["steps"][1]["need_search"] is False
+
+    def test_enforce_web_search_adds_default_step(self):
+        """Test that enforce_web_search adds default research step if no steps exist."""
+        plan = {"steps": []}
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        assert len(result["steps"]) == 1
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][0]["need_search"] is True
+        assert "title" in result["steps"][0]
+        assert "description" in result["steps"][0]
+
+    def test_enforce_web_search_without_steps_key(self):
+        """Test enforce_web_search when steps key is missing."""
+        plan = {"locale": "en-US"}
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        assert len(result.get("steps", [])) > 0
+        assert result["steps"][0]["step_type"] == "research"
+
+
+class TestValidateAndFixPlanIntegration:
+    """Integration tests for step_type repair and web search enforcement together."""
+
+    def test_repair_and_enforce_together(self):
+        """Test that step_type repair and web search enforcement work together."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research Step",
+                    "description": "Gather",
+                    # MISSING step_type
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing Step",
+                    "description": "Analyze",
+                    # MISSING step_type, but enforce_web_search won't change it
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        # step_type should be repaired
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][1]["step_type"] == "processing"
+
+        # First research step should have web search (already has it)
+        assert result["steps"][0]["need_search"] is True
+
+    def test_repair_then_enforce_cascade(self):
+        """Test complex scenario with repair and enforcement cascading."""
+        plan = {
+            "steps": [
+                {
+                    "need_search": False,
+                    "title": "Step 1",
+                    "description": "Do something",
+                    # MISSING step_type
+                },
+                {
+                    "need_search": False,
+                    "title": "Step 2",
+                    "description": "Do something else",
+                    # MISSING step_type
+                },
+            ]
+        }
+
+        result = validate_and_fix_plan(plan, enforce_web_search=True)
+
+        # Step 1: Originally processing but converted to research with web search enforcement
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][0]["need_search"] is True
+
+        # Step 2: Should remain as processing since enforcement already satisfied by step 1
+        assert result["steps"][1]["step_type"] == "processing"
+        assert result["steps"][1]["need_search"] is False
+
+
+class TestValidateAndFixPlanIssue650:
+    """Specific tests for Issue #650 scenarios."""
+
+    def test_issue_650_water_footprint_scenario_fixed(self):
+        """Test the exact scenario from issue #650 - water footprint query with missing step_type."""
+        # This is a simplified version of the actual error from issue #650
+        plan = {
+            "locale": "en-US",
+            "has_enough_context": False,
+            "title": "Research Plan — Water Footprint of 1 kg of Beef",
+            "thought": "You asked: 'How many liters of water are required to produce 1 kg of beef?'",
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Authoritative estimates",
+                    "description": "Collect peer-reviewed estimates",
+                    # MISSING step_type - this caused the error in issue #650
+                },
+                {
+                    "need_search": True,
+                    "title": "System-specific data",
+                    "description": "Gather system-level data",
+                    # MISSING step_type
+                },
+                {
+                    "need_search": False,
+                    "title": "Processing and analysis",
+                    "description": "Compute scenario-based estimates",
+                    # MISSING step_type
+                },
+            ],
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        # All steps should now have step_type
+        assert result["steps"][0]["step_type"] == "research"
+        assert result["steps"][1]["step_type"] == "research"
+        assert result["steps"][2]["step_type"] == "processing"
+
+    def test_issue_650_scenario_passes_pydantic_validation(self):
+        """Test that fixed plan can be validated by Pydantic schema."""
+        from src.prompts.planner_model import Plan as PlanModel
+
+        plan = {
+            "locale": "en-US",
+            "has_enough_context": False,
+            "title": "Research Plan",
+            "thought": "Test thought",
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Research",
+                    "description": "Gather data",
+                    # MISSING step_type
+                },
+            ],
+        }
+
+        # First validate and fix
+        fixed_plan = validate_and_fix_plan(plan)
+
+        # Then try Pydantic validation (should not raise)
+        validated = PlanModel.model_validate(fixed_plan)
+
+        assert validated.steps[0].step_type == "research"
+        assert validated.steps[0].need_search is True
+
+    def test_issue_650_multiple_validation_errors_fixed(self):
+        """Test that plan with multiple missing step_types (like in issue #650) all get fixed."""
+        plan = {
+            "locale": "en-US",
+            "has_enough_context": False,
+            "title": "Complex Plan",
+            "thought": "Research plan",
+            "steps": [
+                {
+                    "need_search": True,
+                    "title": "Step 0",
+                    "description": "Data gathering",
+                },
+                {
+                    "need_search": True,
+                    "title": "Step 1",
+                    "description": "More gathering",
+                },
+                {
+                    "need_search": False,
+                    "title": "Step 2",
+                    "description": "Processing",
+                },
+            ],
+        }
+
+        result = validate_and_fix_plan(plan)
+
+        # All steps should have step_type now
+        for step in result["steps"]:
+            assert "step_type" in step
+            assert step["step_type"] in ["research", "processing"]
+
+    def test_issue_650_no_exceptions_raised(self):
+        """Test that validate_and_fix_plan handles all edge cases without raising exceptions."""
+        test_cases = [
+            {"steps": []},
+            {"steps": [{"need_search": True}]},
+            {"steps": [None, {}]},
+            {"steps": ["invalid"]},
+            {"steps": [{"need_search": True, "step_type": ""}]},
+            "not a dict",
+        ]
+
+        for plan in test_cases:
+            try:
+                result = validate_and_fix_plan(plan)
+                # Should succeed without exception - result may be returned as-is for non-dict
+                # but the function should not raise
+                # No assertion needed; test passes if no exception is raised
+            except Exception as e:
+                pytest.fail(f"validate_and_fix_plan raised exception for {plan}: {e}")