feat(gateway): enable 2-devs-per-cell parallelism + split-before-claim sizing

- Intake / Main-PM / Cell-PM prompts: enumerate independently-shippable work
  units, inherit the breakdown down the chain, and dispatch independents in
  parallel (dependency order, never one-at-a-time).
- Raise the code-spine concurrency cap from 1 to 2 per parent (one per cell
  developer) so both devs build in parallel; keep the same-assignee guard and
  the planning/documentation cap at 1, plus the cross-team planning exemption.
- Split-before-claim: hard-block an egregiously-bundled code leaf at delegate
  time so the PM splits it before any dev claims it; nudge the moderate band
  in the delegate success envelope.

tests/unit/gateway/test_delegate_sizing.py

@@ -0,0 +1,79 @@
+"""Split-before-claim: a code leaf must not bundle too many concerns.
+
+A `code` subtask carrying more acceptance criteria than the hard cap bundles
+multiple independent concerns into one leaf — QA can't pass a partial and
+criteria get dropped. The gateway rejects it at delegate time so the PM splits
+the bundle before any dev can claim it. Moderate bundling is allowed but flagged
+in the success envelope (the nudge). `planning` briefs are exempt.
+"""
+
+from __future__ import annotations
+
+from roboco.services.gateway.choreographer._impl import Choreographer, DelegateInputs
+
+
+def _inputs(*, task_type: str, ac_count: int) -> DelegateInputs:
+    return DelegateInputs(
+        title="t",
+        description="d",
+        assigned_to="be-dev-1",
+        team="backend",
+        task_type=task_type,
+        nature="technical",
+        acceptance_criteria=[f"criterion {i}" for i in range(ac_count)],
+    )
+
+
+def test_small_code_leaf_is_allowed() -> None:
+    """A focused code leaf (<= hard cap) passes the sizing guard."""
+    assert (
+        Choreographer._delegate_sizing_guard(_inputs(task_type="code", ac_count=4))
+        is None
+    )
+
+
+def test_code_leaf_at_hard_cap_is_allowed() -> None:
+    """Exactly at the hard cap is still allowed; only strictly-above is blocked."""
+    cap = Choreographer._SIZING_HARD_AC_COUNT
+    assert (
+        Choreographer._delegate_sizing_guard(_inputs(task_type="code", ac_count=cap))
+        is None
+    )
+
+
+def test_egregiously_bundled_code_leaf_is_rejected() -> None:
+    """A code leaf above the hard cap is rejected with split guidance."""
+    cap = Choreographer._SIZING_HARD_AC_COUNT
+    env = Choreographer._delegate_sizing_guard(
+        _inputs(task_type="code", ac_count=cap + 4)
+    )
+    assert env is not None
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+    assert "Split this into smaller code subtasks" in (env.remediate or "")
+
+
+def test_planning_brief_is_exempt_from_sizing() -> None:
+    """planning subtasks (main_pm -> cell_pm) legitimately carry many criteria."""
+    assert (
+        Choreographer._delegate_sizing_guard(_inputs(task_type="planning", ac_count=20))
+        is None
+    )
+
+
+def test_no_nudge_below_threshold() -> None:
+    assert Choreographer._sizing_hint(_inputs(task_type="code", ac_count=5)) is None
+
+
+def test_nudge_in_moderate_band() -> None:
+    """Above the nudge count and below the hard cap: allowed but flagged."""
+    hint = Choreographer._sizing_hint(_inputs(task_type="code", ac_count=7))
+    assert hint is not None
+    assert "7 acceptance criteria" in hint
+    assert "parallel" in hint
+
+
+def test_no_nudge_for_planning() -> None:
+    assert (
+        Choreographer._sizing_hint(_inputs(task_type="planning", ac_count=20)) is None
+    )

tests/unit/gateway/test_spine_cap_cross_team.py

@@ -1,21 +1,16 @@
-"""Task #157: spine-cap allows cross-team planning fanout.
+"""Spine-cap concurrency rules for delegated subtasks.
 
