feat(gateway): restore Gate Set B delegation-time guards

PARENT_NOT_CLAIMED: Choreographer.delegate now enforces that the parent
task is in_progress AND assigned to the calling PM before allowing
subtask creation. Pre-gateway this was implicit (orchestrator only
spawned PMs after they claimed their parent); the gateway exposes
delegate as a first-class verb so the gate must be explicit.

SUBTASK_CAP: hard-blocks delegation when the parent already has 12
subtasks. Pre-gateway never had this cap because PMs naturally never
created more than a handful per spawn cycle; with delegate as a verb
agents can loop, so a cap is needed.

The _delegate_guard helper was split into _delegate_role_guards,
_delegate_static_guards, and _delegate_lifecycle_guards to keep each
piece below the PLR0911 return-count threshold and make the layered
gating explicit.

Pre-gateway reference: implicit in roboco/runtime/orchestrator.py
spawn flow; restored here as explicit server-side enforcement.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Renn F
2026-05-03 03:26:00 +02:00
co-authored by Claude Opus 4.7
parent 5c0011c90b
commit 466cc8d8f7
3 changed files with 342 additions and 3 deletions
@@ -166,11 +166,17 @@ async def test_delegate_main_pm_to_cell_pm_creates_subtask() -> None:
main_pm_id = uuid4()
parent_id = uuid4()
project_id = uuid4()
parent = MagicMock(id=parent_id, project_id=project_id)
parent = MagicMock(
id=parent_id,
project_id=project_id,
status="in_progress",
assigned_to=main_pm_id,
)
new_task = MagicMock(id=uuid4())
task_svc = AsyncMock()
task_svc.get.return_value = parent
task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm")
task_svc.get_subtasks.return_value = []
task_svc.create_subtask.return_value = new_task
deps = _make_deps(task=task_svc)
c = Choreographer(deps)
@@ -198,11 +204,17 @@ async def test_delegate_cell_pm_to_team_dev_creates_subtask() -> None:
cell_pm_id = uuid4()
parent_id = uuid4()
project_id = uuid4()
parent = MagicMock(id=parent_id, project_id=project_id)
parent = MagicMock(
id=parent_id,
project_id=project_id,
status="in_progress",
assigned_to=cell_pm_id,
)
new_task = MagicMock(id=uuid4())
task_svc = AsyncMock()
task_svc.get.return_value = parent
task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend")
task_svc.get_subtasks.return_value = []
task_svc.create_subtask.return_value = new_task
deps = _make_deps(task=task_svc)
c = Choreographer(deps)