diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py
index 857112c0..3b59fcc5 100644
--- a/src/forge/workflow/base.py
+++ b/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):
diff --git a/src/forge/workflow/bug/graph.py b/src/forge/workflow/bug/graph.py
index ed615dff..f9ccee44 100644
--- a/src/forge/workflow/bug/graph.py
+++ b/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")
diff --git a/src/forge/workflow/bug/state.py b/src/forge/workflow/bug/state.py
index 3dac40c3..94718cc1 100644
--- a/src/forge/workflow/bug/state.py
+++ b/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,
diff --git a/src/forge/workflow/feature/graph.py b/src/forge/workflow/feature/graph.py
index a51f010e..19d571c2 100644
--- a/src/forge/workflow/feature/graph.py
+++ b/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(
diff --git a/src/forge/workflow/feature/state.py b/src/forge/workflow/feature/state.py
index d67e84d9..69543d2f 100644
--- a/src/forge/workflow/feature/state.py
+++ b/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,
diff --git a/src/forge/workflow/nodes/git_persistence.py b/src/forge/workflow/nodes/git_persistence.py
new file mode 100644
index 00000000..7858e1e7
--- /dev/null
+++ b/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,
+    }
diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py
index 6bf02cd6..5514cc26 100644
--- a/src/forge/workflow/nodes/implementation.py
+++ b/src/forge/workflow/nodes/implementation.py
@@ -19,10 +19,15 @@
 from forge.models.workflow import TicketType
 from forge.sandbox import ContainerRunner
 from forge.workflow.feature.state import FeatureState as WorkflowState
+from forge.workflow.nodes.git_persistence import (
+    PushPersistenceError,
+    build_persistence_error_state,
+    push_to_fork_with_retry,
+)
+from forge.workflow.nodes.workspace_setup import prepare_workspace
 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
-from forge.workspace.manager import Workspace
 
 logger = logging.getLogger(__name__)
 
@@ -48,15 +53,59 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
     current_task = state.get("current_task_key")
     task_keys = state.get("task_keys", [])
     implementation_node = _implementation_node_name(state)
+    recorded_workspace = state.get("workspace_path")
+    local_workspace_survived = bool(recorded_workspace and Path(recorded_workspace).exists())
 
-    if not workspace_path:
-        logger.error(f"No workspace for implementation on {ticket_key}")
+    try:
+        git: GitOperations
+        workspace_path, git = prepare_workspace(state)
+        state = {**state, "workspace_path": workspace_path}
+    except Exception as exc:
+        logger.error("Unable to prepare implementation workspace for %s: %s", ticket_key, exc)
         return {
             **state,
-            "last_error": "Workspace not set up",
+            "last_error": str(exc),
             "current_node": implementation_node,
         }
 
+    same_workspace_survived = local_workspace_survived and workspace_path == recorded_workspace
+    if state.get("implementation_push_pending") and same_workspace_survived:
+        try:
+            await push_to_fork_with_retry(git)
+        except PushPersistenceError as exc:
+            return update_state_timestamp(
+                build_persistence_error_state(state, exc, retry_node=implementation_node)
+            )
+
+        pending_task = state.get("implementation_push_pending_task")
+        implemented = list(state.get("implemented_tasks", []))
+        if pending_task and pending_task not in implemented:
+            implemented.append(pending_task)
+        return update_state_timestamp(
+            {
+                **state,
+                "current_task_key": None,
+                "implemented_tasks": implemented,
+                "current_node": implementation_node,
+                "last_error": None,
+                "implementation_push_pending": False,
+                "implementation_push_pending_task": None,
+                "persistence_retry_count": 0,
+            }
+        )
+    if state.get("implementation_push_pending"):
+        logger.warning(
+            "Pending implementation push for %s cannot be recovered on this worker; "
+            "rerunning implementation",
+            ticket_key,
+        )
+        state = {
+            **state,
+            "implementation_push_pending": False,
+            "implementation_push_pending_task": None,
+            "last_error": None,
+        }
+
     # Get next task to implement if not set
     if not current_task and task_keys:
         # Get tasks for current repo
@@ -72,20 +121,10 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
     if not current_task:
         logger.info(f"All tasks implemented for {ticket_key}")
 
-        # Fallback: commit any files the container agent left uncommitted.
-        # The container is responsible for committing, but this catches edge
-        # cases where it exited before the final commit step.
-        if workspace_path:
-            branch_name = state.get("context", {}).get("branch_name", "")
-            current_repo = state.get("current_repo", "")
-            git = GitOperations(
-                Workspace(
-                    path=Path(workspace_path),
-                    repo_name=current_repo,
-                    branch_name=branch_name,
-                    ticket_key=ticket_key,
-                )
-            )
+        try:
+            # Fallback: commit any files the container agent left uncommitted.
+            # The container is responsible for committing, but this catches edge
+            # cases where it exited before the final commit step.
             if git.has_uncommitted_changes():
                 logger.warning(
                     f"Uncommitted changes found after all tasks for {ticket_key} — "
@@ -93,6 +132,25 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
                 )
                 git.stage_all()
                 git.commit(f"[{ticket_key}] chore: commit uncommitted changes after implementation")
