Sync worktree with remote before pipeline phase execution

orchestrator/gateway_client.py

@@ -615,6 +615,66 @@ def push_worktree_branch(
                 except Exception:
                     pass
 
+    def fetch_worktree_branch(
+        self,
+        pipeline_id: str,
+        repo_path: str,
+    ) -> bool:
+        """Fetch latest remote state into a worktree using a temporary session.
+
+        Best-effort operation to sync remote changes into a worktree —
+        called before phase execution to ensure the worktree has all state
+        from previous phases (e.g., after orchestrator restart where the
+        local branch diverged from remote).
+
+        Args:
+            pipeline_id: Pipeline ID (used as container_id for the temp session)
+            repo_path: Path to the worktree repo directory
+
+        Returns:
+            True if fetch succeeded, False otherwise
+        """
+        temp_container_id = f"{pipeline_id}-failsafe-fetch"
+        session_token: str | None = None
+        try:
+            session = self.register_session(
+                container_id=temp_container_id,
+                container_ip="127.0.0.1",
+                mode="local",
+                pipeline_id=pipeline_id,
+            )
+            session_token = session.session_token
+
+            self._make_request(
+                "/api/v1/git/fetch",
+                method="POST",
+                data={
+                    "repo_path": repo_path,
+                    "remote": "origin",
+                    "container_id": temp_container_id,
+                },
+                bearer_token=session_token,
+            )
+
+            logger.info(
+                "Fetched remote state into worktree",
+                pipeline_id=pipeline_id,
+            )
+            return True
+        except Exception as e:
+            logger.warning(
+                "Best-effort fetch failed (continuing with local state)",
+                pipeline_id=pipeline_id,
+                error=str(e),
+            )
+            return False
+        finally:
+            if session_token:
+                try:
+                    self.delete_session(session_token)
+                except Exception:
+                    pass
+
 
 class GatewayError(Exception):
     """Error from gateway operations."""

orchestrator/routes/pipelines.py

@@ -10,7 +10,7 @@
 import threading
 from datetime import datetime
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import yaml
 from docker.errors import DockerException
@@ -79,6 +79,12 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
         get_state_store,
     )
 
+if TYPE_CHECKING:
+    try:
+        from ..container_spawner import ContainerSpawner
+    except ImportError:
+        from container_spawner import ContainerSpawner  # type: ignore
+
 logger = get_logger("orchestrator.pipelines")
 
 # Base directory where the gateway creates per-pipeline worktrees.
