RBAC is necessary in multi-agent RL Envs

from agentenv_protocol import (AgentEnvEnvironment, ToolContext,
                               environment_card, tool)


@environment_card(name="gmail")
class GmailEnv(AgentEnvEnvironment):
    ...

    @tool(name="{environment_name}_list_inbox")
    async def list_inbox(self, context: ToolContext) -> dict:
        """The caller's inbox, newest first."""
        me = self.directory.get(context.role)  # "pm" -> "bob@example.com"
        mine = [e for e in self.emails if me in e["to"]]
        return {"emails": mine[::-1]}

    @tool(name="{environment_name}_send")
    async def send(self, to: list[str], subject: str, body: str,
                   context: ToolContext) -> dict:
        """Send an email from the caller's address."""
        me = self.directory.get(context.role)
        if me is None:
            raise PermissionError("No mailbox for this role")
        self.emails.append({"from": me, "to": to,
                            "subject": subject, "body": body})
        return {"sent": True}

from agentenv_protocol import (AgentEnvEnvironment, ToolContext,
                               environment_card, tool)


@environment_card(name="slack")
class SlackEnv(AgentEnvEnvironment):
    ...

    @tool(name="{environment_name}_read_channel")
    async def read_channel(self, channel: str,
                           context: ToolContext) -> dict:
        """Messages in a private channel the caller is in."""
        me = self.directory.get(context.role)  # "pm" -> "bob@example.com"
        room = self.channels.get(channel)
        if room is None or me not in room["members"]:
            raise LookupError("channel_not_found")
        return {"messages": room["messages"]}

    @tool(name="{environment_name}_read_dms")
    async def read_dms(self, context: ToolContext) -> dict:
        """The caller's direct messages."""
        me = self.directory.get(context.role)
        return {"dms": [d for d in self.dms if me in d["members"]]}

    @tool(name="{environment_name}_post")
    async def post(self, channel: str, text: str,
                   context: ToolContext) -> dict:
        """Post to a channel as the caller."""
        me = self.directory.get(context.role)
        room = self.channels.get(channel)
        if room is None or me not in room["members"]:
            raise PermissionError("not_in_channel")
        room["messages"].append({"user": me, "text": text})
        return {"ok": True}

from agentenv_protocol import (AgentEnvEnvironment, ToolContext,
                               environment_card, tool)


@environment_card(name="github")
class GitHubEnv(AgentEnvEnvironment):
    ...

    def visible(self, repo: dict, me: str | None) -> bool:
        return not repo["private"] or me in repo["collaborators"]

    @tool(name="{environment_name}_list_repos")
    async def list_repos(self, context: ToolContext) -> dict:
        """Public repos, and the private ones the caller works on."""
        me = self.directory.get(context.role)  # "pm" -> "bob@example.com"
        return {"repos": [name for name, repo in self.repos.items()
                          if self.visible(repo, me)]}

    @tool(name="{environment_name}_open_pr")
    async def open_pr(self, repo: str, title: str, head: str,
                      context: ToolContext) -> dict:
        """Open a pull request as the caller."""
        me = self.directory.get(context.role)
        r = self.repos.get(repo)
        if r is None or not self.visible(r, me):
            raise LookupError("404 Not Found")
        r["pulls"].append({"title": title, "head": head, "author": me})
        return {"number": len(r["pulls"])}

from agentenv_protocol import (AgentEnvEnvironment, ToolContext,
                               environment_card, tool)


@environment_card(name="drive")
class DriveEnv(AgentEnvEnvironment):
    ...

    @tool(name="{environment_name}_list_docs")
    async def list_docs(self, context: ToolContext) -> dict:
        """Docs shared with the caller, and their access to each."""
        me = self.directory.get(context.role)  # "pm" -> "bob@example.com"
        shared = [d for d in self.docs if me in d["sharing"]]
        return {"docs": [{"id": d["id"], "title": d["title"],
                          "access": d["sharing"][me]} for d in shared]}

    @tool(name="{environment_name}_edit_doc")
    async def edit_doc(self, doc_id: str, body: str,
                       context: ToolContext) -> dict:
        """Replace a doc's body, if the caller can edit it."""
        me = self.directory.get(context.role)
        doc = next((d for d in self.docs if d["id"] == doc_id), None)
        if doc is None or me not in doc["sharing"]:
            raise LookupError("File not found")
        if doc["sharing"][me] not in ("owner", "editor"):
            raise PermissionError("You need edit access")
        doc.update(body=body, last_edited_by=me)
        return {"edited": doc_id}