+            await push_to_fork_with_retry(git)
+        except PushPersistenceError as exc:
+            return update_state_timestamp(
+                build_persistence_error_state(state, exc, retry_node=implementation_node)
+            )
+        except Exception as exc:
+            logger.error(
+                "Unable to persist completed implementation for %s: %s",
+                ticket_key,
+                exc,
+            )
+            return update_state_timestamp(
+                {
+                    **state,
+                    "last_error": str(exc),
+                    "current_node": implementation_node,
+                    "retry_count": state.get("retry_count", 0) + 1,
+                }
+            )
 
         return update_state_timestamp(
             {
@@ -155,6 +213,29 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
         if result.success:
             logger.info(f"Container completed successfully for {current_task}")
 
+            # Persist each task commit before checkpointing. A subsequent task
+            # or local review may resume on a worker with a different filesystem.
+            try:
+                await push_to_fork_with_retry(git)
+            except PushPersistenceError as exc:
+                pending_state = {
+                    **state,
+                    "implementation_push_pending": True,
+                    "implementation_push_pending_task": current_task,
+                }
+                return update_state_timestamp(
+                    build_persistence_error_state(
+                        pending_state,
+                        exc,
+                        retry_node=implementation_node,
+                    )
+                )
+
+            # Persist workflow bookkeeping immediately after the durable push.
+            implemented = list(state.get("implemented_tasks", []))
+            if current_task not in implemented:
+                implemented.append(current_task)
+
             # Post status comment at task implementation completion
             await post_status_comment(
                 jira,
@@ -163,9 +244,6 @@ async def implement_task(state: WorkflowState) -> WorkflowState:
             )
 
             # Track implemented tasks
-            implemented = state.get("implemented_tasks", [])
-            implemented.append(current_task)
-
             return update_state_timestamp(
                 {
                     **state,
diff --git a/src/forge/workflow/nodes/local_reviewer.py b/src/forge/workflow/nodes/local_reviewer.py
index 9917cf17..26c1d0d7 100644
--- a/src/forge/workflow/nodes/local_reviewer.py
+++ b/src/forge/workflow/nodes/local_reviewer.py
@@ -9,16 +9,21 @@
 from forge.prompts import load_prompt
 from forge.sandbox import ContainerRunner
 from forge.workflow.feature.state import FeatureState as WorkflowState
+from forge.workflow.nodes.git_persistence import (
+    PushPersistenceError,
+    build_persistence_error_state,
+    push_to_fork_with_retry,
+)
 from forge.workflow.nodes.review_utils import (
     next_review_attempt,
     parse_review_verdict,
     review_attempts_exhausted,
     run_review_container,
 )
+from forge.workflow.nodes.workspace_setup import prepare_workspace
 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
-from forge.workspace.manager import Workspace
 
 logger = logging.getLogger(__name__)
 
@@ -96,25 +101,58 @@ async def local_review_changes(state: WorkflowState) -> WorkflowState:
         Updated state routing to create_pr or implement_bug_fix.
     """
     ticket_key = state["ticket_key"]
-    workspace_path = state.get("workspace_path")
     ticket_type = state.get("ticket_type")
+    recorded_workspace = state.get("workspace_path")
+    local_workspace_survived = bool(recorded_workspace and Path(recorded_workspace).exists())
 
-    if not workspace_path:
-        logger.info(f"No workspace for local review on {ticket_key}, skipping")
-        return update_state_timestamp({**state, "current_node": "create_pr"})
+    try:
+        workspace_path, git = prepare_workspace(state)
+        state = {**state, "workspace_path": workspace_path}
+    except Exception as exc:
+        logger.error("Unable to prepare local-review workspace for %s: %s", ticket_key, exc)
+        return update_state_timestamp(
+            {**state, "current_node": "create_pr", "last_error": str(exc)}
+        )
+
+    same_workspace_survived = local_workspace_survived and workspace_path == recorded_workspace
+    if state.get("review_push_pending") and same_workspace_survived:
+        try:
+            await push_to_fork_with_retry(git)
+        except PushPersistenceError as exc:
+            return _review_persistence_error_state(state, exc)
+        updates = state.get("review_push_pending_updates", {})
+        return update_state_timestamp(
+            {
+                **state,
+                **updates,
+                "review_push_pending": False,
+                "review_push_pending_updates": {},
+                "persistence_retry_count": 0,
+            }
+        )
+    if state.get("review_push_pending"):
+        logger.warning(
+            "Pending review push for %s cannot be recovered on this worker; rerunning review",
+            ticket_key,
+        )
+        state = {
+            **state,
+            "review_push_pending": False,
+            "review_push_pending_updates": {},
+            "last_error": None,
+        }
 
     if ticket_type == TicketType.BUG:
-        return await _run_bug_review(state)
+        return await _run_bug_review(state, git)
     else:
-        return await _run_feature_review(state)
+        return await _run_feature_review(state, git)
 
 
-async def _run_bug_review(state: WorkflowState) -> WorkflowState:
+async def _run_bug_review(state: WorkflowState, git: GitOperations) -> WorkflowState:
     """Run qualitative local review for bug tickets."""
     ticket_key = state["ticket_key"]
     workspace_path = state["workspace_path"]
     current_repo = state.get("current_repo", "")
-    branch_name = state.get("context", {}).get("branch_name", "")
     qualitative_retry_count = state.get("qualitative_retry_count", 0)
 
     rca_content = state.get("rca_content") or ""
@@ -143,15 +181,6 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
             repo_name=current_repo,
         )
 
-        git = GitOperations(
-            Workspace(
-                path=Path(workspace_path),
-                repo_name=current_repo,
-                branch_name=branch_name,
-                ticket_key=ticket_key,
-            )
-        )
-
         if git.has_uncommitted_changes():
             git.stage_all()
             git.commit(f"[{ticket_key}] fix: address review feedback")
@@ -165,15 +194,16 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
 
         if verdict == "adequate":
             logger.info(f"Bug qualitative review passed for {ticket_key}")
-            return update_state_timestamp(
+            return await _persist_review_result(
+                state,
+                git,
                 {
-                    **state,
                     "local_review_verdict": verdict,
                     "qualitative_feedback": feedback or None,
                     "qualitative_retry_count": qualitative_retry_count,
                     "current_node": "create_pr",
                     "last_error": None,
-                }
+                },
             )
 
         # Non-adequate verdict
@@ -182,16 +212,17 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
                 f"Qualitative review cap ({_QUALITATIVE_CAP}) reached for {ticket_key}, "
                 f"proceeding with warning"
             )
-            return update_state_timestamp(
+            return await _persist_review_result(
+                state,
+                git,
                 {
-                    **state,
                     "local_review_verdict": verdict,
                     "qualitative_feedback": feedback or None,
                     "qualitative_retry_count": new_retry_count,
                     "qualitative_review_failed": True,
                     "current_node": "create_pr",
                     "last_error": None,
-                }
+                },
             )
 
         logger.info(
@@ -199,9 +230,10 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
             f"retry {new_retry_count}/{_QUALITATIVE_CAP}"
         )
         linked_task_keys = state.get("linked_task_keys") or state.get("task_keys") or []
-        return update_state_timestamp(
+        return await _persist_review_result(
+            state,
+            git,
             {
-                **state,
                 "local_review_verdict": verdict,
                 "qualitative_feedback": feedback or None,
                 "qualitative_retry_count": new_retry_count,
@@ -210,7 +242,7 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
                 # Reset so implement_task re-runs the container instead of seeing "all done"
                 "implemented_tasks": [],
                 "current_task_key": linked_task_keys[0] if linked_task_keys else None,
-            }
+            },
         )
 
     except Exception as e:
@@ -225,13 +257,12 @@ async def _run_bug_review(state: WorkflowState) -> WorkflowState:
         )
 
 
-async def _run_feature_review(state: WorkflowState) -> WorkflowState:
+async def _run_feature_review(state: WorkflowState, git: GitOperations) -> WorkflowState:
     """Run mechanical local review for non-bug tickets (existing behavior)."""
     ticket_key = state["ticket_key"]
     workspace_path = state["workspace_path"]
     review_attempts = state.get("local_review_attempts", 0)
     current_repo = state.get("current_repo", "")
-    branch_name = state.get("context", {}).get("branch_name", "")
     raw_pass_number = state.get("local_review_pass_number", 1)
     validated_pass = _validate_pass_number(raw_pass_number)
 
@@ -306,15 +337,6 @@ async def _run_feature_review(state: WorkflowState) -> WorkflowState:
             repo_name=current_repo,
         )
 
-        git = GitOperations(
-            Workspace(
-                path=Path(workspace_path),
-                repo_name=current_repo,
-                branch_name=branch_name,
-                ticket_key=ticket_key,
-            )
-        )
-
         if git.has_uncommitted_changes():
             git.stage_all()
             git.commit(f"[{ticket_key}] fix: address breaking issues found in local review")
@@ -327,13 +349,15 @@ async def _run_feature_review(state: WorkflowState) -> WorkflowState:
                 f"Breaking issues remain after review attempt {review_attempts + 1}, retrying"
             )
             next_pass = (validated_pass or 1) + 1
-            return update_state_timestamp(
+            return await _persist_review_result(
+                state,
+                git,
                 {
-                    **state,
                     "local_review_attempts": review_attempts + 1,
                     "local_review_pass_number": next_pass,
                     "current_node": "local_review",
-                }
+                    "last_error": None,
+                },
             )
 
         if has_unfixed:
@@ -344,13 +368,14 @@ async def _run_feature_review(state: WorkflowState) -> WorkflowState:
         else:
             logger.info(f"Local review passed for {ticket_key}")
 
-        return update_state_timestamp(
+        return await _persist_review_result(
+            state,
+            git,
             {
-                **state,
                 "local_review_attempts": 0,
                 "current_node": "create_pr",
                 "last_error": None,
-            }
+            },
         )
 
     except Exception as e:
@@ -369,3 +394,43 @@ def _has_unfixed_breaking_issues(output: str) -> bool:
     """Check if the review output indicates unfixed breaking issues remain."""
     lower = output.lower()
     return "unfixed" in lower and "breaking" in lower
+
+
+async def _persist_review_result(
+    state: WorkflowState,
+    git: GitOperations,
+    updates: dict,
+) -> WorkflowState:
+    """Persist review changes before applying the review's routing decision."""
+    try:
+        await push_to_fork_with_retry(git)
+    except PushPersistenceError as exc:
+        pending_state = {
+            **state,
+            "review_push_pending": True,
+            "review_push_pending_updates": updates,
+        }
+        return _review_persistence_error_state(pending_state, exc)
+    return update_state_timestamp(
+        {
+            **state,
+            **updates,
+            "review_push_pending": False,
+            "review_push_pending_updates": {},
+            "persistence_retry_count": 0,
+        }
+    )
+
+
+def _review_persistence_error_state(
+    state: WorkflowState,
+    error: PushPersistenceError,
+) -> WorkflowState:
+    return update_state_timestamp(
+        build_persistence_error_state(
+            state,
+            error,
+            retry_node="local_review",
+            escalation_node="escalate_blocked",
+        )
+    )
diff --git a/src/forge/workflow/nodes/task_takeover_execution.py b/src/forge/workflow/nodes/task_takeover_execution.py
index 36d8d65b..708fe785 100644
--- a/src/forge/workflow/nodes/task_takeover_execution.py
+++ b/src/forge/workflow/nodes/task_takeover_execution.py
@@ -7,6 +7,11 @@
 from forge.config import get_settings
 from forge.integrations.jira.client import JiraClient
 from forge.sandbox.runner import ContainerConfig, ContainerRunner
+from forge.workflow.nodes.git_persistence import (
+    PushPersistenceError,
+    build_persistence_error_state,
+    push_to_fork_with_retry,
+)
 from forge.workflow.nodes.workspace_setup import prepare_workspace
 from forge.workflow.task_takeover.state import TaskTakeoverState
 from forge.workflow.utils import update_state_timestamp
@@ -26,6 +31,8 @@ async def execute_task_changes(state: TaskTakeoverState) -> TaskTakeoverState:
     ticket_key = state["ticket_key"]
     current_repo = state.get("current_repo", "")
     current_task = state.get("current_task_key") or ticket_key
+    recorded_workspace = state.get("workspace_path")
+    local_workspace_survived = bool(recorded_workspace and Path(recorded_workspace).exists())
 
     settings = get_settings()
     jira = JiraClient(settings)
@@ -37,6 +44,47 @@ async def execute_task_changes(state: TaskTakeoverState) -> TaskTakeoverState:
         workspace_path, git = prepare_workspace(state)
         state = {**state, "workspace_path": workspace_path}
 
+        same_workspace_survived = local_workspace_survived and workspace_path == recorded_workspace
+        if state.get("implementation_push_pending") and same_workspace_survived:
+            try:
+                await push_to_fork_with_retry(git)
+            except PushPersistenceError as exc:
+                return cast(
+                    TaskTakeoverState,
+                    update_state_timestamp(
+                        build_persistence_error_state(
+                            state,
+                            exc,
+                            retry_node="execute_task_changes",
+                        )
+                    ),
+                )
+            return cast(
+                TaskTakeoverState,
+                update_state_timestamp(
+                    {
+                        **state,
+                        "last_error": None,
+                        "implementation_push_pending": False,
+                        "implementation_push_pending_task": None,
+                        "persistence_retry_count": 0,
+                        "current_node": "execute_task_changes",
+                    }
+                ),
+            )
+        if state.get("implementation_push_pending"):
+            logger.warning(
+                "Pending task-takeover push for %s cannot be recovered on this worker; "
+                "rerunning implementation",
+                ticket_key,
+            )
+            state = {
+                **state,
+                "implementation_push_pending": False,
+                "implementation_push_pending_task": None,
+                "last_error": None,
+            }
+
         # Get details from Jira for task implementation context
         task_issue = await jira.get_issue(current_task)
         task_description = task_issue.description or ""
@@ -90,33 +138,57 @@ async def execute_task_changes(state: TaskTakeoverState) -> TaskTakeoverState:
             committed = git.commit(commit_message)
 
         current_sha = git.get_current_sha()
+        execution_state = {
+            **state,
+            "task_execution_results": {
+                "success": result.success,
+                "exit_code": result.exit_code,
+                "error_message": result.error_message,
+            },
+            "task_execution_logs": {
+                "stdout": result.stdout,
+                "stderr": result.stderr,
+            },
+            "commit_info": {
+                "sha": current_sha,
+                "message": commit_message,
+                "committed": committed,
+            },
+            "current_node": "execute_task_changes",
+            "last_error": None if result.success else result.error_message,
+            "retry_count": 0 if result.success else state.get("retry_count", 0) + 1,
+        }
+
         # Review may be consumed by another worker with a different local
-        # filesystem.  Persist the exact commit before checkpointing this node.
-        git.push_to_fork()
+        # filesystem. Persist the exact commit before checkpointing this node.
+        try:
+            await push_to_fork_with_retry(git)
+        except PushPersistenceError as exc:
+            pending_state = {
+                **execution_state,
+                "implementation_push_pending": True,
+                "implementation_push_pending_task": current_task,
+            }
+            return cast(
+                TaskTakeoverState,
+                update_state_timestamp(
+                    build_persistence_error_state(
+                        pending_state,
+                        exc,
+                        retry_node="execute_task_changes",
+                    )
+                ),
+            )
 
         # Store results, logs, and commit info in state
         return cast(
             TaskTakeoverState,
             update_state_timestamp(
                 {
-                    **state,
-                    "task_execution_results": {
-                        "success": result.success,
-                        "exit_code": result.exit_code,
-                        "error_message": result.error_message,
-                    },
-                    "task_execution_logs": {
-                        "stdout": result.stdout,
-                        "stderr": result.stderr,
-                    },
-                    "commit_info": {
-                        "sha": current_sha,
-                        "message": commit_message,
-                        "committed": committed,
-                    },
-                    "current_node": "execute_task_changes",
-                    "last_error": None if result.success else result.error_message,
-                    "retry_count": 0 if result.success else state.get("retry_count", 0) + 1,
+                    **execution_state,
+                    "implementation_push_pending": False,
+                    "implementation_push_pending_task": None,
+                    "persistence_retry_count": 0,
                 }
             ),
         )
diff --git a/src/forge/workflow/task_takeover/graph.py b/src/forge/workflow/task_takeover/graph.py
index e2ffc06d..455fb3ad 100644
--- a/src/forge/workflow/task_takeover/graph.py
+++ b/src/forge/workflow/task_takeover/graph.py
@@ -129,6 +129,16 @@ def _route_after_answer(state: TaskTakeoverState) -> str:
     return "task_plan_approval_gate"
 
 
+def _route_after_execution(state: TaskTakeoverState) -> str:
+    """Never review an implementation whose branch was not persisted."""
+    last_error = state.get("last_error")
+    if not last_error:
+        return "run_qualitative_review"
+    if state.get("persistence_retry_count", 0) >= 3 or state.get("retry_count", 0) >= 3:
+        return "escalate_blocked"
+    return "execute_task_changes"
+
+
 def _route_after_qualitative_review(state: TaskTakeoverState) -> str:
     """Route after run_qualitative_review considering qualitative verdict and retry count.
 
@@ -335,7 +345,15 @@ def build_task_takeover_graph() -> StateGraph[TaskTakeoverState, Any, Any]:
 
     # Execution flow
     graph.add_edge("setup_workspace", "execute_task_changes")
-    graph.add_edge("execute_task_changes", "run_qualitative_review")
+    graph.add_conditional_edges(
+        "execute_task_changes",
+        _route_after_execution,
+        {
+            "execute_task_changes": "execute_task_changes",
+            "run_qualitative_review": "run_qualitative_review",
+            "escalate_blocked": "escalate_blocked",
+        },
+    )
     graph.add_conditional_edges(
         "run_qualitative_review",
         _route_after_qualitative_review,
diff --git a/src/forge/workflow/task_takeover/state.py b/src/forge/workflow/task_takeover/state.py
index 252d17fd..f3d3a3e8 100644
--- a/src/forge/workflow/task_takeover/state.py
+++ b/src/forge/workflow/task_takeover/state.py
@@ -62,6 +62,9 @@ def create_initial_task_takeover_state(ticket_key: str, **kwargs: Any) -> TaskTa
         "review_feedback": None,
         "qualitative_review_retry_count": 0,
         "qualitative_review_failed": False,
+        "implementation_push_pending": False,
+        "implementation_push_pending_task": None,
+        "persistence_retry_count": 0,
     }
     defaults.update(kwargs)
     return cast(TaskTakeoverState, defaults)
diff --git a/tests/unit/workflow/bug/test_graph.py b/tests/unit/workflow/bug/test_graph.py
index a7783001..5cad4a8d 100644
--- a/tests/unit/workflow/bug/test_graph.py
+++ b/tests/unit/workflow/bug/test_graph.py
@@ -39,42 +39,45 @@ 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"),
+            ("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),
+        ],
+    )
     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)
@@ -263,6 +266,14 @@ def test_decompose_plan_failure_routes_to_blocked(self):
 class TestLocalReviewRouting:
     """_route_after_local_review routes based on qualitative verdict."""
 
+    def test_persistence_escalation_takes_priority_over_review_cap(self):
+        state = _bug_state(
+            current_node="escalate_blocked",
+            last_error="authentication failed",
+            qualitative_retry_count=2,
+        )
+        assert _route_after_local_review(state) == "escalate_blocked"
+
     def test_adequate_verdict_routes_to_create_pr(self):
         state = _bug_state(local_review_verdict="adequate", qualitative_retry_count=0)
         assert _route_after_local_review(state) == "update_documentation"
@@ -309,9 +320,16 @@ def test_all_new_nodes_present(self):
         graph = build_bug_graph()
         compiled = graph.compile()
         expected_nodes = {
-            "triage_check", "triage_gate", "analyze_bug", "reflect_rca",
-            "rca_option_gate", "regenerate_rca", "plan_bug_fix",
-            "plan_approval_gate", "regenerate_plan", "decompose_plan",
+            "triage_check",
+            "triage_gate",
+            "analyze_bug",
+            "reflect_rca",
+            "rca_option_gate",
+            "regenerate_rca",
+            "plan_bug_fix",
+            "plan_approval_gate",
+            "regenerate_plan",
+            "decompose_plan",
             "post_merge_summary",
         }
         for node in expected_nodes:
diff --git a/tests/unit/workflow/feature/test_workflow.py b/tests/unit/workflow/feature/test_workflow.py
index aa4c46ae..101c23c9 100644
--- a/tests/unit/workflow/feature/test_workflow.py
+++ b/tests/unit/workflow/feature/test_workflow.py
@@ -1,6 +1,5 @@
 """Tests for FeatureWorkflow."""
 
-
 from langgraph.graph import END
 
 from forge.models.workflow import TicketType
@@ -12,6 +11,7 @@
     _route_after_single_task_update,
     _route_after_spec_regeneration,
     _route_after_task_regeneration,
+    _route_implementation,
     build_feature_graph,
     route_by_ticket_type,
 )
@@ -20,6 +20,16 @@
 class TestFeatureWorkflow:
     """Tests for FeatureWorkflow class."""
 
+    def test_implementation_error_never_falls_through_to_review(self):
+        state = {
+            "last_error": "push failed",
+            "retry_count": 1,
+            "current_repo": "owner/repo",
+            "tasks_by_repo": {"owner/repo": ["TASK-1"]},
+            "implemented_tasks": ["TASK-1"],
+        }
+        assert _route_implementation(state) == "implement_task"
+
     def test_workflow_has_name(self):
         """FeatureWorkflow has name attribute."""
         from forge.workflow.feature import FeatureWorkflow
diff --git a/tests/unit/workflow/nodes/conftest.py b/tests/unit/workflow/nodes/conftest.py
new file mode 100644
index 00000000..7c2a80f9
--- /dev/null
+++ b/tests/unit/workflow/nodes/conftest.py
@@ -0,0 +1,48 @@
+"""Opt-in isolation fixtures for workflow-node unit tests."""
+
+from unittest.mock import MagicMock, Mock, patch
+
+import pytest
+
+
+def _workspace_path(state):
+    workspace_path = state.get("workspace_path")
+    if not workspace_path:
+        raise ValueError("Workspace not set up")
+    return workspace_path
+
+
+@pytest.fixture
+def mock_implementation_workspace_recovery():
+    """Prevent implementation-node tests from cloning a real remote."""
+
+    def _prepare(state):
+        from forge.workflow.nodes import implementation
+
+        git = (
+            implementation.GitOperations.return_value
+            if isinstance(implementation.GitOperations, Mock)
+            else MagicMock()
+        )
+        return _workspace_path(state), git
+
+    with patch("forge.workflow.nodes.implementation.prepare_workspace", side_effect=_prepare):
+        yield
+
+
+@pytest.fixture
+def mock_review_workspace_recovery():
+    """Prevent local-review tests from cloning a real remote."""
+
+    def _prepare(state):
+        from forge.workflow.nodes import local_reviewer
+
+        git = (
+            local_reviewer.GitOperations.return_value
+            if isinstance(local_reviewer.GitOperations, Mock)
+            else MagicMock()
+        )
+        return _workspace_path(state), git
+
+    with patch("forge.workflow.nodes.local_reviewer.prepare_workspace", side_effect=_prepare):
+        yield
diff --git a/tests/unit/workflow/nodes/test_git_persistence.py b/tests/unit/workflow/nodes/test_git_persistence.py
new file mode 100644
index 00000000..cfa2a599
--- /dev/null
+++ b/tests/unit/workflow/nodes/test_git_persistence.py
@@ -0,0 +1,50 @@
+"""Tests for durable Git push policy."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from forge.workflow.nodes.git_persistence import (
+    PushFailureKind,
+    PushPersistenceError,
+    classify_push_failure,
+    push_to_fork_with_retry,
+)
+
+
+@pytest.mark.parametrize(
+    ("message", "expected"),
+    [
+        ("connection reset by peer", PushFailureKind.TRANSIENT),
+        ("request rejected by rate limiter", PushFailureKind.TRANSIENT),
+        ("Authentication failed", PushFailureKind.AUTH),
+        ("rejected (non-fast-forward)", PushFailureKind.NON_FAST_FORWARD),
+        ("invalid refspec", PushFailureKind.PERMANENT),
+    ],
+)
+def test_classify_push_failure(message, expected) -> None:
+    assert classify_push_failure(RuntimeError(message)) == expected
+
+
+@pytest.mark.asyncio
+async def test_transient_push_is_retried_without_rerunning_work() -> None:
+    git = MagicMock()
+    git.push_to_fork.side_effect = [RuntimeError("connection reset"), None]
+
+    with patch("forge.workflow.nodes.git_persistence.asyncio.sleep") as sleep:
+        await push_to_fork_with_retry(git)
+
+    assert git.push_to_fork.call_count == 2
+    sleep.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_non_fast_forward_fails_without_retry() -> None:
+    git = MagicMock()
+    git.push_to_fork.side_effect = RuntimeError("rejected (non-fast-forward)")
+
+    with pytest.raises(PushPersistenceError) as raised:
+        await push_to_fork_with_retry(git)
+
+    assert raised.value.kind == PushFailureKind.NON_FAST_FORWARD
+    git.push_to_fork.assert_called_once()
diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py
index 64aeac11..90de8ba8 100644
--- a/tests/unit/workflow/nodes/test_implementation.py
+++ b/tests/unit/workflow/nodes/test_implementation.py
@@ -6,6 +6,8 @@
 
 from forge.models.workflow import TicketType
 
+pytestmark = pytest.mark.usefixtures("mock_implementation_workspace_recovery")
+
 
 def _make_state(
     ticket_key="BUG-123",
@@ -56,7 +58,6 @@ def _make_successful_runner():
 
 
 class TestImplementTaskStartedComment:
-
     @pytest.mark.asyncio
     async def test_posts_comment_on_task_ticket_before_container(self):
         """A comment is posted on the task ticket (not parent) when implementation starts."""
@@ -184,7 +185,6 @@ async def test_passes_trace_context_to_container_runner(self):
 
 
 class TestImplementationNodeRouting:
-
     @pytest.mark.asyncio
     async def test_feature_missing_workspace_uses_feature_implementation_node(self):
         """Feature implementation failures must resume at implement_task."""
@@ -242,6 +242,106 @@ async def test_feature_container_failure_uses_feature_implementation_node(self):
         assert result["last_error"] == "container failed"
         assert result["retry_count"] == 1
 
+    @pytest.mark.asyncio
+    async def test_successful_implementation_is_pushed_before_checkpoint(self) -> None:
+        """A different worker can recover the implementation commit from the fork."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        state = _make_state()
+        mock_git = MagicMock()
+        mock_jira = _make_mock_jira()
+        runner = _make_successful_runner()
+
+        with (
+            patch(
+                "forge.workflow.nodes.implementation.prepare_workspace",
+                return_value=(state["workspace_path"], mock_git),
+            ),
+            patch("forge.workflow.nodes.implementation.JiraClient", return_value=mock_jira),
+            patch("forge.workflow.nodes.implementation.ContainerRunner", return_value=runner),
+            patch("forge.workflow.nodes.implementation.get_settings"),
+        ):
+            result = await implement_task(state)
+
+        assert result["last_error"] is None
+        mock_git.push_to_fork.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_final_push_failure_is_recorded_for_retry(self) -> None:
+        """The all-tasks-done path must not leak a push exception from the graph node."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        state = _make_state(current_task_key=None)
+        state["task_keys"] = []
+        mock_git = MagicMock()
+        mock_git.has_uncommitted_changes.return_value = False
+        mock_git.push_to_fork.side_effect = RuntimeError("fork unavailable")
+
+        with patch(
+            "forge.workflow.nodes.implementation.prepare_workspace",
+            return_value=(state["workspace_path"], mock_git),
+        ):
+            result = await implement_task(state)
+
+        assert result["current_node"] == "implement_bug_fix"
+        assert result["last_error"] == "fork unavailable"
+        assert result["retry_count"] == 0
+        assert result["persistence_retry_count"] == 3
+
+    @pytest.mark.asyncio
+    async def test_pending_push_retries_without_rerunning_container(self, tmp_path) -> None:
+        """A surviving workspace resumes at persistence, not implementation."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        state = _make_state(workspace_path=str(tmp_path))
+        state["implementation_push_pending"] = True
+        state["implementation_push_pending_task"] = "TASK-456"
+        mock_git = MagicMock()
+
+        with (
+            patch(
+                "forge.workflow.nodes.implementation.prepare_workspace",
+                return_value=(str(tmp_path), mock_git),
+            ),
+            patch("forge.workflow.nodes.implementation.ContainerRunner") as runner,
+        ):
+            result = await implement_task(state)
+
+        runner.assert_not_called()
+        mock_git.push_to_fork.assert_called_once()
+        assert result["implemented_tasks"] == ["TASK-456"]
+        assert result["implementation_push_pending"] is False
+
+    @pytest.mark.asyncio
+    async def test_recreated_workspace_does_not_mark_pending_task_complete(self, tmp_path) -> None:
+        """A replacement clone cannot stand in for the workspace holding the commit."""
+        from forge.workflow.nodes.implementation import implement_task
+
+        old_workspace = tmp_path / "old"
+        old_workspace.mkdir()
+        new_workspace = tmp_path / "new"
+        new_workspace.mkdir()
+        state = _make_state(workspace_path=str(old_workspace))
+        state["implementation_push_pending"] = True
+        state["implementation_push_pending_task"] = "TASK-456"
+        mock_git = MagicMock()
+        mock_jira = _make_mock_jira()
+        runner = _make_successful_runner()
+
+        with (
+            patch(
+                "forge.workflow.nodes.implementation.prepare_workspace",
+                return_value=(str(new_workspace), mock_git),
+            ),
+            patch("forge.workflow.nodes.implementation.JiraClient", return_value=mock_jira),
+            patch("forge.workflow.nodes.implementation.ContainerRunner", return_value=runner),
+            patch("forge.workflow.nodes.implementation.get_settings"),
+        ):
+            result = await implement_task(state)
+
+        runner.run.assert_awaited_once()
+        assert result["implementation_push_pending"] is False
+
     @pytest.mark.asyncio
     async def test_bug_container_failure_keeps_bug_implementation_node(self):
         """Bug container failures keep the bug graph retry node."""
diff --git a/tests/unit/workflow/nodes/test_implementation_status_instrumentation.py b/tests/unit/workflow/nodes/test_implementation_status_instrumentation.py
index 03ef06aa..482f8470 100644
--- a/tests/unit/workflow/nodes/test_implementation_status_instrumentation.py
+++ b/tests/unit/workflow/nodes/test_implementation_status_instrumentation.py
@@ -13,6 +13,8 @@
 from forge.workflow.feature.state import create_initial_feature_state
 from forge.workflow.nodes.implementation import implement_task
 
+pytestmark = pytest.mark.usefixtures("mock_implementation_workspace_recovery")
+
 
 def create_mock_jira_client():
     """Create a mock JiraClient with required methods."""
@@ -333,4 +335,3 @@ async def test_multiple_tasks_mixed_success_failure_correct_task_keys(self):
         assert (
             mock_post_status2.call_args_list[0][0][2] == "🔨 Forge started implementing [TASK-2]: Task summary"
         )
-
diff --git a/tests/unit/workflow/nodes/test_local_review_fix_pass_comment.py b/tests/unit/workflow/nodes/test_local_review_fix_pass_comment.py
index 9380b50f..f4c6f3e9 100644
--- a/tests/unit/workflow/nodes/test_local_review_fix_pass_comment.py
+++ b/tests/unit/workflow/nodes/test_local_review_fix_pass_comment.py
@@ -12,6 +12,8 @@
 from forge.workflow.feature.state import create_initial_feature_state
 from forge.workflow.nodes.local_reviewer import local_review_changes
 
+pytestmark = pytest.mark.usefixtures("mock_review_workspace_recovery")
+
 
 def create_mock_jira_client():
     """Create a mock JiraClient with required methods."""
diff --git a/tests/unit/workflow/nodes/test_local_review_pass_number.py b/tests/unit/workflow/nodes/test_local_review_pass_number.py
index c4e90334..a43bbf86 100644
--- a/tests/unit/workflow/nodes/test_local_review_pass_number.py
+++ b/tests/unit/workflow/nodes/test_local_review_pass_number.py
@@ -9,6 +9,11 @@
 from forge.workflow.nodes.implementation import implement_task
 from forge.workflow.nodes.local_reviewer import local_review_changes
 
+pytestmark = pytest.mark.usefixtures(
+    "mock_implementation_workspace_recovery",
+    "mock_review_workspace_recovery",
+)
+
 
 def create_mock_jira_client():
     """Create a mock JiraClient with required methods."""
diff --git a/tests/unit/workflow/nodes/test_local_review_pass_tracking_errors.py b/tests/unit/workflow/nodes/test_local_review_pass_tracking_errors.py
index c4c73845..2d03b252 100644
--- a/tests/unit/workflow/nodes/test_local_review_pass_tracking_errors.py
+++ b/tests/unit/workflow/nodes/test_local_review_pass_tracking_errors.py
@@ -8,6 +8,8 @@
 
 from forge.workflow.nodes.local_reviewer import _validate_pass_number, local_review_changes
 
+pytestmark = pytest.mark.usefixtures("mock_review_workspace_recovery")
+
 
 class TestValidatePassNumber:
     """Test the _validate_pass_number function."""
diff --git a/tests/unit/workflow/nodes/test_local_review_status_comments_comprehensive.py b/tests/unit/workflow/nodes/test_local_review_status_comments_comprehensive.py
index b8cad415..fc12d529 100644
--- a/tests/unit/workflow/nodes/test_local_review_status_comments_comprehensive.py
+++ b/tests/unit/workflow/nodes/test_local_review_status_comments_comprehensive.py
@@ -16,6 +16,8 @@
 from forge.workflow.feature.state import create_initial_feature_state
 from forge.workflow.nodes.local_reviewer import local_review_changes
 
+pytestmark = pytest.mark.usefixtures("mock_review_workspace_recovery")
+
 
 def create_mock_jira_client():
     """Create a mock JiraClient with required methods."""
diff --git a/tests/unit/workflow/nodes/test_local_reviewer.py b/tests/unit/workflow/nodes/test_local_reviewer.py
index 7a78a0ff..a343c622 100644
--- a/tests/unit/workflow/nodes/test_local_reviewer.py
+++ b/tests/unit/workflow/nodes/test_local_reviewer.py
@@ -1,6 +1,6 @@
 """Unit tests for local_review_changes bug-specific enhancements."""
 
-from unittest.mock import MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
@@ -11,6 +11,8 @@
     route_local_review,
 )
 
+pytestmark = pytest.mark.usefixtures("mock_review_workspace_recovery")
+
 
 @pytest.fixture
 def base_bug_review_state():
@@ -140,7 +142,6 @@ async def run(self, workspace_path, task_description="", **_kwargs):  # noqa: AR
                 return result
 
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch(
@@ -148,7 +149,6 @@ async def run(self, workspace_path, task_description="", **_kwargs):  # noqa: AR
                 return_value=_CapturingRunner(),
             ),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             await local_review_changes(base_bug_review_state)
 
@@ -162,17 +162,54 @@ async def test_adequate_verdict_routes_to_create_pr(self, base_bug_review_state)
         """'adequate' verdict → routes to create_pr."""
         runner = _make_mock_runner("verdict: adequate\n\nfeedback: Looks good.")
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
         assert result["current_node"] == "create_pr"
         assert result["local_review_verdict"] == "adequate"
+        mock_git.push_to_fork.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_transient_push_resumes_without_rerunning_review(
+        self, base_bug_review_state, tmp_path
+    ):
+        """A review verdict is checkpointed while only its push is retried."""
+        base_bug_review_state["workspace_path"] = str(tmp_path)
+        runner = _make_mock_runner("verdict: adequate\nfeedback: Looks good.")
+        mock_git = _make_mock_git()
+        mock_git.push_to_fork.side_effect = [
+            RuntimeError("connection reset"),
+            RuntimeError("connection reset"),
+            RuntimeError("connection reset"),
+            None,
+        ]
+
+        with (
+            patch(
+                "forge.workflow.nodes.local_reviewer.prepare_workspace",
+                return_value=(str(tmp_path), mock_git),
+            ),
+            patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
+            patch(
+                "forge.workflow.nodes.git_persistence.asyncio.sleep",
+                new=AsyncMock(),
+            ),
+        ):
+            pending = await local_review_changes(base_bug_review_state)
+            assert pending["review_push_pending"] is True
+            assert pending["current_node"] == "local_review"
+
+            with patch("forge.workflow.nodes.local_reviewer.ContainerRunner") as rerun:
+                completed = await local_review_changes(pending)
+
+        rerun.assert_not_called()
+        assert completed["review_push_pending"] is False
+        assert completed["current_node"] == "create_pr"
+        assert completed["local_review_verdict"] == "adequate"
 
     @pytest.mark.asyncio
     async def test_tests_incomplete_increments_retry(self, base_bug_review_state):
@@ -181,12 +218,10 @@ async def test_tests_incomplete_increments_retry(self, base_bug_review_state):
             "verdict: tests_incomplete\n\nfeedback: Tests do not fail without fix."
         )
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -200,12 +235,10 @@ async def test_symptom_only_increments_retry(self, base_bug_review_state):
         """'symptom_only' verdict → qualitative_retry_count incremented, routes to implement_bug_fix."""
         runner = _make_mock_runner("verdict: symptom_only\n\nfeedback: Root cause not addressed.")
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -219,12 +252,10 @@ async def test_retry_uses_task_keys_when_linked_task_keys_empty(self, base_bug_r
         base_bug_review_state["task_keys"] = ["TASK-789"]
         runner = _make_mock_runner("verdict: tests_incomplete\n\nfeedback: Missing tests.")
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -236,12 +267,10 @@ async def test_cap_at_two_retries_routes_to_create_pr(self, base_bug_review_stat
         base_bug_review_state["qualitative_retry_count"] = 1  # Already 1, will become 2 → cap
         runner = _make_mock_runner("verdict: tests_incomplete\n\nfeedback: Still missing tests.")
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -273,14 +302,11 @@ class _FailingRunner:
             async def run(self, **_kwargs):
                 raise RuntimeError("Container crashed")
 
-        mock_workspace = MagicMock()
-
         with (
             patch(
                 "forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=_FailingRunner()
             ),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=MagicMock()),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -298,14 +324,11 @@ class _FailingRunner:
             async def run(self, **_kwargs):
                 raise RuntimeError("OOM")
 
-        mock_workspace = MagicMock()
-
         with (
             patch(
                 "forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=_FailingRunner()
             ),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=MagicMock()),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_bug_review_state)
 
@@ -332,7 +355,6 @@ async def run(self, workspace_path, task_description="", **_kwargs):  # noqa: AR
                 return result
 
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch(
@@ -340,24 +362,22 @@ async def run(self, workspace_path, task_description="", **_kwargs):  # noqa: AR
                 return_value=_CapturingRunner(),
             ),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_feature_review_state)
 
         assert result["current_node"] == "create_pr"
         assert result.get("local_review_verdict") is None
+        mock_git.push_to_fork.assert_called_once()
 
     @pytest.mark.asyncio
     async def test_feature_no_qualitative_fields_set(self, base_feature_review_state):
         """Feature ticket → qualitative_retry_count and verdict not modified."""
         runner = _make_mock_runner("verdict: adequate\n\nfeedback: Good.")
         mock_git = _make_mock_git()
-        mock_workspace = MagicMock()
 
         with (
             patch("forge.workflow.nodes.local_reviewer.ContainerRunner", return_value=runner),
             patch("forge.workflow.nodes.local_reviewer.GitOperations", return_value=mock_git),
-            patch("forge.workflow.nodes.local_reviewer.Workspace", return_value=mock_workspace),
         ):
             result = await local_review_changes(base_feature_review_state)
 
diff --git a/tests/unit/workflow/nodes/test_task_takeover_execution.py b/tests/unit/workflow/nodes/test_task_takeover_execution.py
index 9f20d682..2eb049e3 100644
--- a/tests/unit/workflow/nodes/test_task_takeover_execution.py
+++ b/tests/unit/workflow/nodes/test_task_takeover_execution.py
@@ -202,3 +202,91 @@ async def test_unexpected_exception(self) -> None:
         assert result_state["last_error"] == "Jira Connection Error"
         assert result_state["current_node"] == "execute_task_changes"
         assert result_state["retry_count"] == 1
+
+    @pytest.mark.asyncio
+    async def test_push_failure_does_not_consume_implementation_retry(self) -> None:
+        state = _make_state()
+        mock_jira = _make_mock_jira()
+        mock_runner = _make_mock_runner()
+        mock_git = _make_mock_git()
+        mock_git.push_to_fork.side_effect = RuntimeError("authentication failed")
+
+        with (
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.JiraClient",
+                return_value=mock_jira,
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.ContainerRunner",
+                return_value=mock_runner,
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                return_value=("/tmp/ws", mock_git),
+            ),
+            patch("forge.workflow.nodes.task_takeover_execution.get_settings"),
+        ):
+            result = await execute_task_changes(state)
+
+        assert result["implementation_push_pending"] is True
+        assert result["retry_count"] == 0
+        assert result["persistence_retry_count"] == 3
+
+    @pytest.mark.asyncio
+    async def test_pending_push_does_not_rerun_container(self, tmp_path) -> None:
+        state = _make_state(workspace_path=str(tmp_path))
+        state["implementation_push_pending"] = True
+        state["implementation_push_pending_task"] = "TASK-123"
+        mock_git = _make_mock_git()
+        mock_jira = _make_mock_jira()
+
+        with (
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.JiraClient",
+                return_value=mock_jira,
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                return_value=(str(tmp_path), mock_git),
+            ),
+            patch("forge.workflow.nodes.task_takeover_execution.ContainerRunner") as runner,
+        ):
+            result = await execute_task_changes(state)
+
+        runner.assert_not_called()
+        mock_git.push_to_fork.assert_called_once()
+        assert result["implementation_push_pending"] is False
+        assert result["last_error"] is None
+
+    @pytest.mark.asyncio
+    async def test_recreated_workspace_reruns_task_instead_of_marking_push_done(
+        self, tmp_path
+    ) -> None:
+        old_workspace = tmp_path / "old"
+        old_workspace.mkdir()
+        new_workspace = tmp_path / "new"
+        new_workspace.mkdir()
+        state = _make_state(workspace_path=str(old_workspace))
+        state["implementation_push_pending"] = True
+        mock_git = _make_mock_git()
+        mock_jira = _make_mock_jira()
+        runner = _make_mock_runner()
+
+        with (
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.prepare_workspace",
+                return_value=(str(new_workspace), mock_git),
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.JiraClient",
+                return_value=mock_jira,
+            ),
+            patch(
+                "forge.workflow.nodes.task_takeover_execution.ContainerRunner",
+                return_value=runner,
+            ),
+        ):
+            result = await execute_task_changes(state)
+
+        runner.run.assert_awaited_once()
+        assert result["implementation_push_pending"] is False
diff --git a/tests/workflow/test_task_takeover_graph.py b/tests/workflow/test_task_takeover_graph.py
index 1c49aa35..862315ae 100644
--- a/tests/workflow/test_task_takeover_graph.py
+++ b/tests/workflow/test_task_takeover_graph.py
@@ -10,6 +10,7 @@
 from forge.workflow.gates.task_plan_approval import route_task_plan_approval
 from forge.workflow.task_takeover.graph import (
     _route_after_answer,
+    _route_after_execution,
     _route_after_generate_plan,
     _route_after_triage_check,
     _route_ci_evaluation,
@@ -157,6 +158,19 @@ def test_route_after_generate_plan_retry_cap_routes_to_blocked(self) -> None:
         assert _route_after_generate_plan(state) == "escalate_blocked"
 
 
+class TestExecutionRouting:
+    def test_success_routes_to_review(self) -> None:
+        assert _route_after_execution(make_task_state(last_error=None)) == "run_qualitative_review"
+
+    def test_push_failure_retries_execution_node(self) -> None:
+        state = make_task_state(last_error="network timeout", persistence_retry_count=1)
+        assert _route_after_execution(state) == "execute_task_changes"
+
+    def test_persistence_cap_blocks_without_review(self) -> None:
+        state = make_task_state(last_error="authentication failed", persistence_retry_count=3)
+        assert _route_after_execution(state) == "escalate_blocked"
+
+
 class TestQualitativeReviewRouting:
     """Test routing after run_qualitative_review."""