fix: fail closed when review analysis omits its plan

src/forge/workflow/nodes/implement_review.py

@@ -27,6 +27,31 @@
 )
 
 
+def _review_plan_has_actionable_items(plan_text: str) -> bool:
+    """Validate a review plan and report whether it contains actionable items."""
+    lines = [line.strip() for line in plan_text.splitlines()]
+    nonempty_lines = [line for line in lines if line]
+    if not nonempty_lines or nonempty_lines[0] != "# Review Plan":
+        raise ValueError("Review plan is missing the required '# Review Plan' heading")
+
+    try:
+        actionable_start = lines.index("## Actionable Items") + 1
+    except ValueError:
+        return False
+
+    actionable_lines = []
+    for line in lines[actionable_start:]:
+        if line.startswith("## "):
+            break
+        actionable_lines.append(line)
+
+    if not any(line.startswith("### Item ") for line in actionable_lines):
+        raise ValueError(
+            "Review plan has an Actionable Items section but no '### Item ...' entries"
+        )
+    return True
+
+
 def review_response_gate(state: WorkflowState) -> WorkflowState:
     """Pause workflow awaiting human confirmation of contested review comments."""
     ticket_key = state["ticket_key"]
@@ -181,6 +206,18 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
             repo_name=current_repo,
         )
 
+        plan_path = Path(workspace_path) / _REVIEW_PLAN_FILE
+        if not plan_path.exists():
+            raise RuntimeError(
+                "Review analysis completed without producing the required "
+                f"{_REVIEW_PLAN_FILE} artifact"
+            )
+
+        plan_text = plan_path.read_text().strip()
+        if not plan_text:
+            raise RuntimeError(f"Review analysis produced an empty {_REVIEW_PLAN_FILE} artifact")
+        has_actionable_items = _review_plan_has_actionable_items(plan_text)
+
         # ── Check for objections ──────────────────────────────────────────────
         objections_path = Path(workspace_path) / _REVIEW_OBJECTIONS_FILE
         if objections_path.exists():
@@ -205,10 +242,7 @@ async def implement_review(state: WorkflowState) -> WorkflowState:
 
         # ── Phase 2: Implementation container ────────────────────────────────
         # Only runs if the analysis produced actionable items.
-        plan_path = Path(workspace_path) / _REVIEW_PLAN_FILE
-        plan_text = plan_path.read_text().strip() if plan_path.exists() else ""
-
-        if not plan_text or plan_text == "# No actionable items":
+        if not has_actionable_items:
             logger.info(f"No actionable review items for {ticket_key} — nothing to implement")
         else:
             fix_prompt = load_prompt("implement-review-fix", ticket_key=ticket_key)

tests/unit/workflow/test_implement_review.py

@@ -11,23 +11,26 @@
 
 
 class TestReviewStateFields:
-
     def test_review_comments_in_review_integration_state(self):
         """review_comments must be a field in ReviewIntegrationState."""
         from forge.workflow.base import ReviewIntegrationState
+
         assert "review_comments" in ReviewIntegrationState.__annotations__
 
     def test_contested_comments_in_review_integration_state(self):
         from forge.workflow.base import ReviewIntegrationState
+
         assert "contested_comments" in ReviewIntegrationState.__annotations__
 
     def test_review_response_posted_in_review_integration_state(self):
         from forge.workflow.base import ReviewIntegrationState
+
         assert "review_response_posted" in ReviewIntegrationState.__annotations__
 
     def test_initial_feature_state_has_empty_review_fields(self):
         from forge.models.workflow import TicketType
         from forge.workflow.feature.state import create_initial_feature_state
+
         state = create_initial_feature_state(
             thread_id="t", ticket_key="TEST-1", ticket_type=TicketType.FEATURE
         )
@@ -40,7 +43,6 @@ def test_initial_feature_state_has_empty_review_fields(self):
 
 
 class TestHumanReviewRoutingToImplementReview:
-
     def test_changes_requested_routes_to_implement_review_not_implement_task(self):
         """On changes_requested, route to implement_review, not implement_task."""
         from forge.workflow.nodes.human_review import route_human_review
@@ -79,7 +81,6 @@ def test_paused_still_routes_to_end(self):
 
 
 class TestReviewResponseGate:
-
     def test_review_response_gate_pauses_workflow(self):
         """review_response_gate sets is_paused=True."""
         from forge.workflow.nodes.implement_review import review_response_gate
