[w4] Label every fleet PR with its org-structure role (#526)

Pure derive_pr_labels (foundation/policy/pr_labels.py) maps a PR's shape
to a stable org-structure label set: to master/to slave (is_root_pr
discriminator), root, MegaTask, and the owning layer (main-pm /
cell/{team} / subtask/{team}). Mirrors batch.py: object|None inputs,
enum-or-string normalization, no DB/I/O. Full slave-targeting semantics
(base_branch vs default_branch) land with the slave/master wiring (W-H);
YAGNI now.

GitService._apply_pr_labels posts the result to the GitHub labels API
best-effort (create-before-add, swallow 422/409, never raises) so a label
failure can never block PR creation. Wired at all three PR-opening sites:
create_pr (gateway path), create_pull_request (REST/task path), and
_push_and_open_conventions_pr (static chore label). Existing PR tests
mock _apply_pr_labels so they never hit the real labels API.

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

roboco/foundation/policy/pr_labels.py

@@ -0,0 +1,73 @@
+"""PR-label derivation — pure predicates from task/PR shape to GitHub labels.
+
+The org-structure label vocabulary every fleet PR carries so a human can triage
+the queue at a glance: which tree a PR targets (``to master`` / ``to slave``),
+whether it is an assembled root PR (``root``), a MegaTask member (``MegaTask``),
+and which layer owns it (``main-pm`` / ``cell/{team}`` / ``subtask/{team}``).
+
+Pure + DB-free so it is unit-testable; the git service's best-effort
+``_apply_pr_labels`` helper posts the result to the GitHub labels API. Inputs are
+typed ``object | None`` because callers pass ORM enum members or ``.value``
+strings (mirrors ``batch.py``).
+
+``to master`` / ``to slave`` is correct-by-construction today — every root PR
+targets the default branch (``to master``) and every cell/leaf PR targets an
+integration/parent branch (``to slave``), so ``is_root_pr`` is the discriminator.
+Full slave-targeting semantics land with the slave/master fleet wiring (W-H); when
+that arrives the call sites can pass the PR base vs the project default branch and
+this predicate grows a ``base_branch``/``default_branch`` pair then.
+"""
+
+from __future__ import annotations
+
+from roboco.foundation.identity import Team
+
+# A project-level conventions scaffold/restore PR carries no task and no org
+# layer, so it gets a single conventional-commit-kind label (its branch is
+# ``chore/roboco-conventions-scaffold``, its title ``chore(conventions): ...``).
+CONVENTIONS_PR_LABELS: list[str] = ["chore"]
+
+
+def _team_value(team: object | None) -> str:
+    if team is None:
+        return ""
+    return str(getattr(team, "value", team)).lower()
+
+
+def _layer_label(team: str, has_children: bool) -> str:
+    """The owning-layer label for a task-bearing PR."""
+    if team == Team.MAIN_PM.value:
+        return "main-pm"
+    if has_children:
+        return f"cell/{team}"
+    return f"subtask/{team}"
+
+
+def derive_pr_labels(
+    *,
+    is_root_pr: bool,
+    task_team: object | None,
+    batch_id: object | None,
+    has_children: bool,
+) -> list[str]:
+    """The org-structure labels for a PR, in a stable order, de-duplicated.
+
+    - ``to master`` vs ``to slave`` — today the only master-targeting PRs are the
+      assembled root->master PRs, so ``is_root_pr`` is the discriminator; real
+      slave-branch targeting lands with the slave/master fleet wiring (W-H).
+    - ``root`` — an assembled root->master PR (``is_root_pr``).
+    - ``MegaTask`` — the task carries a ``batch_id``.
+    - layer label — ``main-pm`` for a Main-PM coordination root, ``cell/{team}``
+      for a cell-assembled PR (``has_children``), else ``subtask/{team}`` for a
+      leaf dev PR. Absent when the PR has no task (a freeform PR).
+    """
+    labels: list[str] = ["to master" if is_root_pr else "to slave"]
+    if is_root_pr:
+        labels.append("root")
+    if batch_id is not None:
+        labels.append("MegaTask")
+    team = _team_value(task_team)
+    if team:
+        labels.append(_layer_label(team, has_children))
+    # de-dup preserving first-seen order (a MegaTask root PR can otherwise repeat)
+    return list(dict.fromkeys(labels))

roboco/services/git.py

@@ -51,6 +51,7 @@ from roboco.exceptions import (
     MergeConflictError,
 )
 from roboco.foundation.policy import lifecycle
+from roboco.foundation.policy.pr_labels import CONVENTIONS_PR_LABELS, derive_pr_labels
 from roboco.models.base import AgentRole, TaskStatus
 from roboco.models.env_branches import head_branch
 from roboco.services.base import (
@@ -2360,6 +2361,92 @@ class GitService(BaseService):
                 {"owner": owner, "repo": repo, "head": payload.get("head")},
             ) from e
 
+    # A single neutral color — labels are distinguished by name, not hue, and
+    # GitHub's create-label endpoint requires a color (it won't auto-assign).
+    _PR_LABEL_COLOR = "5e6ad2"
+
+    async def _ensure_label_exists(
+        self, owner: str, repo: str, git_token: str, name: str
+    ) -> None:
+        """Create a repo label if missing (GitHub's add-label API 404s on an
+        unknown label instead of auto-creating). Swallow 'already exists'
+        (422/409). Best-effort: logs and never raises — a missing label must not
+        block PR creation."""
+        try:
+            async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+                resp = await client.post(
+                    f"{_api_base()}/repos/{owner}/{repo}/labels",
+                    headers={
+                        "Authorization": f"Bearer {git_token}",
+                        "Accept": "application/vnd.github+json",
+                        "X-GitHub-Api-Version": "2022-11-28",
+                    },
+                    json={"name": name, "color": self._PR_LABEL_COLOR},
+                )
+        except httpx.HTTPError as e:
+            self.log.warning("PR label ensure HTTP error", label=name, error=str(e))
+            return
+        # 422 (already_exists) / 409 (conflict) = the label is already present.
+        if resp.is_success or resp.status_code in (409, 422):
+            return
+        self.log.warning(
+            "could not ensure PR label exists",
+            label=name,
+            status=resp.status_code,
+            body=(resp.text or "")[:200],
+        )
+
+    async def _apply_pr_labels(
+        self,
+        owner: str,
+        repo: str,
+        git_token: str,
+        pr_number: int,
+        labels: list[str],
+    ) -> None:
+        """Best-effort: create each label (GitHub won't auto-create on add) then
+        add them to the PR. Re-adding is a no-op, so the 422 'PR already exists'
+        path is safe to re-label. Never raises — labeling must not block PR
+        creation (same posture as ``_record_pr_atomically``)."""
+        if not labels:
+            return
+        for name in labels:
+            await self._ensure_label_exists(owner, repo, git_token, name)
+        try:
+            async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+                resp = await client.post(
+                    f"{_api_base()}/repos/{owner}/{repo}/issues/{pr_number}/labels",
+                    headers={
+                        "Authorization": f"Bearer {git_token}",
+                        "Accept": "application/vnd.github+json",
+                        "X-GitHub-Api-Version": "2022-11-28",
+                    },
+                    json={"labels": labels},
+                )
+        except httpx.HTTPError as e:
+            self.log.warning("add PR labels HTTP error", pr=pr_number, error=str(e))
+            return
+        if not resp.is_success:
+            self.log.warning(
+                "could not add PR labels",
+                pr=pr_number,
+                status=resp.status_code,
+                body=(resp.text or "")[:200],
+            )
+
+    async def _task_has_children(self, task_id: UUID) -> bool:
+        """True iff the task has any subtask (a one-row probe). PR creation is
+        rare; the query is negligible and keeps ``has_children`` honest instead
+        of assumed per call site."""
+        from sqlalchemy import select
+
+        from roboco.db.tables import TaskTable
+
+        result = await self.session.execute(
+            select(TaskTable.id).where(TaskTable.parent_task_id == task_id).limit(1)
+        )
+        return result.first() is not None
+
     async def _pr_base_on_remote(
         self,
         workspace: Path,
@@ -2445,10 +2532,13 @@ class GitService(BaseService):
             },
         )
 
