fix(messaging): restore channel-access RBAC in gateway say()

post_to_channel was calling send_message without agent_slug, which
disabled the validate_channel_access check. Forward the slug; convert
ChannelAccessDeniedError to a friendly not_authorized Envelope with
the agent's writable-channel list (pre-gateway behaviour).

roboco/services/gateway/content_actions.py

@@ -36,10 +36,7 @@ def _ownership_violation(task_id: UUID) -> Envelope:
     gateway exposes task_id parameters so the explicit gate is required.
     """
     return Envelope.not_authorized(
-        message=(
-            f"you are not the assignee of {task_id}; "
-            "cannot post content to it"
-        ),
+        message=(f"you are not the assignee of {task_id}; cannot post content to it"),
         remediate=(
             "only the task's assignee may attach content (commit/note/say/"
             "dm/evidence) to it. Use a different task_id or omit task_id "
@@ -195,7 +192,19 @@ class ContentActions:
         text: str,
         task_id: UUID | None = None,
     ) -> Envelope:
-        """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.
+
+        Channel-write RBAC is enforced inside `messaging.post_to_channel`
+        (which forwards the agent's slug to `send_message` so
+        `validate_channel_access` runs). A denial bubbles up as
+        `ChannelAccessDeniedError`; we convert it into a friendly
+        `not_authorized` Envelope listing the agent's writable channels.
+        """
+        from roboco.enforcement.channel_access import (
+            ChannelAccessDeniedError,
+            get_agent_channels,
+        )
+
         if task_id is not None:
             if reject := await self._verify_explicit_task_ownership(agent_id, task_id):
                 return reject
@@ -203,12 +212,23 @@ class ContentActions:
             t = await self.task.get_active_task_for_agent(agent_id)
             if t is not None:
                 task_id = t.id
-        await self.messaging.post_to_channel(
-            agent_id=agent_id,
-            channel_slug=channel,
-            content=text,
-            task_id=task_id,
-        )
+        try:
+            await self.messaging.post_to_channel(
+                agent_id=agent_id,
+                channel_slug=channel,
+                content=text,
+                task_id=task_id,
+            )
+        except ChannelAccessDeniedError as e:
+            writable = get_agent_channels(e.agent_id, action="write")
+            writable_str = ", ".join(writable) if writable else "(none)"
+            return Envelope.not_authorized(
+                message=(
+                    f"agent '{e.agent_id}' may not write to channel '{e.channel_slug}'"
+                ),
+                remediate=f"writable channels for your role: {writable_str}",
+                context_briefing={},
+            )
         return Envelope.ok(
             status="posted",
             task_id=str(task_id) if task_id else None,

roboco/services/messaging.py

@@ -1808,18 +1808,24 @@ class MessagingService(BaseService):
         message via `send_message`.
 
         Channel access is enforced inside `send_message` (channel writers
-        list + role rules); this adapter never bypasses that check.
+        list + role rules) when `agent_slug` is supplied — this adapter
+        looks up the slug from `agent_id` and forwards it so the check
+        always runs.
         """
+        from roboco.services.repositories import get_agent_slug
+
         channel = await self.get_channel_by_slug_or_raise(channel_slug)
         group = await self._default_group_for_channel(channel)
         session = await self.get_or_create_active_session(cast("UUID", group.id))
+        agent_slug = await get_agent_slug(self.session, agent_id)
         return await self.send_message(
             MessageCreateRequest(
                 agent_id=agent_id,
                 session_id=cast("UUID", session.id),
                 content=content,
                 task_id=task_id,
-            )
+            ),
+            agent_slug=agent_slug,
         )
 
 

tests/unit/gateway/test_say_channel_rbac.py

@@ -0,0 +1,92 @@
+"""ContentActions.say must enforce channel write-access via send_message.
+
+The gateway path (post_to_channel -> send_message) historically called
+send_message WITHOUT agent_slug, which bypassed validate_channel_access.
+Pre-gateway returned a friendly ChannelAccessDeniedError listing the
+agent's writable channels. These tests pin the restored behaviour:
+
+1. say() converts ChannelAccessDeniedError into a not_authorized Envelope
+   that names the offending channel and includes a remediation hint
+   listing the writable channels for the agent's role.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+from roboco.enforcement.channel_access import ChannelAccessDeniedError
+from roboco.services.gateway.content_actions import (
+    ContentActions,
+    ContentActionsDeps,
+)
+
+
+def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
+    """Mirrors test_content_actions._make_deps."""
+    if "task" in overrides:
+        task = overrides["task"]
+    else:
+        task = AsyncMock()
+        task.get_active_task_for_agent.return_value = None
+
+    git = overrides.get("git", AsyncMock())
+    messaging = overrides.get("messaging", AsyncMock())
+    a2a = overrides.get("a2a", AsyncMock())
+    journal = overrides.get("journal", AsyncMock())
+    workspace = overrides.get("workspace", AsyncMock())
+    return ContentActionsDeps(
+        task=task,
+        git=git,
+        messaging=messaging,
+        a2a=a2a,
+        journal=journal,
+        workspace=workspace,
+    )
+
+
+@pytest.mark.asyncio
+async def test_say_returns_not_authorized_envelope_on_access_denied() -> None:
+    """When messaging.post_to_channel raises ChannelAccessDeniedError, say()
+    converts it into a not_authorized Envelope with a writable-channels hint.
+    """
+    aid = uuid4()
+    msg_svc = AsyncMock()
+    msg_svc.post_to_channel.side_effect = ChannelAccessDeniedError(
+        agent_id="be-dev-1",
+        channel_slug="announcements",
+        action="write",
+    )
+    deps = _make_deps(messaging=msg_svc)
+    actions = ContentActions(deps)
+
+    env = await actions.say(agent_id=aid, channel="announcements", text="hi")
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    assert "announcements" in body["message"]
+    # Remediation should hint at the agent's writable channels (or that
+    # they have none). The role/list is resolved from CHANNEL_ACCESS via
+    # get_agent_channels using the agent's slug.
+    assert body["remediate"] is not None
+    assert "writable" in body["remediate"].lower()
+
+
+@pytest.mark.asyncio
+async def test_say_success_path_returns_posted_envelope() -> None:
+    """Sanity: when messaging accepts the post, say() returns the
+    standard posted/continue Envelope (no regression in the happy path).
+    """
+    aid = uuid4()
+    msg_svc = AsyncMock()
+    deps = _make_deps(messaging=msg_svc)
+    actions = ContentActions(deps)
+
+    env = await actions.say(agent_id=aid, channel="dev-all", text="hello")
+    body = env.as_dict()
+
+    assert body["error"] is None
+    assert body["status"] == "posted"
+    assert body["next"] == "continue"
+    msg_svc.post_to_channel.assert_awaited_once()