diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py index 853534508d..4429ba6d0e 100644 --- a/orchestrator/gateway_client.py +++ b/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.""" diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 56fa1342e6..3b0a298b1c 100644 --- a/orchestrator/routes/pipelines.py +++ b/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( diff --git a/orchestrator/tests/test_gateway_client.py b/orchestrator/tests/test_gateway_client.py index bd3770d1c7..d259b76ca9 100644 --- a/orchestrator/tests/test_gateway_client.py +++ b/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.""" diff --git a/orchestrator/tests/test_sync_worktree.py b/orchestrator/tests/test_sync_worktree.py new file mode 100644 index 0000000000..10597880a0 --- /dev/null +++ b/orchestrator/tests/test_sync_worktree.py @@ -0,0 +1,153 @@ +"""Tests for _sync_worktree_with_remote.""" + +import subprocess +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +_orchestrator_path = Path(__file__).parent.parent +if str(_orchestrator_path) not in sys.path: + sys.path.insert(0, str(_orchestrator_path)) + +_shared_path = Path(__file__).parent.parent.parent / "shared" +if _shared_path.exists() and str(_shared_path) not in sys.path: + sys.path.insert(0, str(_shared_path)) + +# Mock docker before importing modules that depend on it +sys.modules.setdefault("docker", MagicMock()) +sys.modules.setdefault("docker.errors", MagicMock()) +sys.modules.setdefault("docker.types", MagicMock()) + +from routes.pipelines import _sync_worktree_with_remote + + +def _make_spawner(fetch_ok: bool = True) -> MagicMock: + """Create a mock spawner with gateway.fetch_worktree_branch.""" + spawner = MagicMock() + spawner.gateway.fetch_worktree_branch.return_value = fetch_ok + return spawner + + +def _make_subprocess_result( + returncode: int = 0, + stdout: str = "", + stderr: str = "", +) -> subprocess.CompletedProcess: + return subprocess.CompletedProcess( + args=[], returncode=returncode, stdout=stdout, stderr=stderr + ) + + +class TestSyncWorktreeWithRemote: + """Tests for _sync_worktree_with_remote.""" + + def test_returns_early_when_fetch_fails(self): + """If gateway fetch fails, function returns without running git commands.""" + spawner = _make_spawner(fetch_ok=False) + with patch("routes.pipelines.subprocess.run") as mock_run: + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + mock_run.assert_not_called() + + def test_returns_early_on_detached_head(self): + """If branch --show-current returns empty, function returns.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + # Step 2: empty branch (detached HEAD) + mock_run.return_value = _make_subprocess_result(stdout="") + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + # Only step 2 should have been called + assert mock_run.call_count == 1 + + def test_returns_early_when_remote_branch_missing(self): + """If origin/{branch} does not exist, function returns.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + # Step 2: branch name + _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse fails (remote branch missing) + _make_subprocess_result(returncode=128), + ] + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert mock_run.call_count == 2 + + def test_skips_reset_when_local_ahead(self): + """If local has commits not on remote, skip the reset.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + # Step 2: branch name + _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse succeeds + _make_subprocess_result(returncode=0), + # Step 3b: local is 2 ahead, 0 behind + _make_subprocess_result(stdout="2\t0\n"), + ] + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + # Should NOT proceed to step 4 (reset) + assert mock_run.call_count == 3 + + def test_skips_reset_when_local_diverged(self): + """If local has diverged from remote (ahead AND behind), skip the reset.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + # Step 2: branch name + _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse succeeds + _make_subprocess_result(returncode=0), + # Step 3b: local is 2 ahead, 3 behind (diverged) + _make_subprocess_result(stdout="2\t3\n"), + ] + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + # Should NOT proceed to step 4 (reset) — local_ahead > 0 + assert mock_run.call_count == 3 + + def test_successful_reset(self): + """Happy path: fetch, detect branch, verify remote, reset.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = [ + # Step 2: branch name + _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse succeeds + _make_subprocess_result(returncode=0), + # Step 3b: local is 0 ahead, 3 behind + _make_subprocess_result(stdout="0\t3\n"), + # Step 4: reset succeeds + _make_subprocess_result(returncode=0), + ] + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + assert mock_run.call_count == 4 + # Verify step 4 was git reset --hard + reset_call = mock_run.call_args_list[3] + assert "reset" in reset_call[0][0] + assert "--hard" in reset_call[0][0] + + def test_logs_warning_on_failed_reset(self): + """If git reset --hard fails, a warning is logged.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run, \ + patch("routes.pipelines.logger") as mock_logger: + mock_run.side_effect = [ + # Step 2: branch name + _make_subprocess_result(stdout="egg/issue-42\n"), + # Step 3: rev-parse succeeds + _make_subprocess_result(returncode=0), + # Step 3b: local is 0 ahead, 1 behind + _make_subprocess_result(stdout="0\t1\n"), + # Step 4: reset fails + _make_subprocess_result(returncode=1, stderr="permission denied"), + ] + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo")) + mock_logger.warning.assert_called() + warning_msg = mock_logger.warning.call_args[0][0] + assert "Failed to reset" in warning_msg + + def test_handles_subprocess_timeout(self): + """If subprocess raises TimeoutExpired, function handles gracefully.""" + spawner = _make_spawner() + with patch("routes.pipelines.subprocess.run") as mock_run: + mock_run.side_effect = subprocess.TimeoutExpired(cmd="git", timeout=10) + # Should not raise + _sync_worktree_with_remote(spawner, "pipe-1", Path("/tmp/repo"))