@@ -1442,6 +1448,119 @@ def _aggregate_review_verdicts(
     return overall, combined
 
 
+def _sync_worktree_with_remote(
+    spawner: "ContainerSpawner",
+    pipeline_id: str,
+    worktree_repo_path: Path,
+) -> None:
+    """Sync a worktree with its remote branch (best-effort).
+
+    After an orchestrator restart, the local worktree branch may be behind
+    the remote: commits pushed during previous phases (contracts, drafts,
+    statefiles) exist on origin but not in the local checkout.  This function
+    fetches those commits and resets the worktree so that all downstream code
+    (contract loading, draft reading, etc.) sees the full pipeline state.
+
+    Only resets if the remote branch exists and local has not diverged —
+    skips the reset when local is ahead of or has diverged from remote.
+    Safe to call on every pipeline start because it is idempotent when the
+    local branch is already up to date.
+    """
+    git_base = ["git", "-c", "core.hooksPath=/dev/null", "-C", str(worktree_repo_path)]
+
+    # Step 1: Authenticated fetch via gateway (gateway holds GitHub credentials)
+    fetch_ok = spawner.gateway.fetch_worktree_branch(
+        pipeline_id=pipeline_id,
+        repo_path=str(worktree_repo_path),
+    )
+    if not fetch_ok:
+        return
+
+    # Step 2: Determine current branch
+    try:
+        result = subprocess.run(
+            [*git_base, "branch", "--show-current"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+        branch = result.stdout.strip()
+        if not branch:
+            return  # Detached HEAD — nothing to sync
+    except Exception:
+        return
+
+    # Step 3: Verify remote tracking branch exists
+    try:
+        result = subprocess.run(
+            [*git_base, "rev-parse", "--verify", f"origin/{branch}"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+        if result.returncode != 0:
+            return  # Remote branch not yet published (first pipeline run)
+    except Exception:
+        return
+
+    # Step 3b: Check if local has diverged from or is ahead of remote.
+    # If local has commits not on remote (e.g., auto-commit hook didn't fire),
+    # skip the reset to avoid discarding local work.
+    try:
+        result = subprocess.run(
+            [*git_base, "rev-list", "--left-right", "--count", f"HEAD...origin/{branch}"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+            check=False,
+        )
+        if result.returncode == 0:
+            parts = result.stdout.strip().split()
+            if len(parts) == 2:
+                local_ahead = int(parts[0])
+                if local_ahead > 0:
+                    logger.info(
+                        "Local branch has commits not on remote — skipping reset",
+                        pipeline_id=pipeline_id,
+                        branch=branch,
+                        local_ahead=local_ahead,
+                    )
+                    return
+    except Exception:
+        pass  # If check fails, proceed with reset (best-effort)
+
+    # Step 4: Reset local branch to remote (local changes from a crashed agent
+    # should already be committed + pushed by the gateway's auto-commit hook)
+    try:
+        result = subprocess.run(
+            [*git_base, "reset", "--hard", f"origin/{branch}"],
+            capture_output=True,
+            text=True,
+            timeout=30,
+            check=False,
+        )
+        if result.returncode != 0:
+            logger.warning(
+                "Failed to reset worktree to remote (continuing with local state)",
+                pipeline_id=pipeline_id,
+                error=result.stderr.strip(),
+            )
+        else:
+            logger.info(
+                "Synced worktree with remote branch",
+                pipeline_id=pipeline_id,
+                branch=branch,
+            )
+    except Exception as sync_err:
+        logger.warning(
+            "Failed to reset worktree to remote (continuing with local state)",
+            pipeline_id=pipeline_id,
+            error=str(sync_err),
+        )
+
+
 def _commit_statefiles_to_worktree(
     worktree_path: Path,
     message: str,
@@ -4115,6 +4234,15 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None:
                 f"worktree creation is required"
             )
 
+        # Sync worktree with remote before starting pipeline phases.  After an
+        # orchestrator restart, the local worktree branch may be behind origin:
+        # commits pushed by agents in previous phases (contracts, drafts,
+        # statefiles) exist on the remote but not in the local checkout.
+        # Fetching and resetting ensures downstream code (contract loading,
+        # draft reading) sees the full pipeline state from prior phases.
+        if worktree_repo_path != repo_path:
+            _sync_worktree_with_remote(spawner, pipeline_id, worktree_repo_path)
+
         # Resolve the certs named volume for gateway CA trust.
         # The docker-compose stack creates ${COMPOSE_PROJECT_NAME:-egg}-certs.
         certs_volume_raw = os.environ.get(

orchestrator/tests/test_gateway_client.py

@@ -67,6 +67,8 @@ def do_POST(self):
             self._handle_worktree_delete(data)
         elif self.path == "/api/v1/git/push":
             self._handle_git_push(data)
+        elif self.path == "/api/v1/git/fetch":
+            self._handle_git_fetch(data)
         else:
             self._send_error(404, "Not found")
 
@@ -229,6 +231,25 @@ def _handle_git_push(self, data):
             }
         )
 
+    def _handle_git_fetch(self, data):
+        """Handle git fetch (POST /api/v1/git/fetch)."""
+        auth_header = self.headers.get("Authorization", "")
+        if not auth_header.startswith("Bearer "):
+            self._send_error(401, "Unauthorized")
+            return
+
+        token = auth_header[7:]
+        if not token:
+            self._send_error(401, "Unauthorized")
+            return
+
+        self._send_json(
+            {
+                "success": True,
+                "message": "Fetch successful",
+            }
+        )
+
     def _send_json(self, data, status=200):
         """Send JSON response."""
         self.send_response(status)
@@ -685,6 +706,43 @@ def test_push_worktree_branch_cleans_up_session(self, gateway_client, mock_gatew
             mock_delete.assert_called_once_with("test-token-12345")
 
 
+class TestFetchWorktreeBranch:
+    """Tests for fetch_worktree_branch method."""
+
+    def test_fetch_worktree_branch_success(self, gateway_client, mock_gateway_server):
+        """Test successful fetch of worktree branch."""
+        result = gateway_client.fetch_worktree_branch(
+            pipeline_id="issue-42",
+            repo_path="/home/egg/.egg-worktrees/issue-42/repo",
+        )
+        assert result is True
+
+    def test_fetch_worktree_branch_gateway_unreachable(self):
+        """Test fetch fails gracefully when gateway is unreachable."""
+        client = GatewayClient(
+            gateway_host="localhost",
+            gateway_port=19999,
+            launcher_secret="test-secret",
+            timeout=1,
+        )
+
+        result = client.fetch_worktree_branch(
+            pipeline_id="issue-42",
+            repo_path="/some/path",
+        )
+        assert result is False
+
+    def test_fetch_worktree_branch_cleans_up_session(self, gateway_client, mock_gateway_server):
+        """Test that temp session is cleaned up after fetch."""
+        with patch.object(gateway_client, "delete_session") as mock_delete:
+            gateway_client.fetch_worktree_branch(
+                pipeline_id="issue-42",
+                repo_path="/some/path",
+            )
+            # Session should be cleaned up
+            mock_delete.assert_called_once_with("test-token-12345")
+
+
 class TestSingletonClient:
     """Tests for singleton client."""