From 3a2498a609c36226ee8f019c652ce7b25c45325f Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 3 May 2026 09:19:08 +0200 Subject: [PATCH] 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 | 4 +- roboco/api/routes/v2/do.py | 18 ++ roboco/api/schemas/v2/do.py | 7 + roboco/mcp/do_server.py | 29 +- roboco/services/gateway/content_actions.py | 85 +++++ roboco/services/gateway/role_config.py | 6 +- roboco/services/notification.py | 34 ++ tests/unit/gateway/test_commit_prefix.py | 2 + tests/unit/gateway/test_content_actions.py | 2 + .../gateway/test_content_actions_ownership.py | 2 + tests/unit/gateway/test_notify.py | 306 ++++++++++++++++++ tests/unit/gateway/test_say_channel_rbac.py | 2 + 12 files changed, 491 insertions(+), 6 deletions(-) create mode 100644 tests/unit/gateway/test_notify.py diff --git a/roboco/api/deps.py b/roboco/api/deps.py index 4181a510..633a5077 100644 --- a/roboco/api/deps.py +++ b/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(), ) ) diff --git a/roboco/api/routes/v2/do.py b/roboco/api/routes/v2/do.py index 9d1ac29b..eb692ea4 100644 --- a/roboco/api/routes/v2/do.py +++ b/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, diff --git a/roboco/api/schemas/v2/do.py b/roboco/api/schemas/v2/do.py index 76afbcc4..d4b3ddf1 100644 --- a/roboco/api/schemas/v2/do.py +++ b/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 diff --git a/roboco/mcp/do_server.py b/roboco/mcp/do_server.py index 5cc0657e..8e400127 100644 --- a/roboco/mcp/do_server.py +++ b/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, } diff --git a/roboco/services/gateway/content_actions.py b/roboco/services/gateway/content_actions.py index e3aaf570..1310fd7d 100644 --- a/roboco/services/gateway/content_actions.py +++ b/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, *, diff --git a/roboco/services/gateway/role_config.py b/roboco/services/gateway/role_config.py index 597ff4e1..21f08434 100644 --- a/roboco/services/gateway/role_config.py +++ b/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", diff --git a/roboco/services/notification.py b/roboco/services/notification.py index 27039313..e73e4abb 100644 --- a/roboco/services/notification.py +++ b/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, diff --git a/tests/unit/gateway/test_commit_prefix.py b/tests/unit/gateway/test_commit_prefix.py index 552670e3..8af4014a 100644 --- a/tests/unit/gateway/test_commit_prefix.py +++ b/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, ) diff --git a/tests/unit/gateway/test_content_actions.py b/tests/unit/gateway/test_content_actions.py index 683aa030..1e26d89a 100644 --- a/tests/unit/gateway/test_content_actions.py +++ b/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, ) diff --git a/tests/unit/gateway/test_content_actions_ownership.py b/tests/unit/gateway/test_content_actions_ownership.py index 676f313f..9a15eca5 100644 --- a/tests/unit/gateway/test_content_actions_ownership.py +++ b/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, ) diff --git a/tests/unit/gateway/test_notify.py b/tests/unit/gateway/test_notify.py new file mode 100644 index 00000000..9ae30c22 --- /dev/null +++ b/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" diff --git a/tests/unit/gateway/test_say_channel_rbac.py b/tests/unit/gateway/test_say_channel_rbac.py index b465804b..bfa9e3f9 100644 --- a/tests/unit/gateway/test_say_channel_rbac.py +++ b/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, )