fix: restore pre-gateway structured verb surfaces (5 fixes)

Smoke run 2026-05-11 showed five regressions stemming from the gateway
consolidating multiple typed endpoints into thin verbs with collapsed
signatures. The choreography is fine; the verb signatures lost the
structured shape that pre-gateway forced agents to fill. Each fix
restores a structured surface so the LLM's tool schema again carries
the constraints that prevent the observed bugs.

A) do_server: list valid channel slugs in say()/dm() docstrings.
   Stops invented channels (`backend`, `backend-dev`) — the LLM now
   sees the closed set in the tool schema.

B) choreographer: add _delegate_sibling_dedup_guard.
   Rejects a delegate that would create a non-terminal sibling with
   the same assigned_to + task_type under the same parent — the dupe
   shape observed on smoke (Main PM creating two planning tasks for
   be-pm; Cell PM creating two code tasks for be-dev-1).

C) choreographer: extend _validate_assignee_task_type to all roles.
   Devs may only get code|documentation|research (not planning/design/
   administrative). QA gets code only. Documenters get documentation
   only. Catches the misroute observed on smoke (Cell PM gave
   be-dev-2 a 'research' coordination task that should have stayed
   with the PM).

D) i_will_plan: thread approach / technical_considerations / risks /
   open_questions from MCP through to TaskService.set_plan as a
   TaskPlan-shaped dict. Empty default keeps back-compat. Panel's
   Plan tab now renders Approach / Sub-Tasks / Technical
   Considerations / Risks / Open Questions instead of an empty pane.

E) note(): scope-specific structured fields restored.
   For 'decision' scope: context, options[], chosen, rationale,
   consequences. For 'reflect' scope: what_done, what_learned,
   what_struggled, next_steps. Rendered as markdown sections into
   the journal entry content so the Decisions and Reflections views
   show named blocks instead of a one-line phrase. Pre-gateway parity.

Files changed:
- roboco/mcp/do_server.py (A, E)
- roboco/mcp/flow_server.py (D)
- roboco/services/gateway/choreographer/_impl.py (B, C, D)
- roboco/services/gateway/content_actions.py (E)
- roboco/api/schemas/v2/flow.py (D)
- roboco/api/schemas/v2/do.py (E)
- roboco/api/routes/v2/flow_main_pm.py (D)
- roboco/api/routes/v2/flow_cell_pm.py (D)
- roboco/api/routes/v2/do.py (E)

Quality: ruff + mypy clean. 89 unit tests pass on the touched surfaces.

roboco/api/routes/v2/do.py

@@ -50,6 +50,18 @@ async def do_note(
         text=body.text,
         scope=body.scope,
         task_id=body.task_id,
+        structured={
+            "title": body.title,
+            "context": body.context,
+            "options": body.options,
+            "chosen": body.chosen,
+            "rationale": body.rationale,
+            "consequences": body.consequences,
+            "what_done": body.what_done,
+            "what_learned": body.what_learned,
+            "what_struggled": body.what_struggled,
+            "next_steps": body.next_steps,
+        },
     )
     return envelope_to_response(env, request)
 

roboco/api/routes/v2/flow_cell_pm.py

@@ -51,7 +51,17 @@ async def i_will_plan(
     x_agent_id: _AgentIdHeader,
     choreographer: _ChoreographerDep,
 ) -> dict:
-    env = await choreographer.i_will_plan(x_agent_id, body.task_id, body.plan)
+    env = await choreographer.i_will_plan(
+        x_agent_id,
+        body.task_id,
+        body.plan,
+        rich_plan={
+            "approach": body.approach,
+            "technical_considerations": body.technical_considerations,
+            "risks": body.risks,
+            "open_questions": body.open_questions,
+        },
+    )
     return envelope_to_response(env, request)
 
 

roboco/api/routes/v2/flow_main_pm.py

