diff --git a/alembic/versions/012_align_agentrole_team_with_foundation.py b/alembic/versions/012_align_agentrole_foundation.py similarity index 92% rename from alembic/versions/012_align_agentrole_team_with_foundation.py rename to alembic/versions/012_align_agentrole_foundation.py index 61ef4342..5e8176c3 100644 --- a/alembic/versions/012_align_agentrole_team_with_foundation.py +++ b/alembic/versions/012_align_agentrole_foundation.py @@ -5,7 +5,7 @@ enum values cannot be removed without a destructive recreation, so the inverse direction (postgres has extras the foundation lacks) is handled in foundation by keeping the legacy value (e.g., Team.MARKETING). -Revision ID: 012_align_agentrole_team_with_foundation +Revision ID: 012_align_agentrole_foundation Revises: 011_drop_quarantined_state Create Date: 2026-05-10 """ @@ -14,7 +14,7 @@ from __future__ import annotations from alembic import context, op -revision = "012_align_agentrole_team_with_foundation" +revision = "012_align_agentrole_foundation" down_revision = "011_drop_quarantined_state" branch_labels = None depends_on = None diff --git a/pyproject.toml b/pyproject.toml index 3424628e..6a283633 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -158,7 +158,11 @@ select = [ # Lazy imports to avoid circular dependencies [tool.ruff.lint.per-file-ignores] -"roboco/mcp/**/*.py" = ["PLC0415"] +# MCP tool surfaces ARE the LLM-facing contract — every parameter must be +# top-level + typed so the SDK exposes it as a discrete schema field with +# enum constraints. Bundling into a dataclass would hide enum hints from +# the LLM and bring back invented values like nature='standard'. +"roboco/mcp/**/*.py" = ["PLC0415", "PLR0913"] "roboco/services/*.py" = ["PLC0415"] "roboco/api/routes/*.py" = ["PLC0415"] "roboco/runtime/*.py" = ["PLC0415"] diff --git a/roboco/events/stream_bus.py b/roboco/events/stream_bus.py index 69ddd96f..cab65328 100644 --- a/roboco/events/stream_bus.py +++ b/roboco/events/stream_bus.py @@ -230,6 +230,21 @@ class StreamEventBus: except asyncio.CancelledError: break + except ResponseError as e: + # NOGROUP: consumer group disappeared (e.g. Redis FLUSHALL by + # an external cleanup while the orchestrator is still up). + # Re-bootstrap the groups so we self-heal instead of spamming + # the same error every block-cycle. + if "NOGROUP" in str(e): + logger.warning( + "Stream consumer group missing; recreating", + group=self.group_name, + ) + for stream in streams: + await self._ensure_consumer_group(stream) + continue + logger.error("Error in stream event loop", error=str(e)) + await asyncio.sleep(1) except Exception as e: logger.error("Error in stream event loop", error=str(e)) await asyncio.sleep(1) diff --git a/roboco/mcp/flow_server.py b/roboco/mcp/flow_server.py index eab9e221..cd4f10bc 100644 --- a/roboco/mcp/flow_server.py +++ b/roboco/mcp/flow_server.py @@ -331,20 +331,44 @@ def i_will_plan(task_id: str, plan: str) -> dict[str, Any]: def delegate( - parent_task_id: str, title: str, description: str, body: dict + parent_task_id: str, + title: str, + description: str, + assigned_to: str, + team: str, + task_type: str, + nature: str, + acceptance_criteria: list[str], + estimated_complexity: str = "medium", ) -> dict[str, Any]: """PM: create a subtask of parent_task_id. - Required body keys: ``assigned_to``, ``team``. Optional: ``task_type``, - ``acceptance_criteria``, ``estimated_complexity``. + Args: + parent_task_id: UUID of the parent task. + title: Short imperative title. + description: Multi-paragraph description with context (>=20 chars). + assigned_to: Agent slug receiving the task (e.g. "be-dev-1"). + team: One of "backend" | "frontend" | "ux_ui" | "board" | "main_pm". + task_type: One of "code" | "documentation" | "research" | "planning" + | "design" | "administrative". + nature: One of "technical" | "non_technical". + acceptance_criteria: Non-empty list of verifiable outcome strings. + estimated_complexity: One of "low" | "medium" | "high". Default "medium". """ - payload: dict[str, Any] = { - "parent_task_id": parent_task_id, - "title": title, - "description": description, - } - payload.update(body) - return _post(_role_path("delegate"), payload) + return _post( + _role_path("delegate"), + { + "parent_task_id": parent_task_id, + "title": title, + "description": description, + "assigned_to": assigned_to, + "team": team, + "task_type": task_type, + "nature": nature, + "acceptance_criteria": acceptance_criteria, + "estimated_complexity": estimated_complexity, + }, + ) def submit_up(task_id: str, notes: str) -> dict[str, Any]: diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 9ba0ec6b..417a3c4e 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -1923,15 +1923,9 @@ class Choreographer: task_id=parent_task_id, verb="delegate", ) - new_task = await self._create_subtask_from_inputs( - pm_agent_id, parent_task_id, parent, inputs + return await self._create_subtask_and_envelope( + pm_agent_id, parent, inputs, briefing, role_str ) - return Envelope.ok( - status="created", - task_id=str(new_task.id), - next=spec_module._INTENT_VERBS["delegate"].next_hint(new_task), - context_briefing=briefing, - ).with_introspection(task=new_task, role=role_str) # 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``. @@ -2175,6 +2169,54 @@ class Choreographer: context_briefing=briefing, ).with_introspection(task=parent, role=role_str) + async def _create_subtask_and_envelope( + self, + pm_agent_id: UUID, + parent: Any, + inputs: DelegateInputs, + briefing: dict[str, Any], + role_str: str, + ) -> Envelope: + """Run subtask creation and translate completeness raises into envelopes. + + The defensive raises inside `_create_subtask_from_inputs` (Task 18) + catch under-filled payloads that slipped past the gateway gate. Without + this translator they surface as Starlette 500s — which means the agent + never sees `field_hints`, retries indefinitely, and looks like a + runaway. Converting to `Envelope.incomplete_input` here closes that + loop so the agent gets the same interrogation-pattern reply it would + have gotten from the upfront completeness check. + """ + from roboco.foundation.policy.task_completeness import TaskCompletenessError + + parent_task_id = parent.id + try: + new_task = await self._create_subtask_from_inputs( + pm_agent_id, parent_task_id, parent, inputs + ) + except TaskCompletenessError as exc: + return await self._emit_rejection( + Envelope.incomplete_input( + missing=exc.missing, + field_hints=exc.field_hints, + remediate=( + "re-issue delegate(...) with corrected fields: " + f"{', '.join(exc.missing)}. Each field's required " + "shape is in `field_hints`." + ), + context_briefing=briefing, + ).with_introspection(task=parent, role=role_str), + agent_id=pm_agent_id, + task_id=parent_task_id, + verb="delegate", + ) + return Envelope.ok( + status="created", + task_id=str(new_task.id), + next=spec_module._INTENT_VERBS["delegate"].next_hint(new_task), + context_briefing=briefing, + ).with_introspection(task=new_task, role=role_str) + async def _create_subtask_from_inputs( self, pm_agent_id: UUID,