+        labels = await self._labels_for_pr_request(request)
+
         existing = await self._existing_pr_tuple(
             resp, (owner, repo), (source_branch, target_branch), git_token, pr_title
         )
         if existing is not None:
+            await self._apply_pr_labels(owner, repo, git_token, existing[0], labels)
             return existing
 
         if not resp.is_success:
@@ -2459,8 +2549,10 @@ class GitService(BaseService):
             )
 
         pr_data = resp.json()
+        pr_number = int(pr_data["number"])
+        await self._apply_pr_labels(owner, repo, git_token, pr_number, labels)
         return (
-            int(pr_data["number"]),
+            pr_number,
             str(pr_data["html_url"]),
             pr_title or "",
             source_branch,
@@ -2615,6 +2707,35 @@ class GitService(BaseService):
         data = resp.json()
         return {"number": int(data["number"]), "url": str(data.get("html_url", ""))}
 
+    async def _labels_for_pr_request(
+        self,
+        request: GitCreatePRRequest,
+    ) -> list[str]:
+        """The org-structure labels for the REST/task PR path. A task PR derives
+        team / batch / has_children from the task; a freeform PR (``task_id``
+        None) carries only the tree + root flags."""
+        if request.task_id is None:
+            return derive_pr_labels(
+                is_root_pr=request.is_root_pr,
+                task_team=None,
+                batch_id=None,
+                has_children=False,
+            )
+        task = await get_task_service(self.session).get(request.task_id)
+        if task is None:
+            return derive_pr_labels(
+                is_root_pr=request.is_root_pr,
+                task_team=None,
+                batch_id=None,
+                has_children=False,
+            )
+        return derive_pr_labels(
+            is_root_pr=request.is_root_pr,
+            task_team=task.team,
+            batch_id=task.batch_id,
+            has_children=await self._task_has_children(UUID(str(task.id))),
+        )
+
     async def _resolve_new_pr_context(
         self,
         workspace: Path,
@@ -4178,6 +4299,15 @@ class GitService(BaseService):
             },
         )
 
