diff --git a/src/forge/workflow/nodes/implement_review.py b/src/forge/workflow/nodes/implement_review.py index d37a9eed..2ee8d6ec 100644 --- a/src/forge/workflow/nodes/implement_review.py +++ b/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) diff --git a/tests/unit/workflow/test_implement_review.py b/tests/unit/workflow/test_implement_review.py index e22627af..72f4c15f 100644 --- a/tests/unit/workflow/test_implement_review.py +++ b/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,10 +138,10 @@ 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 @@ -148,6 +149,7 @@ def test_implement_review_is_a_node(self): 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 @@ -155,23 +157,19 @@ def test_review_response_gate_is_a_node(self): 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 @@ -179,21 +177,19 @@ def test_implement_task_not_reachable_from_human_review_gate(self): 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 @@ -201,24 +197,27 @@ def test_human_review_gate_routes_to_implement_review_in_bug_graph(self): 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)