fix: unblock smoke run (gateway envelope + alembic + redis + MCP)

Four bugs surfaced by the 2026-05-11 smoke run, all on the path from
Main PM's first delegate to the cell PM accepting a subtask:

- gateway: TaskCompletenessError from _create_subtask_from_inputs leaked
  through Starlette as a 500; agents retried in a tight loop because
  they never saw field_hints. Wrap the call in _create_subtask_and_envelope,
  catch the error, return Envelope.incomplete_input with the
  interrogation-pattern reply the upfront completeness check produces.

- alembic: migration 012 used a 40-char revision id which exceeds
  alembic_version.version_num varchar(32). Upgrade fell back to create_all
  on every boot, silently skipping the migration. Rename to
  012_align_agentrole_foundation (30 chars). File rename + revision string.

- events/stream_bus: external Redis FLUSHALL while orchestrator is running
  (e.g. reset_runtime_state.sh) drops the consumer group; the listen loop
  then spams NOGROUP every block-cycle forever. Catch ResponseError with
  NOGROUP in the message and rebootstrap the group via
  _ensure_consumer_group, then continue. Self-heals without restart.

- mcp/flow_server: delegate took body: dict with no schema, so the LLM
  invented values like nature='standard' and the SDK threw 'unhashable
  type: dict' on nested args. Flatten to typed top-level parameters with
  docstring listing valid enum values for team / task_type / nature /
  estimated_complexity. PLR0913 per-file ignore added for roboco/mcp/**
  because MCP tool signatures ARE the LLM contract — bundling into a
  dataclass would hide the enum hints that prevent the invention bug.

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

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"]

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)

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]:

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,