@@ -51,7 +51,17 @@ async def i_will_plan(
     x_agent_id: _AgentIdHeader,
     choreographer: _ChoreographerDep,
 ) -> dict:
-    env = await choreographer.i_will_plan(x_agent_id, body.task_id, body.plan)
+    env = await choreographer.i_will_plan(
+        x_agent_id,
+        body.task_id,
+        body.plan,
+        rich_plan={
+            "approach": body.approach,
+            "technical_considerations": body.technical_considerations,
+            "risks": body.risks,
+            "open_questions": body.open_questions,
+        },
+    )
     return envelope_to_response(env, request)
 
 

roboco/api/schemas/v2/do.py

@@ -11,9 +11,36 @@ class CommitRequest(BaseModel):
 
 
 class NoteRequest(BaseModel):
+    """Journal entry. ``text`` is always the short summary line.
+
+    Scope-specific fields are optional but pre-gateway parity expected
+    them filled for `decision` and `reflect`:
+
+    - decision: ``context``, ``options``, ``chosen``, ``rationale``,
+      ``consequences``
+    - reflect: ``what_done``, ``what_learned``, ``what_struggled``,
+      ``next_steps``
+
+    When provided, these are formatted into the journal entry's content
+    as structured markdown — the panel UI's decision/reflect views show
+    them as named sections instead of a one-line phrase.
+    """
+
     text: str = Field(..., min_length=1)
     scope: str = "note"
     task_id: UUID | None = None
+    title: str | None = None
+    # decision scope
+    context: str | None = None
+    options: list[str] | None = None
+    chosen: str | None = None
+    rationale: str | None = None
+    consequences: str | None = None
+    # reflect scope
+    what_done: str | None = None
+    what_learned: str | None = None
+    what_struggled: str | None = None
+    next_steps: str | None = None
 
 
 class SayRequest(BaseModel):

roboco/api/schemas/v2/flow.py

@@ -91,6 +91,14 @@ class EscalateToCeoRequest(BaseModel):
 class IWillPlanRequest(BaseModel):
     task_id: UUID
     plan: str = Field(..., min_length=1)
+    # Optional rich-plan fields. These persist into Task.plan as a structured
+    # dict matching roboco.models.task.TaskPlan, so the panel's Plan tab
+    # shows Approach / Sub-Tasks / Technical Considerations / Risks /
+    # Open Questions instead of an empty pane. Pre-gateway parity.
+    approach: str = ""
+    technical_considerations: list[str] = Field(default_factory=list)
+    risks: list[dict[str, str]] = Field(default_factory=list)
+    open_questions: list[dict[str, str | bool]] = Field(default_factory=list)
 
 
 class DelegateRequest(BaseModel):

roboco/mcp/do_server.py

@@ -86,13 +86,70 @@ def commit(message: str, files: list[str] | None = None) -> dict[str, Any]:
     return _post("/api/v2/do/commit", {"message": message, "files": files})
 
 
