feat(gateway): notify(target, text, scope?) for PMs and Board

Pre-gateway PMs/Board sent formal notifications (require ack); gateway
had say/dm only. Now PMs and Board can issue ack-required notifications
via NotificationService through the standard envelope path.

roboco/api/deps.py

@@ -29,6 +29,7 @@ from roboco.services.gateway.evidence_repo import EvidenceRepo
 from roboco.services.git import GitService
 from roboco.services.journal import JournalService
 from roboco.services.messaging import MessagingService
+from roboco.services.notification import NotificationService
 from roboco.services.permissions import AgentContext, PermissionService
 from roboco.services.repositories import resolve_agent_identity, resolve_agent_uuid
 from roboco.services.task import TaskService
@@ -513,7 +514,7 @@ async def get_choreographer(
 async def get_content_actions(
     db_session: DbSession,
 ) -> ContentActions:
-    """Build a ContentActions with all 6 service dependencies wired up."""
+    """Build a ContentActions with all 7 service dependencies wired up."""
     return ContentActions(
         ContentActionsDeps(
             task=TaskService(db_session),
@@ -522,6 +523,7 @@ async def get_content_actions(
             a2a=A2AService(db_session),
             journal=JournalService(db_session),
             workspace=WorkspaceService(db_session),
+            notifications=NotificationService(),
         )
     )
 

roboco/api/routes/v2/do.py

@@ -12,6 +12,7 @@ from roboco.api.schemas.v2.do import (
     DmRequest,
     EvidenceRequest,
     NoteRequest,
+    NotifyRequest,
     SayRequest,
 )
 from roboco.services.gateway.content_actions import ContentActions
@@ -86,6 +87,23 @@ async def do_dm(
     return envelope_to_response(env, request)
 
 
+@router.post("/notify")
+async def do_notify(
+    request: Request,
+    body: NotifyRequest,
+    x_agent_id: _AgentIdHeader,
+    actions: _ContentActionsDep,
+) -> dict:
+    env = await actions.notify(
+        agent_id=x_agent_id,
+        target=body.target,
+        text=body.text,
+        priority=body.priority,
+        task_id=body.task_id,
+    )
+    return envelope_to_response(env, request)
+
+
 @router.post("/evidence")
 async def do_evidence(
     request: Request,

roboco/api/schemas/v2/do.py

@@ -29,5 +29,12 @@ class DmRequest(BaseModel):
     skill: str | None = None
 
 
+class NotifyRequest(BaseModel):
+    target: str  # agent slug
+    text: str = Field(..., min_length=1)
+    priority: str = "normal"  # normal | high | urgent
+    task_id: UUID | None = None
+
+
 class EvidenceRequest(BaseModel):
     task_id: UUID

roboco/mcp/do_server.py

@@ -3,8 +3,9 @@
 Forwards to /api/v2/do/* on the orchestrator. Tools are role-scoped at *spawn*
 time: the orchestrator writes ``do_tools`` into the per-agent manifest and we
 register only those names on this server. The orchestrator's API is not
-role-scoped here (any allowed role can call commit/note/say/dm/evidence), so
-the path is fixed (no role segment).
+role-scoped here (any allowed role can call commit/note/say/dm/notify/evidence),
+so the path is fixed (no role segment). Per-tool role gates (e.g., notify
+restricting to PMs/Board) live inside the gateway verbs.
 
 If the manifest is missing or unreadable (local test runs without the bind
 mount) the full registry is registered as a failsafe and a warning is logged.
@@ -93,6 +94,29 @@ def dm(
     )
 
 
+def notify(
+    target: str,
+    text: str,
+    priority: str = "normal",
+    task_id: str | None = None,
+) -> dict[str, Any]:
+    """Send a formal ack-required notification (PMs and Board only).
+
+    Distinct from say (channel post) and dm (informal A2A): notify creates
+    a notification the recipient must acknowledge. priority in
+    normal|high|urgent. task_id auto-injected from active task when omitted.
+    """
+    return _post(
+        "/api/v2/do/notify",
+        {
+            "target": target,
+            "text": text,
+            "priority": priority,
+            "task_id": task_id,
+        },
+    )
+
+
 def evidence(task_id: str) -> dict[str, Any]:
     """Inspect a task's PR diff, commits, files. Fetches dev branch into workspace."""
     return _post("/api/v2/do/evidence", {"task_id": task_id})
@@ -108,6 +132,7 @@ _TOOLS: dict[str, Any] = {
     "note": note,
     "say": say,
     "dm": dm,
+    "notify": notify,
     "evidence": evidence,
 }
 

roboco/services/gateway/content_actions.py

@@ -56,6 +56,17 @@ class ContentActionsDeps:
     a2a: Any
     journal: Any
     workspace: Any
+    notifications: Any
+
+
+# Roles authorized to issue formal ack-required notifications via `notify`.
+# Pre-gateway, NotificationService callers were gated by the same set
+# (PMs and Board members); the gateway re-asserts that gate at the verb
+# layer because the do.py router is shared by all roles.
+_NOTIFY_ALLOWED_ROLES: frozenset[str] = frozenset(
+    {"cell_pm", "main_pm", "product_owner", "head_marketing"}
+)
+_VALID_NOTIFY_PRIORITIES: frozenset[str] = frozenset({"normal", "high", "urgent"})
 
 
 class ContentActions:
@@ -86,6 +97,10 @@ class ContentActions:
     def workspace(self) -> Any:
         return self._deps.workspace
 
+    @property
+    def notifications(self) -> Any:
+        return self._deps.notifications
+
     async def commit(
         self,
         *,
@@ -277,6 +292,76 @@ class ContentActions:
             context_briefing={},
         )
 
+    async def notify(
+        self,
+        *,
+        agent_id: UUID,
+        target: str,
+        text: str,
+        priority: str = "normal",
+        task_id: UUID | None = None,
+    ) -> Envelope:
+        """Send a formal ack-required notification (PMs and Board only).
+
+        Distinct from `say` (channel post, no ack) and `dm` (informal A2A):
+        a notification is a formal signal that the recipient must
+        acknowledge. Pre-gateway, NotificationService restricted senders
+        to PMs/Board; the gateway re-asserts that gate here because the
+        do.py router is shared by all roles (no router-level dep).
+
+        ``target`` is an agent slug ("be-dev-1", "main-pm", "ceo");
+        NotificationService resolves it to a UUID at insert time.
+        ``priority`` is one of normal|high|urgent. ``task_id`` is
+        auto-filled from the caller's active task when omitted, but
+        omission is permitted for off-task notifications (e.g., Board
+        broadcasts).
+        """
+        from roboco.models import NotificationPriority
+
+        if priority not in _VALID_NOTIFY_PRIORITIES:
+            return Envelope.invalid_state(
+                message=f"invalid priority {priority!r}",
+                remediate=(
+                    f"priority must be one of: {sorted(_VALID_NOTIFY_PRIORITIES)}"
+                ),
+                context_briefing={},
+            )
+        agent = await self.task.agent_for(agent_id)
+        caller_role = agent.role if agent is not None else None
+        if caller_role not in _NOTIFY_ALLOWED_ROLES:
+            return Envelope.not_authorized(
+                message=(
+                    f"role {caller_role!r} cannot send formal notifications; "
+                    "only PMs and Board may issue ack-required signals"
+                ),
+                remediate=(
+                    "use say() for channel posts or dm() for informal A2A. "
+                    "notify() is reserved for cell_pm, main_pm, "
+                    "product_owner, and head_marketing."
+                ),
+                context_briefing={},
+            )
+        if task_id is not None:
+            if reject := await self._verify_explicit_task_ownership(agent_id, task_id):
+                return reject
+        else:
+            t = await self.task.get_active_task_for_agent(agent_id)
+            if t is not None:
+                task_id = t.id
+        await self.notifications.send_ack_notification(
+            from_agent=agent_id,
+            to_agent=target,
+            body=text,
+            priority=NotificationPriority(priority),
+            task_id=task_id,
+        )
+        return Envelope.ok(
+            status="sent",
+            task_id=str(task_id) if task_id else None,
+            next="continue",
+            context_briefing={},
+        )
+
     async def evidence(
         self,
         *,

roboco/services/gateway/role_config.py

@@ -69,7 +69,7 @@ _CELL_PM_FLOW = (
     "resume",
     "i_am_idle",
 )
-_CELL_PM_DO = ("note", "say", "dm", "evidence")
+_CELL_PM_DO = ("note", "say", "dm", "notify", "evidence")
 
 _MAIN_PM_FLOW = (
     "give_me_work",
@@ -84,14 +84,14 @@ _MAIN_PM_FLOW = (
     "resume",
     "i_am_idle",
 )
-_MAIN_PM_DO = ("note", "say", "dm", "evidence")
+_MAIN_PM_DO = ("note", "say", "dm", "notify", "evidence")
 
 _BOARD_FLOW = (
     "triage",
     "escalate_to_ceo",
     "i_am_idle",
 )
-_BOARD_DO = ("note", "say", "dm", "evidence")
+_BOARD_DO = ("note", "say", "dm", "notify", "evidence")
 
 _AUDITOR_FLOW = (
     "triage",

roboco/services/notification.py

@@ -206,6 +206,40 @@ class NotificationService:
             )
         )
 
+    async def send_ack_notification(
+        self,
+        *,
+        from_agent: UUID | str,
+        to_agent: str,
+        body: str,
+        priority: NotificationPriority = NotificationPriority.NORMAL,
+        task_id: UUID | str | None = None,
+    ) -> None:
+        """Send a free-form ack-required notification (PM/Board only).
+
+        Used by the gateway `notify` content-tool. Distinguishes from
+        the typed `send_*_notification` helpers above, which carry
+        lifecycle semantics (blocker, qa-ready, etc.). Here the caller
+        supplies the body verbatim. ALERT type is used so consumers
+        treat it as a high-attention formal signal rather than
+        conflating with task-state-driven notifications. The subject
+        is derived from the first line of `body` (truncated), matching
+        how `say`/`dm` derive a subject from free text.
+        """
+        subject = body.split("\n", 1)[0][:200] or "Notification"
+        related_task_id = str(task_id) if task_id is not None else None
+        await self._create_notification(
+            CreateNotificationParams(
+                notification_type=NotificationType.ALERT,
+                priority=priority,
+                from_agent=str(from_agent),
+                to_agents=[to_agent],
+                subject=subject,
+                body=body,
+                related_task_id=related_task_id,
+            )
+        )
+
     async def send_a2a_notification(
         self,
         task_id: str,

tests/unit/gateway/test_commit_prefix.py

@@ -34,6 +34,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
     a2a = overrides.get("a2a", AsyncMock())
     journal = overrides.get("journal", AsyncMock())
     workspace = overrides.get("workspace", AsyncMock())
+    notifications = overrides.get("notifications", AsyncMock())
     return ContentActionsDeps(
         task=task,
         git=git,
@@ -41,6 +42,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
         a2a=a2a,
         journal=journal,
         workspace=workspace,
+        notifications=notifications,
     )
 
 

tests/unit/gateway/test_content_actions.py

@@ -29,6 +29,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
     a2a = overrides.get("a2a", AsyncMock())
     journal = overrides.get("journal", AsyncMock())
     workspace = overrides.get("workspace", AsyncMock())
+    notifications = overrides.get("notifications", AsyncMock())
     return ContentActionsDeps(
         task=task,
         git=git,
@@ -36,6 +37,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
         a2a=a2a,
         journal=journal,
         workspace=workspace,
+        notifications=notifications,
     )
 
 

tests/unit/gateway/test_content_actions_ownership.py

@@ -39,6 +39,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
     a2a = overrides.get("a2a", AsyncMock())
     journal = overrides.get("journal", AsyncMock())
     workspace = overrides.get("workspace", AsyncMock())
+    notifications = overrides.get("notifications", AsyncMock())
     return ContentActionsDeps(
         task=task,
         git=git,
@@ -46,6 +47,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
         a2a=a2a,
         journal=journal,
         workspace=workspace,
+        notifications=notifications,
     )
 
 

tests/unit/gateway/test_notify.py

@@ -0,0 +1,306 @@
+"""Tests for ContentActions.notify — formal ack-required notifications.
+
+Pre-gateway, PMs and Board could issue formal notifications requiring
+acknowledgment via NotificationService. Gateway only had say/dm
+(informal). This verb fills the gap by composing NotificationService
+into the standard envelope path, role-gated to PMs and Board only
+(content tools share one router, so the role check lives in the verb).
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.content_actions import ContentActions, ContentActionsDeps
+
+
+def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
+    if "task" in overrides:
+        task = overrides["task"]
+    else:
+        task = AsyncMock()
+        task.get_active_task_for_agent.return_value = None
+        task.agent_for.return_value = MagicMock(role="cell_pm")
+
+    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())
+    notifications = overrides.get("notifications", AsyncMock())
+    return ContentActionsDeps(
+        task=task,
+        git=git,
+        messaging=messaging,
+        a2a=a2a,
+        journal=journal,
+        workspace=workspace,
+        notifications=notifications,
+    )
+
+
+@pytest.mark.asyncio
+async def test_notify_pm_creates_ack_required_notification() -> None:
+    """Cell PM calls notify(); NotificationService.send_ack_notification fired."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-dev-1",
+        text="Please review the new acceptance criteria before resuming.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    assert body["status"] == "sent"
+    notif_svc.send_ack_notification.assert_awaited_once()
+    call_kwargs = notif_svc.send_ack_notification.call_args.kwargs
+    assert call_kwargs["from_agent"] == agent_id
+    assert call_kwargs["to_agent"] == "be-dev-1"
+    assert "acceptance criteria" in call_kwargs["body"]
+
+
+@pytest.mark.asyncio
+async def test_notify_main_pm_succeeds() -> None:
+    """Main PM is also allowed."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="main_pm")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="fe-pm",
+        text="Please align frontend cell with new release timeline.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    notif_svc.send_ack_notification.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_notify_board_product_owner_succeeds() -> None:
+    """Product Owner (Board) is allowed."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="product_owner")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="main-pm",
+        text="Roadmap priorities updated; please reflect in Q2 plan.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    notif_svc.send_ack_notification.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_notify_board_head_marketing_succeeds() -> None:
+    """Head of Marketing (Board) is allowed."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="head_marketing")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="main-pm",
+        text="Marketing launch dates confirmed; coordinate engineering deliverables.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    notif_svc.send_ack_notification.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_notify_developer_rejected_with_not_authorized() -> None:
+    """Developer cannot send formal notifications; envelope is not_authorized."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="developer")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-pm",
+        text="Heads up — I think the staging deploy is broken.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    assert "developer" in body["message"]
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_qa_rejected_with_not_authorized() -> None:
+    """QA cannot send formal notifications."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="qa")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-pm",
+        text="QA cannot proceed without environment access.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_documenter_rejected_with_not_authorized() -> None:
+    """Documenter cannot send formal notifications."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="documenter")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-pm",
+        text="Documentation review requested.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_auditor_rejected_with_not_authorized() -> None:
+    """Auditor is read-only — cannot communicate outwardly via notifications."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="auditor")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="ceo",
+        text="Quality concern detected.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_auto_fills_task_id_from_active_task() -> None:
+    """When the PM has an active task, notify auto-attaches it."""
+    agent_id = uuid4()
+    task_id = uuid4()
+    task_obj = MagicMock(id=task_id, status="awaiting_pm_review")
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = task_obj
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-dev-1",
+        text="Heads up: this task has been escalated for CEO approval.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    assert body["task_id"] == str(task_id)
+    call_kwargs = notif_svc.send_ack_notification.call_args.kwargs
+    assert call_kwargs["task_id"] == task_id
+
+
+@pytest.mark.asyncio
+async def test_notify_unknown_role_rejected() -> None:
+    """If task.agent_for returns None, treat as unknown role and reject."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = None
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-dev-1",
+        text="Test message.",
+    )
+    body = env.as_dict()
+
+    assert body["error"] == "not_authorized"
+    notif_svc.send_ack_notification.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notify_priority_high_passed_through() -> None:
+    """Optional priority='high' is forwarded to NotificationService."""
+    agent_id = uuid4()
+    task_svc = AsyncMock()
+    task_svc.get_active_task_for_agent.return_value = None
+    task_svc.agent_for.return_value = MagicMock(role="cell_pm")
+    notif_svc = AsyncMock()
+
+    deps = _make_deps(task=task_svc, notifications=notif_svc)
+    ca = ContentActions(deps)
+
+    env = await ca.notify(
+        agent_id=agent_id,
+        target="be-dev-1",
+        text="Critical: production deployment failed; please join war room.",
+        priority="high",
+    )
+    body = env.as_dict()
+
+    assert body["error"] is None
+    call_kwargs = notif_svc.send_ack_notification.call_args.kwargs
+    assert call_kwargs["priority"] == "high"

tests/unit/gateway/test_say_channel_rbac.py

@@ -36,6 +36,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
     a2a = overrides.get("a2a", AsyncMock())
     journal = overrides.get("journal", AsyncMock())
     workspace = overrides.get("workspace", AsyncMock())
+    notifications = overrides.get("notifications", AsyncMock())
     return ContentActionsDeps(
         task=task,
         git=git,
@@ -43,6 +44,7 @@ def _make_deps(**overrides: AsyncMock) -> ContentActionsDeps:
         a2a=a2a,
         journal=journal,
         workspace=workspace,
+        notifications=notifications,
     )