fix: persist Feature and Bug review handoffs across workers

src/forge/workflow/base.py

@@ -58,6 +58,11 @@ class PRIntegrationState(TypedDict, total=False):
     merge_conflicts: list[str]
     local_review_attempts: int
     local_review_pass_number: int
+    implementation_push_pending: bool
+    implementation_push_pending_task: str | None
+    persistence_retry_count: int
+    review_push_pending: bool
+    review_push_pending_updates: dict[str, Any]
 
 
 class CIIntegrationState(TypedDict, total=False):

src/forge/workflow/bug/graph.py

@@ -295,6 +295,12 @@ def _route_after_local_review(state: BugState) -> str:
 
     verdict = state.get("local_review_verdict")
     retry_count = state.get("qualitative_retry_count", 0)
+    current_node = state.get("current_node", "update_documentation")
+
+    if current_node == "escalate_blocked":
+        return "escalate_blocked"
+    if state.get("last_error"):
+        return current_node
 
     if verdict == "adequate" or retry_count >= _QUALITATIVE_CAP:
         return "update_documentation"
@@ -304,7 +310,7 @@ def _route_after_local_review(state: BugState) -> str:
     # to prevent infinite loops if current_node is "local_review".
     if state.get("local_review_attempts", 0) >= MAX_REVIEW_ATTEMPTS:
         return "update_documentation"
-    return state.get("current_node", "update_documentation")
+    return current_node
 
 
 def _route_after_workspace_setup(
@@ -333,6 +339,10 @@ def _route_after_implementation(
     max_retries = 3
     last_error = state.get("last_error")
 
+    if last_error and state.get("persistence_retry_count", 0) >= 3:
+        logger.error(f"Git persistence retry limit exceeded: {last_error}")
+        return "escalate_blocked"
+
     if last_error:
         if retry_count >= max_retries:
             logger.error(f"Implementation retry limit ({max_retries}) exceeded: {last_error}")
@@ -623,6 +633,7 @@ def build_bug_graph() -> StateGraph:
             "update_documentation": "update_documentation",
             "create_pr": "create_pr",
             "implement_bug_fix": "implement_bug_fix",
+            "escalate_blocked": "escalate_blocked",
         },
     )
     graph.add_edge("update_documentation", "create_pr")

src/forge/workflow/bug/state.py

@@ -86,6 +86,11 @@ def create_initial_bug_state(ticket_key: str, **kwargs: Any) -> BugState:
         "merge_conflicts": [],
         "local_review_attempts": 0,
         "local_review_pass_number": 1,
+        "implementation_push_pending": False,
+        "implementation_push_pending_task": None,
+        "persistence_retry_count": 0,
+        "review_push_pending": False,
+        "review_push_pending_updates": {},
         "tdd_approach": False,
         "ci_status": None,
         "current_pr_url": None,

src/forge/workflow/feature/graph.py

