diff --git a/roboco/services/gateway/choreographer.py b/roboco/services/gateway/choreographer.py index bd9eabb9..99256745 100644 --- a/roboco/services/gateway/choreographer.py +++ b/roboco/services/gateway/choreographer.py @@ -950,6 +950,10 @@ class Choreographer: context_briefing=await self._briefing_for(pm_agent_id, parent_task_id), ) + # Gate Set B subtask cap (pre-gateway implicit, made explicit here). + # Soft warn at 8, hard block at 13. Cap enforced by ``_subtask_cap_guard``. + _SUBTASK_HARD_CAP: int = 12 + async def _delegate_guard( self, pm_agent_id: UUID, @@ -959,8 +963,27 @@ class Choreographer: inputs: DelegateInputs, ) -> Envelope | None: """Return rejection Envelope if a delegate precondition fails; else None.""" - from roboco.seeds.initial_data import AGENT_UUIDS + if guard := await self._delegate_role_guards( + pm_agent_id, parent_task_id, agent, inputs + ): + return guard + if guard := await self._delegate_static_guards( + pm_agent_id, parent_task_id, parent, inputs + ): + return guard + # Gate Set B: PARENT_NOT_CLAIMED + SUBTASK_CAP + return await self._delegate_lifecycle_guards( + pm_agent_id, parent_task_id, parent + ) + async def _delegate_role_guards( + self, + pm_agent_id: UUID, + parent_task_id: UUID, + agent: Any, + inputs: DelegateInputs, + ) -> Envelope | None: + """Role + delegation-chain guards (the original two).""" if agent is None or agent.role not in ("cell_pm", "main_pm"): return Envelope.not_authorized( message="only cell_pm or main_pm may delegate", @@ -978,6 +1001,18 @@ class Choreographer: ), context_briefing=await self._briefing_for(pm_agent_id, parent_task_id), ) + return None + + async def _delegate_static_guards( + self, + pm_agent_id: UUID, + parent_task_id: UUID, + parent: Any, + inputs: DelegateInputs, + ) -> Envelope | None: + """Slug / project_id / enum guards. Pure data-shape checks.""" + from roboco.seeds.initial_data import AGENT_UUIDS + if inputs.assigned_to not in AGENT_UUIDS: return Envelope.invalid_state( message=f"unknown agent slug: {inputs.assigned_to!r}", @@ -1000,6 +1035,61 @@ class Choreographer: ) return None + async def _delegate_lifecycle_guards( + self, + pm_agent_id: UUID, + parent_task_id: UUID, + parent: Any, + ) -> Envelope | None: + """Gate Set B: PARENT_NOT_CLAIMED + SUBTASK_CAP. + + Pre-gateway, the orchestrator enforced both implicitly: a PM only + ever called task_create after the orchestrator spawned them + post-claim, and naturally never created more than a handful of + subtasks in one spawn cycle. + + With the gateway exposing ``delegate`` as a first-class verb, + these gates must be explicit. + """ + if str(parent.status) != "in_progress": + return Envelope.invalid_state( + message=( + f"parent task {parent_task_id} is in {parent.status}; " + "must be in_progress to accept subtasks" + ), + remediate=( + f"call i_will_plan(parent_task_id='{parent_task_id}'," + " plan='...') before delegating subtasks" + ), + context_briefing=await self._briefing_for(pm_agent_id, parent_task_id), + ) + if parent.assigned_to != pm_agent_id: + return Envelope.not_authorized( + message=( + f"parent task {parent_task_id} is assigned to " + f"{parent.assigned_to}, not you" + ), + remediate=( + f"call i_will_plan(parent_task_id='{parent_task_id}'," + " plan='...') to claim before delegating subtasks" + ), + context_briefing=await self._briefing_for(pm_agent_id, parent_task_id), + ) + existing = await self.task.get_subtasks(parent_task_id) + if len(existing) >= self._SUBTASK_HARD_CAP: + return Envelope.invalid_state( + message=( + f"parent already has {len(existing)} subtasks; " + f"cap is {self._SUBTASK_HARD_CAP}" + ), + remediate=( + "consolidate or split into a separate parent task before" + " adding more subtasks" + ), + context_briefing=await self._briefing_for(pm_agent_id, parent_task_id), + ) + return None + @staticmethod def _resolve_delegate_enums(inputs: DelegateInputs) -> tuple[Any, Any, Any]: """Convert string inputs to Team/TaskType/Complexity enums. diff --git a/tests/unit/gateway/test_choreographer_delegate_guards.py b/tests/unit/gateway/test_choreographer_delegate_guards.py new file mode 100644 index 00000000..a98afb39 --- /dev/null +++ b/tests/unit/gateway/test_choreographer_delegate_guards.py @@ -0,0 +1,237 @@ +"""Gate Set B: delegation-time guards in Choreographer.delegate. + +Pre-gateway behavior: a PM could only call ``task_create`` while spawned +in-context, which only happened after the orchestrator saw them claim +and start a parent task. That implicit gate is restored explicitly here: + +- PARENT_NOT_CLAIMED: parent must be in_progress AND assigned_to PM. +- SUBTASK_CAP: 8 children = soft warn (allowed), >12 = hard block. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import ( + Choreographer, + ChoreographerDeps, + DelegateInputs, +) + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + repo = base["evidence_repo"] + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(repo, method).return_value = [] + return ChoreographerDeps(**base) + + +def _delegate_inputs() -> DelegateInputs: + return DelegateInputs( + title="Implement endpoint", + description="Add /v1/foo endpoint with tests", + assigned_to="be-dev-1", + team="backend", + ) + + +@pytest.mark.asyncio +async def test_delegate_blocks_when_parent_not_in_progress() -> None: + """Parent in 'pending' status (PM never called i_will_plan) blocks delegate.""" + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="pending", + assigned_to=pm_id, + ) + 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 = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] == "invalid_state" + assert "i_will_plan" in body["remediate"] + task_svc.create_subtask.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_blocks_when_parent_assigned_to_other_agent() -> None: + """Parent claimed by a different PM cannot be delegated against by us.""" + pm_id = uuid4() + other_pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=other_pm_id, + ) + 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 = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] == "not_authorized" + assert "i_will_plan" in body["remediate"] + task_svc.create_subtask.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_allows_when_parent_in_progress_and_owned() -> None: + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=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) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] is None + assert body["status"] == "created" + task_svc.create_subtask.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delegate_blocks_when_subtask_cap_exceeded() -> None: + """13+ subtasks = hard block (cap is 12).""" + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=pm_id, + ) + too_many = [MagicMock(id=uuid4()) for _ in range(13)] + 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 = too_many + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] == "invalid_state" + assert "13" in body["message"] or "consolidate" in body["remediate"].lower() + task_svc.create_subtask.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_delegate_allows_when_subtask_cap_within_soft_zone() -> None: + """8-12 subtasks: warn-but-allow.""" + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=pm_id, + ) + many = [MagicMock(id=uuid4()) for _ in range(10)] + 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 = many + task_svc.create_subtask.return_value = new_task + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] is None + task_svc.create_subtask.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delegate_allows_at_zero_subtasks() -> None: + """Empty subtask list is the common case.""" + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=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) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] is None + + +@pytest.mark.asyncio +async def test_delegate_blocks_at_exact_cap_plus_one() -> None: + """Cap is 12; 13th attempt blocks.""" + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock( + id=parent_id, + project_id=uuid4(), + status="in_progress", + assigned_to=pm_id, + ) + # Already 12 children — adding the 13th must be blocked. + twelve = [MagicMock(id=uuid4()) for _ in range(12)] + 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 = twelve + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate(pm_id, parent_id, _delegate_inputs()) + body = env.as_dict() + assert body["error"] == "invalid_state" + task_svc.create_subtask.assert_not_awaited() diff --git a/tests/unit/gateway/test_choreographer_pm_extras.py b/tests/unit/gateway/test_choreographer_pm_extras.py index eb199de0..5c825978 100644 --- a/tests/unit/gateway/test_choreographer_pm_extras.py +++ b/tests/unit/gateway/test_choreographer_pm_extras.py @@ -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)