fix(delegate): carry the permission mode into a delegated run

A delegated turn gets a session id synthesised per (source, target, root):

    return f"delegate_{source}_{target}_{sha256(root_session)[:16]}"

Nothing ever writes prefs under that id, so on the way back through
`AgentBridge.get_agent` -> `apply_session_prefs` the lookup misses and the
target is handed `apply_permission_mode(None)`, which leaves it on whatever
`global_mode()` says -- `full-access` by default. A conversation the user had
set to read-only could therefore hand a task to a teammate, and the teammate
would answer with full write access.

`agent/subagent/runner.py:131-136` already does the equivalent for sub agents,
and says why:

    # Same working directory and same permission mode as the parent: the tool
    # copies already point at the parent's cwd, and delegating to a sub agent
    # must not become a way around the session's permissions.

and `apply_session_prefs`' own docstring states the intent: "Permission stays
shared either way, because it bounds what this conversation may change
regardless of who is speaking."

Read the mode in force with `peek_agent` (which does not build an Agent, so
asking never spins up MCP connections and skills), put it on the delegated
context, and let `apply_session_prefs` prefer it. It is threaded through
`get_agent(permission_mode=...)`, defaulted to None so every other caller keeps
reading prefs.

Note the read-only gate is left as it is on purpose. `subagent` is in
`_KNOWN_TOOLS` and so is allowed under read-only -- correctly, because the mode
is copied onto the child. `agent_delegate` was allowed too, but by omission:
absent from the table, and `_MUTATING_NAME_RE` does not match the verb. Now
that the mode travels, both are allowed for the same reason, and tightening the
gate instead would break legitimate read-only delegation without closing
anything.
This commit is contained in:
c020627 committed 2026-10-05 17:46:43 +08:00
1 parent e350c71138
commit bb52ed68ee
3 files changed
+201 -2

No files matched your search