@@ -103,8 +104,8 @@ def test_route_review_response_confirmed_resumes_implement_review(self):
         state = make_workflow_state(
             current_node="review_response_gate",
             is_paused=False,
-            revision_requested=True,   # human confirmed — implement it
-            contested_comments=[],     # cleared by worker
+            revision_requested=True,  # human confirmed — implement it
+            contested_comments=[],  # cleared by worker
         )
         assert route_review_response(state) == "implement_review"
 
@@ -137,88 +138,86 @@ def test_route_review_response_paused_returns_end(self):
 
 
 class TestImplementReviewInFeatureGraph:
-
     def test_implement_review_is_a_node(self):
         """implement_review must be a node in the feature graph."""
         from forge.workflow.feature.graph import build_feature_graph
+
         graph = build_feature_graph()
         compiled = graph.compile()
         assert "implement_review" in compiled.get_graph().nodes
 
     def test_review_response_gate_is_a_node(self):
         """review_response_gate must be a node in the feature graph."""
         from forge.workflow.feature.graph import build_feature_graph
+
         graph = build_feature_graph()
         compiled = graph.compile()
         assert "review_response_gate" in compiled.get_graph().nodes
 
     def test_human_review_gate_has_implement_review_edge(self):
         """human_review_gate must have an edge to implement_review."""
         from forge.workflow.feature.graph import build_feature_graph
+
         graph = build_feature_graph()
         compiled = graph.compile()
-        targets = {
-            e.target for e in compiled.get_graph().edges
-            if e.source == "human_review_gate"
-        }
+        targets = {e.target for e in compiled.get_graph().edges if e.source == "human_review_gate"}
         assert "implement_review" in targets
 
     def test_implement_task_not_reachable_from_human_review_gate(self):
         """implement_task must NOT be a direct target of human_review_gate."""
         from forge.workflow.feature.graph import build_feature_graph
+
         graph = build_feature_graph()
         compiled = graph.compile()
-        targets = {
-            e.target for e in compiled.get_graph().edges
-            if e.source == "human_review_gate"
-        }
+        targets = {e.target for e in compiled.get_graph().edges if e.source == "human_review_gate"}
         assert "implement_task" not in targets
 
 
 # ── implement_review in bug graph ────────────────────────────────────────────
 
 
 class TestImplementReviewInBugGraph:
-
     def test_implement_review_is_a_node_in_bug_graph(self):
         from forge.workflow.bug.graph import build_bug_graph
+
         graph = build_bug_graph()
         compiled = graph.compile()
         assert "implement_review" in compiled.get_graph().nodes
 
     def test_human_review_gate_routes_to_implement_review_in_bug_graph(self):
         from forge.workflow.bug.graph import build_bug_graph
+
         graph = build_bug_graph()
         compiled = graph.compile()
-        targets = {
-            e.target for e in compiled.get_graph().edges
-            if e.source == "human_review_gate"
-        }
+        targets = {e.target for e in compiled.get_graph().edges if e.source == "human_review_gate"}
         assert "implement_review" in targets
 
 
 # ── resume routing ────────────────────────────────────────────────────────────
 
 
 class TestResumeRoutingForReviewNodes:
-
     def test_feature_resumes_at_implement_review(self):
         from forge.workflow.feature.graph import route_by_ticket_type
+
         state = make_workflow_state(current_node="implement_review")
         assert route_by_ticket_type(state) == "implement_review"
 
     def test_feature_resumes_at_review_response_gate(self):
         from forge.workflow.feature.graph import route_by_ticket_type
+
         state = make_workflow_state(current_node="review_response_gate")
         assert route_by_ticket_type(state) == "review_response_gate"
 
     def test_bug_resumes_at_implement_review(self):
         from forge.workflow.bug.graph import route_entry
+
         state = make_workflow_state(current_node="implement_review")
         assert route_entry(state) == "implement_review"
 
     def test_bug_resumes_at_review_response_gate(self):
         from forge.workflow.bug.graph import route_entry
+
         state = make_workflow_state(current_node="review_response_gate")
         assert route_entry(state) == "review_response_gate"
 
@@ -227,7 +226,6 @@ def test_bug_resumes_at_review_response_gate(self):
 
 
 class TestImplementReviewErrorHandling:
-
     @pytest.mark.asyncio
     async def test_workspace_prepare_failure_increments_retry_count(self):
         """ValueError from prepare_workspace increments retry_count."""
@@ -251,9 +249,107 @@ async def test_workspace_prepare_failure_increments_retry_count(self):
         assert result["retry_count"] == 2
         assert "workspace gone" in result["last_error"]
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("plan_contents", [None, ""])
+    async def test_missing_or_empty_review_plan_is_an_analysis_failure(
+        self, tmp_path, plan_contents
+    ):
+        """Analysis must fail closed when the required plan artifact is invalid."""
+        from forge.workflow.nodes.implement_review import implement_review
 
