Fix multi-repository PR lifecycle tracking

src/forge/orchestrator/worker.py

@@ -29,6 +29,13 @@
 from forge.skills.utils import extract_project_key
 from forge.utils.redaction import redact_secrets
 from forge.workflow.nodes.error_handler import notify_error
+from forge.workflow.pr_state import (
+    activate_pull_request_for_event,
+    all_pull_requests_merged,
+    event_targets_pull_request,
+    mark_active_pull_request_merged,
+    save_active_pull_request,
+)
 from forge.workflow.registry import create_default_router
 from forge.workflow.router import WorkflowRouter
 from forge.workflow.utils.automated_review_triage import (
@@ -439,6 +446,15 @@ async def _process_workflow(self, message: QueueMessage) -> None:
                 # Run the workflow from the beginning
                 result = await compiled_workflow.ainvoke(state, config=config)
 
+            # Nodes continue to use scalar PR fields as a compatibility view.
+            # Persist that view back into the selected per-repository record
+            # after every invocation so subsequent webhooks restore fresh CI,
+            # review, and merge state for the PR they target.
+            persisted_result = save_active_pull_request(result)
+            if persisted_result != result:
+                await compiled_workflow.aupdate_state(config, persisted_result)
+                result = persisted_result
+
             final_node = result.get("current_node", "unknown")
             is_paused = result.get("is_paused", False)
             logger.info(
@@ -480,6 +496,8 @@ async def _handle_resume_event(
             Updated state for workflow resumption.
         """
         payload = message.payload
+        current_state = activate_pull_request_for_event(current_state, payload)
+        targets_implementation_pr = event_targets_pull_request(current_state, payload)
         changelog = payload.get("changelog", {})
         comment = payload.get("comment", {})
 
@@ -499,6 +517,7 @@ async def _handle_resume_event(
         automated_review_revision_pending = None
         proposal_review_threads: list[dict[str, Any]] = []
         proposal_review_decisions: list[dict[str, Any]] = []
+        implementation_pr_approved = False
 
         current_node = current_state.get("current_node", "")
 
@@ -556,12 +575,12 @@ async def _handle_resume_event(
         # GitHub fires check_suite webhooks for created/in_progress/completed — evaluating
         # 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 (
+        event = message.event_type
+        is_check_event = "check_suite" in event or "check_run" in event
+        if message.source == EventSource.GITHUB and (
             current_node in ("wait_for_ci_gate", "ci_evaluator")
-            and message.source == EventSource.GITHUB
+            or (targets_implementation_pr and is_check_event)
         ):
-            event = message.event_type
-            is_check_event = "check_suite" in event or "check_run" in event
             if is_check_event:
                 suite_status = payload.get("check_suite", {}).get("status") or payload.get(
                     "check_run", {}
@@ -574,7 +593,11 @@ async def _handle_resume_event(
                 else:
                     is_ci_webhook = True
                     logger.info(f"Detected GitHub CI webhook signal for {current_node}")
-            elif "issue_comment" not in event:
+            elif (
+                "issue_comment" not in event
+                and "pull_request_review" not in event
+                and payload.get("pull_request", {}).get("merged") is not True
+            ):
                 is_ci_webhook = True
                 logger.info(f"Detected GitHub CI webhook signal for {current_node}")
 
@@ -1320,15 +1343,16 @@ async def _handle_resume_event(
         if (
             message.source == EventSource.GITHUB
             and "pull_request_review" in message.event_type
-            and current_node in _REVIEW_GATES
+            and (current_node in _REVIEW_GATES or targets_implementation_pr)
             and current_state.get("is_paused", True)
         ):
             review = payload.get("review", {})
             review_state = review.get("state", "").lower()
             review_body = review.get("body", "") or ""
 
             if review_state == "approved":
-                # PR approved — advance to complete_tasks (via route_human_review default)
+                if targets_implementation_pr:
+                    implementation_pr_approved = True
                 is_approved = True
                 logger.info(f"Detected PR review approval for {message.ticket_key}")
             elif review_state in ("changes_requested", "commented"):
@@ -1383,7 +1407,7 @@ async def _handle_resume_event(
             message.source == EventSource.GITHUB
             and "pull_request" in message.event_type
             and payload.get("pull_request", {}).get("merged") is True
-            and current_node in _REVIEW_GATES
+            and (current_node in _REVIEW_GATES or targets_implementation_pr)
         ):
             is_approved = True
             pr_merged = True
@@ -1401,6 +1425,12 @@ async def _handle_resume_event(
                 "payload": payload,
             },
         }
+        if targets_implementation_pr and is_ci_webhook:
+            updated_state["current_node"] = "ci_evaluator"
+        elif targets_implementation_pr and (
+            "pull_request_review" in message.event_type or pr_merged
+        ):
+            updated_state["current_node"] = "human_review_gate"
 
         was_errored = _is_workflow_errored(current_state)
 
@@ -1506,12 +1536,19 @@ async def _handle_resume_event(
             updated_state["feedback_comment"] = None
             updated_state["last_error"] = None
         elif is_approved:
-            updated_state["is_paused"] = False
+            updated_state["is_paused"] = implementation_pr_approved
             updated_state["revision_requested"] = False
             updated_state["feedback_comment"] = None
             updated_state["last_error"] = None
+            if implementation_pr_approved:
+                updated_state["human_review_status"] = "approved"
             if pr_merged:
                 updated_state["pr_merged"] = True
+                if event_targets_pull_request(updated_state, payload):
+                    updated_state = mark_active_pull_request_merged(updated_state)
+                    updated_state["pr_merged"] = all_pull_requests_merged(updated_state)
+                    if not updated_state["pr_merged"]:
+                        updated_state["is_paused"] = True
                 if is_prd_review:
                     # Specification review is a separate artifact cycle and must
                     # receive its own automated revision budget.
@@ -1634,7 +1671,7 @@ async def _handle_resume_event(
                 )
                 return current_state
 
-        return updated_state
+        return save_active_pull_request(updated_state)
 
     async def _post_resume_ack_comment(
         self,

src/forge/workflow/base.py

@@ -9,6 +9,7 @@
 from langgraph.graph.message import add_messages
 
 from forge.models.workflow import TicketType
+from forge.workflow.pr_state import PullRequestState
 
 
 class BaseState(TypedDict, total=False):
@@ -47,6 +48,7 @@ class PRIntegrationState(TypedDict, total=False):
 
     workspace_path: str | None
     pr_urls: list[str]
+    pull_requests: dict[str, PullRequestState]
     current_pr_url: str | None
     current_pr_number: int | None
     current_repo: str | None

src/forge/workflow/bug/state.py

@@ -81,6 +81,7 @@ def create_initial_bug_state(ticket_key: str, **kwargs: Any) -> BugState:
         "bug_fix_implemented": False,
         "workspace_path": None,
         "pr_urls": [],
+        "pull_requests": {},
         "fork_owner": None,
         "fork_repo": None,
         "merge_conflicts": [],

src/forge/workflow/feature/state.py

@@ -94,6 +94,7 @@ def create_initial_feature_state(ticket_key: str, **kwargs: Any) -> FeatureState
         "tasks_by_repo": {},
         "workspace_path": None,
         "pr_urls": [],
+        "pull_requests": {},
         "fork_owner": None,
         "fork_repo": None,
         "merge_conflicts": [],

src/forge/workflow/nodes/ci_evaluator.py

@@ -45,7 +45,27 @@ async def evaluate_ci_status(state: WorkflowState) -> WorkflowState:
         Updated state with ci_status.
     """
     ticket_key = state["ticket_key"]
-    pr_urls = state.get("pr_urls", [])
+    current_pr_url = state.get("current_pr_url")
+    pull_requests = state.get("pull_requests", {})
+    active_pr = pull_requests.get(state.get("current_repo", ""))
+    if pull_requests:
+        if not (
+            isinstance(active_pr, dict)
+            and active_pr.get("number") == state.get("current_pr_number")
+            and current_pr_url
+        ):
+            logger.error("Cannot evaluate CI: active PR state is inconsistent for %s", ticket_key)
+            return update_state_timestamp(
+                {
+                    **state,
+                    "ci_status": "failed",
+                    "current_node": "ci_evaluator",
+                    "last_error": "Active pull request state is inconsistent",
+                }
+            )
+        pr_urls = [current_pr_url]
+    else:
+        pr_urls = state.get("pr_urls", [])
     ci_fix_attempt = state.get("ci_fix_attempt", 0)
     ci_fix_max = state.get("ci_fix_max_attempts", 5)
     settings = get_settings()

src/forge/workflow/nodes/pr_creation.py

@@ -15,6 +15,7 @@
 from forge.prompts import load_prompt
 from forge.workflow.nodes.code_review import sync_pr_description
 from forge.workflow.nodes.post_merge_summary import _extract_impact
+from forge.workflow.pr_state import save_active_pull_request
 from forge.workflow.utils import update_state_timestamp
 from forge.workflow.utils.jira_status import post_status_comment
 from forge.workspace.git_ops import GitOperations
@@ -329,16 +330,18 @@ async def create_pull_request(state: WorkflowState) -> WorkflowState:
                     logger.warning(f"Failed to append review exhaustion section: {e}")
 
         return update_state_timestamp(
-            {
-                **state,
-                "pr_urls": pr_urls,
-                "current_pr_url": pr_url,
-                "current_pr_number": pr_number,
-                "fork_owner": pr_target.fork_owner,
-                "fork_repo": pr_target.fork_repo,
-                "current_node": "teardown_workspace",
-                "last_error": None,
-            }
+            save_active_pull_request(
+                {
+                    **state,
+                    "pr_urls": pr_urls,
+                    "current_pr_url": pr_url,
+                    "current_pr_number": pr_number,
+                    "fork_owner": pr_target.fork_owner,
+                    "fork_repo": pr_target.fork_repo,
+                    "current_node": "teardown_workspace",
+                    "last_error": None,
+                }
+            )
         )
 
     except Exception as e: