feat(git): coordination root cuts a Main-PM integration branch per repo (#58)

The coordination/fan-out root carries a product (cell->repo map) but no
project of its own, and was forced branchless — so a cell's parent-branch
resolution fell back to the project default (master), and cell completion
merged each cell straight to master, bypassing the Main-PM integration
point and the CEO merge gate.

Per the locked branch model (master <- feature/main_pm/{root} <- cell <-
dev), the root is now the Main-PM integration point: on claim it cuts
feature/main_pm/{root} off master in EACH distinct repo the product spans
(monorepo => 1, multi-repo => N). Cells then branch off it via the existing
ancestor-branch resolution, so cell work never targets master.

- ProductService.distinct_project_ids: enumerate the repos a product spans
- TaskService._create_branch_in_project: project-parameterized branch
  creation split out of _auto_create_branch
- TaskService._ensure_coordination_root_branches: cut the integration
  branch in each repo; graceful empty when the product has no cell map yet
- _ensure_branch_for_task routes a product-backed root here, not to no-op

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

roboco/services/product.py

@@ -100,6 +100,25 @@ class ProductService(BaseService):
         )
         return result.scalar_one_or_none()
 
+    async def distinct_project_ids(self, product_id: UUID) -> list[UUID]:
+        """Distinct repos a product spans — one Main-PM integration branch each.
+
+        The product's cell->project map may point several teams at the same
+        Project (the monorepo case) or at different ones (multi-repo). The
+        Main-PM root cuts one ``feature/main_pm/{root}`` integration branch per
+        DISTINCT project, so cells in that repo branch off it instead of master.
+        Ordered by the first team that references each project for determinism.
+        """
+        result = await self.session.execute(
+            select(ProductProjectTable.project_id)
+            .where(ProductProjectTable.product_id == product_id)
+            .order_by(ProductProjectTable.team)
+        )
+        seen: dict[UUID, None] = {}
+        for project_id in result.scalars().all():
+            seen.setdefault(project_id, None)
+        return list(seen)
+
     async def _replace_cells(
         self, product: ProductTable, cells: list[ProductCellMapping]
     ) -> None:

roboco/services/task.py

@@ -690,11 +690,13 @@ class TaskService(BaseService):
 
         if not task.project_id:
             # A coordination/fan-out task carries a product (a cell->project
-            # map) but no repo of its own, so it does no git work and has no
-            # branch — its cell subtasks each resolve a real project and get
-            # their own branches. Only a task with neither is misconfigured.
+            # map) but no repo of its own. Per the CEO-locked branch model it is
+            # the Main-PM integration point: it cuts feature/main_pm/{root} off
+            # master in EACH repo the product spans, so cells branch off it
+            # (not off master) and only the CEO merges the root into master.
+            # Only a task with neither project nor product is misconfigured.
             if task.product_id:
-                return ""
+                return await self._ensure_coordination_root_branches(task, agent_id)
             raise ValueError(
                 "Task requires a project_id (a repo) or a product_id (a "
                 "cell->project map) to create a branch. Assign one before "
@@ -832,16 +834,32 @@ class TaskService(BaseService):
         Raises:
             ValueError: If branch cannot be created
         """
-        from roboco.api.schemas.git import GitCreateBranchRequest
-        from roboco.services.git import get_git_service
         from roboco.services.project import get_project_service
 
-        git_service = get_git_service(self.session)
         project_service = get_project_service(self.session)
-
         project = await project_service.get(UUID(str(task.project_id)))
         if not project:
             raise ValueError(f"Project {task.project_id} not found")
+        return await self._create_branch_in_project(task, agent_id, project)
+
+    async def _create_branch_in_project(
+        self,
+        task: TaskTable,
+        agent_id: UUID,
+        project: Any,
+    ) -> str:
+        """Create the task's hierarchical branch inside one resolved repo.
+
+        Split out of :meth:`_auto_create_branch` so a coordination root can cut
+        the same ``feature/main_pm/{root}`` integration branch in EACH repo its
+        product spans (monorepo: one call; multi-repo: one per repo). The branch
+        name is hierarchy-derived so it is identical across repos; the physical
+        branch is created in each.
+        """
+        from roboco.api.schemas.git import GitCreateBranchRequest
+        from roboco.services.git import get_git_service
+
+        git_service = get_git_service(self.session)
 
         parent_branch = await self._resolve_parent_branch(task, project)
         workspace = await git_service.get_workspace(project.slug, agent_id)
@@ -863,11 +881,47 @@ class TaskService(BaseService):
         self.log.info(
             "Auto-created hierarchical branch",
             task_id=str(task.id),
+            project_slug=project.slug,
             branch_name=branch_name,
             parent_branch=parent_branch or "default",
         )
         return branch_name
 
+    async def _ensure_coordination_root_branches(
+        self,
+        task: TaskTable,
+        agent_id: UUID,
+    ) -> str:
+        """Cut the Main-PM integration branch in every repo the product spans.
+
+        The coordination root carries a product (a cell->repo map) but no
+        project of its own. Per the CEO-locked model, the Main-PM root branches
+        ``feature/main_pm/{root}`` OFF master in each distinct repo; cells then
+        branch off it (via the parent-branch resolution) instead of off master,
+        so cell work never targets master — only the CEO merges the root branch
+        into master, per repo. Monorepo => one branch; multi-repo => N.
+
+        Returns the shared branch name (identical across repos), or ``""`` when
+        the product has no cell->repo map yet (delegation then falls back to the
+        parent's project per the routing spec, and the root stays branchless).
+        """
+        from roboco.services.product import get_product_service
+        from roboco.services.project import get_project_service
+
+        product_service = get_product_service(self.session)
+        project_service = get_project_service(self.session)
+
+        project_ids = await product_service.distinct_project_ids(
+            UUID(str(task.product_id))
+        )
+        branch_name = ""
+        for project_id in project_ids:
+            project = await project_service.get(project_id)
+            if project is None:
+                continue
+            branch_name = await self._create_branch_in_project(task, agent_id, project)
+        return branch_name
+
     async def get(self, task_id: UUID) -> TaskTable | None:
         """Get a task by ID."""
         result = await self.session.execute(

tests/integration/test_product_service.py

@@ -102,6 +102,44 @@ async def test_shared_project_across_cells(product_setup: dict) -> None:
         assert await svc.project_for(product.id, c) == shared
 
 
+@pytest.mark.asyncio
+async def test_distinct_project_ids_monorepo_and_multirepo(product_setup: dict) -> None:
+    """One integration branch per DISTINCT repo: monorepo => 1, multi-repo => N."""
+    svc = product_setup["svc"]
+    projects = product_setup["projects"]
+    shared = projects[Team.BACKEND].id
+
+    mono = await svc.create(
+        ProductCreate(
+            name="Mono",
+            slug=f"mono-{uuid4().hex[:6]}",
+            cells=[
+                ProductCellMapping(team=c, project_id=shared)
+                for c in (Team.BACKEND, Team.FRONTEND, Team.UX_UI)
+            ],
+        ),
+        created_by=product_setup["creator"],
+    )
+    assert await svc.distinct_project_ids(mono.id) == [shared]
+
+    multi = await svc.create(
+        ProductCreate(
+            name="Multi",
+            slug=f"multi-{uuid4().hex[:6]}",
+            cells=[
+                ProductCellMapping(team=c, project_id=projects[c].id)
+                for c in (Team.BACKEND, Team.FRONTEND, Team.UX_UI)
+            ],
+        ),
+        created_by=product_setup["creator"],
+    )
+    assert set(await svc.distinct_project_ids(multi.id)) == {
+        projects[Team.BACKEND].id,
+        projects[Team.FRONTEND].id,
+        projects[Team.UX_UI].id,
+    }
+
+
 @pytest.mark.asyncio
 async def test_duplicate_slug_conflicts(product_setup: dict) -> None:
     svc = product_setup["svc"]

tests/unit/services/test_task.py

@@ -8,7 +8,7 @@ session boundary and checks the method's contract.
 from __future__ import annotations
 
 from datetime import datetime
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import uuid4
 
 import pytest
@@ -644,11 +644,32 @@ async def test_ensure_branch_returns_existing_branch() -> None:
 
 
 @pytest.mark.asyncio
-async def test_ensure_branch_skips_coordination_task() -> None:
-    """A product-backed task with no repo of its own gets no branch (not raised)."""
+async def test_ensure_branch_coordination_root_cuts_integration_branch() -> None:
+    """A product-backed root cuts feature/main_pm/{root} in each product repo."""
     svc = TaskService(MagicMock())
     task = MagicMock(branch_name=None, project_id=None, product_id=uuid4())
-    assert await svc._ensure_branch_for_task(task, uuid4()) == ""
+    create_in_project = AsyncMock(return_value="feature/main_pm/root1234")
+    _bind(svc, "_create_branch_in_project", create_in_project)
+    product_svc = MagicMock(distinct_project_ids=AsyncMock(return_value=[uuid4()]))
+    project_svc = MagicMock(get=AsyncMock(return_value=MagicMock()))
+    with (
+        patch("roboco.services.product.get_product_service", return_value=product_svc),
+        patch("roboco.services.project.get_project_service", return_value=project_svc),
+    ):
+        result = await svc._ensure_branch_for_task(task, uuid4())
+    assert result == "feature/main_pm/root1234"
+    create_in_project.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_ensure_branch_coordination_root_no_cell_map_stays_branchless() -> None:
+    """A product with no cell->repo map yet stays branchless (graceful fallback)."""
+    svc = TaskService(MagicMock())
+    task = MagicMock(branch_name=None, project_id=None, product_id=uuid4())
+    product_svc = MagicMock(distinct_project_ids=AsyncMock(return_value=[]))
+    with patch("roboco.services.product.get_product_service", return_value=product_svc):
+        result = await svc._ensure_branch_for_task(task, uuid4())
+    assert result == ""
 
 
 @pytest.mark.asyncio