From b6210e6561fbf75f8dc064c2f1aafa0113951943 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 11:45:07 -0400 Subject: [PATCH 01/19] feat: add pending_ci_event state field and update resume map for backward compat --- src/forge/workflow/base.py | 1 + src/forge/workflow/utils/__init__.py | 2 +- tests/unit/workflow/test_base.py | 25 +++++++++++++++++++++++++ 3 files changed, 27 insertions(+), 1 deletion(-) diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index 3b59fcc5..c6b38146 100644 --- a/src/forge/workflow/base.py +++ b/src/forge/workflow/base.py @@ -73,6 +73,7 @@ class CIIntegrationState(TypedDict, total=False): ci_skipped_checks: list[str] ci_fix_attempt: int ci_fix_max_attempts: int + pending_ci_event: bool class ReviewIntegrationState(TypedDict, total=False): diff --git a/src/forge/workflow/utils/__init__.py b/src/forge/workflow/utils/__init__.py index 80a10fb1..03a751b5 100644 --- a/src/forge/workflow/utils/__init__.py +++ b/src/forge/workflow/utils/__init__.py @@ -25,7 +25,7 @@ "human_review_gate": "human_review_gate", "implement_review": "implement_review", "review_response_gate": "review_response_gate", - "wait_for_ci_gate": "wait_for_ci_gate", + "wait_for_ci_gate": "human_review_gate", # backward compat for in-flight tickets "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "rebase_pr": "rebase_pr", diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py index 4df75da1..6529f1f1 100644 --- a/tests/unit/workflow/test_base.py +++ b/tests/unit/workflow/test_base.py @@ -78,6 +78,20 @@ def test_ci_state_has_required_fields(self): assert "ci_failed_checks" in hints assert "ci_fix_attempt" in hints + def test_ci_state_has_pending_ci_event_field(self): + """CIIntegrationState includes pending_ci_event field.""" + from forge.workflow.base import CIIntegrationState + + hints = get_type_hints(CIIntegrationState) + assert "pending_ci_event" in hints + + def test_pending_ci_event_is_bool_type(self): + """pending_ci_event field is typed as bool.""" + from forge.workflow.base import CIIntegrationState + + hints = get_type_hints(CIIntegrationState) + assert hints["pending_ci_event"] is bool + class TestReviewIntegrationState: """Tests for review integration mixin.""" @@ -156,3 +170,14 @@ def build_graph(self): # Kwargs preserved assert state["extra_field"] == "value" + + +class TestSharedResumeMap: + """Tests for resolve_shared_resume_node backward compat.""" + + def test_wait_for_ci_gate_resolves_to_human_review_gate(self): + """In-flight tickets with wait_for_ci_gate resume at human_review_gate.""" + from forge.workflow.utils import resolve_shared_resume_node + + result = resolve_shared_resume_node("wait_for_ci_gate") + assert result == "human_review_gate" From 1e9e1c311032085a270132caad5065343ef360a4 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 11:49:40 -0400 Subject: [PATCH 02/19] feat: evaluate_ci_status always clears pending_ci_event --- src/forge/workflow/nodes/ci_evaluator.py | 8 +++ .../workflow/nodes/test_ci_attribution.py | 61 +++++++++++++++++++ 2 files changed, 69 insertions(+) create mode 100644 tests/unit/workflow/nodes/test_ci_attribution.py diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py index 84eb5812..710b45f5 100644 --- a/src/forge/workflow/nodes/ci_evaluator.py +++ b/src/forge/workflow/nodes/ci_evaluator.py @@ -57,6 +57,7 @@ async def evaluate_ci_status(state: WorkflowState) -> WorkflowState: **state, "ci_status": "no_prs", "current_node": "complete", + "pending_ci_event": False, } ) @@ -110,6 +111,7 @@ def _is_skipped(check: dict) -> bool: **state, "ci_status": "pending", "current_node": "ci_evaluator", # Stay here, wait for webhook + "pending_ci_event": False, } ) @@ -163,6 +165,7 @@ def _is_skipped(check: dict) -> bool: "current_node": "human_review_gate", "last_error": None, "ci_fix_attempt": 0, + "pending_ci_event": False, } ) @@ -178,6 +181,7 @@ def _is_skipped(check: dict) -> bool: **state, "ci_status": "pending", "current_node": "ci_evaluator", + "pending_ci_event": False, } ) @@ -190,6 +194,7 @@ def _is_skipped(check: dict) -> bool: **state, "ci_status": "pending", "current_node": "ci_evaluator", + "pending_ci_event": False, } ) @@ -204,6 +209,7 @@ def _is_skipped(check: dict) -> bool: "ci_failed_checks": failed_checks, "current_node": "ci_evaluator", "last_error": "CI fix attempt limit reached", + "pending_ci_event": False, } ) @@ -216,6 +222,7 @@ def _is_skipped(check: dict) -> bool: "ci_failed_checks": failed_checks, "ci_fix_attempt": next_attempt, "current_node": "attempt_ci_fix", + "pending_ci_event": False, } ) @@ -226,6 +233,7 @@ def _is_skipped(check: dict) -> bool: "last_error": str(e), "current_node": "ci_evaluator", "retry_count": state.get("retry_count", 0) + 1, + "pending_ci_event": False, } finally: await github.close() diff --git a/tests/unit/workflow/nodes/test_ci_attribution.py b/tests/unit/workflow/nodes/test_ci_attribution.py new file mode 100644 index 00000000..7958620b --- /dev/null +++ b/tests/unit/workflow/nodes/test_ci_attribution.py @@ -0,0 +1,61 @@ +"""Tests for CI attribution and evaluate_ci_status pending_ci_event clearing.""" + +import pytest +from unittest.mock import AsyncMock, MagicMock, patch + + +BASE_STATE = { + "ticket_key": "TEST-1", + "pr_urls": ["https://github.com/org/repo/pull/1"], + "ci_fix_attempt": 0, + "ci_fix_max_attempts": 5, + "pending_ci_event": True, + "ci_skipped_checks": [], +} + + +@pytest.mark.asyncio +@patch("forge.workflow.nodes.ci_evaluator.GitHubClient") +async def test_evaluate_ci_status_clears_pending_ci_event_on_pending(MockGitHub): + """evaluate_ci_status returns pending_ci_event=False when CI is still running.""" + from forge.workflow.nodes.ci_evaluator import evaluate_ci_status + + mock_github = AsyncMock() + MockGitHub.return_value = mock_github + mock_github.get_pull_request = AsyncMock(return_value={"head": {"sha": "abc123"}}) + mock_github.get_check_runs = AsyncMock(return_value=[]) # No checks yet = pending + mock_github.close = AsyncMock() + + state = {**BASE_STATE, "pending_ci_event": True} + result = await evaluate_ci_status(state) + + assert result.get("ci_status") == "pending" + assert result.get("pending_ci_event", True) is False + + +@pytest.mark.asyncio +@patch("forge.workflow.nodes.ci_evaluator.GitHubClient") +@patch("forge.workflow.nodes.ci_evaluator.JiraClient") +async def test_evaluate_ci_status_clears_pending_ci_event_on_passed(MockJira, MockGitHub): + """evaluate_ci_status returns pending_ci_event=False when all CI passes.""" + from forge.workflow.nodes.ci_evaluator import evaluate_ci_status + + mock_github = AsyncMock() + MockGitHub.return_value = mock_github + mock_github.get_pull_request = AsyncMock(return_value={"head": {"sha": "abc123"}}) + mock_github.get_check_runs = AsyncMock( + return_value=[ + {"name": "unit-tests", "status": "completed", "conclusion": "success"} + ] + ) + mock_github.close = AsyncMock() + + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + + state = {**BASE_STATE, "pending_ci_event": True} + result = await evaluate_ci_status(state) + + assert result.get("ci_status") == "passed" + assert result.get("pending_ci_event", True) is False From 5decca320dba39c88a22713e151f5c802b389c34 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 11:58:12 -0400 Subject: [PATCH 03/19] feat: human_review_gate handles initial-entry housekeeping and CI routing --- src/forge/workflow/nodes/human_review.py | 59 +++++++-- .../workflow/nodes/test_human_review_gate.py | 112 ++++++++++++++++++ 2 files changed, 159 insertions(+), 12 deletions(-) create mode 100644 tests/unit/workflow/nodes/test_human_review_gate.py diff --git a/src/forge/workflow/nodes/human_review.py b/src/forge/workflow/nodes/human_review.py index 2950d6be..736a3bb6 100644 --- a/src/forge/workflow/nodes/human_review.py +++ b/src/forge/workflow/nodes/human_review.py @@ -7,27 +7,57 @@ from forge.integrations.jira.client import JiraClient from forge.models.workflow import ForgeLabel, JiraStatus from forge.workflow.feature.state import FeatureState as WorkflowState -from forge.workflow.utils import set_paused, update_state_timestamp -from forge.workflow.utils.jira_status import post_status_comment +from forge.workflow.utils import update_state_timestamp +from forge.workflow.utils.jira_status import ( + post_status_comment, + remove_implementing_label, + set_ci_pending_label, +) logger = logging.getLogger(__name__) -def human_review_gate(state: WorkflowState) -> WorkflowState: - """Pause workflow for human code review. +async def human_review_gate(state: WorkflowState) -> WorkflowState: + """Pause workflow for CI and human review. - Args: - state: Current workflow state. - - Returns: - State with is_paused=True. + On initial entry (ci_status is None), posts a PR creation status comment + and transitions Jira labels from forge:implementing to forge:ci-pending. + On subsequent entries (CI cycle or review cycle re-entries), skips the + Jira comment to avoid duplicate noise. """ ticket_key = state["ticket_key"] - pr_urls = state.get("pr_urls", []) + ci_status = state.get("ci_status") - logger.info(f"Human review gate: pausing for {ticket_key} ({len(pr_urls)} PRs)") + if ci_status is None: + jira = JiraClient() + try: + pr_number = state.get("current_pr_number") + if pr_number is not None: + message = ( + f"🚀 Pull request #{pr_number} created and submitted. " + "Waiting for CI checks and human review." + ) + else: + message = ( + "🚀 Pull request created and submitted. " + "Waiting for CI checks and human review." + ) + await post_status_comment(jira, ticket_key, message) + await remove_implementing_label(jira, ticket_key) + await set_ci_pending_label(jira, ticket_key) + finally: + await jira.close() + logger.info(f"Pausing {ticket_key} at human_review_gate after PR creation") + else: + logger.info(f"Pausing {ticket_key} at human_review_gate (ci_status={ci_status!r})") - return set_paused(state, "human_review_gate") + return update_state_timestamp( + { + **state, + "is_paused": True, + "current_node": "human_review_gate", + } + ) def route_human_review(state: WorkflowState) -> str: @@ -39,6 +69,11 @@ def route_human_review(state: WorkflowState) -> str: Returns: Next node name or END. """ + # CI webhook arrived while paused at this gate — route through CI cycle + if state.get("pending_ci_event"): + logger.info(f"CI event pending for {state['ticket_key']}, routing to ci_evaluator") + return "ci_evaluator" + # Check if changes were requested — route to dedicated review implementation node if state.get("revision_requested") and state.get("feedback_comment"): logger.info(f"Changes requested for {state['ticket_key']}") diff --git a/tests/unit/workflow/nodes/test_human_review_gate.py b/tests/unit/workflow/nodes/test_human_review_gate.py new file mode 100644 index 00000000..bca84437 --- /dev/null +++ b/tests/unit/workflow/nodes/test_human_review_gate.py @@ -0,0 +1,112 @@ +"""Tests for the updated human_review_gate and route_human_review.""" + +import pytest +from unittest.mock import AsyncMock, patch +from langgraph.graph import END + + +BASE_STATE = { + "ticket_key": "TEST-1", + "pr_urls": ["https://github.com/org/repo/pull/1"], + "current_pr_number": 42, + "is_paused": False, + "pending_ci_event": False, + "ci_status": None, + "revision_requested": False, + "pr_merged": False, + "feedback_comment": None, +} + + +class TestRouteHumanReview: + def test_pending_ci_event_routes_to_ci_evaluator(self): + """pending_ci_event=True routes to ci_evaluator before all other checks.""" + from forge.workflow.nodes.human_review import route_human_review + + state = {**BASE_STATE, "pending_ci_event": True, "is_paused": True} + assert route_human_review(state) == "ci_evaluator" + + def test_pending_ci_event_overrides_revision_requested(self): + """CI event takes priority even if revision is also pending.""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + **BASE_STATE, + "pending_ci_event": True, + "revision_requested": True, + "feedback_comment": "please fix", + "is_paused": False, + } + assert route_human_review(state) == "ci_evaluator" + + def test_no_ci_event_paused_returns_end(self): + """With no pending CI event, paused=True returns END.""" + from forge.workflow.nodes.human_review import route_human_review + + state = {**BASE_STATE, "pending_ci_event": False, "is_paused": True} + assert route_human_review(state) == END + + def test_revision_requested_routes_to_implement_review(self): + """review feedback routes to implement_review when not paused.""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + **BASE_STATE, + "pending_ci_event": False, + "is_paused": False, + "revision_requested": True, + "feedback_comment": "please refactor", + } + assert route_human_review(state) == "implement_review" + + def test_pr_merged_routes_to_complete_tasks(self): + """pr_merged=True routes to complete_tasks.""" + from forge.workflow.nodes.human_review import route_human_review + + state = {**BASE_STATE, "pending_ci_event": False, "is_paused": False, "pr_merged": True} + assert route_human_review(state) == "complete_tasks" + + +class TestHumanReviewGate: + @pytest.mark.asyncio + @patch("forge.workflow.nodes.human_review.remove_implementing_label", new_callable=AsyncMock) + @patch("forge.workflow.nodes.human_review.set_ci_pending_label", new_callable=AsyncMock) + @patch("forge.workflow.nodes.human_review.post_status_comment", new_callable=AsyncMock) + @patch("forge.workflow.nodes.human_review.JiraClient") + async def test_initial_entry_posts_comment_and_updates_labels( + self, MockJira, mock_post, mock_set_label, mock_remove_label + ): + """On initial entry (ci_status=None), gate posts comment and updates labels.""" + from forge.workflow.nodes.human_review import human_review_gate + + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + + state = {**BASE_STATE, "ci_status": None, "pending_ci_event": False} + result = await human_review_gate(state) + + mock_post.assert_called_once() + comment_text = mock_post.call_args[0][2] + assert "42" in comment_text # PR number in comment + mock_remove_label.assert_called_once() + mock_set_label.assert_called_once() + assert result["is_paused"] is True + assert result["current_node"] == "human_review_gate" + + @pytest.mark.asyncio + @patch("forge.workflow.nodes.human_review.post_status_comment", new_callable=AsyncMock) + @patch("forge.workflow.nodes.human_review.JiraClient") + async def test_subsequent_entry_skips_comment(self, MockJira, mock_post): + """On re-entry (ci_status already set), gate skips Jira comment.""" + from forge.workflow.nodes.human_review import human_review_gate + + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + + state = {**BASE_STATE, "ci_status": "pending", "pending_ci_event": False} + result = await human_review_gate(state) + + mock_post.assert_not_called() + assert result["is_paused"] is True From 2d0e77c3f3359cc0974bce0499031951c957ebf1 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:10:33 -0400 Subject: [PATCH 04/19] feat: add Phase 0 CI attribution check, update attempt_ci_fix, delete wait_for_ci_gate - Create src/forge/prompts/v1/ci-attribution.md with attribution prompt - Rewrite attempt_ci_fix: Phase 0 runs attribution container before Phase 1; external failures return ci_status="external_failure" without incrementing ci_fix_attempt; all paths set pending_ci_event=False; return target changed from wait_for_ci_gate to human_review_gate throughout - Delete wait_for_ci_gate function from ci_evaluator.py - Remove wait_for_ci_gate from nodes/__init__.py (graph builders updated in Task 6) - Add 4 new attribution tests to test_ci_attribution.py Co-Authored-By: Claude Sonnet 4.6 --- src/forge/prompts/v1/ci-attribution.md | 50 ++++ src/forge/workflow/nodes/__init__.py | 2 - src/forge/workflow/nodes/ci_evaluator.py | 226 +++++++++--------- .../workflow/nodes/test_ci_attribution.py | 138 +++++++++++ 4 files changed, 300 insertions(+), 116 deletions(-) create mode 100644 src/forge/prompts/v1/ci-attribution.md diff --git a/src/forge/prompts/v1/ci-attribution.md b/src/forge/prompts/v1/ci-attribution.md new file mode 100644 index 00000000..c6139627 --- /dev/null +++ b/src/forge/prompts/v1/ci-attribution.md @@ -0,0 +1,50 @@ +You are analyzing CI failures to determine whether they are caused by changes +in this pull request or by external factors. + +## CI Failure Information + +The failed checks are described in: +{failures_file_path} + +All available log files are in `.forge/logs/`. + +## Your Task + +Read the failure logs carefully. Compare the failing test names, file paths, +and error messages against the files changed in this PR. + +To see which files this PR touches, run: +``` +git diff --name-only HEAD~1 HEAD +``` +or inspect the most recent commit: +``` +git show --stat HEAD +``` + +Determine whether each failing check is caused by changes introduced in this PR, +or by external factors such as: +- Flaky or intermittently failing tests unrelated to the diff +- Broken test infrastructure (container images, network, environment config) +- Pre-existing failures in files not touched by this PR +- Tests that fail because of unrelated upstream changes + +Write your verdict to `.forge/ci-attribution.json` in exactly this format: + +```json +{ + "attributable": true, + "reason": "one sentence explaining why", + "confidence": "high" +} +``` + +Set `attributable` to `true` if the failure logs reference files, functions, +or test cases that appear in the PR diff. Set it to `false` if the failures +are clearly unrelated to the diff. + +When confidence is low, set `attributable` to `true` (fail-safe: the fix +pipeline will attempt a fix, and the human can always use `/forge skip-gate` +to bypass a stuck check). + +Do not attempt to fix anything. Write only the attribution JSON file. diff --git a/src/forge/workflow/nodes/__init__.py b/src/forge/workflow/nodes/__init__.py index 32863101..ac709747 100644 --- a/src/forge/workflow/nodes/__init__.py +++ b/src/forge/workflow/nodes/__init__.py @@ -4,7 +4,6 @@ attempt_ci_fix, escalate_to_blocked, evaluate_ci_status, - wait_for_ci_gate, ) from forge.workflow.nodes.docs_updater import update_documentation from forge.workflow.nodes.epic_decomposition import ( @@ -114,7 +113,6 @@ "attempt_ci_fix", "escalate_to_blocked", "evaluate_ci_status", - "wait_for_ci_gate", # Documentation update "update_documentation", # Local code review diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py index 710b45f5..da41e1b1 100644 --- a/src/forge/workflow/nodes/ci_evaluator.py +++ b/src/forge/workflow/nodes/ci_evaluator.py @@ -1,6 +1,7 @@ """CI/CD evaluator node for monitoring and responding to CI results.""" import io +import json import logging import zipfile from pathlib import Path @@ -20,8 +21,6 @@ from forge.workflow.utils import update_state_timestamp from forge.workflow.utils.jira_status import ( post_status_comment, - remove_implementing_label, - set_ci_pending_label, set_review_pending_label, ) from forge.workspace.git_ops import GitOperations @@ -242,39 +241,28 @@ def _is_skipped(check: dict) -> bool: async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: """Attempt to autonomously fix CI failures. - This node: - 1. Extracts error information from failed checks - 2. Invokes Claude to generate a fix - 3. Applies the fix and pushes - 4. Routes back to CI evaluation - - Args: - state: Current workflow state with ci_failed_checks. - - Returns: - Updated state after fix attempt. + Phase 0 determines whether the failure is caused by this PR before + burning a fix attempt. If attribution is external, posts a comment and + returns to human_review_gate without incrementing ci_fix_attempt. """ ticket_key = state["ticket_key"] failed_checks = state.get("ci_failed_checks", []) - workspace_path = state.get("workspace_path") if not failed_checks: return update_state_timestamp( { **state, - "current_node": "ci_evaluator", + "current_node": "human_review_gate", + "pending_ci_event": False, } ) - logger.info(f"Attempting CI fix for {ticket_key}") - - # Post status comment to feature ticket at start of CI fix attempt ci_fix_attempt = state.get("ci_fix_attempt", 0) ci_fix_max = state.get("ci_fix_max_attempts", 5) jira = JiraClient() try: - message = f"🔧 CI checks failed. Analyzing failure and attempting fix ({ci_fix_attempt}/{ci_fix_max})." + message = f"🔧 CI checks failed. Analyzing failure attribution ({ci_fix_attempt}/{ci_fix_max})." await post_status_comment(jira, ticket_key, message) finally: await jira.close() @@ -287,45 +275,119 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: try: workspace_path, _ = prepare_workspace(state) state = {**state, "workspace_path": workspace_path} - except Exception as _setup_err: logger.error(f"Workspace setup failed for {ticket_key}: {_setup_err}") return { **state, "last_error": str(_setup_err), "current_node": "attempt_ci_fix", + "pending_ci_event": False, } try: - # ── Phase 1: Analysis ──────────────────────────────────────────────── - # ForgeAgent with MCP access fetches the Prow logs and produces a - # ── Phase 1: Analysis (container) ──────────────────────────────────── - # Runs in a container so the agent has gh CLI and shell access to fetch - # real GitHub Actions / Prow logs. Writes the fix plan to a file so - # Phase 2 can read it without hitting token limits. - logger.info(f"Phase 1: Analyzing CI failures for {ticket_key} (attempt {attempt})") - failures_file = Path(workspace_path) / ".forge" / "ci-failures.md" + attribution_file = Path(workspace_path) / ".forge" / "ci-attribution.json" fix_plan_file = Path(workspace_path) / ".forge" / "fix-plan.md" logs_dir = Path(workspace_path) / ".forge" / "logs" failures_file.parent.mkdir(parents=True, exist_ok=True) - # Download job logs and artifacts before the container starts so the - # agent can read them directly without needing gh CLI authentication. await _fetch_ci_logs_and_artifacts(failed_checks, logs_dir, GitHubClient()) - failures_file.write_text(_collect_error_info(failed_checks)) + # ── Phase 0: Attribution ───────────────────────────────────────────── + logger.info(f"Phase 0: Attributing CI failures for {ticket_key}") + attribution_prompt = load_prompt( + "ci-attribution", + failures_file_path=str(failures_file), + ) + runner = ContainerRunner(settings) + await runner.run( + workspace_path=Path(workspace_path), + task_summary=f"Attribute CI failure (attempt {attempt})", + task_description=attribution_prompt, + ticket_key=ticket_key, + task_key=f"{ticket_key}-ci-attribution", + repo_name=state.get("current_repo", ""), + ) + + attributable = True # fail-safe default + attribution_reason = "" + if attribution_file.exists(): + try: + verdict = json.loads(attribution_file.read_text()) + attributable = verdict.get("attributable", True) + attribution_reason = verdict.get("reason", "") + except (ValueError, KeyError): + logger.warning( + f"Could not parse ci-attribution.json for {ticket_key}, assuming attributable" + ) + + if not attributable: + logger.info( + f"CI failure attributed as external for {ticket_key}: {attribution_reason}" + ) + jira = JiraClient() + try: + check_names = ", ".join(c.get("name", "unknown") for c in failed_checks) + await post_status_comment( + jira, + ticket_key, + f"⚠️ CI check(s) `{check_names}` are failing due to issues outside this PR " + f"({attribution_reason}). " + "The workflow will continue waiting. " + "Use `/forge skip-gate ` on the PR to bypass if needed.", + ) + finally: + await jira.close() + return update_state_timestamp( + { + **state, + "ci_status": "external_failure", + "current_node": "human_review_gate", + "pending_ci_event": False, + "last_error": None, + } + ) + + # ── Phase 1: Analysis ──────────────────────────────────────────────── + next_attempt = ci_fix_attempt + 1 + logger.info( + f"Phase 1: Analyzing CI failures for {ticket_key} (fix attempt {next_attempt}/{ci_fix_max})" + ) + + if ci_fix_attempt >= ci_fix_max: + logger.warning(f"CI fix attempt limit ({ci_fix_max}) reached for {ticket_key}") + return update_state_timestamp( + { + **state, + "ci_status": "failed", + "current_node": "ci_evaluator", + "pending_ci_event": False, + "last_error": "CI fix attempt limit reached", + } + ) + + state = {**state, "ci_fix_attempt": next_attempt} + + jira = JiraClient() + try: + await post_status_comment( + jira, + ticket_key, + f"🔧 Attempting CI fix ({next_attempt}/{ci_fix_max}).", + ) + finally: + await jira.close() + analysis_prompt = load_prompt( "analyze-ci", failures_file_path=str(failures_file), - attempt=attempt, + attempt=next_attempt, ) - runner = ContainerRunner(settings) await runner.run( workspace_path=Path(workspace_path), - task_summary=f"Analyze CI failures (attempt {attempt})", + task_summary=f"Analyze CI failures (attempt {next_attempt})", task_description=analysis_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-analyze", @@ -337,7 +399,8 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: return update_state_timestamp( { **state, - "current_node": "wait_for_ci_gate", + "current_node": "human_review_gate", + "pending_ci_event": False, "last_error": None, } ) @@ -346,14 +409,12 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: logger.info(f"Phase 1 complete: fix plan generated ({len(fix_plan)} chars)") # ── Phase 2: Fix ───────────────────────────────────────────────────── - # Container applies the fix plan written by Phase 1. logger.info(f"Phase 2: Applying fixes for {ticket_key}") fix_prompt = load_prompt("fix-ci", fix_plan=fix_plan) - runner = ContainerRunner(settings) await runner.run( workspace_path=Path(workspace_path), - task_summary=f"Apply CI fix plan (attempt {attempt})", + task_summary=f"Apply CI fix plan (attempt {next_attempt})", task_description=fix_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-fix", @@ -367,15 +428,14 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: ticket_key=ticket_key, ) git = GitOperations(workspace) - branch_name = state.get("context", {}).get("branch_name", "") - # Commit any files the container left uncommitted (safety net) if git.has_uncommitted_changes(): git.stage_all() - git.commit(f"[{ticket_key}] fix: address CI failures (attempt {attempt})") + git.commit( + f"[{ticket_key}] fix: address CI failures (attempt {next_attempt})" + ) - # Check for changes before doing anything expensive if fork_owner and fork_repo: git.add_fork_remote(fork_owner, fork_repo) remote_ref = f"fork/{branch_name}" @@ -387,9 +447,10 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: ).stdout.strip() if not unpushed: - logger.warning(f"Container made no changes for {ticket_key} (attempt {attempt})") + logger.warning( + f"Container made no changes for {ticket_key} (attempt {next_attempt})" + ) else: - # Only run the expensive review pass when the fix actually changed code await run_post_change_review( workspace_path=str(workspace_path), ticket_key=ticket_key, @@ -397,19 +458,16 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: branch_name=branch_name, spec_content=state.get("spec_content", ""), guardrails=state.get("context", {}).get("guardrails", ""), - label=f"ci-fix-{attempt}", + label=f"ci-fix-{next_attempt}", ) - - # Push all commits (CI fix + any review corrections) if fork_owner and fork_repo: git.push_to_fork(force=False) else: logger.warning("Fork info not in state — pushing to origin instead") git.push(force=False) - logger.info(f"CI fix pushed for {ticket_key} (attempt {attempt})") + logger.info(f"CI fix pushed for {ticket_key} (attempt {next_attempt})") record_ci_fix_attempt(repo=state.get("current_repo", "unknown"), result="pushed") - # Sync PR description to reflect what actually changed _repo = state.get("current_repo", "/") _owner, _repo_name = (_repo.split("/") + [""])[:2] await sync_pr_description( @@ -418,13 +476,14 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: owner=_owner, repo=_repo_name, pr_number=state.get("current_pr_number"), - attempt=attempt, + attempt=next_attempt, ) return update_state_timestamp( { **state, - "current_node": "wait_for_ci_gate", + "current_node": "human_review_gate", + "pending_ci_event": False, "last_error": None, } ) @@ -434,70 +493,9 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: return { **state, "last_error": str(e), - "current_node": "attempt_ci_fix", # preserved so retry resumes here, - } - - -async def wait_for_ci_gate(state: WorkflowState) -> WorkflowState: - """Pause the workflow until a GitHub CI webhook arrives. - - Inserted after PR creation and after each CI fix push. The workflow - resumes when GitHub sends a check_run or check_suite webhook, at which - point ci_evaluator re-checks the actual CI results without any - polling delay. - - On initial entry (after PR creation), posts a status comment to the feature - ticket and transitions labels from forge:implementing to forge:ci-pending. - - Args: - state: Current workflow state. - - Returns: - Updated state with is_paused=True. - """ - ticket_key = state["ticket_key"] - ci_fix_attempt = state.get("ci_fix_attempt", 0) - - # Detect initial entry (after PR creation) vs re-entry (after CI fix) - is_initial_entry = ci_fix_attempt == 0 - - if is_initial_entry: - # Post status comment and update labels on initial entry - jira = JiraClient() - try: - # Build status message with PR number if available - pr_number = state.get("current_pr_number") - if pr_number is not None: - message = f"🚀 Pull request #{pr_number} created and submitted. Waiting for CI checks to complete." - else: - message = ( - "🚀 Pull request created and submitted. Waiting for CI checks to complete." - ) - - # Post status comment to feature ticket - await post_status_comment(jira, ticket_key, message) - - # Transition labels: remove forge:implementing, add forge:ci-pending - await remove_implementing_label(jira, ticket_key) - await set_ci_pending_label(jira, ticket_key) - - finally: - await jira.close() - - logger.info(f"Pausing {ticket_key} after PR creation, waiting for GitHub CI webhook") - else: - logger.info( - f"Pausing {ticket_key} after CI fix attempt {ci_fix_attempt}, " - "waiting for GitHub CI webhook" - ) - - return update_state_timestamp( - { - **state, - "is_paused": True, - "current_node": "wait_for_ci_gate", + "current_node": "attempt_ci_fix", + "pending_ci_event": False, } - ) async def escalate_to_blocked(state: WorkflowState) -> WorkflowState: diff --git a/tests/unit/workflow/nodes/test_ci_attribution.py b/tests/unit/workflow/nodes/test_ci_attribution.py index 7958620b..948e5369 100644 --- a/tests/unit/workflow/nodes/test_ci_attribution.py +++ b/tests/unit/workflow/nodes/test_ci_attribution.py @@ -13,6 +13,19 @@ "ci_skipped_checks": [], } +ATTEMPT_BASE_STATE = { + "ticket_key": "TEST-1", + "ci_failed_checks": [ + {"name": "unit-tests", "conclusion": "failure", "log_url": "", "output": {}} + ], + "ci_fix_attempt": 1, + "ci_fix_max_attempts": 5, + "current_node": "attempt_ci_fix", + "current_repo": "org/repo", + "context": {"branch_name": "forge/TEST-1"}, + "pending_ci_event": False, +} + @pytest.mark.asyncio @patch("forge.workflow.nodes.ci_evaluator.GitHubClient") @@ -59,3 +72,128 @@ async def test_evaluate_ci_status_clears_pending_ci_event_on_passed(MockJira, Mo assert result.get("ci_status") == "passed" assert result.get("pending_ci_event", True) is False + + +@pytest.mark.asyncio +@patch("forge.workflow.nodes.ci_evaluator.ContainerRunner") +@patch("forge.workflow.nodes.ci_evaluator.prepare_workspace") +@patch("forge.workflow.nodes.ci_evaluator.JiraClient") +@patch("forge.workflow.nodes.ci_evaluator.GitHubClient") +@patch("forge.workflow.nodes.ci_evaluator._fetch_ci_logs_and_artifacts", new_callable=AsyncMock) +async def test_attribution_external_skips_fix( + mock_fetch, MockGitHub, MockJira, mock_prep, MockRunner, tmp_path +): + """Phase 0 verdict attributable=false → external_failure, no fix attempt increment.""" + from forge.workflow.nodes.ci_evaluator import attempt_ci_fix + + # Phase 0 container writes attribution file + forge_dir = tmp_path / ".forge" + forge_dir.mkdir() + attribution_file = forge_dir / "ci-attribution.json" + attribution_file.write_text('{"attributable": false, "reason": "Flaky infra", "confidence": "high"}') + + mock_prep.return_value = (str(tmp_path), None) + + mock_runner = AsyncMock() + MockRunner.return_value = mock_runner + mock_runner.run = AsyncMock() + + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + + mock_github = AsyncMock() + MockGitHub.return_value = mock_github + + state = {**ATTEMPT_BASE_STATE, "workspace_path": str(tmp_path)} + result = await attempt_ci_fix(state) + + assert result["ci_status"] == "external_failure" + assert result["ci_fix_attempt"] == 1 # NOT incremented + assert result["current_node"] == "human_review_gate" + assert result.get("pending_ci_event", True) is False + + +@pytest.mark.asyncio +@patch("forge.workflow.nodes.ci_evaluator.ContainerRunner") +@patch("forge.workflow.nodes.ci_evaluator.prepare_workspace") +@patch("forge.workflow.nodes.ci_evaluator.JiraClient") +@patch("forge.workflow.nodes.ci_evaluator.GitHubClient") +@patch("forge.workflow.nodes.ci_evaluator._fetch_ci_logs_and_artifacts", new_callable=AsyncMock) +async def test_attribution_attributable_proceeds_to_fix( + mock_fetch, MockGitHub, MockJira, mock_prep, MockRunner, tmp_path +): + """Phase 0 verdict attributable=true → proceeds to fix, increments attempt.""" + from forge.workflow.nodes.ci_evaluator import attempt_ci_fix + + forge_dir = tmp_path / ".forge" + forge_dir.mkdir() + attribution_file = forge_dir / "ci-attribution.json" + attribution_file.write_text( + '{"attributable": true, "reason": "auth.py modified", "confidence": "high"}' + ) + # Phase 1 writes no fix plan → short-circuit back to gate + # (fix_plan_file does not exist) + + mock_prep.return_value = (str(tmp_path), None) + + mock_runner = AsyncMock() + MockRunner.return_value = mock_runner + mock_runner.run = AsyncMock() + + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + + mock_github = AsyncMock() + MockGitHub.return_value = mock_github + + state = {**ATTEMPT_BASE_STATE, "workspace_path": str(tmp_path)} + result = await attempt_ci_fix(state) + + # Proceeds: returns to gate, pending_ci_event cleared + assert result["current_node"] == "human_review_gate" + assert result.get("pending_ci_event", True) is False + + +@pytest.mark.asyncio +@patch("forge.workflow.nodes.ci_evaluator.ContainerRunner") +@patch("forge.workflow.nodes.ci_evaluator.prepare_workspace") +@patch("forge.workflow.nodes.ci_evaluator.JiraClient") +@patch("forge.workflow.nodes.ci_evaluator.GitHubClient") +@patch("forge.workflow.nodes.ci_evaluator._fetch_ci_logs_and_artifacts", new_callable=AsyncMock) +async def test_attribution_missing_file_proceeds_as_attributable( + mock_fetch, MockGitHub, MockJira, mock_prep, MockRunner, tmp_path +): + """Missing ci-attribution.json is treated as attributable (fail-safe).""" + from forge.workflow.nodes.ci_evaluator import attempt_ci_fix + + forge_dir = tmp_path / ".forge" + forge_dir.mkdir() + # No attribution file written — container failed silently + + mock_prep.return_value = (str(tmp_path), None) + mock_runner = AsyncMock() + MockRunner.return_value = mock_runner + mock_runner.run = AsyncMock() + mock_jira = AsyncMock() + MockJira.return_value = mock_jira + mock_jira.close = AsyncMock() + mock_github = AsyncMock() + MockGitHub.return_value = mock_github + + state = {**ATTEMPT_BASE_STATE, "workspace_path": str(tmp_path)} + result = await attempt_ci_fix(state) + + # Fail-safe: assumes attributable, proceeds + assert result.get("ci_status") != "external_failure" + assert result["current_node"] == "human_review_gate" + + +def test_wait_for_ci_gate_does_not_exist(): + """wait_for_ci_gate has been deleted from ci_evaluator module.""" + import forge.workflow.nodes.ci_evaluator as mod + + assert not hasattr(mod, "wait_for_ci_gate"), ( + "wait_for_ci_gate must be deleted — it no longer exists as a graph node" + ) From 786daacbb509ad4cc2a8c8cdc93d178e1d1e12dd Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:15:37 -0400 Subject: [PATCH 05/19] fix: remove double-increment of ci_fix_attempt in attempt_ci_fix --- src/forge/workflow/nodes/ci_evaluator.py | 23 ++++++++++------------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py index da41e1b1..7cf077bf 100644 --- a/src/forge/workflow/nodes/ci_evaluator.py +++ b/src/forge/workflow/nodes/ci_evaluator.py @@ -350,9 +350,8 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: ) # ── Phase 1: Analysis ──────────────────────────────────────────────── - next_attempt = ci_fix_attempt + 1 logger.info( - f"Phase 1: Analyzing CI failures for {ticket_key} (fix attempt {next_attempt}/{ci_fix_max})" + f"Phase 1: Analyzing CI failures for {ticket_key} (fix attempt {ci_fix_attempt}/{ci_fix_max})" ) if ci_fix_attempt >= ci_fix_max: @@ -367,14 +366,12 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: } ) - state = {**state, "ci_fix_attempt": next_attempt} - jira = JiraClient() try: await post_status_comment( jira, ticket_key, - f"🔧 Attempting CI fix ({next_attempt}/{ci_fix_max}).", + f"🔧 Attempting CI fix ({ci_fix_attempt}/{ci_fix_max}).", ) finally: await jira.close() @@ -382,12 +379,12 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: analysis_prompt = load_prompt( "analyze-ci", failures_file_path=str(failures_file), - attempt=next_attempt, + attempt=ci_fix_attempt, ) runner = ContainerRunner(settings) await runner.run( workspace_path=Path(workspace_path), - task_summary=f"Analyze CI failures (attempt {next_attempt})", + task_summary=f"Analyze CI failures (attempt {ci_fix_attempt})", task_description=analysis_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-analyze", @@ -414,7 +411,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: runner = ContainerRunner(settings) await runner.run( workspace_path=Path(workspace_path), - task_summary=f"Apply CI fix plan (attempt {next_attempt})", + task_summary=f"Apply CI fix plan (attempt {ci_fix_attempt})", task_description=fix_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-fix", @@ -433,7 +430,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: if git.has_uncommitted_changes(): git.stage_all() git.commit( - f"[{ticket_key}] fix: address CI failures (attempt {next_attempt})" + f"[{ticket_key}] fix: address CI failures (attempt {ci_fix_attempt})" ) if fork_owner and fork_repo: @@ -448,7 +445,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: if not unpushed: logger.warning( - f"Container made no changes for {ticket_key} (attempt {next_attempt})" + f"Container made no changes for {ticket_key} (attempt {ci_fix_attempt})" ) else: await run_post_change_review( @@ -458,14 +455,14 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: branch_name=branch_name, spec_content=state.get("spec_content", ""), guardrails=state.get("context", {}).get("guardrails", ""), - label=f"ci-fix-{next_attempt}", + label=f"ci-fix-{ci_fix_attempt}", ) if fork_owner and fork_repo: git.push_to_fork(force=False) else: logger.warning("Fork info not in state — pushing to origin instead") git.push(force=False) - logger.info(f"CI fix pushed for {ticket_key} (attempt {next_attempt})") + logger.info(f"CI fix pushed for {ticket_key} (attempt {ci_fix_attempt})") record_ci_fix_attempt(repo=state.get("current_repo", "unknown"), result="pushed") _repo = state.get("current_repo", "/") @@ -476,7 +473,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: owner=_owner, repo=_repo_name, pr_number=state.get("current_pr_number"), - attempt=next_attempt, + attempt=ci_fix_attempt, ) return update_state_timestamp( From 2bdd3d65e6827d359e645aa7b4d8c1832c1c2d01 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:21:51 -0400 Subject: [PATCH 06/19] feat: add shared post-PR lifecycle module with add_post_pr_nodes/edges --- src/forge/workflow/post_pr.py | 171 ++++++++++++++++++++++++++++ tests/unit/workflow/test_post_pr.py | 84 ++++++++++++++ 2 files changed, 255 insertions(+) create mode 100644 src/forge/workflow/post_pr.py create mode 100644 tests/unit/workflow/test_post_pr.py diff --git a/src/forge/workflow/post_pr.py b/src/forge/workflow/post_pr.py new file mode 100644 index 00000000..fbb5e8a0 --- /dev/null +++ b/src/forge/workflow/post_pr.py @@ -0,0 +1,171 @@ +"""Shared post-PR lifecycle nodes and edges for all workflow types. + +All three workflow types (feature, bug, task_takeover) share identical +CI/review node wiring after PR creation. This module provides that shared +wiring so each graph builder calls two functions instead of duplicating ~60 +lines of node and edge declarations. +""" + +import logging +from collections.abc import Callable +from typing import Any + +from langgraph.graph import END, StateGraph + +logger = logging.getLogger(__name__) + + +def _route_after_teardown(state: dict[str, Any]) -> str: + """Route after workspace teardown: next repo or human_review_gate.""" + repos_to_process = state.get("repos_to_process", []) + repos_completed = state.get("repos_completed", []) + remaining = [r for r in repos_to_process if r not in repos_completed] + return "setup_workspace" if remaining else "human_review_gate" + + +def _route_after_pr_creation(state: dict[str, Any]) -> str: + """Route after PR creation: teardown on success, escalate on failure.""" + last_error = state.get("last_error") + pr_urls = state.get("pr_urls", []) + if last_error and not pr_urls: + return "escalate_blocked" + return "teardown_workspace" + + +def _route_ci_evaluation(state: dict[str, Any]) -> str: + """Route after ci_evaluator. + + 'fixing' → attempt_ci_fix. + 'failed' or 'blocked' → escalate_blocked. + All other statuses ('pending', 'passed', 'external_failure', etc.) → human_review_gate + so the gate re-pauses and waits for the next CI or review webhook. + Unknown/unexpected statuses → escalate_blocked as safety net. + """ + ci_status = state.get("ci_status", "") + if ci_status == "fixing": + return "attempt_ci_fix" + if ci_status in ("failed", "blocked"): + return "escalate_blocked" + # All known non-fix/non-fail statuses go back to gate + if ci_status in ("pending", "passed", "external_failure", "no_prs"): + return "human_review_gate" + # Unknown status — safety escalation + logger.warning(f"Unexpected ci_status: {ci_status!r}, routing to escalate_blocked") + return "escalate_blocked" + + +def add_post_pr_nodes(graph: StateGraph) -> None: + """Add all shared CI/review nodes to a workflow graph.""" + from forge.workflow.nodes.ci_evaluator import ( + attempt_ci_fix, + escalate_to_blocked, + evaluate_ci_status, + ) + from forge.workflow.nodes.human_review import human_review_gate + from forge.workflow.nodes.implement_review import implement_review, review_response_gate + from forge.workflow.nodes.rebase import rebase_pr + + graph.add_node("ci_evaluator", evaluate_ci_status) + graph.add_node("attempt_ci_fix", attempt_ci_fix) + graph.add_node("escalate_blocked", escalate_to_blocked) + graph.add_node("human_review_gate", human_review_gate) + graph.add_node("implement_review", implement_review) + graph.add_node("review_response_gate", review_response_gate) + graph.add_node("rebase_pr", rebase_pr) + + +def add_post_pr_edges( + graph: StateGraph, + on_complete_node: str, + human_review_routing_fn: Callable | None = None, +) -> None: + """Add all shared CI/review edges to a workflow graph. + + Args: + graph: The StateGraph to wire. + on_complete_node: Node to route to when PR is merged (e.g. "complete_tasks"). + For the feature workflow pass "complete_tasks" (the default route_human_review + return value). For bug pass "post_merge_summary" and supply _route_human_review_bug + as human_review_routing_fn. + human_review_routing_fn: Routing function for human_review_gate conditional edges. + Defaults to route_human_review (used by feature workflow). Bug and task_takeover + pass their workflow-specific wrapper that intercepts pr_merged. + """ + from forge.workflow.nodes.human_review import route_human_review + from forge.workflow.nodes.implement_review import route_review_response + + if human_review_routing_fn is None: + human_review_routing_fn = route_human_review + + # teardown → human_review_gate or next repo + graph.add_conditional_edges( + "teardown_workspace", + _route_after_teardown, + {"setup_workspace": "setup_workspace", "human_review_gate": "human_review_gate"}, + ) + + # CI cycle: ci_evaluator → fix or gate + graph.add_conditional_edges( + "ci_evaluator", + _route_ci_evaluation, + { + "human_review_gate": "human_review_gate", + "attempt_ci_fix": "attempt_ci_fix", + "escalate_blocked": "escalate_blocked", + }, + ) + + # attempt_ci_fix returns via current_node in state + graph.add_conditional_edges( + "attempt_ci_fix", + lambda s: s.get("current_node", "human_review_gate"), + { + "human_review_gate": "human_review_gate", + "escalate_blocked": "escalate_blocked", + "ci_evaluator": "ci_evaluator", + "attempt_ci_fix": "escalate_blocked", # safety: loop → escalate + }, + ) + + # Gate routing — accepts CI, review, and merge signals + # on_complete_node is the workflow-specific merge destination. + # "complete_tasks" is also mapped to on_complete_node as a defensive catch + # for callers that pass a custom routing fn which still uses route_human_review + # as a fallthrough. + human_review_targets: dict[str, str] = { + "ci_evaluator": "ci_evaluator", + "implement_review": "implement_review", + on_complete_node: on_complete_node, + END: END, + } + if on_complete_node != "complete_tasks": + human_review_targets["complete_tasks"] = on_complete_node + graph.add_conditional_edges( + "human_review_gate", + human_review_routing_fn, + human_review_targets, + ) + + # Review cycle + graph.add_conditional_edges( + "implement_review", + lambda s: s.get("current_node", "human_review_gate"), + { + "human_review_gate": "human_review_gate", + "review_response_gate": "review_response_gate", + "implement_review": "implement_review", + "escalate_blocked": "escalate_blocked", + }, + ) + graph.add_conditional_edges( + "review_response_gate", + route_review_response, + { + "implement_review": "implement_review", + "human_review_gate": "human_review_gate", + END: END, + }, + ) + + # Terminal + graph.add_edge("escalate_blocked", END) diff --git a/tests/unit/workflow/test_post_pr.py b/tests/unit/workflow/test_post_pr.py new file mode 100644 index 00000000..e9d01246 --- /dev/null +++ b/tests/unit/workflow/test_post_pr.py @@ -0,0 +1,84 @@ +"""Tests for the shared post-PR lifecycle module.""" + +import pytest +from langgraph.graph import END + + +class TestRouteAfterTeardown: + def test_more_repos_remaining_routes_to_setup(self): + from forge.workflow.post_pr import _route_after_teardown + + state = {"repos_to_process": ["org/a", "org/b"], "repos_completed": ["org/a"]} + assert _route_after_teardown(state) == "setup_workspace" + + def test_all_repos_done_routes_to_human_review_gate(self): + from forge.workflow.post_pr import _route_after_teardown + + state = {"repos_to_process": ["org/a"], "repos_completed": ["org/a"]} + assert _route_after_teardown(state) == "human_review_gate" + + def test_no_repos_routes_to_human_review_gate(self): + from forge.workflow.post_pr import _route_after_teardown + + state = {"repos_to_process": [], "repos_completed": []} + assert _route_after_teardown(state) == "human_review_gate" + + +class TestRouteCiEvaluation: + def test_fixing_routes_to_attempt_ci_fix(self): + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "fixing"}) == "attempt_ci_fix" + + def test_failed_routes_to_escalate_blocked(self): + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "failed"}) == "escalate_blocked" + + def test_blocked_routes_to_escalate_blocked(self): + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "blocked"}) == "escalate_blocked" + + def test_passed_returns_to_human_review_gate(self): + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "passed"}) == "human_review_gate" + + def test_pending_returns_to_human_review_gate(self): + """pending CI no longer returns END — gate re-pauses and waits.""" + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "pending"}) == "human_review_gate" + + def test_external_failure_returns_to_human_review_gate(self): + """External CI failure stays at gate for human to decide.""" + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "external_failure"}) == "human_review_gate" + + def test_unknown_status_routes_to_escalate_blocked(self): + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "unknown_junk"}) == "escalate_blocked" + + +class TestRouteAfterPrCreation: + def test_success_routes_to_teardown(self): + from forge.workflow.post_pr import _route_after_pr_creation + + state = {"pr_urls": ["https://github.com/org/repo/pull/1"], "last_error": None} + assert _route_after_pr_creation(state) == "teardown_workspace" + + def test_failure_with_no_prs_routes_to_escalate(self): + from forge.workflow.post_pr import _route_after_pr_creation + + state = {"pr_urls": [], "last_error": "PR creation failed"} + assert _route_after_pr_creation(state) == "escalate_blocked" + + def test_partial_success_routes_to_teardown(self): + """If some PRs were created despite an error, proceed to teardown.""" + from forge.workflow.post_pr import _route_after_pr_creation + + state = {"pr_urls": ["https://github.com/org/repo/pull/1"], "last_error": "partial"} + assert _route_after_pr_creation(state) == "teardown_workspace" From 16f8c845b2c0037b4800a5c86340bb98d6192f61 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:33:37 -0400 Subject: [PATCH 07/19] refactor: replace duplicated post-PR wiring with shared add_post_pr_nodes/edges - Replaced duplicated CI/review node and edge declarations in all three workflow graphs (feature, bug, task_takeover) with shared add_post_pr_nodes() and add_post_pr_edges() - Updated backward compat mapping: wait_for_ci_gate -> ci_evaluator (was human_review_gate) - Fixed tests to reflect wait_for_ci_gate deletion and new routing - Updated rebase_pr edges to remove wait_for_ci_gate from all three graphs - Added rebase_pr support to task_takeover workflow (previously missing) --- src/forge/workflow/bug/graph.py | 156 ++--------------- src/forge/workflow/feature/graph.py | 168 +++---------------- src/forge/workflow/task_takeover/graph.py | 131 ++++----------- src/forge/workflow/utils/__init__.py | 2 +- tests/unit/workflow/bug/test_graph.py | 61 +++++-- tests/unit/workflow/feature/test_workflow.py | 2 +- tests/unit/workflow/test_base.py | 6 +- tests/unit/workflow/test_cleanup.py | 2 +- 8 files changed, 118 insertions(+), 410 deletions(-) diff --git a/src/forge/workflow/bug/graph.py b/src/forge/workflow/bug/graph.py index f9ccee44..fd616209 100644 --- a/src/forge/workflow/bug/graph.py +++ b/src/forge/workflow/bug/graph.py @@ -9,22 +9,8 @@ from langgraph.graph import END, StateGraph from forge.workflow.bug.state import BugState -from forge.workflow.nodes.ci_evaluator import ( - attempt_ci_fix, - escalate_to_blocked, - evaluate_ci_status, - wait_for_ci_gate, -) from forge.workflow.nodes.docs_updater import update_documentation -from forge.workflow.nodes.human_review import ( - human_review_gate, - route_human_review, -) -from forge.workflow.nodes.implement_review import ( - implement_review, - review_response_gate, - route_review_response, -) +from forge.workflow.nodes.human_review import route_human_review from forge.workflow.nodes.implementation import implement_task from forge.workflow.nodes.local_reviewer import local_review_changes from forge.workflow.nodes.plan_bug_fix import ( @@ -44,9 +30,13 @@ regenerate_rca, route_rca_option, ) -from forge.workflow.nodes.rebase import rebase_pr from forge.workflow.nodes.triage import route_triage_gate, triage_check, triage_gate from forge.workflow.nodes.workspace_setup import setup_workspace +from forge.workflow.post_pr import ( + _route_after_pr_creation, + add_post_pr_edges, + add_post_pr_nodes, +) from forge.workflow.utils import resolve_shared_resume_node logger = logging.getLogger(__name__) @@ -129,8 +119,6 @@ def route_entry(state: BugState) -> str: return "create_pr" elif current_node == "teardown_workspace": return "teardown_workspace" - elif current_node == "wait_for_ci_gate": - return "wait_for_ci_gate" elif current_node == "ai_review": return "human_review_gate" elif current_node == "escalate_blocked": @@ -354,47 +342,6 @@ def _route_after_implementation( return "local_review" -def _route_after_pr_creation( - state: BugState, -) -> Literal["teardown_workspace", "escalate_blocked"]: - """Route after PR creation attempt.""" - last_error = state.get("last_error") - pr_urls = state.get("pr_urls", []) - - if last_error and not pr_urls: - return "escalate_blocked" - - return "teardown_workspace" - - -def _route_after_teardown(state: BugState) -> str: - """Route after workspace teardown. - - If more repos remain in repos_to_process, loop back to setup_workspace. - Otherwise proceed to wait_for_ci_gate (matching feature workflow pattern) - so that Jira labels are updated and the workflow pauses until CI fires. - """ - repos_to_process = state.get("repos_to_process", []) - repos_completed = state.get("repos_completed", []) - remaining = [r for r in repos_to_process if r not in repos_completed] - if remaining: - return "setup_workspace" - return "wait_for_ci_gate" - - -def _route_ci_evaluation( - state: BugState, -) -> Literal["human_review_gate", "attempt_ci_fix", "escalate_blocked", "__end__"]: - """Route based on CI evaluation results.""" - ci_status = state.get("ci_status", "") - routes = { - "passed": "human_review_gate", - "fixing": "attempt_ci_fix", - "pending": "__end__", - } - return routes.get(ci_status, "escalate_blocked") - - def build_bug_graph() -> StateGraph: """Create the Bug workflow graph. @@ -450,16 +397,8 @@ def build_bug_graph() -> StateGraph: graph.add_node("create_pr", create_pull_request) graph.add_node("teardown_workspace", teardown_and_route) - # ── CI/CD ── - graph.add_node("wait_for_ci_gate", wait_for_ci_gate) - graph.add_node("ci_evaluator", evaluate_ci_status) - graph.add_node("attempt_ci_fix", attempt_ci_fix) - graph.add_node("escalate_blocked", escalate_to_blocked) - - # ── Review ── - graph.add_node("human_review_gate", human_review_gate) - graph.add_node("implement_review", implement_review) - graph.add_node("review_response_gate", review_response_gate) + # ── Post-PR nodes (CI/review) - shared across all workflows ── + add_post_pr_nodes(graph) # ── Set entry point ── graph.set_entry_point("route_entry") @@ -485,7 +424,6 @@ def build_bug_graph() -> StateGraph: "update_documentation": "update_documentation", "create_pr": "create_pr", "teardown_workspace": "teardown_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "human_review_gate": "human_review_gate", "implement_review": "implement_review", @@ -645,80 +583,15 @@ def build_bug_graph() -> StateGraph: "escalate_blocked": "escalate_blocked", }, ) - graph.add_conditional_edges( - "teardown_workspace", - _route_after_teardown, - { - "setup_workspace": "setup_workspace", # multi-repo loop-back - "wait_for_ci_gate": "wait_for_ci_gate", - }, - ) - - # ── CI/CD flow ── - graph.add_conditional_edges( - "wait_for_ci_gate", - lambda s: END if s.get("is_paused") else "ci_evaluator", - {END: END, "ci_evaluator": "ci_evaluator"}, - ) - graph.add_conditional_edges( - "ci_evaluator", - _route_ci_evaluation, - { - "human_review_gate": "human_review_gate", - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - END: END, - }, - ) - graph.add_conditional_edges( - "attempt_ci_fix", - lambda s: s.get("current_node", "wait_for_ci_gate"), - { - "wait_for_ci_gate": "wait_for_ci_gate", - "escalate_blocked": "escalate_blocked", - "ci_evaluator": "ci_evaluator", - "attempt_ci_fix": "escalate_blocked", - }, - ) - graph.add_edge("escalate_blocked", END) - - # ── Review flow (merge path → post_merge_summary) ── - # "complete_tasks" is the feature-workflow merge return from route_human_review; - # in the bug graph it should never be reached (pr_merged check intercepts first), - # but map it to post_merge_summary defensively. - graph.add_conditional_edges( - "human_review_gate", - _route_human_review_bug, - { - "implement_review": "implement_review", - "post_merge_summary": "post_merge_summary", - "complete_tasks": "post_merge_summary", - END: END, - }, - ) - graph.add_conditional_edges( - "implement_review", - lambda s: s.get("current_node", "wait_for_ci_gate"), - { - "wait_for_ci_gate": "wait_for_ci_gate", - "review_response_gate": "review_response_gate", - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", - "escalate_blocked": "escalate_blocked", - }, - ) - graph.add_conditional_edges( - "review_response_gate", - route_review_response, - { - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", - END: END, - }, + # ── Post-PR edges (CI/review) - shared across all workflows ── + add_post_pr_edges( + graph, + on_complete_node="post_merge_summary", + human_review_routing_fn=_route_human_review_bug, ) # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── - graph.add_node("rebase_pr", rebase_pr) + # Note: rebase_pr node is added by add_post_pr_nodes graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), @@ -732,7 +605,6 @@ def build_bug_graph() -> StateGraph: "update_documentation": "update_documentation", "create_pr": "create_pr", "teardown_workspace": "teardown_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "human_review_gate": "human_review_gate", diff --git a/src/forge/workflow/feature/graph.py b/src/forge/workflow/feature/graph.py index 19d571c2..10c49a16 100644 --- a/src/forge/workflow/feature/graph.py +++ b/src/forge/workflow/feature/graph.py @@ -22,42 +22,35 @@ from forge.workflow.nodes import ( aggregate_epic_status, aggregate_feature_status, - attempt_ci_fix, complete_tasks, create_pull_request, decompose_epics, - escalate_to_blocked, - evaluate_ci_status, generate_prd, generate_spec, generate_tasks, - human_review_gate, implement_task, local_review_changes, regenerate_all_epics, regenerate_prd_with_feedback, regenerate_spec_with_feedback, - route_human_review, route_tasks_by_repo, route_tasks_parallel, setup_workspace, teardown_and_route, update_documentation, update_single_epic, - wait_for_ci_gate, -) -from forge.workflow.nodes.implement_review import ( - implement_review, - review_response_gate, - route_review_response, ) from forge.workflow.nodes.qa_handler import answer_question -from forge.workflow.nodes.rebase import rebase_pr from forge.workflow.nodes.task_generation import ( regenerate_all_tasks, regenerate_epic_tasks, update_single_task, ) +from forge.workflow.post_pr import ( + _route_after_pr_creation, + add_post_pr_edges, + add_post_pr_nodes, +) from forge.workflow.utils import resolve_shared_resume_node logger = logging.getLogger(__name__) @@ -121,8 +114,6 @@ def route_by_ticket_type(state: FeatureState) -> str: return "regenerate_epic_tasks" elif current_node == "task_approval_gate": return "task_approval_gate" - elif current_node == "wait_for_ci_gate": - return "wait_for_ci_gate" elif current_node in ("implement_task", "implementation", "implement_bug_fix"): return "implement_task" elif current_node == "setup_workspace": @@ -336,56 +327,6 @@ def _route_implementation( return "implement_task" -def _route_after_pr_creation( - state: FeatureState, -) -> Literal["teardown_workspace", "escalate_blocked"]: - """Route after PR creation attempt. - - On success: proceed to teardown workspace - On failure: escalate to blocked status - - Args: - state: Current workflow state. - - Returns: - Next node name. - """ - last_error = state.get("last_error") - pr_urls = state.get("pr_urls", []) - - # If there's an error and no PR was created, escalate - if last_error and not pr_urls: - return "escalate_blocked" - - # Success or partial success - proceed to teardown - return "teardown_workspace" - - -def _route_after_teardown(state: FeatureState) -> Literal["setup_workspace", "wait_for_ci_gate"]: - """Route after workspace teardown.""" - repos_to_process = state.get("repos_to_process", []) - repos_completed = state.get("repos_completed", []) - - remaining = [r for r in repos_to_process if r not in repos_completed] - if remaining: - return "setup_workspace" - return "wait_for_ci_gate" - - -def _route_ci_evaluation( - state: FeatureState, -) -> Literal["human_review_gate", "attempt_ci_fix", "escalate_blocked", "__end__"]: - """Route based on CI evaluation results.""" - ci_status = state.get("ci_status", "") - - routes = { - "passed": "human_review_gate", - "fixing": "attempt_ci_fix", - "pending": "__end__", # Pause workflow until CI webhook - } - return routes.get(ci_status, "escalate_blocked") - - def _route_after_answer(state: FeatureState) -> str: """Route back to the original gate after answering a question. @@ -474,16 +415,10 @@ def build_feature_graph() -> StateGraph: # Documentation update node (pre-PR, updates stale docs) graph.add_node("update_documentation", update_documentation) - # CI/CD Validation nodes (US7) - graph.add_node("wait_for_ci_gate", wait_for_ci_gate) - graph.add_node("ci_evaluator", evaluate_ci_status) - graph.add_node("attempt_ci_fix", attempt_ci_fix) - graph.add_node("escalate_blocked", escalate_to_blocked) + # Post-PR nodes (CI/review) - shared across all workflows + add_post_pr_nodes(graph) - # Human Review nodes (US9) - graph.add_node("human_review_gate", human_review_gate) - graph.add_node("implement_review", implement_review) - graph.add_node("review_response_gate", review_response_gate) + # Feature workflow completion nodes graph.add_node("complete_tasks", complete_tasks) graph.add_node("aggregate_epic_status", aggregate_epic_status) graph.add_node("aggregate_feature_status", aggregate_feature_status) @@ -525,7 +460,6 @@ def build_feature_graph() -> StateGraph: # Resume routing for pre-PR and CI/review stages "local_review": "local_review", "update_documentation": "update_documentation", - "wait_for_ci_gate": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "human_review_gate": "human_review_gate", "implement_review": "implement_review", @@ -721,76 +655,10 @@ def build_feature_graph() -> StateGraph: "escalate_blocked": "escalate_blocked", }, ) - graph.add_conditional_edges( - "teardown_workspace", - _route_after_teardown, - { - "setup_workspace": "setup_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", - }, - ) + # Post-PR edges (CI/review) - shared across all workflows + add_post_pr_edges(graph, on_complete_node="complete_tasks") - # CI/CD flow (US7) - graph.add_conditional_edges( - "wait_for_ci_gate", - lambda s: END if s.get("is_paused") else "ci_evaluator", - {END: END, "ci_evaluator": "ci_evaluator"}, - ) - graph.add_conditional_edges( - "ci_evaluator", - _route_ci_evaluation, - { - "human_review_gate": "human_review_gate", - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - END: END, # Pause: CI still running, wait for next webhook - }, - ) - graph.add_conditional_edges( - "attempt_ci_fix", - lambda s: s.get("current_node", "wait_for_ci_gate"), - { - "wait_for_ci_gate": "wait_for_ci_gate", - "escalate_blocked": "escalate_blocked", - "ci_evaluator": "ci_evaluator", - # Exception handlers set current_node="attempt_ci_fix" so resume routing - # can map it back to ci_evaluator. Escalate within the current invocation - # so the graph doesn't crash on an unmapped key. - "attempt_ci_fix": "escalate_blocked", - }, - ) - graph.add_edge("escalate_blocked", END) - - # Human Review flow (US9) - graph.add_conditional_edges( - "human_review_gate", - route_human_review, - { - "implement_review": "implement_review", - "complete_tasks": "complete_tasks", - END: END, # Pause workflow until review webhook - }, - ) - graph.add_conditional_edges( - "implement_review", - lambda s: s.get("current_node", "wait_for_ci_gate"), - { - "wait_for_ci_gate": "wait_for_ci_gate", - "review_response_gate": "review_response_gate", - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", - "escalate_blocked": "escalate_blocked", - }, - ) - graph.add_conditional_edges( - "review_response_gate", - route_review_response, - { - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", - END: END, - }, - ) + # Feature workflow completion chain graph.add_edge("complete_tasks", "aggregate_epic_status") graph.add_edge("aggregate_epic_status", "aggregate_feature_status") graph.add_edge("aggregate_feature_status", END) @@ -808,15 +676,26 @@ def build_feature_graph() -> StateGraph: ) # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── - graph.add_node("rebase_pr", rebase_pr) + # Note: rebase_pr node is added by add_post_pr_nodes graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), { + "generate_prd": "generate_prd", "prd_approval_gate": "prd_approval_gate", + "regenerate_prd": "regenerate_prd", + "generate_spec": "generate_spec", "spec_approval_gate": "spec_approval_gate", + "regenerate_spec": "regenerate_spec", + "decompose_epics": "decompose_epics", "plan_approval_gate": "plan_approval_gate", + "regenerate_all_epics": "regenerate_all_epics", + "update_single_epic": "update_single_epic", + "generate_tasks": "generate_tasks", "task_approval_gate": "task_approval_gate", + "regenerate_all_tasks": "regenerate_all_tasks", + "update_single_task": "update_single_task", + "regenerate_epic_tasks": "regenerate_epic_tasks", "task_router": "task_router", "setup_workspace": "setup_workspace", "implement_task": "implement_task", @@ -824,7 +703,6 @@ def build_feature_graph() -> StateGraph: "update_documentation": "update_documentation", "create_pr": "create_pr", "teardown_workspace": "teardown_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "human_review_gate": "human_review_gate", diff --git a/src/forge/workflow/task_takeover/graph.py b/src/forge/workflow/task_takeover/graph.py index b2cacfcb..c0a09636 100644 --- a/src/forge/workflow/task_takeover/graph.py +++ b/src/forge/workflow/task_takeover/graph.py @@ -16,24 +16,21 @@ ) from forge.workflow.nodes import ( answer_question, - attempt_ci_fix, create_pull_request, - escalate_to_blocked, - evaluate_ci_status, execute_task_changes, generate_plan, - human_review_gate, - implement_review, - review_response_gate, route_human_review, - route_review_response, route_triage_gate, run_qualitative_review, setup_workspace, teardown_and_route, triage_gate, triage_task, - wait_for_ci_gate, +) +from forge.workflow.post_pr import ( + _route_after_pr_creation, + add_post_pr_edges, + add_post_pr_nodes, ) from forge.workflow.task_takeover.state import TaskTakeoverState from forge.workflow.utils import resolve_shared_resume_node, update_state_timestamp @@ -193,36 +190,6 @@ def _route_after_qualitative_review(state: TaskTakeoverState) -> str: return "execute_task_changes" -def _route_after_pr_creation(state: TaskTakeoverState) -> str: - """Route after PR creation: teardown on success, escalate on failure.""" - last_error = state.get("last_error") - pr_urls = state.get("pr_urls", []) - if last_error and not pr_urls: - return "escalate_blocked" - return "teardown_workspace" - - -def _route_after_teardown(state: TaskTakeoverState) -> str: - """Route after teardown: next repo or wait for CI.""" - repos_to_process = state.get("repos_to_process", []) - repos_completed = state.get("repos_completed", []) - remaining = [r for r in repos_to_process if r not in repos_completed] - if remaining: - return "setup_workspace" - return "wait_for_ci_gate" - - -def _route_ci_evaluation(state: TaskTakeoverState) -> str: - """Route based on CI evaluation results.""" - ci_status = state.get("ci_status", "") - routes = { - "passed": "human_review_gate", - "fixing": "attempt_ci_fix", - "pending": END, - } - return routes.get(ci_status, "escalate_blocked") - - def _route_human_review_task_takeover(state: TaskTakeoverState) -> str: """Route after human_review_gate for a standalone Task/Epic PR.""" if state.get("pr_merged"): @@ -275,21 +242,17 @@ def build_task_takeover_graph() -> StateGraph[TaskTakeoverState, Any, Any]: graph.add_node("triage_gate", triage_gate) graph.add_node("generate_plan", generate_plan) graph.add_node("task_plan_approval_gate", task_plan_approval_gate) - graph.add_node("escalate_blocked", escalate_to_blocked) graph.add_node("answer_question", answer_question) graph.add_node("setup_workspace", setup_workspace) graph.add_node("execute_task_changes", execute_task_changes) graph.add_node("run_qualitative_review", run_qualitative_review) graph.add_node("create_pr", create_pull_request) graph.add_node("teardown_workspace", teardown_and_route) - graph.add_node("wait_for_ci_gate", wait_for_ci_gate) - graph.add_node("ci_evaluator", evaluate_ci_status) - graph.add_node("attempt_ci_fix", attempt_ci_fix) - graph.add_node("human_review_gate", human_review_gate) - graph.add_node("implement_review", implement_review) - graph.add_node("review_response_gate", review_response_gate) graph.add_node("complete_task_takeover", complete_task_takeover) + # Post-PR nodes (CI/review) - shared across all workflows + add_post_pr_nodes(graph) + # Set entry point graph.set_entry_point("route_entry") @@ -307,12 +270,12 @@ def build_task_takeover_graph() -> StateGraph[TaskTakeoverState, Any, Any]: "run_qualitative_review": "run_qualitative_review", "create_pr": "create_pr", "teardown_workspace": "teardown_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "human_review_gate": "human_review_gate", "implement_review": "implement_review", "review_response_gate": "review_response_gate", + "rebase_pr": "rebase_pr", "escalate_blocked": "escalate_blocked", END: END, }, @@ -388,69 +351,39 @@ def build_task_takeover_graph() -> StateGraph[TaskTakeoverState, Any, Any]: "escalate_blocked": "escalate_blocked", }, ) - graph.add_conditional_edges( - "teardown_workspace", - _route_after_teardown, - { - "setup_workspace": "setup_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", - }, - ) - graph.add_conditional_edges( - "wait_for_ci_gate", - lambda s: END if s.get("is_paused") else "ci_evaluator", - {END: END, "ci_evaluator": "ci_evaluator"}, - ) - graph.add_conditional_edges( - "ci_evaluator", - _route_ci_evaluation, - { - "human_review_gate": "human_review_gate", - "attempt_ci_fix": "attempt_ci_fix", - "escalate_blocked": "escalate_blocked", - END: END, - }, + # Post-PR edges (CI/review) - shared across all workflows + add_post_pr_edges( + graph, + on_complete_node="complete_task_takeover", + human_review_routing_fn=_route_human_review_task_takeover, ) + + graph.add_edge("complete_task_takeover", END) + + # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── + # Note: rebase_pr node is added by add_post_pr_nodes graph.add_conditional_edges( - "attempt_ci_fix", - lambda s: s.get("current_node", "wait_for_ci_gate"), + "rebase_pr", + lambda s: s.get("current_node", END), { - "wait_for_ci_gate": "wait_for_ci_gate", - "escalate_blocked": "escalate_blocked", + "triage_gate": "triage_gate", + "generate_plan": "generate_plan", + "task_plan_approval_gate": "task_plan_approval_gate", + "setup_workspace": "setup_workspace", + "execute_task_changes": "execute_task_changes", + "run_qualitative_review": "run_qualitative_review", + "create_pr": "create_pr", + "teardown_workspace": "teardown_workspace", "ci_evaluator": "ci_evaluator", - "attempt_ci_fix": "escalate_blocked", - }, - ) - graph.add_conditional_edges( - "human_review_gate", - _route_human_review_task_takeover, - { + "attempt_ci_fix": "ci_evaluator", + "human_review_gate": "human_review_gate", "implement_review": "implement_review", - "complete_task_takeover": "complete_task_takeover", - END: END, - }, - ) - graph.add_conditional_edges( - "implement_review", - lambda s: s.get("current_node", "wait_for_ci_gate"), - { - "wait_for_ci_gate": "wait_for_ci_gate", "review_response_gate": "review_response_gate", - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", + "complete_task_takeover": "complete_task_takeover", "escalate_blocked": "escalate_blocked", - }, - ) - graph.add_conditional_edges( - "review_response_gate", - route_review_response, - { - "implement_review": "implement_review", - "human_review_gate": "human_review_gate", END: END, }, ) - graph.add_edge("complete_task_takeover", END) # Q&A routing graph.add_conditional_edges( diff --git a/src/forge/workflow/utils/__init__.py b/src/forge/workflow/utils/__init__.py index 03a751b5..6b539f0a 100644 --- a/src/forge/workflow/utils/__init__.py +++ b/src/forge/workflow/utils/__init__.py @@ -25,7 +25,7 @@ "human_review_gate": "human_review_gate", "implement_review": "implement_review", "review_response_gate": "review_response_gate", - "wait_for_ci_gate": "human_review_gate", # backward compat for in-flight tickets + "wait_for_ci_gate": "ci_evaluator", # backward compat: wait_for_ci_gate deleted, maps to ci_evaluator "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "rebase_pr": "rebase_pr", diff --git a/tests/unit/workflow/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py index 5cad4a8d..a180d2e8 100644 --- a/tests/unit/workflow/bug/test_graph.py +++ b/tests/unit/workflow/bug/test_graph.py @@ -78,6 +78,42 @@ class TestRouteEntry: ("aggregate_feature_status", END), ], ) + @pytest.mark.parametrize("node,expected", [ + # New nodes + ("triage_check", "triage_check"), + ("triage_gate", "triage_gate"), + ("analyze_bug", "analyze_bug"), + ("reflect_rca", "reflect_rca"), + ("rca_option_gate", "rca_option_gate"), + ("plan_bug_fix", "plan_bug_fix"), + ("plan_approval_gate", "plan_approval_gate"), + ("regenerate_plan", "regenerate_plan"), + ("decompose_plan", "decompose_plan"), + ("post_merge_summary", "post_merge_summary"), + # Backward compat: old rca_approval_gate value maps to rca_option_gate + ("rca_approval_gate", "rca_option_gate"), + # regenerate_rca performs cleanup before routing through analyze_bug + ("regenerate_rca", "regenerate_rca"), + # Preserved existing nodes + ("setup_workspace", "setup_workspace"), + ("implement_bug_fix", "implement_bug_fix"), + ("local_review", "local_review"), + ("update_documentation", "update_documentation"), + ("create_pr", "create_pr"), + ("teardown_workspace", "teardown_workspace"), + ("ci_evaluator", "ci_evaluator"), + ("attempt_ci_fix", "ci_evaluator"), + ("wait_for_ci_gate", "ci_evaluator"), # Backward compat: wait_for_ci_gate deleted, maps to ci_evaluator + ("ai_review", "human_review_gate"), + ("human_review_gate", "human_review_gate"), + ("implement_review", "implement_review"), + ("review_response_gate", "review_response_gate"), + ("escalate_blocked", "escalate_blocked"), + ("complete", END), + ("complete_tasks", END), + ("aggregate_epic_status", END), + ("aggregate_feature_status", END), + ]) def test_route_entry_mapping(self, node, expected): """route_entry maps each current_node to the correct resume target.""" state = _bug_state(current_node=node) @@ -341,32 +377,21 @@ def test_post_merge_summary_in_graph(self): compiled = graph.compile() assert "post_merge_summary" in compiled.nodes - def test_ci_fix_routes_through_wait_for_ci_gate(self): + def test_ci_fix_routes_to_human_review_gate(self): + """attempt_ci_fix can route to human_review_gate after fixing.""" graph = build_bug_graph() compiled = graph.compile() targets = {e.target for e in compiled.get_graph().edges if e.source == "attempt_ci_fix"} - assert "wait_for_ci_gate" in targets + assert "human_review_gate" in targets - def test_review_fix_routes_through_wait_for_ci_gate(self): + def test_review_fix_routes_to_human_review_gate(self): + """implement_review can route to human_review_gate after implementing changes.""" graph = build_bug_graph() compiled = graph.compile() targets = {e.target for e in compiled.get_graph().edges if e.source == "implement_review"} - assert "wait_for_ci_gate" in targets - - def test_wait_for_ci_gate_paused_routes_to_end(self): - graph = build_bug_graph() - compiled = graph.compile() - ci_gate_edges = compiled.get_graph().edges - targets = {e.target for e in ci_gate_edges if e.source == "wait_for_ci_gate"} - assert END in targets - - def test_wait_for_ci_gate_not_paused_routes_to_ci_evaluator(self): - graph = build_bug_graph() - compiled = graph.compile() - targets = {e.target for e in compiled.get_graph().edges if e.source == "wait_for_ci_gate"} - assert "ci_evaluator" in targets + assert "human_review_gate" in targets def test_attempt_ci_fix_escalates_on_self_referential_failure(self): graph = build_bug_graph() @@ -380,7 +405,7 @@ def test_rebase_can_return_to_post_pr_nodes(self): targets = {e.target for e in compiled.get_graph().edges if e.source == "rebase_pr"} assert { - "wait_for_ci_gate", + "ci_evaluator", "implement_review", "review_response_gate", "create_pr", diff --git a/tests/unit/workflow/feature/test_workflow.py b/tests/unit/workflow/feature/test_workflow.py index 101c23c9..401e1dca 100644 --- a/tests/unit/workflow/feature/test_workflow.py +++ b/tests/unit/workflow/feature/test_workflow.py @@ -357,7 +357,7 @@ def test_rebase_can_return_to_post_pr_nodes(self): targets = {e.target for e in compiled.get_graph().edges if e.source == "rebase_pr"} assert { - "wait_for_ci_gate", + "ci_evaluator", "implement_review", "review_response_gate", "create_pr", diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py index 6529f1f1..4498286d 100644 --- a/tests/unit/workflow/test_base.py +++ b/tests/unit/workflow/test_base.py @@ -175,9 +175,9 @@ def build_graph(self): class TestSharedResumeMap: """Tests for resolve_shared_resume_node backward compat.""" - def test_wait_for_ci_gate_resolves_to_human_review_gate(self): - """In-flight tickets with wait_for_ci_gate resume at human_review_gate.""" + def test_wait_for_ci_gate_resolves_to_ci_evaluator(self): + """In-flight tickets with wait_for_ci_gate resume at ci_evaluator.""" from forge.workflow.utils import resolve_shared_resume_node result = resolve_shared_resume_node("wait_for_ci_gate") - assert result == "human_review_gate" + assert result == "ci_evaluator" diff --git a/tests/unit/workflow/test_cleanup.py b/tests/unit/workflow/test_cleanup.py index a63cceff..a49df26a 100644 --- a/tests/unit/workflow/test_cleanup.py +++ b/tests/unit/workflow/test_cleanup.py @@ -101,7 +101,7 @@ def test_existing_nodes_still_mapped(self): "teardown_workspace": "teardown_workspace", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", - "wait_for_ci_gate": "wait_for_ci_gate", + "wait_for_ci_gate": "ci_evaluator", # Backward compat: wait_for_ci_gate deleted, maps to ci_evaluator "ai_review": "human_review_gate", "human_review_gate": "human_review_gate", "implement_review": "implement_review", From 089a4216969c4cba5013b5d510fca8c00bf3b891 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:35:07 -0400 Subject: [PATCH 08/19] =?UTF-8?q?fix:=20restore=20wait=5Ffor=5Fci=5Fgate?= =?UTF-8?q?=E2=86=92human=5Freview=5Fgate=20backward=20compat,=20delete=20?= =?UTF-8?q?orphaned=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/forge/workflow/utils/__init__.py | 2 +- tests/unit/workflow/test_base.py | 6 +- .../unit/workflow/test_pr_status_comments.py | 535 ------------------ 3 files changed, 4 insertions(+), 539 deletions(-) delete mode 100644 tests/unit/workflow/test_pr_status_comments.py diff --git a/src/forge/workflow/utils/__init__.py b/src/forge/workflow/utils/__init__.py index 6b539f0a..03a751b5 100644 --- a/src/forge/workflow/utils/__init__.py +++ b/src/forge/workflow/utils/__init__.py @@ -25,7 +25,7 @@ "human_review_gate": "human_review_gate", "implement_review": "implement_review", "review_response_gate": "review_response_gate", - "wait_for_ci_gate": "ci_evaluator", # backward compat: wait_for_ci_gate deleted, maps to ci_evaluator + "wait_for_ci_gate": "human_review_gate", # backward compat for in-flight tickets "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "rebase_pr": "rebase_pr", diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py index 4498286d..6529f1f1 100644 --- a/tests/unit/workflow/test_base.py +++ b/tests/unit/workflow/test_base.py @@ -175,9 +175,9 @@ def build_graph(self): class TestSharedResumeMap: """Tests for resolve_shared_resume_node backward compat.""" - def test_wait_for_ci_gate_resolves_to_ci_evaluator(self): - """In-flight tickets with wait_for_ci_gate resume at ci_evaluator.""" + def test_wait_for_ci_gate_resolves_to_human_review_gate(self): + """In-flight tickets with wait_for_ci_gate resume at human_review_gate.""" from forge.workflow.utils import resolve_shared_resume_node result = resolve_shared_resume_node("wait_for_ci_gate") - assert result == "ci_evaluator" + assert result == "human_review_gate" diff --git a/tests/unit/workflow/test_pr_status_comments.py b/tests/unit/workflow/test_pr_status_comments.py deleted file mode 100644 index 7a5deaf5..00000000 --- a/tests/unit/workflow/test_pr_status_comments.py +++ /dev/null @@ -1,535 +0,0 @@ -"""Unit tests for PR status comment and label transition logic. - -These tests verify the core logic of PR creation status comments and label -transitions in the wait_for_ci_gate node, focusing on: -- PR number extraction (valid, missing, malformed) -- PR status comment posting with/without PR number -- Label removal (forge:implementing) with success and failure cases -- Label addition (forge:ci-pending) with success and failure cases -- Error suppression and logging for all operations -- Workflow continuation after failures -""" - -from unittest.mock import AsyncMock, MagicMock, patch - -import pytest - -from forge.workflow.feature.state import create_initial_feature_state -from forge.workflow.nodes.ci_evaluator import wait_for_ci_gate - - -def create_mock_jira_client(): - """Create a mock JiraClient with all required methods.""" - mock = MagicMock() - mock.close = AsyncMock() - mock.add_comment = AsyncMock() - mock.remove_labels = AsyncMock() - mock.set_workflow_label = AsyncMock() - return mock - - -class TestPRNumberExtraction: - """Test PR number extraction from workflow state.""" - - @pytest.mark.asyncio - async def test_pr_number_extraction_with_valid_response(self): - """Verify PR number is correctly extracted when present in state.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-100", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 42 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify PR number used in status comment - assert mock_jira.add_comment.call_count == 1 - comment_call = mock_jira.add_comment.call_args - assert "Pull request #42" in comment_call[0][1] - - @pytest.mark.asyncio - async def test_pr_number_extraction_with_missing_pr_number(self): - """Verify fallback message when PR number is None.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-101", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = None - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify fallback message used - assert mock_jira.add_comment.call_count == 1 - comment_call = mock_jira.add_comment.call_args - assert comment_call[0][1] == "🚀 Pull request created and submitted. Waiting for CI checks to complete." - assert "#" not in comment_call[0][1] - - @pytest.mark.asyncio - async def test_pr_number_extraction_with_malformed_response(self): - """Verify handling when current_pr_number key is missing from state.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-102", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - # Don't set current_pr_number at all - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify fallback message used when key is missing - assert mock_jira.add_comment.call_count == 1 - comment_call = mock_jira.add_comment.call_args - assert comment_call[0][1] == "🚀 Pull request created and submitted. Waiting for CI checks to complete." - - -class TestPRStatusCommentPosting: - """Test PR status comment posting logic.""" - - @pytest.mark.asyncio - async def test_status_comment_posted_with_pr_number_present(self): - """Verify status comment posted with PR number when available.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-200", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 999 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify comment posted to correct ticket with correct message - mock_jira.add_comment.assert_called_once_with( - "TEST-200", - "🚀 Pull request #999 created and submitted. Waiting for CI checks to complete." - ) - - @pytest.mark.asyncio - async def test_status_comment_posted_with_pr_number_absent(self): - """Verify fallback status comment posted when PR number absent.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-201", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = None - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify fallback comment posted to correct ticket - mock_jira.add_comment.assert_called_once_with( - "TEST-201", - "🚀 Pull request created and submitted. Waiting for CI checks to complete." - ) - - @pytest.mark.asyncio - async def test_status_comment_not_posted_on_reentry(self): - """Verify status comment only posted on initial entry, not after CI fix.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-202", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 123 - state["ci_fix_attempt"] = 1 # Re-entry after CI fix - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify NO comment posted on re-entry - mock_jira.add_comment.assert_not_called() - - -class TestLabelRemoval: - """Test forge:implementing label removal logic.""" - - @pytest.mark.asyncio - async def test_label_removal_success(self): - """Verify forge:implementing label removed successfully.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-300", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 100 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify remove_labels called with correct parameters - mock_jira.remove_labels.assert_called_once_with( - "TEST-300", - ["forge:implementing"] - ) - # Verify workflow continues - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - - @pytest.mark.asyncio - async def test_label_removal_label_not_present(self, caplog): - """Verify workflow continues when label is not present on ticket.""" - mock_jira = create_mock_jira_client() - # Simulate label not present (Jira API might raise exception) - mock_jira.remove_labels.side_effect = Exception("Label not found") - - state = create_initial_feature_state( - ticket_key="TEST-301", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 101 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues despite error - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify error logged (via post_status_comment utility) - assert any("Failed to remove implementing label" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_label_removal_api_error(self, caplog): - """Verify workflow continues when label removal API fails.""" - mock_jira = create_mock_jira_client() - # Simulate API error - mock_jira.remove_labels.side_effect = Exception("Jira API timeout") - - state = create_initial_feature_state( - ticket_key="TEST-302", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 102 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues despite API error - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify error logged at WARNING level - assert any("Failed to remove implementing label" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_label_removal_not_called_on_reentry(self): - """Verify label removal only happens on initial entry, not after CI fix.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-303", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 103 - state["ci_fix_attempt"] = 2 # Re-entry after CI fix - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify remove_labels NOT called on re-entry - mock_jira.remove_labels.assert_not_called() - - -class TestLabelAddition: - """Test forge:ci-pending label addition logic.""" - - @pytest.mark.asyncio - async def test_label_addition_success(self): - """Verify forge:ci-pending label added successfully.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-400", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 200 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify set_workflow_label called with forge:ci-pending - from forge.models.workflow import ForgeLabel - mock_jira.set_workflow_label.assert_called_once_with( - "TEST-400", - ForgeLabel.TASK_CI_PENDING - ) - # Verify workflow continues - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - - @pytest.mark.asyncio - async def test_label_addition_api_error(self, caplog): - """Verify workflow continues when label addition API fails.""" - mock_jira = create_mock_jira_client() - # Simulate API error - mock_jira.set_workflow_label.side_effect = Exception("Jira API connection error") - - state = create_initial_feature_state( - ticket_key="TEST-401", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 201 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues despite API error - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify error logged at WARNING level - assert any("Failed to set ci-pending label" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_label_addition_not_called_on_reentry(self): - """Verify label addition only happens on initial entry, not after CI fix.""" - mock_jira = create_mock_jira_client() - - state = create_initial_feature_state( - ticket_key="TEST-402", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 202 - state["ci_fix_attempt"] = 3 # Re-entry after CI fix - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify set_workflow_label NOT called on re-entry - mock_jira.set_workflow_label.assert_not_called() - - -class TestErrorSuppressionAndLogging: - """Test error suppression and logging for all label operations.""" - - @pytest.mark.asyncio - async def test_comment_posting_error_logged_and_suppressed(self, caplog): - """Verify comment posting errors are logged at WARNING level and suppressed.""" - mock_jira = create_mock_jira_client() - mock_jira.add_comment.side_effect = Exception("Comment API error") - - state = create_initial_feature_state( - ticket_key="TEST-500", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 300 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues (exception suppressed) - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify error logged - assert any("Failed to post status comment" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_label_removal_error_logged_and_suppressed(self, caplog): - """Verify label removal errors are logged at WARNING level and suppressed.""" - mock_jira = create_mock_jira_client() - mock_jira.remove_labels.side_effect = Exception("Remove label API error") - - state = create_initial_feature_state( - ticket_key="TEST-501", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 301 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues - assert result["is_paused"] is True - # Verify error logged - assert any("Failed to remove implementing label" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_label_addition_error_logged_and_suppressed(self, caplog): - """Verify label addition errors are logged at WARNING level and suppressed.""" - mock_jira = create_mock_jira_client() - mock_jira.set_workflow_label.side_effect = Exception("Add label API error") - - state = create_initial_feature_state( - ticket_key="TEST-502", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 302 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues - assert result["is_paused"] is True - # Verify error logged - assert any("Failed to set ci-pending label" in record.message - for record in caplog.records if record.levelname == "WARNING") - - @pytest.mark.asyncio - async def test_all_operations_fail_workflow_still_continues(self, caplog): - """Verify workflow continues when all Jira operations fail.""" - mock_jira = create_mock_jira_client() - # Simulate all operations failing - mock_jira.add_comment.side_effect = Exception("Comment failed") - mock_jira.remove_labels.side_effect = Exception("Remove failed") - mock_jira.set_workflow_label.side_effect = Exception("Add failed") - - state = create_initial_feature_state( - ticket_key="TEST-503", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 303 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow continues despite all failures - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify all errors logged - warning_messages = [record.message for record in caplog.records if record.levelname == "WARNING"] - assert any("Failed to post status comment" in msg for msg in warning_messages) - assert any("Failed to remove implementing label" in msg for msg in warning_messages) - assert any("Failed to set ci-pending label" in msg for msg in warning_messages) - - -class TestWorkflowContinuation: - """Test that workflow continues after comment/label failures.""" - - @pytest.mark.asyncio - async def test_workflow_continues_after_comment_failure(self): - """Verify workflow continues to completion after comment posting fails.""" - mock_jira = create_mock_jira_client() - mock_jira.add_comment.side_effect = Exception("Comment API down") - - state = create_initial_feature_state( - ticket_key="TEST-600", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 400 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify state updated correctly - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - assert result["ticket_key"] == "TEST-600" - - @pytest.mark.asyncio - async def test_workflow_continues_after_label_failures(self): - """Verify workflow continues after both label operations fail.""" - mock_jira = create_mock_jira_client() - mock_jira.remove_labels.side_effect = Exception("Cannot remove") - mock_jira.set_workflow_label.side_effect = Exception("Cannot add") - - state = create_initial_feature_state( - ticket_key="TEST-601", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 401 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify state progresses correctly - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify JiraClient still closed properly - mock_jira.close.assert_called_once() - - @pytest.mark.asyncio - async def test_jira_client_closed_even_after_failures(self): - """Verify JiraClient is properly closed even when operations fail.""" - mock_jira = create_mock_jira_client() - mock_jira.add_comment.side_effect = Exception("Comment failed") - mock_jira.remove_labels.side_effect = Exception("Remove failed") - mock_jira.set_workflow_label.side_effect = Exception("Add failed") - - state = create_initial_feature_state( - ticket_key="TEST-602", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 402 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - await wait_for_ci_gate(state) - - # Verify JiraClient.close() called in finally block - mock_jira.close.assert_called_once() - - @pytest.mark.asyncio - async def test_workflow_continues_with_mixed_success_and_failure(self): - """Verify workflow continues when some operations succeed and some fail.""" - mock_jira = create_mock_jira_client() - # Comment succeeds, remove_labels fails, set_workflow_label succeeds - mock_jira.remove_labels.side_effect = Exception("Remove failed") - - state = create_initial_feature_state( - ticket_key="TEST-603", - current_repo="owner/repo", - task_keys=["TASK-001"], - ) - state["current_pr_number"] = 403 - state["ci_fix_attempt"] = 0 - - with patch("forge.workflow.nodes.ci_evaluator.JiraClient", return_value=mock_jira): - result = await wait_for_ci_gate(state) - - # Verify workflow completes successfully - assert result["is_paused"] is True - assert result["current_node"] == "wait_for_ci_gate" - # Verify successful operations were called - mock_jira.add_comment.assert_called_once() - mock_jira.set_workflow_label.assert_called_once() From 2402423175da1379987d5c23aff2736ddaa37667 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:43:20 -0400 Subject: [PATCH 09/19] fix: replace remaining wait_for_ci_gate references with human_review_gate pr_creation and implement_review still routed to the deleted node; update them and fix tests that expected ci_evaluator for backward compat. Co-Authored-By: Claude Opus 4.6 --- src/forge/workflow/feature/graph.py | 8 ++++---- src/forge/workflow/nodes/implement_review.py | 4 ++-- src/forge/workflow/nodes/pr_creation.py | 2 +- tests/unit/workflow/bug/test_graph.py | 2 +- tests/unit/workflow/test_cleanup.py | 2 +- 5 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/forge/workflow/feature/graph.py b/src/forge/workflow/feature/graph.py index 10c49a16..f63ae307 100644 --- a/src/forge/workflow/feature/graph.py +++ b/src/forge/workflow/feature/graph.py @@ -363,11 +363,11 @@ def build_feature_graph() -> StateGraph: 17. local_review: reviews git diff vs main, fixes breaking issues in-place (up to 2 passes) 18. local_review -> create_pr 19. create_pr -> teardown_workspace - 20. teardown_workspace -> wait_for_ci_gate (pause) or next repo - 21. wait_for_ci_gate: resumes on GitHub CI webhook + 20. teardown_workspace -> human_review_gate (pause) or next repo + 21. human_review_gate: resumes on GitHub CI or review webhook 22. ci_evaluator: checks CI status, attempts autonomous fixes on failure (up to 5 retries) - 23. ci_evaluator (passed) -> human_review_gate - 24. human_review_gate -> complete_tasks + 23. ci_evaluator -> human_review_gate + 24. human_review_gate (approved) -> complete_tasks 25. complete_tasks -> aggregate_epic_status -> aggregate_feature_status -> END Returns: diff --git a/src/forge/workflow/nodes/implement_review.py b/src/forge/workflow/nodes/implement_review.py index d37a9eed..f7592e44 100644 --- a/src/forge/workflow/nodes/implement_review.py +++ b/src/forge/workflow/nodes/implement_review.py @@ -269,8 +269,8 @@ async def implement_review(state: WorkflowState) -> WorkflowState: logger.info(f"No new commits after review implementation for {ticket_key}") # Only re-enter the CI gate if we actually pushed new commits; otherwise - # CI won't re-trigger and wait_for_ci_gate would block forever. - next_node = "wait_for_ci_gate" if unpushed else "human_review_gate" + # CI won't re-trigger and human_review_gate would block forever. + next_node = "human_review_gate" return update_state_timestamp( { diff --git a/src/forge/workflow/nodes/pr_creation.py b/src/forge/workflow/nodes/pr_creation.py index 35bd6d28..fde02052 100644 --- a/src/forge/workflow/nodes/pr_creation.py +++ b/src/forge/workflow/nodes/pr_creation.py @@ -609,6 +609,6 @@ async def teardown_and_route(state: WorkflowState) -> WorkflowState: { **state, "repos_completed": repos_completed, - "current_node": "wait_for_ci_gate", + "current_node": "human_review_gate", } ) diff --git a/tests/unit/workflow/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py index a180d2e8..dae61da5 100644 --- a/tests/unit/workflow/bug/test_graph.py +++ b/tests/unit/workflow/bug/test_graph.py @@ -103,7 +103,7 @@ class TestRouteEntry: ("teardown_workspace", "teardown_workspace"), ("ci_evaluator", "ci_evaluator"), ("attempt_ci_fix", "ci_evaluator"), - ("wait_for_ci_gate", "ci_evaluator"), # Backward compat: wait_for_ci_gate deleted, maps to ci_evaluator + ("wait_for_ci_gate", "human_review_gate"), # Backward compat: in-flight tickets resume at gate ("ai_review", "human_review_gate"), ("human_review_gate", "human_review_gate"), ("implement_review", "implement_review"), diff --git a/tests/unit/workflow/test_cleanup.py b/tests/unit/workflow/test_cleanup.py index a49df26a..63019db1 100644 --- a/tests/unit/workflow/test_cleanup.py +++ b/tests/unit/workflow/test_cleanup.py @@ -101,7 +101,7 @@ def test_existing_nodes_still_mapped(self): "teardown_workspace": "teardown_workspace", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", - "wait_for_ci_gate": "ci_evaluator", # Backward compat: wait_for_ci_gate deleted, maps to ci_evaluator + "wait_for_ci_gate": "human_review_gate", # Backward compat: in-flight tickets resume at gate "ai_review": "human_review_gate", "human_review_gate": "human_review_gate", "implement_review": "implement_review", From 0ab1a02090ef5350c106ce3ee7c14024787a3724 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:48:02 -0400 Subject: [PATCH 10/19] feat: extend worker CI webhook routing to human_review_gate with pending_ci_event Update the worker to accept CI webhooks (check_run/check_suite) when current_node is human_review_gate, setting pending_ci_event=True so route_human_review can route to ci_evaluator. Remove wait_for_ci_gate from _CI_STAGES and _signal_required_nodes, add human_review_gate to _CI_STAGES so /forge skip-gate commands work at the new pause point. Co-Authored-By: Claude Opus 4.6 --- src/forge/orchestrator/worker.py | 16 ++++-- tests/unit/orchestrator/test_worker.py | 69 ++++++++++++++++++++++++ tests/unit/workflow/test_ci_gate_skip.py | 2 +- 3 files changed, 81 insertions(+), 6 deletions(-) diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index dc1a0da9..00fc4f63 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -475,7 +475,7 @@ async def _handle_resume_event( # on the earlier actions would see a partial set of check runs and could # prematurely declare success. Other event types (push, pull_request) always wake up. if ( - current_node in ("wait_for_ci_gate", "ci_evaluator") + current_node in ("ci_evaluator", "human_review_gate") and message.source == EventSource.GITHUB ): event = message.event_type @@ -498,7 +498,7 @@ async def _handle_resume_event( # GitHub issue_comment events: detect /forge skip-gate and /forge unskip-gate # commands posted as PR comments. - _CI_STAGES = ("wait_for_ci_gate", "ci_evaluator", "attempt_ci_fix") + _CI_STAGES = ("ci_evaluator", "attempt_ci_fix", "human_review_gate") if message.source == EventSource.GITHUB and "issue_comment" in message.event_type: gh_comment_body = payload.get("comment", {}).get("body", "").strip() repo_full = payload.get("repository", {}).get("full_name", "") @@ -1203,8 +1203,15 @@ async def _handle_resume_event( } # Keep current_node — workflow resumes from the node that failed elif is_ci_webhook: - # GitHub CI event — unpause the gate and let ci_evaluator check the results - updated_state["is_paused"] = False + # GitHub CI event — unpause and route to ci_evaluator + if current_node == "human_review_gate": + # Keep current_node as human_review_gate so review webhooks arriving + # during the CI cycle are still accepted from the queue. + updated_state["pending_ci_event"] = True + updated_state["is_paused"] = False + else: + # ci_evaluator: old behavior — just unpause + updated_state["is_paused"] = False elif is_yolo: updated_state["yolo_mode"] = True updated_state["is_paused"] = False @@ -1297,7 +1304,6 @@ async def _handle_resume_event( "ci_evaluator", "attempt_ci_fix", "human_review_gate", - "wait_for_ci_gate", ) if ( not current_state.get("is_paused", True) diff --git a/tests/unit/orchestrator/test_worker.py b/tests/unit/orchestrator/test_worker.py index 30a7fe62..f928f77d 100644 --- a/tests/unit/orchestrator/test_worker.py +++ b/tests/unit/orchestrator/test_worker.py @@ -1268,3 +1268,72 @@ async def test_process_workflow_extracts_labels_and_calls_resolve(self): labels=["forge:managed"], event=message.payload, ) + + +class TestCiWebhookAtHumanReviewGate: + """Worker routes CI webhooks to human_review_gate correctly.""" + + @pytest.fixture + def worker(self) -> OrchestratorWorker: + return OrchestratorWorker(consumer_name="test-worker") + + @pytest.mark.asyncio + async def test_ci_webhook_at_review_gate_sets_pending_ci_event(self, worker): + """CI check_suite.completed at human_review_gate sets pending_ci_event=True.""" + current_state = { + "ticket_key": "TEST-1", + "ticket_type": "Feature", + "current_node": "human_review_gate", + "is_paused": True, + "pending_ci_event": False, + "context": {}, + } + message = QueueMessage( + message_id="msg-1", + event_id="evt-1", + source=EventSource.GITHUB, + event_type="check_suite.completed", + ticket_key="TEST-1", + payload={ + "check_suite": { + "status": "completed", + "pull_requests": [{"number": 42}], + } + }, + ) + + result = await worker._handle_resume_event(message, current_state) + + assert result.get("pending_ci_event") is True + assert result.get("is_paused") is False + assert result.get("current_node") == "human_review_gate" # unchanged + + @pytest.mark.asyncio + async def test_ci_webhook_at_ci_evaluator_does_not_set_pending_ci_event(self, worker): + """CI webhook at ci_evaluator does NOT set pending_ci_event (old behavior preserved).""" + current_state = { + "ticket_key": "TEST-1", + "ticket_type": "Feature", + "current_node": "ci_evaluator", + "is_paused": False, + "pending_ci_event": False, + "context": {}, + } + message = QueueMessage( + message_id="msg-2", + event_id="evt-2", + source=EventSource.GITHUB, + event_type="check_suite.completed", + ticket_key="TEST-1", + payload={ + "check_suite": { + "status": "completed", + "pull_requests": [{"number": 42}], + } + }, + ) + + result = await worker._handle_resume_event(message, current_state) + + assert result.get("is_paused") is False + assert result.get("pending_ci_event", False) is False # not set for ci_evaluator diff --git a/tests/unit/workflow/test_ci_gate_skip.py b/tests/unit/workflow/test_ci_gate_skip.py index 89da27a2..dcae4c50 100644 --- a/tests/unit/workflow/test_ci_gate_skip.py +++ b/tests/unit/workflow/test_ci_gate_skip.py @@ -70,7 +70,7 @@ def base_message(): @pytest.fixture def ci_state(): return make_workflow_state( - current_node="wait_for_ci_gate", + current_node="human_review_gate", current_repo="org/repo", current_pr_number=42, ci_failed_checks=[ From dbb22fa928f3eaaa0a225c7e7c8ee16709aa15da Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:50:10 -0400 Subject: [PATCH 11/19] chore: fix stale comment referencing wait_for_ci_gate in worker Co-Authored-By: Claude Opus 4.6 --- src/forge/orchestrator/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index 00fc4f63..6118fa89 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -467,7 +467,7 @@ async def _handle_resume_event( current_node = current_state.get("current_node", "") - # GitHub check_run/check_suite events are the explicit signal for wait_for_ci_gate. + # GitHub check_run/check_suite events are the explicit signal for CI evaluation. # They don't carry Jira labels or comments, so handle them before the label loop. # For check_suite and check_run events (both real GitHub webhooks and poller # forwarded ones), only wake up CI evaluation when the suite is completed. From 69bd98a4606ec4641bcdf1a078d29e35f4e0c6fc Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 12:52:46 -0400 Subject: [PATCH 12/19] test: add smoke tests for concurrent CI/review gate Verify routing logic (route_human_review, _route_ci_evaluation) and graph compilation for all three workflow types after the DRY refactor. Co-Authored-By: Claude Opus 4.6 --- tests/unit/workflow/test_concurrent_gate.py | 106 ++++++++++++++++++++ 1 file changed, 106 insertions(+) create mode 100644 tests/unit/workflow/test_concurrent_gate.py diff --git a/tests/unit/workflow/test_concurrent_gate.py b/tests/unit/workflow/test_concurrent_gate.py new file mode 100644 index 00000000..d16125a5 --- /dev/null +++ b/tests/unit/workflow/test_concurrent_gate.py @@ -0,0 +1,106 @@ +"""Smoke tests for the concurrent CI/review gate design.""" + +import pytest +from langgraph.graph import END + + +class TestRoutingFunctions: + def test_no_ci_event_gate_stays_paused(self): + """Gate with no CI event and no review signal returns END (stays paused).""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + "ticket_key": "TEST-1", + "is_paused": True, + "pending_ci_event": False, + "revision_requested": False, + "pr_merged": False, + "feedback_comment": None, + } + assert route_human_review(state) == END + + def test_ci_event_routes_through_evaluator(self): + """CI webhook (pending_ci_event=True) routes to ci_evaluator.""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + "ticket_key": "TEST-1", + "is_paused": False, + "pending_ci_event": True, + "revision_requested": False, + "pr_merged": False, + "feedback_comment": None, + } + assert route_human_review(state) == "ci_evaluator" + + def test_review_event_routes_to_implement_review(self): + """Review rejection routes to implement_review.""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + "ticket_key": "TEST-1", + "is_paused": False, + "pending_ci_event": False, + "revision_requested": True, + "feedback_comment": "please refactor", + "pr_merged": False, + } + assert route_human_review(state) == "implement_review" + + def test_merge_routes_to_complete_tasks(self): + """PR merge routes to complete_tasks.""" + from forge.workflow.nodes.human_review import route_human_review + + state = { + "ticket_key": "TEST-1", + "is_paused": False, + "pending_ci_event": False, + "revision_requested": False, + "pr_merged": True, + "feedback_comment": None, + } + assert route_human_review(state) == "complete_tasks" + + def test_external_failure_stays_at_gate(self): + """external_failure ci_status routes back to human_review_gate (not escalate).""" + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "external_failure"}) == "human_review_gate" + + def test_passed_ci_routes_to_gate_not_end(self): + """Passed CI routes to human_review_gate (review can now merge).""" + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "passed"}) == "human_review_gate" + + def test_pending_ci_routes_to_gate_not_end(self): + """Pending CI routes to human_review_gate to re-pause and wait.""" + from forge.workflow.post_pr import _route_ci_evaluation + + assert _route_ci_evaluation({"ci_status": "pending"}) == "human_review_gate" + + +class TestGraphCompilation: + def test_feature_graph_compiles(self): + """Feature graph builds without error after DRY refactor.""" + from forge.workflow.feature.graph import build_feature_graph + + graph = build_feature_graph() + compiled = graph.compile() + assert compiled is not None + + def test_bug_graph_compiles(self): + """Bug graph builds without error after DRY refactor.""" + from forge.workflow.bug.graph import build_bug_graph + + graph = build_bug_graph() + compiled = graph.compile() + assert compiled is not None + + def test_task_takeover_graph_compiles(self): + """Task takeover graph builds without error after DRY refactor.""" + from forge.workflow.task_takeover.graph import build_task_takeover_graph + + graph = build_task_takeover_graph() + compiled = graph.compile() + assert compiled is not None From a6e87f78a298f735b13b0477b88d4c09ef91dd8d Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 13:06:32 -0400 Subject: [PATCH 13/19] fix: remove stale comment, vestigial alias, and GitHubClient resource leak Co-Authored-By: Claude Opus 4.6 --- src/forge/workflow/nodes/ci_evaluator.py | 10 ++++++---- src/forge/workflow/nodes/implement_review.py | 2 -- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py index 7cf077bf..03832184 100644 --- a/src/forge/workflow/nodes/ci_evaluator.py +++ b/src/forge/workflow/nodes/ci_evaluator.py @@ -270,8 +270,6 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: settings = get_settings() fork_owner = state.get("fork_owner", "") fork_repo = state.get("fork_repo", "") - attempt = ci_fix_attempt - try: workspace_path, _ = prepare_workspace(state) state = {**state, "workspace_path": workspace_path} @@ -291,7 +289,11 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: logs_dir = Path(workspace_path) / ".forge" / "logs" failures_file.parent.mkdir(parents=True, exist_ok=True) - await _fetch_ci_logs_and_artifacts(failed_checks, logs_dir, GitHubClient()) + github = GitHubClient() + try: + await _fetch_ci_logs_and_artifacts(failed_checks, logs_dir, github) + finally: + await github.close() failures_file.write_text(_collect_error_info(failed_checks)) # ── Phase 0: Attribution ───────────────────────────────────────────── @@ -303,7 +305,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: runner = ContainerRunner(settings) await runner.run( workspace_path=Path(workspace_path), - task_summary=f"Attribute CI failure (attempt {attempt})", + task_summary=f"Attribute CI failure (attempt {ci_fix_attempt})", task_description=attribution_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-attribution", diff --git a/src/forge/workflow/nodes/implement_review.py b/src/forge/workflow/nodes/implement_review.py index f7592e44..d3b1b391 100644 --- a/src/forge/workflow/nodes/implement_review.py +++ b/src/forge/workflow/nodes/implement_review.py @@ -268,8 +268,6 @@ async def implement_review(state: WorkflowState) -> WorkflowState: else: logger.info(f"No new commits after review implementation for {ticket_key}") - # Only re-enter the CI gate if we actually pushed new commits; otherwise - # CI won't re-trigger and human_review_gate would block forever. next_node = "human_review_gate" return update_state_timestamp( From 0e1afecc9274aa313783f83fc142f30fdd941819 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 15:29:37 -0400 Subject: [PATCH 14/19] chore: clean up if statement --- src/forge/orchestrator/worker.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index 6118fa89..3cfbcb4a 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -1203,15 +1203,14 @@ async def _handle_resume_event( } # Keep current_node — workflow resumes from the node that failed elif is_ci_webhook: - # GitHub CI event — unpause and route to ci_evaluator + # GitHub CI event — unpause the gate and let ci_evaluator check the results + updated_state["is_paused"] = False + if current_node == "human_review_gate": # Keep current_node as human_review_gate so review webhooks arriving # during the CI cycle are still accepted from the queue. updated_state["pending_ci_event"] = True - updated_state["is_paused"] = False - else: - # ci_evaluator: old behavior — just unpause - updated_state["is_paused"] = False + elif is_yolo: updated_state["yolo_mode"] = True updated_state["is_paused"] = False From 802faa4e3e9f5fa35c7f75f4f581ac160c45e452 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 15:40:18 -0400 Subject: [PATCH 15/19] chore: refactor _route_after_pr_creation to route_after_pr_creation - _route_after_pr_creation is imported in other modules, so semantically, it should not be prepended with a _ --- src/forge/workflow/bug/graph.py | 4 ++-- src/forge/workflow/feature/graph.py | 4 ++-- src/forge/workflow/post_pr.py | 2 +- src/forge/workflow/task_takeover/graph.py | 4 ++-- tests/flows/bug_workflow/test_complete_bug_flow.py | 12 ++++++------ tests/unit/workflow/test_post_pr.py | 12 ++++++------ 6 files changed, 19 insertions(+), 19 deletions(-) diff --git a/src/forge/workflow/bug/graph.py b/src/forge/workflow/bug/graph.py index fd616209..d0fe691a 100644 --- a/src/forge/workflow/bug/graph.py +++ b/src/forge/workflow/bug/graph.py @@ -33,7 +33,7 @@ from forge.workflow.nodes.triage import route_triage_gate, triage_check, triage_gate from forge.workflow.nodes.workspace_setup import setup_workspace from forge.workflow.post_pr import ( - _route_after_pr_creation, + route_after_pr_creation, add_post_pr_edges, add_post_pr_nodes, ) @@ -577,7 +577,7 @@ def build_bug_graph() -> StateGraph: graph.add_edge("update_documentation", "create_pr") graph.add_conditional_edges( "create_pr", - _route_after_pr_creation, + route_after_pr_creation, { "teardown_workspace": "teardown_workspace", "escalate_blocked": "escalate_blocked", diff --git a/src/forge/workflow/feature/graph.py b/src/forge/workflow/feature/graph.py index f63ae307..12600189 100644 --- a/src/forge/workflow/feature/graph.py +++ b/src/forge/workflow/feature/graph.py @@ -47,7 +47,7 @@ update_single_task, ) from forge.workflow.post_pr import ( - _route_after_pr_creation, + route_after_pr_creation, add_post_pr_edges, add_post_pr_nodes, ) @@ -649,7 +649,7 @@ def build_feature_graph() -> StateGraph: graph.add_edge("update_documentation", "create_pr") graph.add_conditional_edges( "create_pr", - _route_after_pr_creation, + route_after_pr_creation, { "teardown_workspace": "teardown_workspace", "escalate_blocked": "escalate_blocked", diff --git a/src/forge/workflow/post_pr.py b/src/forge/workflow/post_pr.py index fbb5e8a0..59342e0d 100644 --- a/src/forge/workflow/post_pr.py +++ b/src/forge/workflow/post_pr.py @@ -23,7 +23,7 @@ def _route_after_teardown(state: dict[str, Any]) -> str: return "setup_workspace" if remaining else "human_review_gate" -def _route_after_pr_creation(state: dict[str, Any]) -> str: +def route_after_pr_creation(state: dict[str, Any]) -> str: """Route after PR creation: teardown on success, escalate on failure.""" last_error = state.get("last_error") pr_urls = state.get("pr_urls", []) diff --git a/src/forge/workflow/task_takeover/graph.py b/src/forge/workflow/task_takeover/graph.py index c0a09636..be7bee45 100644 --- a/src/forge/workflow/task_takeover/graph.py +++ b/src/forge/workflow/task_takeover/graph.py @@ -28,7 +28,7 @@ triage_task, ) from forge.workflow.post_pr import ( - _route_after_pr_creation, + route_after_pr_creation, add_post_pr_edges, add_post_pr_nodes, ) @@ -345,7 +345,7 @@ def build_task_takeover_graph() -> StateGraph[TaskTakeoverState, Any, Any]: ) graph.add_conditional_edges( "create_pr", - _route_after_pr_creation, + route_after_pr_creation, { "teardown_workspace": "teardown_workspace", "escalate_blocked": "escalate_blocked", diff --git a/tests/flows/bug_workflow/test_complete_bug_flow.py b/tests/flows/bug_workflow/test_complete_bug_flow.py index 41c999c2..b5a2baba 100644 --- a/tests/flows/bug_workflow/test_complete_bug_flow.py +++ b/tests/flows/bug_workflow/test_complete_bug_flow.py @@ -11,7 +11,7 @@ _route_after_answer_bug, _route_after_implementation, _route_after_local_review, - _route_after_pr_creation, + route_after_pr_creation, _route_after_reflect_rca, _route_after_teardown, _route_after_triage_check, @@ -721,35 +721,35 @@ def test_no_verdict_mechanical_below_cap_falls_back_to_current_node(self): class TestRouteAfterPrCreation: - """_route_after_pr_creation routes based on last_error and pr_urls.""" + """route_after_pr_creation routes based on last_error and pr_urls.""" def test_success_routes_to_teardown(self): state = make_workflow_state( ticket_key="BUG-PR1", ticket_type=TicketType.BUG, current_node="create_pr", last_error=None, pr_urls=["https://github.com/org/repo/pull/1"], ) - assert _route_after_pr_creation(state) == "teardown_workspace" + assert route_after_pr_creation(state) == "teardown_workspace" def test_error_with_no_pr_urls_escalates(self): state = make_workflow_state( ticket_key="BUG-PR2", ticket_type=TicketType.BUG, current_node="create_pr", last_error="PR creation failed", pr_urls=[], ) - assert _route_after_pr_creation(state) == "escalate_blocked" + assert route_after_pr_creation(state) == "escalate_blocked" def test_error_with_existing_pr_urls_routes_to_teardown(self): state = make_workflow_state( ticket_key="BUG-PR3", ticket_type=TicketType.BUG, current_node="create_pr", last_error="partial failure", pr_urls=["https://github.com/org/repo/pull/1"], ) - assert _route_after_pr_creation(state) == "teardown_workspace" + assert route_after_pr_creation(state) == "teardown_workspace" def test_no_error_no_pr_urls_routes_to_teardown(self): state = make_workflow_state( ticket_key="BUG-PR4", ticket_type=TicketType.BUG, current_node="create_pr", last_error=None, pr_urls=[], ) - assert _route_after_pr_creation(state) == "teardown_workspace" + assert route_after_pr_creation(state) == "teardown_workspace" class TestRouteAfterTeardown: diff --git a/tests/unit/workflow/test_post_pr.py b/tests/unit/workflow/test_post_pr.py index e9d01246..6e090f4c 100644 --- a/tests/unit/workflow/test_post_pr.py +++ b/tests/unit/workflow/test_post_pr.py @@ -65,20 +65,20 @@ def test_unknown_status_routes_to_escalate_blocked(self): class TestRouteAfterPrCreation: def test_success_routes_to_teardown(self): - from forge.workflow.post_pr import _route_after_pr_creation + from forge.workflow.post_pr import route_after_pr_creation state = {"pr_urls": ["https://github.com/org/repo/pull/1"], "last_error": None} - assert _route_after_pr_creation(state) == "teardown_workspace" + assert route_after_pr_creation(state) == "teardown_workspace" def test_failure_with_no_prs_routes_to_escalate(self): - from forge.workflow.post_pr import _route_after_pr_creation + from forge.workflow.post_pr import route_after_pr_creation state = {"pr_urls": [], "last_error": "PR creation failed"} - assert _route_after_pr_creation(state) == "escalate_blocked" + assert route_after_pr_creation(state) == "escalate_blocked" def test_partial_success_routes_to_teardown(self): """If some PRs were created despite an error, proceed to teardown.""" - from forge.workflow.post_pr import _route_after_pr_creation + from forge.workflow.post_pr import route_after_pr_creation state = {"pr_urls": ["https://github.com/org/repo/pull/1"], "last_error": "partial"} - assert _route_after_pr_creation(state) == "teardown_workspace" + assert route_after_pr_creation(state) == "teardown_workspace" From 1b3ac781c16c0dfbefb4fc27fe395def89283367 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 15:47:47 -0400 Subject: [PATCH 16/19] chore: remove unnecessary comment --- src/forge/workflow/bug/graph.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/forge/workflow/bug/graph.py b/src/forge/workflow/bug/graph.py index d0fe691a..32d76e58 100644 --- a/src/forge/workflow/bug/graph.py +++ b/src/forge/workflow/bug/graph.py @@ -591,7 +591,6 @@ def build_bug_graph() -> StateGraph: ) # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── - # Note: rebase_pr node is added by add_post_pr_nodes graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), From 20ba10eda2cbcdba0473640b5cfc75c0c5d74ec1 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 15:59:18 -0400 Subject: [PATCH 17/19] fix: remove scope creep --- src/forge/workflow/feature/graph.py | 12 ------------ 1 file changed, 12 deletions(-) diff --git a/src/forge/workflow/feature/graph.py b/src/forge/workflow/feature/graph.py index 12600189..cb35cb93 100644 --- a/src/forge/workflow/feature/graph.py +++ b/src/forge/workflow/feature/graph.py @@ -676,26 +676,14 @@ def build_feature_graph() -> StateGraph: ) # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── - # Note: rebase_pr node is added by add_post_pr_nodes graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), { - "generate_prd": "generate_prd", "prd_approval_gate": "prd_approval_gate", - "regenerate_prd": "regenerate_prd", - "generate_spec": "generate_spec", "spec_approval_gate": "spec_approval_gate", - "regenerate_spec": "regenerate_spec", - "decompose_epics": "decompose_epics", "plan_approval_gate": "plan_approval_gate", - "regenerate_all_epics": "regenerate_all_epics", - "update_single_epic": "update_single_epic", - "generate_tasks": "generate_tasks", "task_approval_gate": "task_approval_gate", - "regenerate_all_tasks": "regenerate_all_tasks", - "update_single_task": "update_single_task", - "regenerate_epic_tasks": "regenerate_epic_tasks", "task_router": "task_router", "setup_workspace": "setup_workspace", "implement_task": "implement_task", From 24bd3584fe01fbd066cf8b4295c61386e1d0c813 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Mon, 13 Jul 2026 16:11:36 -0400 Subject: [PATCH 18/19] fix: break backwards compat --- src/forge/workflow/utils/__init__.py | 1 - tests/unit/orchestrator/test_worker.py | 11 ----------- tests/unit/workflow/bug/test_graph.py | 1 - tests/unit/workflow/test_base.py | 8 ++++---- tests/unit/workflow/test_cleanup.py | 1 - 5 files changed, 4 insertions(+), 18 deletions(-) diff --git a/src/forge/workflow/utils/__init__.py b/src/forge/workflow/utils/__init__.py index 03a751b5..be51dea2 100644 --- a/src/forge/workflow/utils/__init__.py +++ b/src/forge/workflow/utils/__init__.py @@ -25,7 +25,6 @@ "human_review_gate": "human_review_gate", "implement_review": "implement_review", "review_response_gate": "review_response_gate", - "wait_for_ci_gate": "human_review_gate", # backward compat for in-flight tickets "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "rebase_pr": "rebase_pr", diff --git a/tests/unit/orchestrator/test_worker.py b/tests/unit/orchestrator/test_worker.py index f928f77d..68569e5f 100644 --- a/tests/unit/orchestrator/test_worker.py +++ b/tests/unit/orchestrator/test_worker.py @@ -934,17 +934,6 @@ async def test_check_suite_recognized_at_ci_evaluator(self, worker): ) assert result["is_paused"] is False - @pytest.mark.asyncio - async def test_check_suite_at_wait_for_ci_gate_still_works(self, worker): - """Existing wait_for_ci_gate behaviour must be preserved.""" - state = self._ci_state("wait_for_ci_gate") - message = self._check_suite_message("success") - - result = await worker._handle_resume_event(message, state) - - assert result is not state - assert result["is_paused"] is False - @pytest.mark.asyncio async def test_incomplete_check_suite_does_not_unpause_at_ci_evaluator(self, worker): """A check_suite with status=in_progress must not wake up the workflow.""" diff --git a/tests/unit/workflow/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py index dae61da5..540861a2 100644 --- a/tests/unit/workflow/bug/test_graph.py +++ b/tests/unit/workflow/bug/test_graph.py @@ -103,7 +103,6 @@ class TestRouteEntry: ("teardown_workspace", "teardown_workspace"), ("ci_evaluator", "ci_evaluator"), ("attempt_ci_fix", "ci_evaluator"), - ("wait_for_ci_gate", "human_review_gate"), # Backward compat: in-flight tickets resume at gate ("ai_review", "human_review_gate"), ("human_review_gate", "human_review_gate"), ("implement_review", "implement_review"), diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py index 6529f1f1..da2b286a 100644 --- a/tests/unit/workflow/test_base.py +++ b/tests/unit/workflow/test_base.py @@ -173,11 +173,11 @@ def build_graph(self): class TestSharedResumeMap: - """Tests for resolve_shared_resume_node backward compat.""" + """Tests for resolve_shared_resume_node.""" - def test_wait_for_ci_gate_resolves_to_human_review_gate(self): - """In-flight tickets with wait_for_ci_gate resume at human_review_gate.""" + def test_wait_for_ci_gate_is_not_mapped(self): + """wait_for_ci_gate was removed — it must not resolve to any node.""" from forge.workflow.utils import resolve_shared_resume_node result = resolve_shared_resume_node("wait_for_ci_gate") - assert result == "human_review_gate" + assert result is None diff --git a/tests/unit/workflow/test_cleanup.py b/tests/unit/workflow/test_cleanup.py index 63019db1..6d51a0fd 100644 --- a/tests/unit/workflow/test_cleanup.py +++ b/tests/unit/workflow/test_cleanup.py @@ -101,7 +101,6 @@ def test_existing_nodes_still_mapped(self): "teardown_workspace": "teardown_workspace", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", - "wait_for_ci_gate": "human_review_gate", # Backward compat: in-flight tickets resume at gate "ai_review": "human_review_gate", "human_review_gate": "human_review_gate", "implement_review": "implement_review", From f1637d3c8290026ba7906fed1ca853cfdc550306 Mon Sep 17 00:00:00 2001 From: Dan Childers Date: Fri, 17 Jul 2026 10:03:38 -0400 Subject: [PATCH 19/19] fix: address review findings across CI routing and tests - Set ci_status="passed" when attempt_ci_fix has no failed checks to prevent human_review_gate from hanging without a wake signal - Route "no_prs" to escalate_blocked instead of human_review_gate to avoid indefinite pause when no PRs exist - Remove unreachable ci_fix_attempt limit guard in attempt_ci_fix (already enforced by evaluate_ci_status) - Remove stale imports and test classes for functions moved to post_pr - Delete duplicate @pytest.mark.parametrize block causing cartesian product of ~800 nonsensical test cases - Remove trailing whitespace in worker.py Co-Authored-By: Claude Opus 4.6 --- src/forge/orchestrator/worker.py | 2 +- src/forge/workflow/nodes/ci_evaluator.py | 13 +--- src/forge/workflow/post_pr.py | 4 +- .../bug_workflow/test_complete_bug_flow.py | 74 ------------------- tests/unit/workflow/bug/test_graph.py | 39 ---------- 5 files changed, 4 insertions(+), 128 deletions(-) diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index 3cfbcb4a..fa87341a 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -1210,7 +1210,7 @@ async def _handle_resume_event( # Keep current_node as human_review_gate so review webhooks arriving # during the CI cycle are still accepted from the queue. updated_state["pending_ci_event"] = True - + elif is_yolo: updated_state["yolo_mode"] = True updated_state["is_paused"] = False diff --git a/src/forge/workflow/nodes/ci_evaluator.py b/src/forge/workflow/nodes/ci_evaluator.py index 03832184..e2c2d1a7 100644 --- a/src/forge/workflow/nodes/ci_evaluator.py +++ b/src/forge/workflow/nodes/ci_evaluator.py @@ -252,6 +252,7 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: return update_state_timestamp( { **state, + "ci_status": "passed", "current_node": "human_review_gate", "pending_ci_event": False, } @@ -356,18 +357,6 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: f"Phase 1: Analyzing CI failures for {ticket_key} (fix attempt {ci_fix_attempt}/{ci_fix_max})" ) - if ci_fix_attempt >= ci_fix_max: - logger.warning(f"CI fix attempt limit ({ci_fix_max}) reached for {ticket_key}") - return update_state_timestamp( - { - **state, - "ci_status": "failed", - "current_node": "ci_evaluator", - "pending_ci_event": False, - "last_error": "CI fix attempt limit reached", - } - ) - jira = JiraClient() try: await post_status_comment( diff --git a/src/forge/workflow/post_pr.py b/src/forge/workflow/post_pr.py index 59342e0d..9b3dc096 100644 --- a/src/forge/workflow/post_pr.py +++ b/src/forge/workflow/post_pr.py @@ -44,10 +44,10 @@ def _route_ci_evaluation(state: dict[str, Any]) -> str: ci_status = state.get("ci_status", "") if ci_status == "fixing": return "attempt_ci_fix" - if ci_status in ("failed", "blocked"): + if ci_status in ("failed", "blocked", "no_prs"): return "escalate_blocked" # All known non-fix/non-fail statuses go back to gate - if ci_status in ("pending", "passed", "external_failure", "no_prs"): + if ci_status in ("pending", "passed", "external_failure"): return "human_review_gate" # Unknown status — safety escalation logger.warning(f"Unexpected ci_status: {ci_status!r}, routing to escalate_blocked") diff --git a/tests/flows/bug_workflow/test_complete_bug_flow.py b/tests/flows/bug_workflow/test_complete_bug_flow.py index b5a2baba..0b99199d 100644 --- a/tests/flows/bug_workflow/test_complete_bug_flow.py +++ b/tests/flows/bug_workflow/test_complete_bug_flow.py @@ -13,10 +13,8 @@ _route_after_local_review, route_after_pr_creation, _route_after_reflect_rca, - _route_after_teardown, _route_after_triage_check, _route_after_workspace_setup, - _route_ci_evaluation, _route_human_review_bug, route_entry, ) @@ -144,7 +142,6 @@ class TestBugWorkflowResumeRouting: ("teardown_workspace", "teardown_workspace"), ("ci_evaluator", "ci_evaluator"), ("attempt_ci_fix", "ci_evaluator"), - ("wait_for_ci_gate", "wait_for_ci_gate"), ("local_review", "local_review"), ("ai_review", "human_review_gate"), ("human_review_gate", "human_review_gate"), @@ -752,77 +749,6 @@ def test_no_error_no_pr_urls_routes_to_teardown(self): assert route_after_pr_creation(state) == "teardown_workspace" -class TestRouteAfterTeardown: - """_route_after_teardown loops back for remaining repos or proceeds to CI.""" - - def test_remaining_repos_loops_to_setup_workspace(self): - state = make_workflow_state( - ticket_key="BUG-TD1", ticket_type=TicketType.BUG, current_node="teardown_workspace", - repos_to_process=["org/a", "org/b"], repos_completed=["org/a"], - ) - assert _route_after_teardown(state) == "setup_workspace" - - def test_all_repos_done_routes_to_wait_for_ci_gate(self): - state = make_workflow_state( - ticket_key="BUG-TD2", ticket_type=TicketType.BUG, current_node="teardown_workspace", - repos_to_process=["org/a"], repos_completed=["org/a"], - ) - assert _route_after_teardown(state) == "wait_for_ci_gate" - - def test_empty_repos_routes_to_wait_for_ci_gate(self): - state = make_workflow_state( - ticket_key="BUG-TD3", ticket_type=TicketType.BUG, current_node="teardown_workspace", - repos_to_process=[], repos_completed=[], - ) - assert _route_after_teardown(state) == "wait_for_ci_gate" - - def test_multiple_remaining_repos_loops(self): - state = make_workflow_state( - ticket_key="BUG-TD4", ticket_type=TicketType.BUG, current_node="teardown_workspace", - repos_to_process=["org/a", "org/b", "org/c"], repos_completed=[], - ) - assert _route_after_teardown(state) == "setup_workspace" - - -class TestRouteCiEvaluation: - """_route_ci_evaluation routes based on ci_status.""" - - def test_passed_routes_to_human_review_gate(self): - state = make_workflow_state( - ticket_key="BUG-CI1", ticket_type=TicketType.BUG, current_node="ci_evaluator", - ci_status="passed", - ) - assert _route_ci_evaluation(state) == "human_review_gate" - - def test_fixing_routes_to_attempt_ci_fix(self): - state = make_workflow_state( - ticket_key="BUG-CI2", ticket_type=TicketType.BUG, current_node="ci_evaluator", - ci_status="fixing", - ) - assert _route_ci_evaluation(state) == "attempt_ci_fix" - - def test_pending_routes_to_end(self): - state = make_workflow_state( - ticket_key="BUG-CI3", ticket_type=TicketType.BUG, current_node="ci_evaluator", - ci_status="pending", - ) - assert _route_ci_evaluation(state) == END - - def test_failed_routes_to_escalate_blocked(self): - state = make_workflow_state( - ticket_key="BUG-CI4", ticket_type=TicketType.BUG, current_node="ci_evaluator", - ci_status="failed", - ) - assert _route_ci_evaluation(state) == "escalate_blocked" - - def test_empty_status_routes_to_escalate_blocked(self): - state = make_workflow_state( - ticket_key="BUG-CI5", ticket_type=TicketType.BUG, current_node="ci_evaluator", - ci_status="", - ) - assert _route_ci_evaluation(state) == "escalate_blocked" - - class TestRouteHumanReviewBug: """_route_human_review_bug intercepts pr_merged for post_merge_summary.""" diff --git a/tests/unit/workflow/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py index 540861a2..0fcc87b2 100644 --- a/tests/unit/workflow/bug/test_graph.py +++ b/tests/unit/workflow/bug/test_graph.py @@ -39,45 +39,6 @@ def _bug_state(**overrides): class TestRouteEntry: """route_entry maps current_node values to correct resume targets.""" - @pytest.mark.parametrize( - "node,expected", - [ - # New nodes - ("triage_check", "triage_check"), - ("triage_gate", "triage_gate"), - ("analyze_bug", "analyze_bug"), - ("reflect_rca", "reflect_rca"), - ("rca_option_gate", "rca_option_gate"), - ("plan_bug_fix", "plan_bug_fix"), - ("plan_approval_gate", "plan_approval_gate"), - ("regenerate_plan", "regenerate_plan"), - ("decompose_plan", "decompose_plan"), - ("post_merge_summary", "post_merge_summary"), - # Backward compat: old rca_approval_gate value maps to rca_option_gate - ("rca_approval_gate", "rca_option_gate"), - # regenerate_rca performs cleanup before routing through analyze_bug - ("regenerate_rca", "regenerate_rca"), - # Preserved existing nodes - ("setup_workspace", "setup_workspace"), - ("implement_bug_fix", "implement_bug_fix"), - ("local_review", "local_review"), - ("update_documentation", "update_documentation"), - ("create_pr", "create_pr"), - ("teardown_workspace", "teardown_workspace"), - ("ci_evaluator", "ci_evaluator"), - ("attempt_ci_fix", "ci_evaluator"), - ("wait_for_ci_gate", "wait_for_ci_gate"), - ("ai_review", "human_review_gate"), - ("human_review_gate", "human_review_gate"), - ("implement_review", "implement_review"), - ("review_response_gate", "review_response_gate"), - ("escalate_blocked", "escalate_blocked"), - ("complete", END), - ("complete_tasks", END), - ("aggregate_epic_status", END), - ("aggregate_feature_status", END), - ], - ) @pytest.mark.parametrize("node,expected", [ # New nodes ("triage_check", "triage_check"),