diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index dc1a0da9..fa87341a 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. @@ -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", "") @@ -1205,6 +1205,12 @@ async def _handle_resume_event( elif is_ci_webhook: # 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 + elif is_yolo: updated_state["yolo_mode"] = True updated_state["is_paused"] = False @@ -1297,7 +1303,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/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/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/bug/graph.py b/src/forge/workflow/bug/graph.py index f9ccee44..32d76e58 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", @@ -639,86 +577,20 @@ 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", }, ) - 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) graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), @@ -732,7 +604,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..cb35cb93 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. @@ -422,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: @@ -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", @@ -715,82 +649,16 @@ 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", }, ) - graph.add_conditional_edges( - "teardown_workspace", - _route_after_teardown, - { - "setup_workspace": "setup_workspace", - "wait_for_ci_gate": "wait_for_ci_gate", - }, - ) - - # 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) + # Post-PR edges (CI/review) - shared across all workflows + add_post_pr_edges(graph, on_complete_node="complete_tasks") - # 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,7 +676,6 @@ def build_feature_graph() -> StateGraph: ) # ── Rebase (merge conflict resolution, triggered by /forge rebase) ── - graph.add_node("rebase_pr", rebase_pr) graph.add_conditional_edges( "rebase_pr", lambda s: s.get("current_node", END), @@ -824,7 +691,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/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 84eb5812..e2c2d1a7 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 @@ -57,6 +56,7 @@ async def evaluate_ci_status(state: WorkflowState) -> WorkflowState: **state, "ci_status": "no_prs", "current_node": "complete", + "pending_ci_event": False, } ) @@ -110,6 +110,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 +164,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 +180,7 @@ def _is_skipped(check: dict) -> bool: **state, "ci_status": "pending", "current_node": "ci_evaluator", + "pending_ci_event": False, } ) @@ -190,6 +193,7 @@ def _is_skipped(check: dict) -> bool: **state, "ci_status": "pending", "current_node": "ci_evaluator", + "pending_ci_event": False, } ) @@ -204,6 +208,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 +221,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 +232,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() @@ -234,39 +241,29 @@ 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", + "ci_status": "passed", + "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() @@ -274,50 +271,111 @@ 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} - 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()) - + 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 ───────────────────────────────────────────── + 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 {ci_fix_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 ──────────────────────────────────────────────── + logger.info( + f"Phase 1: Analyzing CI failures for {ticket_key} (fix attempt {ci_fix_attempt}/{ci_fix_max})" + ) + + jira = JiraClient() + try: + await post_status_comment( + jira, + ticket_key, + f"🔧 Attempting CI fix ({ci_fix_attempt}/{ci_fix_max}).", + ) + finally: + await jira.close() + analysis_prompt = load_prompt( "analyze-ci", failures_file_path=str(failures_file), - attempt=attempt, + attempt=ci_fix_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 {ci_fix_attempt})", task_description=analysis_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-analyze", @@ -329,7 +387,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, } ) @@ -338,14 +397,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 {ci_fix_attempt})", task_description=fix_prompt, ticket_key=ticket_key, task_key=f"{ticket_key}-ci-fix", @@ -359,15 +416,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 {ci_fix_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}" @@ -379,9 +435,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 {ci_fix_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, @@ -389,19 +446,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-{ci_fix_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 {ci_fix_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( @@ -410,13 +464,14 @@ async def attempt_ci_fix(state: WorkflowState) -> WorkflowState: owner=_owner, repo=_repo_name, pr_number=state.get("current_pr_number"), - attempt=attempt, + attempt=ci_fix_attempt, ) return update_state_timestamp( { **state, - "current_node": "wait_for_ci_gate", + "current_node": "human_review_gate", + "pending_ci_event": False, "last_error": None, } ) @@ -426,70 +481,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/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/src/forge/workflow/nodes/implement_review.py b/src/forge/workflow/nodes/implement_review.py index d37a9eed..d3b1b391 100644 --- a/src/forge/workflow/nodes/implement_review.py +++ b/src/forge/workflow/nodes/implement_review.py @@ -268,9 +268,7 @@ 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 wait_for_ci_gate would block forever. - next_node = "wait_for_ci_gate" if unpushed else "human_review_gate" + 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/src/forge/workflow/post_pr.py b/src/forge/workflow/post_pr.py new file mode 100644 index 00000000..9b3dc096 --- /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", "no_prs"): + return "escalate_blocked" + # All known non-fix/non-fail statuses go back to gate + 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") + 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/src/forge/workflow/task_takeover/graph.py b/src/forge/workflow/task_takeover/graph.py index b2cacfcb..be7bee45 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, }, @@ -382,75 +345,45 @@ 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", }, ) - 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 80a10fb1..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": "wait_for_ci_gate", "ci_evaluator": "ci_evaluator", "attempt_ci_fix": "ci_evaluator", "rebase_pr": "rebase_pr", diff --git a/tests/flows/bug_workflow/test_complete_bug_flow.py b/tests/flows/bug_workflow/test_complete_bug_flow.py index 41c999c2..0b99199d 100644 --- a/tests/flows/bug_workflow/test_complete_bug_flow.py +++ b/tests/flows/bug_workflow/test_complete_bug_flow.py @@ -11,12 +11,10 @@ _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, _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"), @@ -721,106 +718,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" - - -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" + assert route_after_pr_creation(state) == "teardown_workspace" class TestRouteHumanReviewBug: diff --git a/tests/unit/orchestrator/test_worker.py b/tests/unit/orchestrator/test_worker.py index 30a7fe62..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.""" @@ -1268,3 +1257,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/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py index 5cad4a8d..0fcc87b2 100644 --- a/tests/unit/workflow/bug/test_graph.py +++ b/tests/unit/workflow/bug/test_graph.py @@ -39,45 +39,41 @@ 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"), + ("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"), + ("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 +337,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 +365,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/nodes/test_ci_attribution.py b/tests/unit/workflow/nodes/test_ci_attribution.py new file mode 100644 index 00000000..948e5369 --- /dev/null +++ b/tests/unit/workflow/nodes/test_ci_attribution.py @@ -0,0 +1,199 @@ +"""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": [], +} + +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") +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 + + +@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" + ) 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 diff --git a/tests/unit/workflow/test_base.py b/tests/unit/workflow/test_base.py index 4df75da1..da2b286a 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.""" + + 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 is None 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=[ diff --git a/tests/unit/workflow/test_cleanup.py b/tests/unit/workflow/test_cleanup.py index a63cceff..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": "wait_for_ci_gate", "ai_review": "human_review_gate", "human_review_gate": "human_review_gate", "implement_review": "implement_review", 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 diff --git a/tests/unit/workflow/test_post_pr.py b/tests/unit/workflow/test_post_pr.py new file mode 100644 index 00000000..6e090f4c --- /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" 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()