From 7fe4c7cc06fbb0d46f6125ddc73768e4d3b0f4f2 Mon Sep 17 00:00:00 2001 From: egg Date: Fri, 13 Feb 2026 07:27:27 +0000 Subject: [PATCH 1/3] Fix orchestrator reading verdict files from worktree path --- docker-compose.yml | 2 + .../local_pipeline/docker-compose.yml | 2 + orchestrator/routes/pipelines.py | 119 +++++++++--------- 3 files changed, 60 insertions(+), 63 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index 76b23bee6a..0cabdf1c31 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -129,6 +129,8 @@ services: - state:/home/egg/.egg-state # Docker socket for container management - /var/run/docker.sock:/var/run/docker.sock + # Worktrees directory (read container-written artifacts: verdicts, drafts, checks) + - ${HOST_HOME:-/home/egg}/.egg-worktrees:/home/egg/.egg-worktrees healthcheck: test: ["CMD", "curl", "-sf", "http://localhost:9849/api/v1/health"] interval: 10s diff --git a/integration_tests/local_pipeline/docker-compose.yml b/integration_tests/local_pipeline/docker-compose.yml index 6f58f24d27..b0a7e847e5 100644 --- a/integration_tests/local_pipeline/docker-compose.yml +++ b/integration_tests/local_pipeline/docker-compose.yml @@ -86,6 +86,8 @@ services: # Per-repo mounts are added via override file (generated by conftest.py) - state:/home/egg/.egg-state - /var/run/docker.sock:/var/run/docker.sock + # Worktrees directory (read container-written artifacts: verdicts, drafts, checks) + - worktrees:/home/egg/.egg-worktrees healthcheck: test: ["CMD", "curl", "-sf", "http://localhost:9849/api/v1/health"] interval: 5s diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 729a51572a..2da4063c2e 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -385,33 +385,8 @@ def create_pipeline() -> tuple[Response, int]: prompt=prompt, ) - # Create companion contract — required for pipeline to function - from egg_contracts.loader import create_local_contract - - try: - create_local_contract( - pipeline_id=pipeline.id, - title=prompt[:100], - repo_root=repo_path, - ) - except Exception as contract_err: - # Clean up the pipeline we just created - store.delete_pipeline(pipeline.id) - logger.error( - "Failed to create contract for local pipeline", - pipeline_id=pipeline.id, - error=str(contract_err), - ) - return make_error_response( - f"Failed to create contract: {contract_err}", - status_code=500, - ) - pipeline.contract_synced = True - store.save_pipeline(pipeline, commit=False) - logger.info( - "Local pipeline contract created", - pipeline_id=pipeline.id, - ) + # Contract creation is deferred to _run_pipeline so it writes + # into the per-pipeline worktree instead of the main repo. logger.info( "Local pipeline created", @@ -454,37 +429,8 @@ def create_pipeline() -> tuple[Response, int]: mode="issue", ) - # Create companion contract — required for pipeline to function - from egg_contracts.loader import create_contract - - issue_url = f"https://github.com/{repo}/issues/{issue_number}" - try: - create_contract( - issue_number=issue_number, - title=f"Issue #{issue_number}", - url=issue_url, - repo_root=repo_path, - ) - except Exception as contract_err: - # Clean up the pipeline we just created - store.delete_pipeline(pipeline.id) - logger.error( - "Failed to create contract for issue pipeline", - pipeline_id=pipeline.id, - issue_number=issue_number, - error=str(contract_err), - ) - return make_error_response( - f"Failed to create contract: {contract_err}", - status_code=500, - ) - pipeline.contract_synced = True - store.save_pipeline(pipeline, commit=False) - logger.info( - "Issue pipeline contract created", - pipeline_id=pipeline.id, - issue_number=issue_number, - ) + # Contract creation is deferred to _run_pipeline so it writes + # into the per-pipeline worktree instead of the main repo. logger.info( "Pipeline created", @@ -1721,6 +1667,7 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # containers in the pipeline share the same working trees. worktree_id = pipeline_id repo_volumes = dict(host_repo_map) # fallback: raw host paths + worktree_repo_path = repo_path # default; overridden when worktrees exist host_uid = int(os.environ.get("HOST_UID", 1000)) host_gid = int(os.environ.get("HOST_GID", 1000)) pipeline_repos = [pipeline.repo] if pipeline.repo else [] @@ -1741,6 +1688,16 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # stripping the owner prefix from "owner/repo" format. This matches # the container mount target at /home/egg/repos/. repo_volumes = wt_result.worktrees + + # Derive the orchestrator-accessible worktree path. + # Reviewer containers write verdict/draft/check files into + # the worktree, so the orchestrator must read from there. + for name in wt_result.worktrees: + candidate = Path(f"/home/egg/.egg-worktrees/{worktree_id}/{name}") + if candidate.exists(): + worktree_repo_path = candidate + break + logger.info( "Worktrees created for pipeline", pipeline_id=pipeline_id, @@ -1781,6 +1738,42 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: else: certs_volume = certs_volume_raw + # Create companion contract in the worktree (deferred from pipeline + # creation so it doesn't pollute the main repo working directory). + if not pipeline.contract_synced: + try: + if pipeline_mode == "local": + from egg_contracts.loader import create_local_contract + + create_local_contract( + pipeline_id=pipeline.id, + title=(pipeline.prompt or "")[:100], + repo_root=worktree_repo_path, + ) + else: + from egg_contracts.loader import create_contract + + issue_url = f"https://github.com/{pipeline.repo}/issues/{pipeline.issue_number}" + create_contract( + issue_number=pipeline.issue_number, + title=f"Issue #{pipeline.issue_number}", + url=issue_url, + repo_root=worktree_repo_path, + ) + pipeline.contract_synced = True + store.save_pipeline(pipeline, commit=False) + logger.info( + "Pipeline contract created in worktree", + pipeline_id=pipeline_id, + mode=pipeline_mode, + ) + except Exception as contract_err: + logger.warning( + "Failed to create contract in worktree, continuing without contract", + pipeline_id=pipeline_id, + error=str(contract_err), + ) + while True: pipeline = store.load_pipeline(pipeline_id) @@ -1983,7 +1976,7 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: ) break - check_results = _read_check_results(repo_path) + check_results = _read_check_results(worktree_repo_path) if check_results is None or check_results.get("all_passed"): logger.info( "All checks passed", @@ -2063,7 +2056,7 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: pipeline.issue_number, pipeline_id, ) - verdict_path = repo_path / verdict_rel + verdict_path = worktree_repo_path / verdict_rel if verdict_path.exists(): try: verdict_path.unlink() @@ -2148,7 +2141,7 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # Read this reviewer's verdict all_verdicts[reviewer_type] = _read_review_verdict( - repo_path, + worktree_repo_path, current_phase.value, reviewer_type=reviewer_type, pipeline_mode=pipeline_mode, @@ -2217,13 +2210,13 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # After plan phase: populate contract with task structure if current_phase.value == "plan": _populate_contract_from_plan( - repo_path, pipeline_id, pipeline_mode, pipeline.issue_number + worktree_repo_path, pipeline_id, pipeline_mode, pipeline.issue_number ) # --- HITL gate: pause for human approval --- if pipeline.config.hitl_gates and current_phase.value in _HITL_GATE_PHASES: draft_content = _read_phase_draft( - repo_path, + worktree_repo_path, current_phase.value, pipeline_mode, pipeline.issue_number, From b69e176963c78bb79b3c797bb817d7ae4c7458c6 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Fri, 13 Feb 2026 07:38:55 +0000 Subject: [PATCH 2/3] Address review feedback on worktree verdict path fix - Fail pipeline on contract creation error instead of silently continuing (was downgraded to warning, now properly sets FAILED status and returns) - Match worktree path against pipeline.repo explicitly instead of taking first arbitrary dict entry, with fallback for safety - Extract hardcoded worktree base path as WORKTREE_BASE_DIR constant - Add clarifying comments for path coupling and volume mount divergence Authored-by: egg --- .../local_pipeline/docker-compose.yml | 5 ++- orchestrator/routes/pipelines.py | 33 +++++++++++++++---- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/integration_tests/local_pipeline/docker-compose.yml b/integration_tests/local_pipeline/docker-compose.yml index b0a7e847e5..3cdc290f94 100644 --- a/integration_tests/local_pipeline/docker-compose.yml +++ b/integration_tests/local_pipeline/docker-compose.yml @@ -86,7 +86,10 @@ services: # Per-repo mounts are added via override file (generated by conftest.py) - state:/home/egg/.egg-state - /var/run/docker.sock:/var/run/docker.sock - # Worktrees directory (read container-written artifacts: verdicts, drafts, checks) + # Worktrees directory (read container-written artifacts: verdicts, drafts, checks). + # Uses a named volume (not a bind mount) because tests run in CI without + # a host filesystem — unlike production docker-compose which bind-mounts + # ${HOST_HOME}/.egg-worktrees. - worktrees:/home/egg/.egg-worktrees healthcheck: test: ["CMD", "curl", "-sf", "http://localhost:9849/api/v1/health"] diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 2da4063c2e..e41ded2489 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -55,6 +55,10 @@ def get_logger(name: str, **kwargs) -> logging.Logger: # type: ignore[misc] logger = get_logger("orchestrator.pipelines") +# Base directory where the gateway creates per-pipeline worktrees. +# Must match the gateway's WORKTREE_BASE_DIR and docker-compose volume mounts. +WORKTREE_BASE_DIR = Path("/home/egg/.egg-worktrees") + # Network constants for sandbox container URLs try: from egg_config import ( @@ -1692,11 +1696,20 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # Derive the orchestrator-accessible worktree path. # Reviewer containers write verdict/draft/check files into # the worktree, so the orchestrator must read from there. - for name in wt_result.worktrees: - candidate = Path(f"/home/egg/.egg-worktrees/{worktree_id}/{name}") + # Match against pipeline.repo explicitly to avoid picking + # the wrong repo in multi-repo pipelines. + repo_short = pipeline.repo.split("/")[-1] if pipeline.repo else None + if repo_short and repo_short in wt_result.worktrees: + candidate = WORKTREE_BASE_DIR / worktree_id / repo_short if candidate.exists(): worktree_repo_path = candidate - break + else: + # Fallback: take the first existing worktree path + for name in wt_result.worktrees: + candidate = WORKTREE_BASE_DIR / worktree_id / name + if candidate.exists(): + worktree_repo_path = candidate + break logger.info( "Worktrees created for pipeline", @@ -1768,11 +1781,15 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: mode=pipeline_mode, ) except Exception as contract_err: - logger.warning( - "Failed to create contract in worktree, continuing without contract", + logger.error( + "Failed to create contract in worktree", pipeline_id=pipeline_id, error=str(contract_err), ) + pipeline.status = PipelineStatus.FAILED + pipeline.error = f"Failed to create contract: {contract_err}" + store.save_pipeline(pipeline) + return while True: pipeline = store.load_pipeline(pipeline_id) @@ -2207,7 +2224,11 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: message=f"Phase {current_phase.value} completed", ) - # After plan phase: populate contract with task structure + # After plan phase: populate contract with task structure. + # NOTE: worktree_repo_path is used for both draft reads and + # contract load/save inside _populate_contract_from_plan. + # The contract was created at worktree_repo_path above, so + # both operations must use the same path. if current_phase.value == "plan": _populate_contract_from_plan( worktree_repo_path, pipeline_id, pipeline_mode, pipeline.issue_number From 2936cc79238dae31e768c82a9fac6659d387d666 Mon Sep 17 00:00:00 2001 From: "egg-reviewer[bot]" <261018737+egg-reviewer[bot]@users.noreply.github.com> Date: Fri, 13 Feb 2026 07:50:25 +0000 Subject: [PATCH 3/3] Fix worktree path resolution dead branch in fallback logic When repo_short matched in wt_result.worktrees but candidate.exists() returned False, worktree_repo_path silently remained as repo_path because the else-fallback was attached to the outer if, not the inner exists() check. Use a matched flag so the fallback loop runs whenever the primary match fails for any reason. Authored-by: egg --- orchestrator/routes/pipelines.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index e41ded2489..7c11cd783c 100644 --- a/orchestrator/routes/pipelines.py +++ b/orchestrator/routes/pipelines.py @@ -1699,11 +1699,13 @@ def _run_pipeline(pipeline_id: str, repo_path: Path) -> None: # Match against pipeline.repo explicitly to avoid picking # the wrong repo in multi-repo pipelines. repo_short = pipeline.repo.split("/")[-1] if pipeline.repo else None + matched = False if repo_short and repo_short in wt_result.worktrees: candidate = WORKTREE_BASE_DIR / worktree_id / repo_short if candidate.exists(): worktree_repo_path = candidate - else: + matched = True + if not matched: # Fallback: take the first existing worktree path for name in wt_result.worktrees: candidate = WORKTREE_BASE_DIR / worktree_id / name