From 46e741919ef953eb45da15699b23042ac7143a8e Mon Sep 17 00:00:00 2001 From: James Wiesebron Date: Sat, 21 Feb 2026 12:47:34 -0800 Subject: [PATCH 1/3] Sync worktree with remote before pipeline phase execution MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit After an orchestrator restart, the local worktree branch can lag behind origin: commits pushed by agents in previous phases (contracts, drafts, statefiles) only exist on the remote. The next pipeline start would find the local checkout at main HEAD, causing contract loads and draft reads to fail even though contract_synced=True. Fix by adding _sync_worktree_with_remote() which runs a gateway-authenticated git fetch followed by a local git reset --hard origin/ at the start of _run_pipeline, before the contract sync check. The operation is best-effort and idempotent — it skips gracefully when the remote branch doesn't yet exist (first run) or when fetch fails. Also adds GatewayClient.fetch_worktree_branch(), modeled on the existing push_worktree_branch(), using the /api/v1/git/fetch gateway endpoint. Fixes the immediate issue-738 failure: ContractNotFoundError: Contract for 738 not found at /home/egg/.egg-worktrees/issue-738/egg/.egg-state/contracts/738.json Co-Authored-By: Claude Sonnet 4.6 --- orchestrator/gateway_client.py | 60 ++++++++++++++++++++++ orchestrator/routes/pipelines.py | 86 ++++++++++++++++++++++++++++++++ 2 files changed, 146 insertions(+) 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..dcd3eee15a 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -1442,6 +1442,83 @@ def _aggregate_review_verdicts( return overall, combined +def _sync_worktree_with_remote( + spawner: object, + 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 is strictly ahead of local — + 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, + timeout=10, + check=False, + ) + if result.returncode != 0: + return # Remote branch not yet published (first pipeline run) + except Exception: + return + + # 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: + subprocess.run( + [*git_base, "reset", "--hard", f"origin/{branch}"], + capture_output=True, + timeout=30, + check=False, + ) + 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 +4192,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( From 91105e3325c74c3a35fc1de943ffe3617a4c7899 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 21 Feb 2026 20:57:40 +0000 Subject: [PATCH 2/3] Address review feedback on worktree sync PR - Check git reset --hard returncode and log warning on failure (issue 1) - Add text=True to all subprocess calls for consistency (issues 1, 5) - Add ahead/behind check to skip reset when local has diverged (issue 2) - Update docstring to match actual behavior (issue 2) - Fix spawner type annotation from object to ContainerSpawner (issue 3) - Add tests for fetch_worktree_branch and _sync_worktree_with_remote (issue 4) --- orchestrator/routes/pipelines.py | 62 ++++++++-- orchestrator/tests/test_gateway_client.py | 58 +++++++++ orchestrator/tests/test_sync_worktree.py | 137 ++++++++++++++++++++++ 3 files changed, 247 insertions(+), 10 deletions(-) create mode 100644 orchestrator/tests/test_sync_worktree.py diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index dcd3eee15a..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. @@ -1443,7 +1449,7 @@ def _aggregate_review_verdicts( def _sync_worktree_with_remote( - spawner: object, + spawner: "ContainerSpawner", pipeline_id: str, worktree_repo_path: Path, ) -> None: @@ -1455,8 +1461,9 @@ def _sync_worktree_with_remote( 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 is strictly ahead of local — - safe to call on every pipeline start because it is idempotent when the + 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)] @@ -1489,6 +1496,7 @@ def _sync_worktree_with_remote( result = subprocess.run( [*git_base, "rev-parse", "--verify", f"origin/{branch}"], capture_output=True, + text=True, timeout=10, check=False, ) @@ -1497,20 +1505,54 @@ def _sync_worktree_with_remote( 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: - subprocess.run( + result = subprocess.run( [*git_base, "reset", "--hard", f"origin/{branch}"], capture_output=True, + text=True, timeout=30, check=False, ) - logger.info( - "Synced worktree with remote branch", - pipeline_id=pipeline_id, - branch=branch, - ) + 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)", 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..6a38097462 --- /dev/null +++ b/orchestrator/tests/test_sync_worktree.py @@ -0,0 +1,137 @@ +"""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_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")) From cb0f6392b96caad342c7c8a18711797b9640fc8d Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Sat, 21 Feb 2026 21:07:42 +0000 Subject: [PATCH 3/3] Add test for diverged branch case in worktree sync --- orchestrator/tests/test_sync_worktree.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/orchestrator/tests/test_sync_worktree.py b/orchestrator/tests/test_sync_worktree.py index 6a38097462..10597880a0 100644 --- a/orchestrator/tests/test_sync_worktree.py +++ b/orchestrator/tests/test_sync_worktree.py @@ -87,6 +87,22 @@ def test_skips_reset_when_local_ahead(self): # 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()