Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions helpers/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ class Settings(TypedDict):

update_check_enabled: bool
chat_inherit_project: bool
subagent_spawn_locked: bool


class PartialSettings(Settings, total=False):
Expand Down Expand Up @@ -487,6 +488,7 @@ def get_default_settings() -> Settings:
litellm_global_kwargs=get_default_value("litellm_global_kwargs", {}),
update_check_enabled=get_default_value("update_check_enabled", True),
chat_inherit_project=get_default_value("chat_inherit_project", True),
subagent_spawn_locked=get_default_value("subagent_spawn_locked", False),
)


Expand Down
147 changes: 147 additions & 0 deletions tests/test_subagent_spawn_lock.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
from __future__ import annotations

import asyncio
import importlib
import sys
import types
from dataclasses import dataclass
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))


@dataclass
class _FakeResponse:
message: str
break_loop: bool
additional: dict | None = None


class _FakeTool:
def __init__(
self,
agent,
name: str,
method: str | None,
args: dict | None,
message: str,
loop_data=None,
**kwargs,
) -> None:
self.agent = agent
self.name = name
self.method = method
self.args = args or {}
self.message = message
self.loop_data = loop_data


class _FakeAgentClass:
DATA_NAME_SUBORDINATE = "_subordinate"
DATA_NAME_SUPERIOR = "_superior"


class _FakeAgent:
def __init__(self) -> None:
self._data: dict = {}
self.number = 0
self.context = types.SimpleNamespace(id="ctx", log=None)

def get_data(self, key: str):
return self._data.get(key)

def set_data(self, key: str, value) -> None:
self._data[key] = value


def _install_call_subordinate_stubs(monkeypatch, *, locked: bool) -> None:
agent_stub = types.ModuleType("agent")
agent_stub.Agent = _FakeAgentClass

class _UserMessage:
def __init__(self, message: str = "", attachments=None) -> None:
self.message = message
self.attachments = attachments or []

agent_stub.UserMessage = _UserMessage
monkeypatch.setitem(sys.modules, "agent", agent_stub)

tool_stub = types.ModuleType("helpers.tool")
tool_stub.Tool = _FakeTool
tool_stub.Response = _FakeResponse
monkeypatch.setitem(sys.modules, "helpers.tool", tool_stub)

initialize_stub = types.ModuleType("initialize")
initialize_stub.initialize_agent = lambda: types.SimpleNamespace(profile="")
monkeypatch.setitem(sys.modules, "initialize", initialize_stub)

ext_pkg = types.ModuleType("extensions")
monkeypatch.setitem(sys.modules, "extensions", ext_pkg)
ext_py = types.ModuleType("extensions.python")
monkeypatch.setitem(sys.modules, "extensions.python", ext_py)
hist_stub = types.ModuleType("extensions.python.hist_add_tool_result")
hist_stub._90_save_tool_call_file = types.SimpleNamespace(LEN_MIN=10**9)
monkeypatch.setitem(
sys.modules, "extensions.python.hist_add_tool_result", hist_stub
)

settings_stub = types.ModuleType("helpers.settings")
settings_stub.get_settings = lambda: {"subagent_spawn_locked": locked}
monkeypatch.setitem(sys.modules, "helpers.settings", settings_stub)

# ensure a fresh import of the tool module with the active stubs
sys.modules.pop("tools.call_subordinate", None)


def _make_delegation(monkeypatch, *, locked: bool):
_install_call_subordinate_stubs(monkeypatch, locked=locked)
module = importlib.import_module("tools.call_subordinate")
tool = module.Delegation(
_FakeAgent(),
"call_subordinate",
None,
{"message": "hello", "reset": "true"},
"",
None,
)
return module, tool


def test_call_subordinate_is_blocked_when_subagent_spawn_locked(monkeypatch):
module, tool = _make_delegation(monkeypatch, locked=True)

response = asyncio.run(tool.execute(**tool.args))

assert isinstance(response, module.Response)
assert response.break_loop is False
assert "locked" in response.message.lower()
# no subordinate must have been created on the agent
assert tool.agent.get_data(_FakeAgentClass.DATA_NAME_SUBORDINATE) is None


def test_call_subordinate_does_not_short_circuit_when_unlocked(monkeypatch):
module, tool = _make_delegation(monkeypatch, locked=False)

# When unlocked, the tool must walk past the gate and try to construct
# a subordinate Agent. Our fake Agent class is not constructible, so we
# detect "did not short-circuit" by observing the resulting TypeError.
try:
asyncio.run(tool.execute(**tool.args))
except TypeError as exc:
assert "_FakeAgentClass() takes no arguments" in str(exc)
else: # pragma: no cover - defensive
raise AssertionError(
"Lock unexpectedly short-circuited when subagent_spawn_locked=False"
)


def test_subagent_spawn_locked_default_is_false_in_settings_source():
"""The setting must default to False so existing installs are unaffected."""
settings_src = (PROJECT_ROOT / "helpers" / "settings.py").read_text(
encoding="utf-8"
)
assert "subagent_spawn_locked: bool" in settings_src
assert 'get_default_value("subagent_spawn_locked", False)' in settings_src
15 changes: 15 additions & 0 deletions tools/call_subordinate.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,27 @@
from agent import Agent, UserMessage
from helpers.tool import Tool, Response
from helpers import settings
from initialize import initialize_agent
from extensions.python.hist_add_tool_result import _90_save_tool_call_file as save_tool_call_file


class Delegation(Tool):

async def execute(self, message="", reset="", **kwargs):
# honor the runtime lock that prevents autonomous subagent spawning so a
# manually designed multi-agent system stays exactly as built
if settings.get_settings().get("subagent_spawn_locked", False):
existing = self.agent.get_data(Agent.DATA_NAME_SUBORDINATE)
if existing is None or str(reset).lower().strip() == "true":
return Response(
message=(
"call_subordinate is locked: autonomous subagent spawning is "
"disabled by the 'subagent_spawn_locked' setting. Continue "
"without delegating, or ask the user to switch the toggle off."
),
break_loop=False,
)

# create subordinate agent using the data object on this agent and set superior agent to his data object
if (
self.agent.get_data(Agent.DATA_NAME_SUBORDINATE) is None
Expand Down
15 changes: 15 additions & 0 deletions webui/components/settings/agent/agent.html
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,21 @@
</label>
</div>
</div>

<div class="field">
<div class="field-label">
<div class="field-title">Lock subagent spawning</div>
<div class="field-description">
Prevent the agent from autonomously spawning new subordinate agents via call_subordinate. Useful for stable, intentionally designed multi-agent pipelines.
</div>
</div>
<div class="field-control">
<label class="toggle">
<input type="checkbox" x-model="$store.settings.settings.subagent_spawn_locked">
<span class="toggler"> </span>
</label>
</div>
</div>
</div>
</div>
</div>
Expand Down