Enforce contract existence before implement phase

.gitignore

@@ -31,6 +31,11 @@ secrets.yaml
 .env.*.local
 # Note: .env may be generated at runtime by egg --compose
 
+# Claude Code
+.claude/worktrees/
+.claude/settings.local.json
+.claude/scheduled_tasks.lock
+
 # IDE
 .idea/
 .vscode/

orchestrator/routes/coordinator.py

@@ -694,7 +694,6 @@ def advance_phase(pipeline_id: str) -> tuple[Response, int]:
                         f"Target phase '{target_phase_str}' is the current phase",
                         status_code=400,
                     )
-                pipeline.current_phase = target_phase
             else:
                 # Advance to next phase in sequence
                 current_idx = phase_order.index(pipeline.current_phase)
@@ -705,7 +704,26 @@ def advance_phase(pipeline_id: str) -> tuple[Response, int]:
                     )
                 target_phase = phase_order[current_idx + 1]
                 action = "advance"
-                pipeline.current_phase = target_phase
+
+            # Enforce contract existence before entering implement phase.
+            # Every pipeline — simple or complex — must have a contract so
+            # reviewers, phase gates, and the reviewer_contract role have a
+            # shared source of truth about what is being built.
+            if (
+                target_phase in (PipelinePhase.IMPLEMENT, PipelinePhase.PR)
+                and not pipeline.contract_synced
+            ):
+                return make_error_response(
+                    f"Cannot advance to '{target_phase.value}' phase: no contract exists for this pipeline. "
+                    "A contract must be created before implementation can begin. "
+                    "The orchestrator creates contracts automatically during pipeline "
+                    "startup — if contract_synced is still false, the contract creation "
+                    "may have failed. Check pipeline logs for details.",
+                    status_code=409,
+                )
+
+            # All validations passed — now mutate state
+            pipeline.current_phase = target_phase
 
             # Record the phase decision in coordinator state
             if pipeline.coordinator_state is None:

orchestrator/tests/test_coordinator_routes_functional.py

@@ -819,6 +819,163 @@ def test_phase_advance_coordinator_disabled(self, mock_repo, mock_lock, mock_sto
         assert response.status_code == 403
 
 
+# ── Contract enforcement tests ─────────────────────────────────────
+
+
+class TestContractEnforcement:
+    """Phase advancement must be blocked when no contract exists."""
+
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_advance_to_implement_blocked_without_contract(
+        self, mock_repo, mock_lock, mock_store_fn, client
+    ):
+        """Advancing from plan to implement must fail if contract_synced is False."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.PLAN)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"reason": "Plan approved"},
+        )
+
+        assert response.status_code == 409
+        assert "contract" in response.get_json()["message"].lower()
+
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_skip_to_implement_blocked_without_contract(
+        self, mock_repo, mock_lock, mock_store_fn, client
+    ):
+        """Skipping directly to implement must fail if contract_synced is False."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.REFINE)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"target_phase": "implement", "reason": "Skip to implement"},
+        )
+
+        assert response.status_code == 409
+        assert "contract" in response.get_json()["message"].lower()
+
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_skip_to_pr_blocked_without_contract(self, mock_repo, mock_lock, mock_store_fn, client):
+        """Skipping to PR phase must also fail if contract_synced is False."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.REFINE)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"target_phase": "pr", "reason": "Skip everything"},
+        )
+
+        assert response.status_code == 409
+        assert "contract" in response.get_json()["message"].lower()
+
+    @patch("routes.coordinator.emit_event")
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_advance_to_implement_allowed_with_contract(
+        self, mock_repo, mock_lock, mock_store_fn, mock_emit, client
+    ):
+        """Advancing to implement succeeds when contract_synced is True."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.PLAN)
+        pipeline.contract_synced = True
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"reason": "Plan approved"},
+        )
+
+        assert response.status_code == 200
+        assert response.get_json()["data"]["current_phase"] == "implement"
+
+    @patch("routes.coordinator.emit_event")
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_advance_to_plan_allowed_without_contract(
+        self, mock_repo, mock_lock, mock_store_fn, mock_emit, client
+    ):
+        """Advancing to plan (before implement) is allowed without a contract."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.REFINE)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"reason": "Refine complete"},
+        )
+
+        assert response.status_code == 200
+        assert response.get_json()["data"]["current_phase"] == "plan"
+
+    @patch("routes.coordinator.get_state_store")
+    @patch("routes.coordinator.get_pipeline_state_lock")
+    @patch("routes.coordinator.get_repo_path")
+    def test_loopback_to_implement_blocked_without_contract(
+        self, mock_repo, mock_lock, mock_store_fn, client
+    ):
+        """Looping back from PR to implement must fail if contract_synced is False."""
+        mock_repo.return_value = Path("/tmp/repo")
+        mock_lock.return_value.__enter__ = MagicMock()
+        mock_lock.return_value.__exit__ = MagicMock(return_value=False)
+
+        pipeline = _make_pipeline(phase=PipelinePhase.PR)
+        pipeline.contract_synced = False
+        store = MagicMock()
+        store.load_pipeline.return_value = pipeline
+        mock_store_fn.return_value = store
+
+        response = client.post(
+            "/api/v1/pipelines/test-pipeline/coordinator/phase",
+            json={"target_phase": "implement", "reason": "Need more implementation"},
+        )
+
+        assert response.status_code == 409
+        assert "contract" in response.get_json()["message"].lower()
+
+
 # ── Escalation endpoint tests ───────────────────────────────────────
 
 

sandbox/.claude/rules/coordinator.md

@@ -28,6 +28,14 @@ Also use standard commands:
 - `egg-orch signal heartbeat` — Send heartbeat
 - `egg-orch message send --to all --type STATUS --subject "..." --body "..."` — Broadcast status
 
+## Contract Requirement (CRITICAL)
+
+Every pipeline MUST have a contract before implementation begins. The orchestrator enforces this: phase advancement to `implement` will be **rejected** if no contract exists.
+
+Contracts are created automatically by the orchestrator during pipeline startup. If contract creation fails, the pipeline will be marked as FAILED. You do not need to create contracts yourself, but you must ensure the pipeline has one before advancing to implement.
+
+If you encounter a contract enforcement error when advancing phases, check `egg-orch coordinator state` to verify `contract_synced` is true. If not, the pipeline setup failed and needs investigation.
+
 ## Phase-Role Mappings (CRITICAL)
 
 When spawning agents, you MUST use the correct roles for the current phase. The orchestrator validates role-phase alignment and will reject mismatches. Primary agents and reviewers run in parallel:

scripts/check-hardcoded-ports.py

@@ -170,10 +170,10 @@ def main() -> int:
         if file_path.suffix.lower() not in extensions:
             continue
 
-        # Skip files in .git, .venv, node_modules
+        # Skip files in .git, .venv, node_modules, .claude (worktrees)
         rel_parts = file_path.relative_to(repo_root).parts
         if any(
-            part in {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache"}
+            part in {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache", ".claude"}
             for part in rel_parts
         ):
             continue