+        # Org-structure labels: create_pr is always an assembled PM PR
+        # (cell->root or root->master), so has_children is True by construction.
+        labels = derive_pr_labels(
+            is_root_pr=is_root_pr,
+            task_team=task.team,
+            batch_id=task.batch_id,
+            has_children=True,
+        )
+
         if resp.status_code == _GH_UNPROCESSABLE and "already exists" in resp.text:
             found = await self._find_existing_pr(
                 owner, repo, branch_name, parent, git_token
@@ -4192,6 +4322,7 @@ class GitService(BaseService):
                 await _await_shielded(
                     self._record_pr_atomically(UUID(str(task.id)), pr_number, pr_url)
                 )
+                await self._apply_pr_labels(owner, repo, git_token, pr_number, labels)
                 return {
                     "pr_number": pr_number,
                     "pr_url": pr_url,
@@ -4217,6 +4348,7 @@ class GitService(BaseService):
         await _await_shielded(
             self._record_pr_atomically(UUID(str(task.id)), pr_number, pr_url)
         )
+        await self._apply_pr_labels(owner, repo, git_token, pr_number, labels)
         return {"pr_number": pr_number, "pr_url": pr_url, "is_root_pr": is_root_pr}
 
     async def _lock_parent_task_for_merge(self, parent_task_id: UUID | None) -> None:
@@ -5585,9 +5717,16 @@ class GitService(BaseService):
         if not resp.is_success:
             return unopened
         data = resp.json()
+        pr_number = data.get("number")
+        if pr_number is not None:
+            # Static label — a project-level scaffold/restore PR has no task or
+            # org layer; best-effort, never blocks.
+            await self._apply_pr_labels(
+                owner, repo, token, int(pr_number), CONVENTIONS_PR_LABELS
+            )
         return {
             "branch": spec.branch,
-            "pr_number": data.get("number"),
+            "pr_number": pr_number,
             "pr_url": data.get("html_url"),
         }
 

tests/integration/test_git_conventions_pr.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import subprocess
 from typing import TYPE_CHECKING, Any
+from unittest.mock import AsyncMock
 from uuid import uuid4
 
 from roboco.config import settings
@@ -158,6 +159,7 @@ async def test_open_conventions_pr_force_pushes_scaffold_branch(
         return _Resp()
 
     monkeypatch.setattr(git, "_post_pr", _fake_post_pr)
+    monkeypatch.setattr(git, "_apply_pr_labels", AsyncMock())
 
     spec = _ConventionsPr(
         content="version: 1\n",

tests/unit/foundation/test_pr_labels.py

@@ -0,0 +1,100 @@
+"""Pure derivation matrix for ``derive_pr_labels`` — no DB, no I/O."""
+
+from __future__ import annotations
+
+from uuid import uuid4
+
+from roboco.foundation.identity import Team
+from roboco.foundation.policy.pr_labels import (
+    CONVENTIONS_PR_LABELS,
+    derive_pr_labels,
+)
+
+
+def test_root_master_megatask_main_pm() -> None:
+    # submit_root on a MegaTask root-subtask: root->master, main_pm, batch member.
+    labels = derive_pr_labels(
+        is_root_pr=True,
+        task_team=Team.MAIN_PM,
+        batch_id=uuid4(),
+        has_children=True,
+    )
+    assert labels == ["to master", "root", "MegaTask", "main-pm"]
+
+
+def test_root_master_main_pm_no_batch() -> None:
+    labels = derive_pr_labels(
+        is_root_pr=True,
+        task_team=Team.MAIN_PM,
+        batch_id=None,
+        has_children=True,
+    )
+    assert labels == ["to master", "root", "main-pm"]
+
+
+def test_cell_to_root_assembled() -> None:
+    # submit_up: cell->root PR, base is the integration branch (not default).
+    labels = derive_pr_labels(
+        is_root_pr=False,
+        task_team=Team.BACKEND,
+        batch_id=None,
+        has_children=True,
+    )
+    assert labels == ["to slave", "cell/backend"]
+
+
+def test_leaf_dev_pr() -> None:
+    labels = derive_pr_labels(
+        is_root_pr=False,
+        task_team=Team.FRONTEND,
+        batch_id=None,
+        has_children=False,
+    )
+    assert labels == ["to slave", "subtask/frontend"]
+
+
+def test_freeform_pr_no_task() -> None:
+    # task_id None: no team, no batch — just the tree + root flags.
+    labels = derive_pr_labels(
+        is_root_pr=False,
+        task_team=None,
+        batch_id=None,
+        has_children=False,
+    )
+    assert labels == ["to slave"]
+
+
+def test_freeform_root_pr_no_task() -> None:
+    labels = derive_pr_labels(
+        is_root_pr=True,
+        task_team=None,
+        batch_id=None,
+        has_children=False,
+    )
+    assert labels == ["to master", "root"]
+
+
+def test_accepts_string_team_value() -> None:
+    # callers pass ORM enum members OR their .value strings (mirrors batch.py).
+    labels = derive_pr_labels(
+        is_root_pr=False,
+        task_team="main_pm",
+        batch_id=None,
+        has_children=True,
+    )
+    assert labels == ["to slave", "main-pm"]
+
+
+def test_conventions_pr_labels_static() -> None:
+    assert CONVENTIONS_PR_LABELS == ["chore"]
+
+
+def test_no_duplicates() -> None:
+    # a shape that could repeat a label still yields a unique list.
+    labels = derive_pr_labels(
+        is_root_pr=True,
+        task_team=Team.MAIN_PM,
+        batch_id=uuid4(),
+        has_children=True,
+    )
+    assert len(labels) == len(set(labels))

tests/unit/services/test_git.py

@@ -503,6 +503,7 @@ async def test_create_pr_returns_pr_dict() -> None:
         "html_url": f"https://github.com/acme/repo/pull/{_EXPECTED_PR_NUMBER}",
     }
     _bind(svc, "_post_pr", AsyncMock(return_value=fake_resp))
+    _bind(svc, "_apply_pr_labels", AsyncMock())
 
     with _patch_project_service(fake_project):
         out = await svc.create_pr(
@@ -550,6 +551,7 @@ async def test_create_pr_records_pr_despite_cancellation_after_post() -> None:
         "html_url": f"https://github.com/acme/repo/pull/{_EXPECTED_PR_NUMBER}",
     }
     _bind(svc, "_post_pr", AsyncMock(return_value=fake_resp))
+    _bind(svc, "_apply_pr_labels", AsyncMock())
 
     with _patch_project_service(fake_project):
         task = asyncio.ensure_future(
@@ -607,6 +609,7 @@ async def test_create_pr_cancellation_waits_out_record_before_reraising() -> Non
         "html_url": f"https://github.com/acme/repo/pull/{_EXPECTED_PR_NUMBER}",
     }
     _bind(svc, "_post_pr", AsyncMock(return_value=fake_resp))
+    _bind(svc, "_apply_pr_labels", AsyncMock())
 
     with _patch_project_service(fake_project):
         task = asyncio.ensure_future(