-Pre-fix:
-    main_pm delegates a planning subtask to be-pm (backend cell). When
-    it then tries to delegate a second planning subtask to fe-pm
-    (frontend cell), the spine-cap rejects because there's already a
-    non-terminal task_type='planning' under the parent. Pre-gateway
-    allowed this parallel cross-cell pattern; the gateway over-applied
-    the over-decomposition cap.
+The per-parent cap is type-aware (`_SPINE_TYPE_CAPS`):
+  - ``code``: 2 — one per cell developer, so both build independent units in
+    parallel. A third concurrent code subtask is rejected; a second code
+    subtask to the SAME developer is rejected (same-assignee rule).
+  - ``planning`` / ``documentation``: 1.
 
-Fix:
-    `_sibling_dup_envelope` skips the spine-cap when:
-      - new task_type == "planning"
-      - new team != sibling team (both non-empty)
-    Other combinations stay capped:
-      - same-team planning: still over-decomposition (real bug)
-      - code / documentation regardless of team: a single repo on one
-        branch shouldn't have two simultaneous code subtasks
+Cross-team exemption:
+  - ``planning`` on a different team does NOT count toward the cap — that's
+    main_pm's legitimate cross-cell fanout (be-pm + fe-pm + ux-pm in parallel).
+  - ``code`` / ``documentation`` count regardless of team (a parent's code
+    spine is bounded by the two devs in its one cell).
 """
 
 from __future__ import annotations
@@ -25,21 +20,27 @@ from unittest.mock import MagicMock
 from roboco.services.gateway.choreographer._impl import Choreographer
 
 
-def _sibling(*, task_type: str, team: str, status: str = "pending") -> MagicMock:
+def _sibling(
+    *,
+    task_type: str,
+    team: str,
+    status: str = "pending",
+    assignee: str = "some-pm",
+) -> MagicMock:
     sib = MagicMock()
     sib.id = "11111111-aaaa-bbbb-cccc-dddddddddddd"
     sib.status = status
     sib.task_type = task_type
     sib.team = team
-    sib.assigned_to = "some-pm"
+    sib.assigned_to = assignee
     return sib
 
 
 def test_cross_team_planning_fanout_is_allowed() -> None:
     """main-pm: planning→be-pm (backend) exists; planning→fe-pm (frontend) must pass."""
-    sib = _sibling(task_type="planning", team="backend")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+    sib = _sibling(task_type="planning", team="backend", assignee="be-pm")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="planning",
         new_team="frontend",
         new_assignee="fe-pm",
@@ -52,9 +53,9 @@ def test_cross_team_planning_fanout_is_allowed() -> None:
 
 def test_cross_team_planning_third_cell_also_allowed() -> None:
     """Third cell (ux_ui) is also a valid cross-team planning fanout target."""
-    sib = _sibling(task_type="planning", team="backend")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+    sib = _sibling(task_type="planning", team="backend", assignee="be-pm")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="planning",
         new_team="ux_ui",
         new_assignee="ux-pm",
@@ -63,42 +64,70 @@ def test_cross_team_planning_third_cell_also_allowed() -> None:
 
 
 def test_same_team_planning_still_rejected() -> None:
-    """Two planning subtasks on the SAME team is the real over-decomp pattern —
-    must still be blocked by the spine-cap."""
-    sib = _sibling(task_type="planning", team="backend")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+    """Two planning subtasks on the SAME team is over-decomposition —
+    must still be blocked by the spine-cap (planning cap is 1)."""
+    sib = _sibling(task_type="planning", team="backend", assignee="be-pm")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="planning",
         new_team="backend",
-        new_assignee="be-pm",
+        new_assignee="be-pm-2",
     )
     assert env is not None
     body = env.as_dict()
     assert body["error"] == "invalid_state", body
 
 
-def test_cross_team_code_still_rejected() -> None:
-    """Code subtasks stay capped regardless of team — only one code task per
-    parent at a time. (Cross-team code under one parent is meaningless;
-    each cell's code work lives under its own cell-PM planning task.)"""
-    sib = _sibling(task_type="code", team="backend")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+def test_one_code_sibling_allows_a_second_parallel_dev() -> None:
+    """Code cap is 2 — one in flight to be-dev-1 must NOT block a second to
+    be-dev-2. This is the two-devs-per-cell parallelism the cap exists to allow."""
+    sib = _sibling(task_type="code", team="backend", assignee="be-dev-1")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="code",
-        new_team="frontend",
-        new_assignee="fe-dev-1",
+        new_team="backend",
+        new_assignee="be-dev-2",
+    )
+    assert env is None, f"A second parallel code subtask must be allowed. Got: {env}"
+
+
+def test_two_code_siblings_reject_a_third() -> None:
+    """Both cell devs busy (2 non-terminal code subtasks) → a third is capped."""
+    sibs = [
+        _sibling(task_type="code", team="backend", assignee="be-dev-1"),
+        _sibling(task_type="code", team="backend", assignee="be-dev-2"),
+    ]
+    env = Choreographer._sibling_cap_envelope(
+        siblings=sibs,
+        new_type="code",
+        new_team="backend",
+        new_assignee="be-dev-1",
     )
     assert env is not None
     body = env.as_dict()
     assert body["error"] == "invalid_state", body
 
 
