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..3cdc290f94 100644 --- a/integration_tests/local_pipeline/docker-compose.yml +++ b/integration_tests/local_pipeline/docker-compose.yml @@ -86,6 +86,11 @@ 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). + # 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"] interval: 5s diff --git a/orchestrator/routes/pipelines.py b/orchestrator/routes/pipelines.py index 729a51572a..7c11cd783c 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 ( @@ -385,33 +389,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 +433,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 +1671,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 +1692,27 @@ 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. + # 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 + 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 + if candidate.exists(): + worktree_repo_path = candidate + break + logger.info( "Worktrees created for pipeline", pipeline_id=pipeline_id, @@ -1781,6 +1753,46 @@ 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.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) @@ -1983,7 +1995,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 +2075,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 +2160,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, @@ -2214,16 +2226,20 @@ 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( - 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,