@@ -276,6 +276,35 @@ class AgentDelegateTool(BaseTool):
digest = hashlib.sha256(root_session_id.encode("utf-8")).hexdigest()[:16]
return f"delegate_{source_agent_id}_{target_agent_id}_{digest}"
def _inherited_permission_mode(self, source_agent_id: str) -> str | None:
"""The mode in force for whoever is delegating, or None to leave it alone.
A delegated run gets a session id of its own, so the target's session
prefs miss and `apply_session_prefs` calls `apply_permission_mode(None)`,
which leaves the target on the *global* mode. Delegating would then be a
way out of whatever the user set for this conversation -- and
`agent_delegate` is deliberately not gated on `subagent`'s pattern
(see `runner.py`, which is why subagent is in `_KNOWN_TOOLS`): the
permission travels with the delegation instead.
Read with `peek_agent`, which does not build an Agent, so asking a
question never spins up MCP connections and skills. `None` means the
source has no live instance (a nested hop, or the turn is mid-build);
the target then keeps its own mode rather than being guessed at.
"""
session_id = (self.current_context.kwargs or {}).get("session_id")
if not session_id:
return None
try:
source = self.agent_bridge.peek_agent(session_id, source_agent_id)
except Exception as e:
logger.debug(f"[AgentDelegate] could not read the source's mode: {e}")
return None
if source is None:
return None
getter = getattr(source, "effective_permission_mode", None)
return getter() if callable(getter) else None
def _team_members(self, context_values: dict, source_agent_id: str) -> list:
"""The teammates the source Agent may delegate to this turn.
@@ -520,6 +549,12 @@ class AgentDelegateTool(BaseTool):
delegated_context["run_id"] = run_id
delegated_context["parent_run_id"] = parent_run_id
delegated_context["task_source"] = TASK_SOURCE
# The permission mode the delegating conversation is under. Read by
# apply_session_prefs, which would otherwise fall back to the global
# mode because this session id has no prefs of its own.
inherited_mode = self._inherited_permission_mode(source.id)
if inherited_mode:
delegated_context["delegated_permission_mode"] = inherited_mode
prompt = delegated_prompt(source.name, source.id, task)
+20 -2
View File
@@ -1247,6 +1247,7 @@ class AgentBridge:
session_id: str = None,
agent_id: str = None,
host_agent_id: str = None,
permission_mode: str = None,
) -> Optional[Agent]:
"""
Get agent instance for the given session
@@ -1303,6 +1304,7 @@ class AgentBridge:
session_id,
host_id,
owns_conversation=resolved_agent_id == host_id,
permission_mode=permission_mode,
)
return agent
@@ -1343,7 +1345,12 @@ class AgentBridge:
logger.debug(f"[AgentBridge] apply_session_project failed: {e}")
def apply_session_prefs(
self, agent, session_id: str, agent_id: str = None, owns_conversation: bool = True
self,
agent,
session_id: str,
agent_id: str = None,
owns_conversation: bool = True,
permission_mode: str = None,
) -> None:
"""Apply a session's model / permission overrides to its agent.
@@ -1378,7 +1385,15 @@ class AgentBridge:
else:
model.set_session_override(None, None)
if hasattr(agent, "apply_permission_mode"):
agent.apply_permission_mode(prefs.get("permission"))
# `permission_mode` is for a delegated turn. Its session id is
# synthesised per (source, target, root) and never has prefs, so
# the lookup above always misses and the target would keep the
# global mode -- which would make delegating a way out of the
# mode this conversation is under. The delegating tool passes
# the mode it saw in force instead, the same way
# agent/subagent/runner.py copies it onto a sub agent. Every
# other caller leaves it None and reads prefs as before.
agent.apply_permission_mode(permission_mode or prefs.get("permission"))
except Exception as e:
logger.debug(f"[AgentBridge] apply_session_prefs failed: {e}")
@@ -1624,6 +1639,9 @@ class AgentBridge:
session_id=session_id,
agent_id=speaker_agent_id,
host_agent_id=resolved_agent_id,
# Set by AgentDelegateTool so the teammate answers under the same
# permission mode as the conversation that handed it the work.
permission_mode=context.get("delegated_permission_mode"),
)
if not agent:
return Reply(ReplyType.ERROR, "Failed to initialize super agent")
+146
View File
@@ -522,3 +522,149 @@ def _query_for(task, order):
if task in query:
return query
raise AssertionError(f"no query recorded for {task}")
# ---------------------------------------------------------------------------
# A delegated run must answer under the same permission mode as the
# conversation that handed it the work. agent/subagent/runner.py:131-136 does
# exactly this for sub agents, and says why:
#
# # Same working directory and same permission mode as the parent: the tool
# # copies already point at the parent's cwd, and delegating to a sub agent
# # must not become a way around the session's permissions.
#
# A delegated turn gets a session id synthesised per (source, target, root), so
# the target's session prefs never exist and the mode fell back to the global
# default -- full-access.
# ---------------------------------------------------------------------------
class _Source:
"""Stands in for the delegating Agent, already in a mode for this session."""
def __init__(self, mode="read-only"):
self._mode = mode
def effective_permission_mode(self):
return self._mode
_DEFAULT = object()
class PeekBridge(FakeBridge):
"""A bridge that has a live source Agent to look up.
``source=None`` means genuinely no live instance, which is what
``peek_agent`` returns for a nested hop; the default is a live source in
read-only. A sentinel keeps those two apart.
"""
def __init__(self, registry=None, source_mode="read-only", source=_DEFAULT):
super().__init__(registry)
self._source = _Source(source_mode) if source is _DEFAULT else source
self.peeked = []
def peek_agent(self, session_id, agent_id=None):
self.peeked.append((session_id, agent_id))
return self._source
def test_a_delegated_run_inherits_the_delegating_permission_mode():
bridge = PeekBridge()
tool = _tool(bridge=bridge)
result = tool.execute({"agent_id": "research", "task": "Check the evidence"})
assert result.status == "success"
_, context, _ = bridge.calls[0]
assert context.get("delegated_permission_mode") == "read-only"
# Read from the source's own session, not the delegated one.
assert bridge.peeked == [("user-session", "primary")]
def test_the_inherited_mode_is_the_one_actually_in_force():
# Not the mode the source was configured with, but the one in force for this
# conversation -- a session may have narrowed it.
bridge = PeekBridge(source_mode="workspace-write")
tool = _tool(bridge=bridge)
tool.execute({"agent_id": "research", "task": "Check the evidence"})
_, context, _ = bridge.calls[0]
assert context.get("delegated_permission_mode") == "workspace-write"
def test_no_live_source_leaves_the_mode_alone():
"""peek_agent returns None when the source has no instance (a nested hop).
Guessing a mode there would either invent a restriction the user never set
or, worse, widen one they did -- so the key is left off and the target keeps
its own.
"""
bridge = PeekBridge(source=None)
tool = _tool(bridge=bridge)
result = tool.execute({"agent_id": "research", "task": "Check the evidence"})
assert result.status == "success"
_, context, _ = bridge.calls[0]
assert "delegated_permission_mode" not in context
def test_a_failing_lookup_does_not_break_the_delegation():
class _Broken(PeekBridge):
def peek_agent(self, session_id, agent_id=None):
raise RuntimeError("registry is locked")
tool = _tool(bridge=_Broken())
result = tool.execute({"agent_id": "research", "task": "Check the evidence"})
assert result.status == "success"
_, context, _ = tool.agent_bridge.calls[0]
assert "delegated_permission_mode" not in context
def test_ordinary_turns_still_read_their_prefs():
"""The control: without delegation the mode comes from session_prefs.
Exercised through the real apply_session_prefs, since that is where the two
paths meet.
"""
class _Target:
def __init__(self):
self.applied = []
self.model = None
def apply_permission_mode(self, mode):
self.applied.append(mode)
from bridge.agent_bridge import AgentBridge
bridge = AgentBridge.__new__(AgentBridge)
target = _Target()
bridge.apply_session_prefs(target, "ordinary-session", "primary")
assert target.applied == [None], "no prefs and no override means 'use global'"
target = _Target()
bridge.apply_session_prefs(
target, "ordinary-session", "primary", permission_mode="read-only"
)
assert target.applied == ["read-only"]
def test_the_read_only_gate_lets_both_delegation_tools_through():
"""Why the fix is a mode hand-off and not a stricter gate.
`subagent` is in _KNOWN_TOOLS and is therefore allowed in read-only --
correctly, because runner.py copies the mode onto the child.
`agent_delegate` was allowed too, but by omission: it is not in the table
and _MUTATING_NAME_RE does not match the verb. Now that the mode travels,
both are allowed for the same reason.
"""
from agent.permission import policy
assert policy.check_tool_call("read-only", "agent_delegate", {}).allowed
assert policy.check_tool_call("read-only", "subagent", {}).allowed