@@ -160,6 +160,7 @@ def _route_after_generation(state: FeatureState) -> str:
         "prd_approval_gate" on success, END on failure.
     """
     last_error = state.get("last_error")
+
     prd_content = state.get("prd_content", "")
 
     if last_error and not prd_content:
@@ -313,10 +314,17 @@ def _route_implementation(
     max_retries = 3  # Max retries per task
     last_error = state.get("last_error")
 
+    if last_error and state.get("persistence_retry_count", 0) >= 3:
+        logger.error(f"Git persistence retry limit exceeded: {last_error}")
+        return "escalate_blocked"
+
     if last_error and retry_count >= max_retries:
         logger.error(f"Implementation retry limit ({max_retries}) exceeded: {last_error}")
         return "escalate_blocked"
 
+    if last_error:
+        return "implement_task"
+
     current_repo = state.get("current_repo", "")
     repo_tasks = state.get("tasks_by_repo", {}).get(current_repo, [])
     implemented = state.get("implemented_tasks", [])
@@ -698,7 +706,11 @@ def build_feature_graph() -> StateGraph:
     graph.add_conditional_edges(
         "local_review",
         lambda s: s.get("current_node", "create_pr"),
-        {"local_review": "local_review", "create_pr": "update_documentation"},
+        {
+            "local_review": "local_review",
+            "create_pr": "update_documentation",
+            "escalate_blocked": "escalate_blocked",
+        },
     )
     graph.add_edge("update_documentation", "create_pr")
     graph.add_conditional_edges(

src/forge/workflow/feature/state.py

@@ -90,6 +90,11 @@ def create_initial_feature_state(ticket_key: str, **kwargs: Any) -> FeatureState
         "merge_conflicts": [],
         "local_review_attempts": 0,
         "local_review_pass_number": 1,
+        "implementation_push_pending": False,
+        "implementation_push_pending_task": None,
+        "persistence_retry_count": 0,
+        "review_push_pending": False,
+        "review_push_pending_updates": {},
         "ci_status": None,
         "current_pr_url": None,
         "current_pr_number": None,

src/forge/workflow/nodes/git_persistence.py

@@ -0,0 +1,120 @@
+"""Durable Git persistence helpers for workflow handoffs."""
+
+import asyncio
+import logging
+from enum import StrEnum
+
+from forge.workspace.git_ops import GitOperations
+
+logger = logging.getLogger(__name__)
+
+
+class PushFailureKind(StrEnum):
+    """Actionable categories for a failed Git push."""
+
+    TRANSIENT = "transient"
+    AUTH = "auth"
+    NON_FAST_FORWARD = "non_fast_forward"
+    PERMANENT = "permanent"
+
+
+class PushPersistenceError(RuntimeError):
+    """Raised when a branch cannot be persisted after the allowed attempts."""
+
+    def __init__(self, message: str, kind: PushFailureKind):
+        super().__init__(message)
+        self.kind = kind
+
+
+def classify_push_failure(error: Exception) -> PushFailureKind:
+    """Classify Git's text error until GitError exposes structured metadata."""
+    message = str(error).lower()
+    if any(
+        marker in message
+        for marker in (
+            "timed out",
+            "timeout",
+            "could not resolve host",
+            "connection reset",
+            "connection refused",
+            "remote end hung up",
+            "network is unreachable",
+            "rate limit",
+            "http 429",
+            "http 502",
+            "http 503",
+            "http 504",
+        )
+    ):
+        return PushFailureKind.TRANSIENT
+    if any(
+        marker in message
+        for marker in (
+            "authentication failed",
+            "permission denied",
+            "could not read username",
+            "repository not found",
+            "http 401",
+            "http 403",
+        )
+    ):
+        return PushFailureKind.AUTH
+    if any(
+        marker in message
+        for marker in (
+            "non-fast-forward",
+            "fetch first",
+            "[rejected]",
+            "failed to push some refs",
+        )
+    ):
+        return PushFailureKind.NON_FAST_FORWARD
+    return PushFailureKind.PERMANENT
+
+
+async def push_to_fork_with_retry(
+    git: GitOperations,
+    *,
+    max_attempts: int = 3,
+    initial_delay_seconds: float = 1.0,
+) -> None:
+    """Push a workflow branch, retrying only failures known to be transient."""
+    for attempt in range(1, max_attempts + 1):
+        try:
+            git.push_to_fork()
+            return
+        except Exception as exc:
+            kind = classify_push_failure(exc)
+            if kind != PushFailureKind.TRANSIENT or attempt >= max_attempts:
+                raise PushPersistenceError(str(exc), kind) from exc
+            delay = initial_delay_seconds * (2 ** (attempt - 1))
+            logger.warning(
+                "Transient fork push failure (%s/%s); retrying in %.1fs: %s",
+                attempt,
+                max_attempts,
+                delay,
+                exc,
+            )
+            await asyncio.sleep(delay)
+
+
+def build_persistence_error_state(
+    state: dict,
+    error: PushPersistenceError,
+    *,
+    retry_node: str,
+    escalation_node: str | None = None,
+    max_workflow_attempts: int = 3,
+) -> dict:
+    """Build consistent workflow state for an exhausted push operation."""
+    previous = state.get("persistence_retry_count", 0)
+    attempts = max_workflow_attempts if error.kind != PushFailureKind.TRANSIENT else previous + 1
+    current_node = (
+        escalation_node if escalation_node and attempts >= max_workflow_attempts else retry_node
+    )
+    return {
+        **state,
+        "last_error": str(error),
+        "current_node": current_node,
+        "persistence_retry_count": attempts,
+    }