-def test_cross_team_documentation_still_rejected() -> None:
-    """Documentation subtasks stay capped regardless of team — single doc
-    pass per parent."""
-    sib = _sibling(task_type="documentation", team="backend")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+def test_second_code_to_same_dev_is_rejected() -> None:
+    """The same developer never holds two code subtasks under one parent
+    (same-assignee rule), even though the cap is 2."""
+    sib = _sibling(task_type="code", team="backend", assignee="be-dev-1")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
+        new_type="code",
+        new_team="backend",
+        new_assignee="be-dev-1",
+    )
+    assert env is not None
+    body = env.as_dict()
+    assert body["error"] == "invalid_state", body
+
+
+def test_documentation_still_capped_at_one() -> None:
+    """Documentation subtasks stay capped at 1 regardless of team."""
+    sib = _sibling(task_type="documentation", team="backend", assignee="be-doc")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="documentation",
         new_team="frontend",
         new_assignee="fe-doc",
@@ -107,22 +136,22 @@ def test_cross_team_documentation_still_rejected() -> None:
 
 
 def test_planning_with_missing_team_still_rejected() -> None:
-    """If either side has no team attribute (defensive), fall back to the
-    strict cap. Don't let an empty-team escape hatch sneak past."""
-    sib_no_team = _sibling(task_type="planning", team="")
-    env = Choreographer._sibling_dup_envelope(
-        sibling=sib_no_team,
+    """If either side has no team (defensive), fall back to the strict cap —
+    don't let an empty-team escape hatch sneak a second planning past."""
+    sib_no_team = _sibling(task_type="planning", team="", assignee="be-pm")
+    env = Choreographer._sibling_cap_envelope(
+        siblings=[sib_no_team],
         new_type="planning",
         new_team="backend",
-        new_assignee="be-pm",
+        new_assignee="be-pm-2",
     )
     assert env is not None, "Empty team on sibling must NOT bypass the cap"
 
-    sib = _sibling(task_type="planning", team="backend")
-    env2 = Choreographer._sibling_dup_envelope(
-        sibling=sib,
+    sib = _sibling(task_type="planning", team="backend", assignee="be-pm")
+    env2 = Choreographer._sibling_cap_envelope(
+        siblings=[sib],
         new_type="planning",
         new_team="",
-        new_assignee="be-pm",
+        new_assignee="be-pm-2",
     )
     assert env2 is not None, "Empty team on new task must NOT bypass the cap"

tests/unit/gateway/test_spine_cap_remediate.py

@@ -27,6 +27,7 @@ def test_spine_cap_remediate_forbids_task_type_workaround() -> None:
         new_type="code",
         sibling=sibling,
         sib_assignee="be-dev-1",
+        cap=2,
     )
     remediate = env.remediate or ""
     assert "DO NOT work around" in remediate, (
@@ -43,6 +44,7 @@ def test_spine_cap_remediate_warns_about_verification_subtasks() -> None:
         new_type="code",
         sibling=sibling,
         sib_assignee="be-dev-1",
+        cap=2,
     )
     remediate = env.remediate or ""
     # Anti-pattern names — must appear so the model pattern-matches its own behavior.
@@ -57,5 +59,6 @@ def test_spine_cap_envelope_is_invalid_state() -> None:
         new_type="code",
         sibling=sibling,
         sib_assignee="be-dev-1",
+        cap=2,
     )
     assert env.error == "invalid_state"