diff --git a/roboco/services/gateway/content_actions.py b/roboco/services/gateway/content_actions.py index e9ea4a1a..cdc40456 100644 --- a/roboco/services/gateway/content_actions.py +++ b/roboco/services/gateway/content_actions.py @@ -88,6 +88,29 @@ _NOTIFY_ALLOWED_ROLES: frozenset[str] = frozenset( r.value for r in _comms.NOTIFY_SENDER_ROLES ) +# Roles with NO agent-comms surface (CLAUDE.md): auditor (silent observer), +# pr_reviewer (posts review findings on the PR itself — no say/dm), prompter +# and secretary (human-only, restricted to note + evidence — no say/dm/notify). +# The spawn manifest already omits say/dm from these roles' tool surfaces, but +# that is convention-only — this frozenset is the handler-level defence-in-depth +# that refuses any call that bypassed the manifest (direct verb dispatch, test +# harness, future routing change), so the no-comms invariant holds regardless of +# how the call arrived. Matches the explicit role-frozenset gates on commit / +# notify / pitch / playbook / open_session. +_NO_COMMS_ROLES: frozenset[str] = frozenset( + {"auditor", "pr_reviewer", "prompter", "secretary"} +) + + +def _no_comms_remediate(role: str) -> str: + """Role-appropriate remediation for a no-comms role blocked at say/dm.""" + if role == "auditor": + return "record observations via note(scope='reflect') instead" + if role == "pr_reviewer": + return "post review findings on the PR itself via pr_pass/pr_fail instead" + # prompter / secretary are human-only (note + evidence). + return "use note() to record; this human-only role has no agent-comms surface" + _DECISION_SECTIONS: tuple[tuple[str, str], ...] = ( ("context", "Context"), @@ -982,17 +1005,23 @@ class ContentActions: get_agent_channels, ) - # Spec §5.5: auditor is silent — defense-in-depth runtime guard. - # The spawn manifest already omits `say` from the auditor's tool - # surface, but that is convention-only. This guard refuses any + # Spec §5.5: silent / no-comms roles — defense-in-depth runtime guard. + # The spawn manifest already omits `say` from these roles' tool + # surfaces, but that is convention-only. This guard refuses any # call that bypassed the manifest (direct verb dispatch, test - # harness, future routing change) so the silent-observer rule - # holds regardless of how the call arrived. + # harness, future routing change) so the no-comms rule holds + # regardless of how the call arrived. Covers auditor (silent + # observer), pr_reviewer (posts findings on the PR), and the + # human-only prompter / secretary (note + evidence only). agent = await self.task.agent_for(agent_id) - if agent is not None and str(agent.role) == "auditor": + caller_role = str(agent.role) if agent is not None else "" + if caller_role in _NO_COMMS_ROLES: return Envelope.not_authorized( - message="auditor is a silent observer; say is not permitted", - remediate="record observations via note(scope='reflect') instead", + message=( + f"role '{caller_role}' is a silent / no-comms role;" + " say is not permitted" + ), + remediate=_no_comms_remediate(caller_role), context_briefing={}, ) @@ -1039,14 +1068,19 @@ class ContentActions: """A2A direct message. Requires task_id (active or explicit).""" if rej := self._reject_soup(text, field="message", min_chars=2): return rej - # Spec §5.5: auditor is silent — defense-in-depth runtime guard. - # See say() above for rationale. Mirrored here because dm() is - # the other channel through which the auditor could "speak". + # Spec §5.5: silent / no-comms roles — defense-in-depth runtime guard. + # See say() for rationale. Mirrored here because dm() is the other + # channel through which a no-comms role could "speak". Covers auditor, + # pr_reviewer, and the human-only prompter / secretary. agent = await self.task.agent_for(agent_id) - if agent is not None and str(agent.role) == "auditor": + caller_role = str(agent.role) if agent is not None else "" + if caller_role in _NO_COMMS_ROLES: return Envelope.not_authorized( - message="auditor is a silent observer; dm is not permitted", - remediate="record observations via note(scope='reflect') instead", + message=( + f"role '{caller_role}' is a silent / no-comms role;" + " dm is not permitted" + ), + remediate=_no_comms_remediate(caller_role), context_briefing={}, ) diff --git a/tests/unit/gateway/test_auditor_silent_guard.py b/tests/unit/gateway/test_auditor_silent_guard.py index 89f0d26b..25adb1af 100644 --- a/tests/unit/gateway/test_auditor_silent_guard.py +++ b/tests/unit/gateway/test_auditor_silent_guard.py @@ -124,3 +124,63 @@ async def test_developer_dm_passes_auditor_guard() -> None: if body.get("error") == "not_authorized": haystack = (body.get("message") or "") + " " + (body.get("remediate") or "") assert "silent" not in haystack.lower() + + +# --------------------------------------------------------------------------- +# The same no-comms invariant covers pr_reviewer / prompter / secretary +# (CLAUDE.md): pr_reviewer "posts its change-request on the PR itself — no +# say/dm"; prompter + secretary are "restricted to note + evidence — no +# say/dm/notify". The auditor guard's own comment claims defence-in-depth for +# "any call that bypassed the manifest" — that rationale must hold for these +# three roles too, or the claimed defence-in-depth is only 1 of 4 silent roles. +# --------------------------------------------------------------------------- + + +_NO_COMMS_ROLES = ("pr_reviewer", "prompter", "secretary") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("role", _NO_COMMS_ROLES) +async def test_no_comms_role_say_returns_not_authorized(role: str) -> None: + """pr_reviewer / prompter / secretary may not say() — the handler-level + guard must refuse them regardless of how the call arrived, same as auditor.""" + deps = _make_deps(role) + actions = ContentActions(deps) + + env = await actions.say(agent_id=uuid4(), channel="backend-cell", text="hi") + body = env.as_dict() + + assert body["error"] == "not_authorized" + # The no-comms signal distinguishes the role guard from any downstream + # reject (channel-access denial) — proves it's the silent-role guard. + haystack = (body.get("message") or "") + " " + (body.get("remediate") or "") + assert "silent" in haystack.lower() + # The guard fires before any downstream call. + deps.messaging.post_to_channel.assert_not_called() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("role", _NO_COMMS_ROLES) +async def test_no_comms_role_dm_returns_not_authorized(role: str) -> None: + """pr_reviewer / prompter / secretary may not dm() — handler-level guard. + + Asserts the no-comms signal ("silent") in the message so the test fails for + the right reason on RED: without the role guard, dm() with an unowned + task_id still returns not_authorized from the ownership check, but that + reject message does NOT carry the silent-role signal. The role guard firing + FIRST (before the ownership check) is what makes "silent" appear.""" + deps = _make_deps(role) + actions = ContentActions(deps) + + env = await actions.dm( + agent_id=uuid4(), + recipient=str(uuid4()), + text="hi", + task_id=uuid4(), + ) + body = env.as_dict() + + assert body["error"] == "not_authorized" + haystack = (body.get("message") or "") + " " + (body.get("remediate") or "") + assert "silent" in haystack.lower() + deps.a2a.send.assert_not_called()