feat(gateway): G8 part b — typed blocker_type + what_needed on i_am_blocked

Pre-gateway parity (G8 part b of the 2026-05-11 design). The pre-gateway
TaskBlockInput at 254cc93:roboco/mcp/schemas/__init__.py required
blocker_type (external|internal|question|dependency) and what_needed
so PMs could triage their inbox by class. Current i_am_blocked dropped
both fields — every blocked task looked the same to the PM.

Now i_am_blocked accepts both as optional kwargs:
- Back-compat: callers that omit them still work (blocker_type defaults
  to None → rendered as flat reason in the struggle entry).
- New: when supplied, the struggle journal entry body is structured
  markdown (## Blocker Type / ## What Needed sections) so the panel's
  journal view renders named blocks instead of one flat sentence.

Validator on blocker_type enforces the enum at the Pydantic boundary
with a clear "must be one of: ..." error if the agent invents a value
(same pattern as the Wave 3 G7 validators).

G8 part a — typed `pause(checkpoint_summary, remaining_work)` — defers.
That gap needs a new IntentSpec in foundation/policy/lifecycle.py
(currently pause is an ActionSpec only; agents auto-pause via i_am_idle)
plus checkpoint wiring through TaskService.add_checkpoint. Material
work, deferred until after the user has deployed and verified G7 +
G8b lands cleanly.

Wired:
- roboco/api/schemas/v2/flow.py — IAmBlockedRequest gains optional
  blocker_type + what_needed; @field_validator enforces the enum
- roboco/api/routes/v2/flow_dev.py — passes the new fields through
- roboco/services/gateway/choreographer/_impl.py — i_am_blocked signature
  + structured struggle-entry rendering
- roboco/mcp/flow_server.py — typed wrapper with the kwargs
- agents/prompts/roles/developer.md — updated verb table
- tests/unit/mcp_servers/test_flow_server.py — updated to expect the
  new optional kwargs as None when omitted

Quality: ruff + mypy clean. 505 tests pass.

agents/prompts/roles/developer.md

@@ -22,7 +22,7 @@ You write code; you do not coordinate. If you find yourself thinking "let me als
 | `commit(message)` | Makes the git commit, auto-prefixes `[task-id]`, records a progress entry. This is the ONLY way to commit — the gateway covers the actual git operation. | Task in `in_progress`; on your branch. |
 | `open_pr(task_id)` | Push your branch and open a PR. Run after your last commit, before `i_am_done`. | Task assigned to you; at least one commit; no PR yet. |
 | `i_am_done(task_id, notes)` | Submit for QA. Auto-runs in_progress→verifying→awaiting_qa. Requires PR already open — run `open_pr` first. | At least one commit; PR open; progress entry; journal `reflect`; every acceptance criterion addressed. |
-| `i_am_blocked(reason)` | Records the blocker, escalates to your PM, idles you. | Task is yours and active. |
+| `i_am_blocked(task_id, reason, blocker_type?, what_needed?)` | Records the blocker, escalates to your PM, idles you. `blocker_type` ∈ `external` (waiting on a 3rd-party API/service), `internal` (a teammate or process), `question` (need clarification), `dependency` (waiting on another task). `what_needed` is a one-sentence concrete unblock request. Both fields are pre-gateway parity — PMs triage by class. | Task is yours and active. |
 | `unclaim(task_id)` | Release this claim back to pending. Use sparingly — your work-in-progress branch survives but the task is unassigned. | Task assigned to you and in claimed/in_progress. |
 | `resume(task_id)` | Resume a paused task. Transitions paused → in_progress. | Task assigned to you and in paused state. |
 | `note(text, scope?)` | Journal entry (`scope ∈ note|decision|reflect|learning|struggle`). | None. |

roboco/api/routes/v2/flow_dev.py

@@ -81,7 +81,13 @@ async def i_am_blocked(
     x_agent_id: _AgentIdHeader,
     choreographer: _ChoreographerDep,
 ) -> dict:
-    env = await choreographer.i_am_blocked(x_agent_id, body.task_id, body.reason)
+    env = await choreographer.i_am_blocked(
+        x_agent_id,
+        body.task_id,
+        body.reason,
+        blocker_type=body.blocker_type,
+        what_needed=body.what_needed,
+    )
     return envelope_to_response(env, request)
 
 

roboco/api/schemas/v2/flow.py

@@ -26,6 +26,37 @@ class IAmDoneRequest(BaseModel):
 class IAmBlockedRequest(BaseModel):
     task_id: UUID
     reason: str = Field(..., min_length=1)
+    # Pre-gateway parity (G8 part b). The old TaskBlockInput at
+    # 0c3d15a:roboco/mcp/schemas/__init__.py required blocker_type and
+    # what_needed so PMs could triage by class. Optional here for
+    # back-compat with i_am_blocked(reason) callers; supplied fields are
+    # rendered into the struggle journal entry so the panel surfaces them.
+    blocker_type: str | None = Field(
+        default=None,
+        description=(
+            "external | internal | question | dependency. Required from "
+            "newly-spawned agents (per the developer.md verb table); "
+            "older agents that don't supply it default to 'internal'."
+        ),
+    )
+    what_needed: str | None = Field(
+        default=None,
+        description="Concrete description of what would unblock the task.",
+    )
+
+    @field_validator("blocker_type", mode="before")
+    @classmethod
+    def _blocker_type_enum(cls, v: object) -> object:
+        if v is None:
+            return v
+        if isinstance(v, str) and v.lower() not in {
+            "external", "internal", "question", "dependency",
+        }:
+            raise ValueError(
+                f"blocker_type must be one of: external | internal | "
+                f"question | dependency. Got {v!r}."
+            )
+        return v
 
 
 class UnclaimRequest(BaseModel):

roboco/mcp/flow_server.py

@@ -227,9 +227,32 @@ def i_am_done(task_id: str, notes: str = "") -> dict[str, Any]:
     return _post(_role_path("i_am_done"), {"task_id": task_id, "notes": notes})
 
 
-def i_am_blocked(task_id: str, reason: str) -> dict[str, Any]:
+def i_am_blocked(
-    """Escalate to PM. Logs a struggle journal entry."""
+    task_id: str,
-    return _post(_role_path("i_am_blocked"), {"task_id": task_id, "reason": reason})
+    reason: str,
+    blocker_type: str | None = None,
+    what_needed: str | None = None,
+) -> dict[str, Any]:
+    """Escalate to PM. Logs a struggle journal entry.
+
+    Args:
+        task_id: UUID of the task you're stuck on.
+        reason: One paragraph describing the blocker.
+        blocker_type: One of ``external`` | ``internal`` | ``question`` |
+            ``dependency``. Optional but strongly preferred — the PM
+            triages by class. Pre-gateway parity.
+        what_needed: Concrete description of what would unblock the
+            task. Pre-gateway parity.
+    """
+    return _post(
+        _role_path("i_am_blocked"),
+        {
+            "task_id": task_id,
+            "reason": reason,
+            "blocker_type": blocker_type,
+            "what_needed": what_needed,
+        },
+    )
 
 
 def unclaim(task_id: str) -> dict[str, Any]:

roboco/services/gateway/choreographer/_impl.py

@@ -1446,7 +1446,12 @@ class Choreographer:
         return preference[0]
 
     async def i_am_blocked(
-        self, agent_id: UUID, task_id: UUID, reason: str
+        self,
+        agent_id: UUID,
+        task_id: UUID,
+        reason: str,
+        blocker_type: str | None = None,
+        what_needed: str | None = None,
     ) -> Envelope:
         """Escalate task_id and write a struggle journal entry; idle the agent.
 
@@ -1505,8 +1510,21 @@ class Choreographer:
         # verb body. Written before the runner dispatches `block` so a
         # later runner failure still leaves an audit trail of the agent's
         # struggle.
+        # Pre-gateway parity (G8 part b): if the agent supplied typed
+        # blocker_type / what_needed, render them as a structured
+        # markdown body so the panel renders Blocker / Type / Needed
+        # blocks instead of one flat sentence. Pre-gateway shape
+        # came from TaskBlockInput at 0c3d15a:roboco/mcp/schemas/__init__.py.
+        struggle_body = reason
+        if blocker_type or what_needed:
+            parts = [reason.strip()] if reason.strip() else []
+            if blocker_type:
+                parts.append(f"## Blocker Type\n{blocker_type}")
+            if what_needed:
+                parts.append(f"## What Needed\n{what_needed}")
+            struggle_body = "\n\n".join(parts)
         await self.journal.write_struggle(
-            agent_id=agent_id, task_id=task_id, content=reason
+            agent_id=agent_id, task_id=task_id, content=struggle_body
         )
         runner = self._verb_runner()
         try:

tests/unit/mcp_servers/test_flow_server.py

@@ -184,7 +184,16 @@ def test_i_am_blocked_sends_reason(flow_module: types.ModuleType) -> None:
 
     assert result == {"status": "blocked"}
     _, kwargs = fake_client.post.call_args
-    assert kwargs["json"] == {"task_id": "task-xyz", "reason": "waiting for env var"}
+    # Pre-gateway parity (G8): i_am_blocked now also carries optional
+    # blocker_type / what_needed — when caller omits them the wrapper
+    # forwards `None` so the backend can fall back to the default
+    # 'internal' classification.
+    assert kwargs["json"] == {
+        "task_id": "task-xyz",
+        "reason": "waiting for env var",
+        "blocker_type": None,
+        "what_needed": None,
+    }
 
 
 def test_i_am_idle_posts_empty_body(flow_module: types.ModuleType) -> None: