[issue-3393][slice-1/6] Repo dimension in the persisted contract...

.egg/schemas/contract.schema.json

@@ -14,7 +14,7 @@
       "type": "string",
       "description": "Schema version for migrations",
       "pattern": "^[0-9]+\\.[0-9]+$",
-      "default": "1.3"
+      "default": "1.4"
     },
     "issue": {
       "oneOf": [

orchestrator/models.py

@@ -12,7 +12,7 @@
 from typing import Any, Literal, NamedTuple
 
 from agent_model_resolution import OVERSEER_TIER_MODELS
-from egg_contracts.models import PipelinePhase
+from egg_contracts.models import PipelinePhase, Slice
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 from slice_id_validation import SLICE_ID_PATTERN
 
@@ -1117,6 +1117,32 @@ def validate_concurrent_phases(cls, v: list[str]) -> list[str]:
         return v
 
 
+class RepoSpec(BaseModel):
+    """One repository a pipeline operates in (#3393, multi-repo pipelines).
+
+    A pipeline carries a *list* of these on ``Pipeline.repos`` — one entry
+    per repository the run coordinates PRs across. Each repo pins its own
+    ``base_branch`` (PRs in that repo are opened against it), so the repo
+    set is genuinely list-shaped: there is no primary+secondary shape baked
+    into the data model, and nothing may assume ``len(repos)`` ∈ {1, 2}.
+
+    The ``repo`` is ``owner/name``-shaped — the full slug, not the bare
+    short name — so two repos with the same short name under different
+    owners stay distinct downstream (the worktree map is re-keyed by the
+    full slug in slice 3).
+    """
+
+    repo: str = Field(..., description="Repository in owner/name format")
+    base_branch: str | None = Field(
+        default=None,
+        description=(
+            "Base branch PRs in this repo are opened against. When None, "
+            "auto-detected from the repo's default branch (mirrors the "
+            "legacy ``Pipeline.base_branch`` semantics)."
+        ),
+    )
+
+
 class Pipeline(BaseModel):
     """Complete state of an SDLC pipeline execution.
 
@@ -1134,6 +1160,23 @@ class Pipeline(BaseModel):
         default=None,
         description="Base branch for PR creation. When None, auto-detected from repo's default branch.",
     )
+    repos: list[RepoSpec] = Field(
+        default_factory=list,
+        description=(
+            "The full list of repositories this pipeline coordinates PRs "
+            "across (#3393, multi-repo pipelines). List-shaped end to end — "
+            "an arbitrary number of repos, each with its own ``base_branch``; "
+            "no two-repo special case and no primary+secondary shape is baked "
+            "in. The legacy singleton ``repo``/``base_branch`` scalars above "
+            "are kept in sync by ``_sync_repos_and_legacy_singleton``: a "
+            "pipeline persisted before this field existed synthesizes "
+            "``repos=[RepoSpec(repo, base_branch)]`` from the singleton on "
+            "load, and ``repos[0]`` is mirrored back onto the scalars so "
+            "legacy readers keep working until slice 3 rewires them. Use the "
+            "``primary_repo`` property (not ``repos[0]``) for naming/"
+            "defaulting; use ``resolve_slice_repo`` to resolve a slice's repo."
+        ),
+    )
     prompt: str | None = Field(default=None, description="User prompt for prompt-driven pipelines")
     status: PipelineStatus = Field(
         default=PipelineStatus.PENDING, description="Overall pipeline status"
@@ -1366,6 +1409,80 @@ def resolve_decision(self, decision_id: str, resolution: str) -> HITLDecision |
                 return decision
         return None
 
+    @model_validator(mode="after")
+    def _sync_repos_and_legacy_singleton(self) -> Pipeline:
+        """Keep ``repos`` and the legacy ``repo``/``base_branch`` in sync (#3393).
+
+        Back-compat bridge for the multi-repo migration. Runs at every load
+        (``mode="after"``) and is idempotent:
+
+        * **Synthesize (legacy → list):** when ``repos`` is empty but the
+          legacy singleton ``repo`` is set — a pipeline persisted before the
+          ``repos`` field existed — synthesize
+          ``repos=[RepoSpec(repo=self.repo, base_branch=self.base_branch)]``
+          so list-shaped consumers see the repo.
+        * **Mirror (list → legacy):** when ``repos`` is non-empty, mirror
+          ``repos[0]`` back onto the legacy ``repo``/``base_branch`` scalars
+          so legacy readers (and the three ``repos[0]`` collapse sites that
+          slice 3 has not yet rewired) keep working.
+
+        No behavioural change for N=1 pipelines: a single-repo pipeline's
+        singleton and its one-element ``repos`` list agree after this runs.
+        A repo-less pipeline (local mode, ``repo is None``) is left untouched
+        — ``repos`` stays empty and the scalars stay ``None``.
+
+        The absent-``Slice.repo``⇒primary default is NOT resolved here; that
+        lives in ``resolve_slice_repo`` (a slice needs the pipeline as a
+        second input, which only the orchestrator layer has).
+        """
+        if not self.repos:
+            if self.repo is not None:
+                self.repos = [RepoSpec(repo=self.repo, base_branch=self.base_branch)]
+        else:
+            primary = self.repos[0]
+            self.repo = primary.repo
+            self.base_branch = primary.base_branch
+        return self
+
+    @property
+    def primary_repo(self) -> str | None:
+        """The pipeline's primary repo — ``repos[0].repo`` (#3393).
+
+        The INTENTIONAL named-primary accessor for naming and defaulting.
+        Explicitly NOT one of the three ``repos[0]`` collapse sites removed
+        in slice 3: those collapse the agent-facing repo *set* to a single
+        repo, discarding the others; this exposes a named primary while the
+        full ``repos`` list stays available to every other consumer.
+
+        Returns ``None`` only for a repo-less pipeline (local mode with no
+        ``repo`` and no ``repos``) — the ``_sync_repos_and_legacy_singleton``
+        validator guarantees a singleton-only pipeline has a populated
+        ``repos`` by the time this is read, so the ``self.repo`` fallback is
+        just belt-and-braces for a pre-validation read.
+        """
+        if self.repos:
+            return self.repos[0].repo
+        return self.repo
+
+
+def resolve_slice_repo(slice: Slice, pipeline: Pipeline) -> str | None:
+    """Resolve the repository a slice operates in (#3393, multi-repo pipelines).
+
+    This is the RUNTIME home of the absent-``Slice.repo``⇒primary default —
+    the contract migration deliberately leaves ``Slice.repo`` as ``None`` on a
+    legacy load because the ``Contract`` model has no repo field and cannot see
+    the orchestrator ``Pipeline`` (risk_analyst R1 / architect aeb3528). This
+    resolver takes the pipeline as a second input, so it is the correct layer
+    to apply the default:
+
+    * an explicit ``slice.repo`` (``owner/name``) wins, else
+    * fall back to ``pipeline.primary_repo``.
+
+    For an N=1 pipeline every slice's ``repo`` is ``None`` and this returns the
+    single repo — behaviourally identical to the pre-multi-repo world.
+    """
+    return slice.repo if slice.repo else pipeline.primary_repo
+
 
 class PipelineEvent(BaseModel):
     """Event emitted during pipeline execution."""

orchestrator/tests/test_models.py

@@ -5,6 +5,7 @@
 from datetime import UTC, datetime
 
 import pytest
+from egg_contracts.models import Slice
 from models import (
     PHASE_CONSENSUS_TIMEOUT_DEFAULTS_MIN,
     AgentExecution,
@@ -20,7 +21,9 @@
     PipelineConfig,
     PipelinePhase,
     PipelineStatus,
+    RepoSpec,
     resolve_consensus_timeout_minutes,
+    resolve_slice_repo,
 )
 
 
@@ -1206,3 +1209,155 @@ def test_apply_phase_exists(self):
         assert PipelinePhase.APPLY == "apply"
         # StrEnum: value-equal to string for serialisation symmetry.
         assert PipelinePhase("apply") == PipelinePhase.APPLY
+
+
+# ---------------------------------------------------------------------------
+# Multi-repo pipelines (#3393, slice-1) — orchestrator layer.
+#
+# The ORCHESTRATOR half of the two-layer repo-dimension design: RepoSpec,
+# Pipeline.repos (+ the legacy-singleton sync validator), the primary_repo
+# accessor, and the runtime resolve_slice_repo default. The CONTRACT half
+# (Slice.repo + the schema 1.4 migration) is covered in
+# shared/egg_contracts/tests/test_models.py.
+# ---------------------------------------------------------------------------
+
+
+class TestRepoSpec:
+    """``RepoSpec`` carries an owner/name repo + an optional base branch."""
+
+    def test_repo_required(self):
+        spec = RepoSpec(repo="jwbron/egg")
+        assert spec.repo == "jwbron/egg"
+
+    def test_base_branch_defaults_to_none(self):
+        spec = RepoSpec(repo="jwbron/egg")
+        assert spec.base_branch is None
+
+    def test_base_branch_can_be_set(self):
+        spec = RepoSpec(repo="jwbron/egg", base_branch="main")
+        assert spec.base_branch == "main"
+
+
+class TestPipelineReposSynthesisAndMirror:
+    """``_sync_repos_and_legacy_singleton`` bridges the legacy scalar ⇄ list (AC-e)."""
+
+    def test_legacy_singleton_synthesizes_one_element_repos(self):
+        # A pipeline persisted before the ``repos`` field existed: only the
+        # legacy ``repo``/``base_branch`` singleton is set. The validator
+        # synthesizes a one-element list from it.
+        pipeline = Pipeline(id="issue-3393", repo="jwbron/egg", base_branch="main")
+        assert len(pipeline.repos) == 1
+        assert pipeline.repos[0].repo == "jwbron/egg"
+        assert pipeline.repos[0].base_branch == "main"
+
+    def test_explicit_repos_mirror_back_onto_legacy_singleton(self):
+        # When ``repos`` is provided, repos[0] is mirrored back onto the
+        # legacy scalars so the not-yet-rewired (slice-3) readers keep working.
+        pipeline = Pipeline(
+            id="issue-3393",
+            repos=[
+                RepoSpec(repo="jwbron/egg", base_branch="main"),
+                RepoSpec(repo="jwbron/other", base_branch="develop"),
+            ],
+        )
+        assert pipeline.repo == "jwbron/egg"
+        assert pipeline.base_branch == "main"
+        # The full list is preserved — nothing collapses to a single repo.
+        assert [r.repo for r in pipeline.repos] == ["jwbron/egg", "jwbron/other"]
+
+    def test_repo_less_pipeline_left_untouched(self):
+        # Local-mode pipeline (no repo, no repos): stays empty, scalars None.
+        pipeline = Pipeline(id="local-run")
+        assert pipeline.repos == []
+        assert pipeline.repo is None
+        assert pipeline.base_branch is None
+
+    def test_synthesis_is_idempotent_on_reload(self):
+        pipeline = Pipeline(id="issue-3393", repo="jwbron/egg", base_branch="main")
+        reloaded = Pipeline.model_validate(pipeline.model_dump())
+        assert len(reloaded.repos) == 1
+        assert reloaded.repos[0].repo == "jwbron/egg"
+        assert reloaded.repo == "jwbron/egg"
+        assert reloaded.base_branch == "main"
+
+    def test_arbitrary_repo_count_supported(self):
+        # No two-repo special case: an arbitrary-length list round-trips.
+        specs = [RepoSpec(repo=f"jwbron/r{i}", base_branch="main") for i in range(5)]
+        pipeline = Pipeline(id="issue-3393", repos=specs)
+        assert [r.repo for r in pipeline.repos] == [f"jwbron/r{i}" for i in range(5)]
+
+
+class TestPrimaryRepo:
+    """``primary_repo`` is the intentional named-primary accessor (AC-f)."""
+
+    def test_primary_repo_is_repos_zero(self):
+        pipeline = Pipeline(
+            id="issue-3393",
+            repos=[
+                RepoSpec(repo="jwbron/egg", base_branch="main"),
+                RepoSpec(repo="jwbron/other", base_branch="develop"),
+            ],
+        )
+        assert pipeline.primary_repo == pipeline.repos[0].repo == "jwbron/egg"
+
+    def test_primary_repo_from_legacy_singleton(self):
+        pipeline = Pipeline(id="issue-3393", repo="jwbron/egg")
+        assert pipeline.primary_repo == "jwbron/egg"
+
+    def test_primary_repo_none_for_repo_less_pipeline(self):
+        pipeline = Pipeline(id="local-run")
+        assert pipeline.primary_repo is None
+
+
+class TestResolveSliceRepo:
+    """``resolve_slice_repo`` applies the absent-Slice.repo⇒primary default (AC-d)."""
+
+    def test_explicit_slice_repo_wins(self):
+        pipeline = Pipeline(
+            id="issue-3393",
+            repos=[
+                RepoSpec(repo="jwbron/egg", base_branch="main"),
+                RepoSpec(repo="jwbron/other", base_branch="develop"),
+            ],
+        )
+        slice_ = Slice(id="slice-1", name="in the secondary repo", repo="jwbron/other")
+        assert resolve_slice_repo(slice_, pipeline) == "jwbron/other"
+
+    def test_absent_slice_repo_falls_back_to_primary(self):
+        pipeline = Pipeline(
+            id="issue-3393",
+            repos=[
+                RepoSpec(repo="jwbron/egg", base_branch="main"),
+                RepoSpec(repo="jwbron/other", base_branch="develop"),
+            ],
+        )
+        slice_ = Slice(id="slice-1", name="no explicit repo")
+        assert slice_.repo is None
+        assert resolve_slice_repo(slice_, pipeline) == "jwbron/egg"
+
+    def test_absent_slice_repo_with_repo_less_pipeline_is_none(self):
+        pipeline = Pipeline(id="local-run")
+        slice_ = Slice(id="slice-1", name="no repo anywhere")
+        assert resolve_slice_repo(slice_, pipeline) is None
+
+
+class TestN1BackCompat:
+    """N=1 pipelines behave identically to the pre-multi-repo world (AC-g)."""
+
+    def test_single_repo_pipeline_resolves_to_that_repo(self):
+        pipeline = Pipeline(id="issue-3393", repo="jwbron/egg", base_branch="main")
+        # One synthesized repo, primary is it, and every (repo-less) slice
+        # resolves to the single repo — no behavioural change.
+        assert len(pipeline.repos) == 1
+        assert pipeline.primary_repo == "jwbron/egg"
+        slice_ = Slice(id="slice-1", name="the only slice")
+        assert resolve_slice_repo(slice_, pipeline) == "jwbron/egg"
+
+    def test_single_repo_pipeline_round_trips_unchanged(self):
+        pipeline = Pipeline(id="issue-3393", repo="jwbron/egg", base_branch="main")
+        dumped = pipeline.model_dump()
+        reloaded = Pipeline.model_validate(dumped)
+        assert reloaded.repo == "jwbron/egg"
+        assert reloaded.base_branch == "main"
+        assert len(reloaded.repos) == 1
+        assert reloaded.primary_repo == "jwbron/egg"