[eric] workflows: enforce a frozen Actions tool set at launch, it was computed and then discarded

This commit is contained in:
ciregenz
2026-07-31 15:52:01 -07:00
parent 61d824b14f
commit e4f917a8c6
5 changed files with 103 additions and 10 deletions
+2 -1
View File
@@ -9,7 +9,8 @@ class AgentConfig(BaseModel):
mode: str = "agent"
provider: str = "anthropic"
system_prompt: Optional[str] = None
allowed_tools: list[str] = Field(default_factory=lambda: ["Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion"])
# None means "whatever the mode allows". A list is an actual restriction and IS enforced at launch, so it must stay None unless the caller really means to narrow the surface.
allowed_tools: Optional[list[str]] = None
max_turns: Optional[int] = None
target_directory: Optional[str] = None
dashboard_id: Optional[str] = None
+16 -1
View File
@@ -29,6 +29,21 @@ from backend.apps.agents.manager.prompt.prompt_context import resolve_mode
logger = logging.getLogger(__name__)
@typechecked
def resolve_launch_tools(mode_tools: List[str], allowed: Optional[List[str]]) -> List[str]:
"""The tool surface a session actually launches with.
None means "whatever the mode allows". A list is a real restriction: a workflow whose Actions
set the user froze. This used to be computed and then discarded, so the Actions page sold a
boundary the runtime never enforced. Intersecting rather than trusting means a stale saved set
can only ever narrow the mode, never widen it.
"""
if allowed is None:
return mode_tools
permitted = set(allowed)
return [t for t in mode_tools if t in permitted]
from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtocol
@@ -50,7 +65,7 @@ class AgentLaunch(AgentManagerProtocol):
config.target_directory = bound
mode_tools, _, mode_folder = resolve_mode(config.mode, get_all_tool_names)
tools = mode_tools
tools = resolve_launch_tools(mode_tools, config.allowed_tools)
global_settings = load_settings()
effective_cwd = (
+2 -3
View File
@@ -273,9 +273,8 @@ async def execute(
mode=wf.mode or "agent",
provider=wf.provider or "anthropic",
system_prompt=_resolve_system_prompt(wf),
allowed_tools=resolved_allowed_tools if resolved_allowed_tools is not None else [
"Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion",
],
# None when the user has not frozen the Actions set, which means the workflow runs with the mode's full surface exactly like a chat does.
allowed_tools=resolved_allowed_tools,
dashboard_id=resolve_workflow_dashboard_id(wf),
workflow_run_id=run.id,
)
+6 -5
View File
@@ -1022,7 +1022,8 @@ async def edit_agent_session(workflow_id: str):
mode=wf.mode or "agent",
provider=wf.provider or "anthropic",
system_prompt=system_prompt,
allowed_tools=[],
# None, not [], since an empty list is now a real restriction and would leave this chat with no tools at all.
allowed_tools=None,
dashboard_id=edit_dashboard_id,
workflow_edit_id=wf.id,
)
@@ -1261,9 +1262,8 @@ async def test_run_workflow(workflow_id: str, body: dict):
mode=wf.mode or "agent",
provider=wf.provider or "anthropic",
system_prompt=executor._resolve_system_prompt(wf),
allowed_tools=resolved_allowed_tools if resolved_allowed_tools is not None else [
"Read", "Edit", "Write", "Bash", "Glob", "Grep", "AskUserQuestion",
],
# A test run has to hit the same tool surface the scheduled run will, or the test proves nothing.
allowed_tools=resolved_allowed_tools,
dashboard_id=test_dashboard_id,
)
session = await agent_manager.launch_agent(config)
@@ -1400,7 +1400,8 @@ async def schedule_agent_session(workflow_id: str):
mode=wf.mode or "agent",
provider=wf.provider or "anthropic",
system_prompt=system_prompt,
allowed_tools=[],
# None, not [], since an empty list is now a real restriction and would leave this chat with no tools at all.
allowed_tools=None,
dashboard_id=wf.dashboard_id,
)
session = await agent_manager.launch_agent(config)
@@ -0,0 +1,77 @@
"""A workflow's frozen Actions set has to actually restrict the run.
It did not. `AgentConfig.allowed_tools` was computed by the executor, passed to launch, and then
dropped on the floor: launch resolved tools purely from the mode. So the Actions page offered a
permission toggle that no dispatch code read, which is worse than offering nothing, because it
sells a boundary that is not there. An unattended 3am run with Bash and nobody to deny an approval
is exactly the case the toggle exists for.
None still means "whatever the mode allows", so an unfrozen workflow keeps the full surface.
Run:
cd backend && .venv/bin/python -m pytest tests/test_frozen_workflow_tools.py -v
"""
from __future__ import annotations
from typing import List
from backend.apps.agents.core.models import AgentConfig
from backend.apps.agents.manager.AgentLaunch import resolve_launch_tools
def p_resolve(config: AgentConfig, mode_tools: List[str]) -> List[str]:
"""Drives the REAL launch-time resolver, not a copy of it. A mirrored implementation here would
keep passing even if launch went back to ignoring the frozen set entirely."""
return resolve_launch_tools(mode_tools, config.allowed_tools)
P_MODE = ["Read", "Edit", "Write", "Bash", "Glob", "Grep", "WebSearch", "AskUserQuestion"]
def test_default_is_unrestricted():
"""Every normal chat posts a config with no allowed_tools. If the default were a list instead of
None, honouring it would silently strip the whole app down to that list."""
assert AgentConfig(name="chat").allowed_tools is None
def test_unfrozen_workflow_keeps_the_full_mode_surface():
assert p_resolve(AgentConfig(name="wf"), P_MODE) == P_MODE
def test_frozen_set_actually_removes_tools():
config = AgentConfig(name="wf", allowed_tools=["Read", "Glob"])
resolved = p_resolve(config, P_MODE)
assert resolved == ["Read", "Glob"]
assert "Bash" not in resolved, "the whole point: a frozen set must be able to withhold Bash"
assert "Write" not in resolved
def test_a_frozen_set_cannot_widen_the_mode():
"""Intersect, never trust. A stale saved set naming a tool the mode does not grant must not
smuggle it back in."""
config = AgentConfig(name="wf", allowed_tools=["Read", "NotebookEdit", "BrowserClick"])
resolved = p_resolve(config, ["Read", "Bash"])
assert resolved == ["Read"]
def test_an_explicitly_empty_set_grants_nothing():
"""Distinct from None on purpose. This is why the workflow edit and scheduling chats had to move
off [] and onto None: under the old dead code [] was harmless, now it means zero tools."""
assert p_resolve(AgentConfig(name="wf", allowed_tools=[]), P_MODE) == []
def test_order_follows_the_mode_not_the_saved_set():
config = AgentConfig(name="wf", allowed_tools=["Grep", "Read"])
assert p_resolve(config, P_MODE) == ["Read", "Grep"]
def test_workflow_call_sites_do_not_pass_an_empty_list():
"""Guards the migration: an [] left behind at any AgentConfig site is now a silently toolless
agent, which reads to the user as the agent being broken."""
import re
from pathlib import Path
root = Path(__file__).resolve().parents[1]
for rel in ("apps/workflows/workflows.py", "apps/workflows/executor.py"):
src = (root / rel).read_text()
assert not re.search(r"allowed_tools=\[\]", src), f"{rel} still passes an empty allowed_tools"