diff --git a/orchestrator/dispatch.py b/orchestrator/dispatch.py index af2723aee9..927b7ffaec 100644 --- a/orchestrator/dispatch.py +++ b/orchestrator/dispatch.py @@ -105,11 +105,22 @@ def __init__(self, pipeline: Pipeline, repo_path: Path): self.repo_path = repo_path self._contract_orchestrator: ContractOrchestrator | None = None + @property + def contract_key(self) -> int | str: + """Return the contract identifier for this pipeline. + + Issue-mode pipelines use issue_number; local-mode pipelines use + the pipeline ID (e.g. ``local-47601d1d``). + """ + if self.pipeline.issue_number is not None: + return self.pipeline.issue_number + return self.pipeline.id + @property def contract_orchestrator(self) -> ContractOrchestrator: """Get or create the contract orchestrator.""" if self._contract_orchestrator is None: - contract = load_contract(self.pipeline.issue_number, self.repo_path) + contract = load_contract(self.contract_key, self.repo_path) self._contract_orchestrator = create_orchestrator(contract) return self._contract_orchestrator diff --git a/orchestrator/routes/__init__.py b/orchestrator/routes/__init__.py index 3c71a42aba..a4f5e18eeb 100644 --- a/orchestrator/routes/__init__.py +++ b/orchestrator/routes/__init__.py @@ -111,3 +111,52 @@ def resolve_repo_path_for_pipeline(pipeline_id: str, base_path: Path) -> Path: ) return base_path + + +# Must match the gateway's WORKTREE_BASE_DIR and docker-compose volume mounts. +_WORKTREE_BASE_DIR = Path("/home/egg/.egg-worktrees") + + +def resolve_worktree_path(pipeline_id: str, repo_path: Path) -> Path: + """Resolve the worktree repo path for a pipeline. + + Contracts and other container-written files live in per-pipeline + worktrees at ``/home/egg/.egg-worktrees///``. + This helper checks for a worktree and returns it when present, + falling back to ``repo_path`` otherwise (e.g. when worktrees have + already been cleaned up or were never created). + + Args: + pipeline_id: Pipeline ID (e.g. ``issue-546``) + repo_path: Main repo path (e.g. ``/home/egg/repos/egg``) + + Returns: + Worktree path if it exists, otherwise ``repo_path`` + """ + wt_pipeline_dir = _WORKTREE_BASE_DIR / pipeline_id + if not wt_pipeline_dir.is_dir(): + return repo_path + + # Match by repo directory name (last component of repo_path) + repo_name = repo_path.name + candidate = wt_pipeline_dir / repo_name + if candidate.is_dir(): + return candidate + + # Fallback: take the first existing subdirectory. + # iterdir() order is non-deterministic; log a warning so operators + # can detect when the heuristic fires (e.g. after a repo rename). + try: + for entry in wt_pipeline_dir.iterdir(): + if entry.is_dir(): + logger.warning( + "Worktree repo name mismatch, using fallback", + pipeline_id=pipeline_id, + expected_repo=repo_name, + fallback_path=str(entry), + ) + return entry + except OSError: + pass + + return repo_path diff --git a/orchestrator/routes/signals.py b/orchestrator/routes/signals.py index 9d41f49fed..cab77c4188 100644 --- a/orchestrator/routes/signals.py +++ b/orchestrator/routes/signals.py @@ -64,7 +64,11 @@ def make_success_response( return jsonify(response), 200 -from routes import get_repo_path, resolve_repo_path_for_pipeline # noqa: E402 — shared helper +from routes import ( # noqa: E402 — shared helper + get_repo_path, + resolve_repo_path_for_pipeline, + resolve_worktree_path, +) @signals_bp.route("//signal", methods=["POST"]) @@ -156,8 +160,11 @@ def handle_complete_signal( store = get_state_store(repo_path) pipeline = store.load_pipeline(pipeline_id) + # Contracts live in per-pipeline worktrees, not the main repo. + contract_path = resolve_worktree_path(pipeline_id, repo_path) + # Create dispatcher and record completion - dispatcher = create_dispatcher(pipeline, repo_path) + dispatcher = create_dispatcher(pipeline, contract_path) commit = data.get("commit") outputs = data.get("handoff_data", {}) @@ -174,7 +181,7 @@ def handle_complete_signal( handoff_data=outputs, metrics=data.get("metrics", {}), ) - save_agent_output(repo_path, output) + save_agent_output(contract_path, output) logger.info( "Agent completed", @@ -280,8 +287,11 @@ def handle_error_signal( store = get_state_store(repo_path) pipeline = store.load_pipeline(pipeline_id) + # Contracts live in per-pipeline worktrees, not the main repo. + contract_path = resolve_worktree_path(pipeline_id, repo_path) + # Mark agent as failed - dispatcher = create_dispatcher(pipeline, repo_path) + dispatcher = create_dispatcher(pipeline, contract_path) dispatcher.fail_agent(agent_role, error_message) dispatcher.save_contract() diff --git a/orchestrator/tests/test_dispatch.py b/orchestrator/tests/test_dispatch.py new file mode 100644 index 0000000000..98bfe75891 --- /dev/null +++ b/orchestrator/tests/test_dispatch.py @@ -0,0 +1,48 @@ +""" +Tests for PipelineDispatcher. +""" + +from pathlib import Path + +from dispatch import PipelineDispatcher +from models import Pipeline + + +class TestContractKey: + """Tests for PipelineDispatcher.contract_key.""" + + def test_issue_mode_returns_issue_number(self): + """Issue-mode pipelines use the issue number as contract key.""" + pipeline = Pipeline( + id="issue-496", + issue_number=496, + repo="owner/repo", + branch="egg/issue-496", + ) + dispatcher = PipelineDispatcher(pipeline, Path("/tmp/repo")) + assert dispatcher.contract_key == 496 + + def test_local_mode_returns_pipeline_id(self): + """Local-mode pipelines use the pipeline ID as contract key.""" + pipeline = Pipeline( + id="local-47601d1d", + repo="owner/repo", + branch="egg/local-47601d1d", + ) + dispatcher = PipelineDispatcher(pipeline, Path("/tmp/repo")) + assert dispatcher.contract_key == "local-47601d1d" + + def test_issue_mode_returns_int(self): + """Issue-mode contract key is an int.""" + pipeline = Pipeline( + id="issue-42", + issue_number=42, + ) + dispatcher = PipelineDispatcher(pipeline, Path("/tmp/repo")) + assert isinstance(dispatcher.contract_key, int) + + def test_local_mode_returns_str(self): + """Local-mode contract key is a str.""" + pipeline = Pipeline(id="local-abc123") + dispatcher = PipelineDispatcher(pipeline, Path("/tmp/repo")) + assert isinstance(dispatcher.contract_key, str)