-def note(text: str, scope: str = "note", task_id: str | None = None) -> dict[str, Any]:
-    """Write a journal entry. scope in note|decision|reflect|learning|struggle."""
-    return _post("/api/v2/do/note", {"text": text, "scope": scope, "task_id": task_id})
+def note(
+    text: str,
+    scope: str = "note",
+    task_id: str | None = None,
+    title: str | None = None,
+    context: str | None = None,
+    options: list[str] | None = None,
+    chosen: str | None = None,
+    rationale: str | None = None,
+    consequences: str | None = None,
+    what_done: str | None = None,
+    what_learned: str | None = None,
+    what_struggled: str | None = None,
+    next_steps: str | None = None,
+) -> dict[str, Any]:
+    """Write a journal entry. scope in note|decision|reflect|learning|struggle.
+
+    ``text`` is always the short summary (one paragraph max). For ``decision``
+    and ``reflect`` scopes, fill the scope-specific structured fields so the
+    panel renders them as named sections — pre-gateway parity:
+
+    - decision: ``context`` (the situation), ``options`` (list of strings,
+      one per alternative considered), ``chosen`` (the alternative you took),
+      ``rationale`` (why), ``consequences`` (what this commits us to)
+    - reflect: ``what_done`` (literal output), ``what_learned`` (new info),
+      ``what_struggled`` (where you got stuck), ``next_steps`` (follow-ups)
+
+    Other scopes (note / learning / struggle) just need ``text``.
+    """
+    return _post(
+        "/api/v2/do/note",
+        {
+            "text": text,
+            "scope": scope,
+            "task_id": task_id,
+            "title": title,
+            "context": context,
+            "options": options,
+            "chosen": chosen,
+            "rationale": rationale,
+            "consequences": consequences,
+            "what_done": what_done,
+            "what_learned": what_learned,
+            "what_struggled": what_struggled,
+            "next_steps": next_steps,
+        },
+    )
 
 
 def say(channel: str, text: str, task_id: str | None = None) -> dict[str, Any]:
