Fix contract lookup for local-mode pipelines

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
 

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/<pipeline_id>/<repo>/``.
+    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

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("/<pipeline_id>/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()
 

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)