-class TestImplementReviewStatusComment:
+        mock_git = MagicMock()
+        mock_runner = MagicMock()
+
+        async def run_analysis(**_kwargs):
+            if plan_contents is not None:
+                plan_path = tmp_path / ".forge" / "review-plan.md"
+                plan_path.write_text(plan_contents)
 
+        mock_runner.run = AsyncMock(side_effect=run_analysis)
+        state = make_workflow_state(
+            current_node="implement_review",
+            retry_count=1,
+            workspace_path=str(tmp_path),
+            current_repo="org/repo",
+            feedback_comment="Fix the tests",
+            current_pr_number=17,
+        )
+
+        with (
+            patch(
+                "forge.workflow.nodes.implement_review.prepare_workspace",
+                return_value=(str(tmp_path), mock_git),
+            ),
+            patch(
+                "forge.workflow.nodes.implement_review._post_review_addressing_comment",
+                new=AsyncMock(),
+            ),
+            patch(
+                "forge.workflow.nodes.implement_review._fetch_pr_review_comments",
+                new=AsyncMock(return_value="# PR Review Feedback\n"),
+            ),
+            patch(
+                "forge.workflow.nodes.implement_review.ContainerRunner",
+                return_value=mock_runner,
+            ),
+        ):
+            result = await implement_review(state)
+
+        assert result["current_node"] == "implement_review"
+        assert result["retry_count"] == 2
+        assert "review-plan.md" in result["last_error"]
+
+
+class TestReviewPlanValidation:
+    def test_plan_without_actionable_section_is_a_valid_noop(self):
+        from forge.workflow.nodes.implement_review import (
+            _review_plan_has_actionable_items,
+        )
+
+        plan = """# Review Plan
+
+## Acknowledged (not addressed)
+
+### Intentional behavior
+
+No change required.
+"""
+
+        assert _review_plan_has_actionable_items(plan) is False
+
+    def test_plan_with_actionable_item_requires_implementation(self):
+        from forge.workflow.nodes.implement_review import (
+            _review_plan_has_actionable_items,
+        )
+
+        plan = """# Review Plan
+
+## Actionable Items
+
+### Item 1: Fix retry handling
+
+**Change:** Fail closed.
+"""
+
+        assert _review_plan_has_actionable_items(plan) is True
+
+    @pytest.mark.parametrize(
+        "plan",
+        [
+            "# No actionable items",
+            "# Review Plan\n\n## Actionable Items\n\nNothing listed.",
+        ],
+    )
+    def test_malformed_plan_is_rejected(self, plan):
+        from forge.workflow.nodes.implement_review import (
+            _review_plan_has_actionable_items,
+        )
+
+        with pytest.raises(ValueError):
+            _review_plan_has_actionable_items(plan)
+
+
+class TestImplementReviewStatusComment:
     @pytest.mark.asyncio
     async def test_posts_addressing_review_comment_when_review_work_starts(self, tmp_path):
         """implement_review posts an informational PR status when work starts."""
@@ -268,7 +364,17 @@ async def test_posts_addressing_review_comment_when_review_work_starts(self, tmp
         mock_github.create_issue_comment = AsyncMock()
         mock_github.close = AsyncMock()
         mock_runner = MagicMock()
-        mock_runner.run = AsyncMock()
+
+        async def write_no_action_plan(**_kwargs):
+            plan_path = tmp_path / ".forge" / "review-plan.md"
+            plan_path.write_text(
+                "# Review Plan\n\n"
+                "## Acknowledged (not addressed)\n\n"
+                "### No changes requested\n\n"
+                "The feedback requires no code changes.\n"
+            )
+
+        mock_runner.run = AsyncMock(side_effect=write_no_action_plan)
 
         state = make_workflow_state(
             current_node="implement_review",
@@ -290,7 +396,9 @@ async def test_posts_addressing_review_comment_when_review_work_starts(self, tmp
                 new=AsyncMock(return_value="# PR Review Feedback\n"),
             ),
             patch("forge.workflow.nodes.implement_review.GitHubClient", return_value=mock_github),
-            patch("forge.workflow.nodes.implement_review.ContainerRunner", return_value=mock_runner),
+            patch(
+                "forge.workflow.nodes.implement_review.ContainerRunner", return_value=mock_runner
+            ),
         ):
             result = await implement_review(state)