-    """Post to a channel. task_id auto-injected if you have an active task."""
+    """Post to a channel. task_id auto-injected if you have an active task.
+
+    Args:
+        channel: Channel slug WITHOUT leading `#`. Valid values:
+            - Cell channels: `backend-cell`, `frontend-cell`, `uxui-cell`
+            - Cross-cell: `dev-all`, `qa-all`, `pm-all`, `doc-all`
+            - Management: `main-pm-board`, `board-private`
+            - Broadcast: `announcements`, `all-hands`
+            Write access varies by role — gateway returns `not_authorized` if
+            you cannot write to the requested channel; the error lists which
+            channels you can write to.
+        text: Message body.
+        task_id: Optional; auto-filled from your active task if omitted.
+    """
     return _post(
         "/api/v2/do/say",
         {"channel": channel, "text": text, "task_id": task_id},
@@ -105,7 +162,14 @@ def dm(
     task_id: str | None = None,
     skill: str | None = None,
 ) -> dict[str, Any]:
-    """A2A message. Auto-creates conversation; auto-resolves skill if needed."""
+    """A2A message. Auto-creates conversation; auto-resolves skill if needed.
+
+    Args:
+        recipient: Target agent slug (e.g. `be-pm`, `be-dev-1`, `ceo`).
+        text: Message body.
+        task_id: Optional; auto-filled from your active task if omitted.
+        skill: Optional skill slug to scope the conversation.
+    """
     return _post(
         "/api/v2/do/dm",
         {"recipient": recipient, "text": text, "task_id": task_id, "skill": skill},

roboco/mcp/flow_server.py

@@ -325,9 +325,38 @@ def escalate_to_ceo(task_id: str, reason: str) -> dict[str, Any]:
 # PM lifecycle so PMs can drive parent tasks instead of stalling.
 
 
-def i_will_plan(task_id: str, plan: str) -> dict[str, Any]:
-    """PM: claim+start a pending parent task with a one-paragraph plan."""
-    return _post(_role_path("i_will_plan"), {"task_id": task_id, "plan": plan})
+def i_will_plan(
+    task_id: str,
+    plan: str,
+    approach: str = "",
+    technical_considerations: list[str] | None = None,
+    risks: list[dict[str, str]] | None = None,
+    open_questions: list[dict[str, str | bool]] | None = None,
+) -> dict[str, Any]:
+    """PM: claim+start a pending parent task with a structured plan.
+
+    Args:
+        task_id: UUID of the task you are planning.
+        plan: One-paragraph narrative (the agent-facing summary).
+        approach: 2-4 sentences describing the high-level approach for the
+            Plan tab. Required for non-trivial tasks; empty string is allowed
+            but produces an unpopulated Plan view.
+        technical_considerations: Bullet list of architectural / library /
+            constraint notes. Each item is a single string.
+        risks: List of {"risk": "...", "mitigation": "..."} entries.
+        open_questions: List of {"question": "...", "answered": false} entries.
+    """
+    return _post(
+        _role_path("i_will_plan"),
+        {
+            "task_id": task_id,
+            "plan": plan,
+            "approach": approach,
+            "technical_considerations": technical_considerations or [],
+            "risks": risks or [],
+            "open_questions": open_questions or [],
+        },
+    )
 
 
 def delegate(

roboco/services/gateway/choreographer/_impl.py

@@ -99,7 +99,7 @@ class _ClaimPlanStartContext:
     task: Any
     role_str: str
     briefing: dict[str, Any]
-    plan: str | None
+    plan: str | dict[str, Any] | None
     verb_name: str
 
 
@@ -1758,7 +1758,11 @@ class Choreographer:
     # claim_doc_task + i_documented moved to ``doc.py`` (audit P2-2).
 
     async def i_will_plan(
-        self, pm_agent_id: UUID, task_id: UUID, plan: str
+        self,
+        pm_agent_id: UUID,
+        task_id: UUID,
+        plan: str,
+        rich_plan: dict[str, Any] | None = None,
     ) -> Envelope:
         """PM mirror of i_will_work_on for parent tasks.
 
@@ -1768,6 +1772,14 @@ class Choreographer:
         the DB. Idempotent re-entry: a respawned PM re-calling on a
         task they already own in claimed/in_progress just refreshes
         the heartbeat.
+
+        The rich-plan kwargs (``approach``, ``technical_considerations``,
+        ``risks``, ``open_questions``) populate the panel's Plan tab
+        (pre-gateway parity). When any are non-empty/non-default they
+        are persisted as a structured ``TaskPlan``-shaped dict via
+        ``TaskService.set_plan``; otherwise ``plan`` is stored as a
+        narrative string. Empty defaults keep behavior backward-compatible
+        for callers that don't pass rich fields.
         """
         t = await self.task.get(task_id)
         if t is None:
@@ -1799,13 +1811,39 @@ class Choreographer:
             actor_slug=getattr(agent, "slug", None) if agent is not None else None,
             original_developer_slug=_extract_original_developer(t),
         )
+        # Persist the structured plan dict (pre-gateway parity for the Plan
+        # tab) when the caller passed any rich field; otherwise fall through
+        # with the bare string so set_plan stores {"text": plan}.
+        effective_plan: str | dict[str, Any]
+        if rich_plan and any(
+            rich_plan.get(k)
+            for k in (
+                "approach",
+                "sub_tasks",
+                "technical_considerations",
+                "risks",
+                "open_questions",
+            )
+        ):
+            effective_plan = {
+                "text": plan,
+                "approach": rich_plan.get("approach", ""),
+                "sub_tasks": rich_plan.get("sub_tasks", []),
+                "technical_considerations": rich_plan.get(
+                    "technical_considerations", []
+                ),
+                "risks": rich_plan.get("risks", []),
+                "open_questions": rich_plan.get("open_questions", []),
+            }
+        else:
+            effective_plan = plan
         ctx = _ClaimPlanStartContext(
             agent_id=pm_agent_id,
             task_id=task_id,
             task=t,
             role_str=role_str,
             briefing=briefing,
-            plan=plan,
+            plan=effective_plan,
             verb_name="i_will_plan",
         )
         # Idempotent re-entry: PM already owns the task in_progress.
@@ -1969,11 +2007,55 @@ class Choreographer:
             pm_agent_id, parent_task_id, parent, inputs
         ):
             return guard
+        # Sibling dedup: catch the PM-decomposition bug where the same
+        # parent gets two subtasks for the same role + task_type
+        # (observed on smoke run 2026-05-11: Main PM created two planning
+        # tasks for be-pm; Cell PM created two code tasks for be-dev-1).
+        if guard := await self._delegate_sibling_dedup_guard(parent_task_id, 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_sibling_dedup_guard(
+        self,
+        parent_task_id: UUID,
+        inputs: DelegateInputs,
+    ) -> Envelope | None:
+        """Block delegation when a non-terminal sibling owns the same slot.
+
+        Same slot = same ``assigned_to`` + same ``task_type``. If a PM has
+        already delegated work to that agent of that type under this parent
+        and it isn't completed/cancelled, the new delegation is almost
+        certainly the PM decomposing twice. Reject with the existing
+        task_id so the PM can finish or cancel that one instead.
+        """
+        terminal = {"completed", "cancelled"}
+        siblings = await self.task.get_subtasks(parent_task_id)
+        for s in siblings:
+            if str(getattr(s, "status", "")) in terminal:
+                continue
+            if (
+                getattr(s, "assigned_to", None) is not None
+                and str(getattr(s, "assigned_to", "")) == str(inputs.assigned_to or "")
+                and str(getattr(s, "task_type", "")) == str(inputs.task_type or "")
+            ):
+                return Envelope.invalid_state(
+                    message=(
+                        f"sibling subtask already assigned to "
+                        f"{inputs.assigned_to!r} with task_type="
+                        f"{inputs.task_type!r}: id={s.id} status={s.status}"
+                    ),
+                    remediate=(
+                        "Either drive the existing sibling to completion / "
+                        "cancel it, or split this work into a subtask of "
+                        "the existing sibling rather than a new sibling."
+                    ),
+                    context_briefing={},
+                )
+        return None
+
     async def _delegate_static_guards(
         self,
         pm_agent_id: UUID,
@@ -2023,18 +2105,50 @@ class Choreographer:
     def _validate_assignee_task_type(assigned_to: str, task_type: str) -> str | None:
         """Reject role-vs-type misclassifications.
 
-        Rule (2026-05-09 smoke Bug B): when delegating to a Cell PM, the
-        subtask must be `planning`-typed. The Cell PM owns the planning
-        of the slice and delegates code execution to devs; a code-typed
-        task assigned to a Cell PM conflates the two layers and made
-        the lifecycle harder to reason about (a code task that nobody
-        will execute, just plan).
+        Rules:
+        - (2026-05-09 smoke Bug B): delegating to a Cell PM requires
+          ``task_type='planning'``. Cell PMs decompose; they don't execute.
+        - (2026-05-11 smoke): delegating to a Developer requires
+          ``task_type in {'code', 'documentation', 'research'}``. Devs
+          implement. Planning/design/administrative belong to PMs/board.
+          The 'research' allowance covers genuine spike work (try a
+          library, prototype an approach) — NOT coordination/handoff,
+          which is PM work.
+        - Delegating to a QA requires ``task_type='code'`` (their work is
+          to review PRs of code changes).
+        - Delegating to a Documenter requires ``task_type='documentation'``.
         """
+        from roboco.foundation.identity import AGENTS, Role
+
         if assigned_to in Choreographer._CELL_PM_SLUGS and task_type != "planning":
             return (
                 f"task_type={task_type!r} is invalid for assignee {assigned_to!r}: "
                 f"Cell PMs own planning tasks, not code/documentation/etc."
             )
+        agent = AGENTS.get(assigned_to)
+        if agent is None:
+            return None
+        if agent.role is Role.DEVELOPER and task_type not in {
+            "code",
+            "documentation",
+            "research",
+        }:
+            return (
+                f"task_type={task_type!r} is invalid for assignee {assigned_to!r}: "
+                f"Developers own code/documentation/research. Coordination, "
+                f"planning, design, and administrative work belong to PMs."
+            )
+        if agent.role is Role.QA and task_type != "code":
+            return (
+                f"task_type={task_type!r} is invalid for assignee {assigned_to!r}: "
+                f"QA reviews code PRs — task_type must be 'code'."
+            )
+        if agent.role is Role.DOCUMENTER and task_type != "documentation":
+            return (
+                f"task_type={task_type!r} is invalid for assignee {assigned_to!r}: "
+                f"Documenters write documentation — task_type must be "
+                f"'documentation'."
+            )
         return None
 
     async def _delegate_lifecycle_guards(

roboco/services/gateway/content_actions.py

@@ -48,6 +48,56 @@ _NOTIFY_ALLOWED_ROLES: frozenset[str] = frozenset(
 )
 
 
+_DECISION_SECTIONS: tuple[tuple[str, str], ...] = (
+    ("context", "Context"),
+    ("options", "Options Considered"),
+    ("chosen", "Chosen"),
+    ("rationale", "Rationale"),
+    ("consequences", "Consequences"),
+)
+
+_REFLECT_SECTIONS: tuple[tuple[str, str], ...] = (
+    ("what_done", "What Done"),
+    ("what_learned", "What Learned"),
+    ("what_struggled", "What Struggled"),
+    ("next_steps", "Next Steps"),
+)
+
+
+def _render_journal_content(scope: str, text: str, structured: dict[str, Any]) -> str:
+    """Build the journal entry body. Pre-gateway parity for decision/reflect.
+
+    For scopes that have a structured shape (``decision``, ``reflect``), append
+    a markdown section for each populated field. Other scopes return ``text``
+    unchanged. The original ``text`` always lands first so consumers that
+    only render flat content still see the summary line.
+    """
+    sections = (
+        _DECISION_SECTIONS
+        if scope == "decision"
+        else _REFLECT_SECTIONS
+        if scope == "reflect"
+        else ()
+    )
+    if not sections:
+        return text
+    body_parts: list[str] = [text.strip()] if text.strip() else []
+    for key, label in sections:
+        value = structured.get(key)
+        if value is None:
+            continue
+        if isinstance(value, list):
+            if not value:
+                continue
+            rendered = "\n".join(f"- {item}" for item in value)
+        else:
+            rendered = str(value).strip()
+            if not rendered:
+                continue
+        body_parts.append(f"## {label}\n{rendered}")
+    return "\n\n".join(body_parts) if body_parts else text
+
+
 def _ownership_violation(task_id: UUID) -> Envelope:
     """Standard envelope for Gate Set D ownership violations.
 
@@ -200,8 +250,21 @@ class ContentActions:
         text: str,
         scope: str = "note",
         task_id: UUID | None = None,
+        structured: dict[str, Any] | None = None,
     ) -> Envelope:
-        """Write a journal entry. scope ∈ note|decision|reflect|learning|struggle."""
+        """Write a journal entry. scope ∈ note|decision|reflect|learning|struggle.
+
+        ``structured`` carries scope-specific fields (pre-gateway parity):
+
+        - decision: context, options[], chosen, rationale, consequences
+        - reflect: what_done, what_learned, what_struggled, next_steps
+
+        Non-None fields are formatted into the entry content as markdown
+        sections so the panel's Decisions / Reflections views render
+        them as named blocks instead of a one-line phrase. The ``title``
+        is taken from ``structured["title"]`` when present, otherwise
+        from the first line of ``text``.
+        """
         if scope not in _VALID_NOTE_SCOPES:
             return Envelope.invalid_state(
                 message=f"invalid scope {scope!r}",
@@ -215,13 +278,15 @@ class ContentActions:
             t = await self.task.get_active_task_for_agent(agent_id)
             if t is not None:
                 task_id = t.id
-        title = text.split("\n", 1)[0][:200]
+        s = structured or {}
+        title = (s.get("title") or text.split("\n", 1)[0])[:200]
+        content = _render_journal_content(scope, text, s)
         await self.journal.write_entry(
             agent_id=agent_id,
             task_id=task_id,
             scope=scope,
             title=title,
-            content=text,
+            content=content,
         )
         return Envelope.ok(
             status="noted",