refactor: move the email alias rule into tool_security; extract the assistant seed constant

Code-quality pass over the PR's own changes:

- The bare<->qualified email aliasing rule lived inline in the generic
  dispatcher (_execute_tool_block_impl). It is policy knowledge, so it
  moves next to BUILTIN_EMAIL_TOOLS as email_tool_policy_names(); the
  dispatcher just consumes it, and the rule gets its own unit test
  (including the mcp__email__<not-a-tool> and mcp__other__ non-alias
  cases).

- The default Assistant's enabled_tools list was an inline literal
  inside the CrewMember seed, and its registry-sync test asserted a
  source-code substring. Extracted to DEFAULT_ASSISTANT_ENABLED_TOOLS
  so the test imports and checks the actual value.

- _fenced_tool_call return type tightened to Optional[Tuple[str, str]].

No behavior change; suite green (3295 passed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
botinate
2026-06-11 22:25:30 +02:00
parent db29046e0b
commit 016ce473f4
5 changed files with 71 additions and 32 deletions
+16 -13
View File
@@ -13,6 +13,21 @@ from src.tool_security import BUILTIN_EMAIL_TOOLS
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# Tools granted to the default Assistant crew member created on first run.
# Email names derive from BUILTIN_EMAIL_TOOLS so a tool added to the email
# server is granted to new default assistants automatically; users can still
# untick tools in the assistant settings UI.
DEFAULT_ASSISTANT_ENABLED_TOOLS = (
"manage_calendar", "manage_notes", "manage_tasks", "manage_memory",
*sorted(BUILTIN_EMAIL_TOOLS),
"resolve_contact",
"search_chats", "web_search", "web_fetch", "read_file",
"create_document", "update_document", "edit_document",
"generate_image", "trigger_research",
"download_model", "serve_model", "list_served_models", "stop_served_model",
"edit_image",
)
def _utcnow() -> datetime: def _utcnow() -> datetime:
"""Return naive UTC for task DB fields without using deprecated APIs.""" """Return naive UTC for task DB fields without using deprecated APIs."""
@@ -2293,19 +2308,7 @@ class TaskScheduler:
model=model, model=model,
endpoint_url=endpoint_url, endpoint_url=endpoint_url,
greeting=None, greeting=None,
enabled_tools=json.dumps([ enabled_tools=json.dumps(list(DEFAULT_ASSISTANT_ENABLED_TOOLS)),
"manage_calendar", "manage_notes", "manage_tasks", "manage_memory",
# Full built-in email tool set (single source of truth in
# tool_security) so a tool added to the email server is
# granted to new default assistants without touching this.
*sorted(BUILTIN_EMAIL_TOOLS),
"resolve_contact",
"search_chats", "web_search", "web_fetch", "read_file",
"create_document", "update_document", "edit_document",
"generate_image", "trigger_research",
"download_model", "serve_model", "list_served_models", "stop_served_model",
"edit_image",
]),
session_id=session_id, session_id=session_id,
is_active=True, is_active=True,
sort_order=0, sort_order=0,
+11 -13
View File
@@ -21,7 +21,12 @@ from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
from src.tool_security import BUILTIN_EMAIL_TOOLS, is_public_blocked_tool, owner_is_admin_or_single_user from src.tool_security import (
BUILTIN_EMAIL_TOOLS,
email_tool_policy_names,
is_public_blocked_tool,
owner_is_admin_or_single_user,
)
from src.tool_policy import ToolPolicy from src.tool_policy import ToolPolicy
from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, MAX_DIFF_LINES, DATA_DIR from src.constants import MAX_OUTPUT_CHARS, MAX_READ_CHARS, MAX_DIFF_LINES, DATA_DIR
from src.tool_utils import _truncate, get_mcp_manager from src.tool_utils import _truncate, get_mcp_manager
@@ -559,18 +564,11 @@ async def _execute_tool_block_impl(
tool = block.tool_type tool = block.tool_type
content = block.content content = block.content
# A bare email tool name is an alias for its MCP-qualified form (the # The block/disable gates below must match every policy-equivalent
# dispatch below routes it to mcp__email__<name>). Policy sources spell # spelling of the tool name (bare email names alias their mcp__email__
# email tools either way — plan mode and the MCP settings toggle write the # form — see email_tool_policy_names), not just the spelling the model
# qualified name into the denylist, chat-level toggles the bare one — so # happened to emit.
# the block/disable gates below must match on BOTH spellings. Gating only policy_names = email_tool_policy_names(tool)
# the name the model happened to emit lets a bare fence slip past a
# qualified denylist entry (and vice versa).
policy_names = {tool}
if tool in BUILTIN_EMAIL_TOOLS:
policy_names.add(f"mcp__email__{tool}")
elif tool.startswith("mcp__email__") and tool[len("mcp__email__"):] in BUILTIN_EMAIL_TOOLS:
policy_names.add(tool[len("mcp__email__"):])
# Misformatted tool call detection: model put JSON inside ```python``` (or # Misformatted tool call detection: model put JSON inside ```python``` (or
# similar) without naming the tool. Common with MiniMax-style outputs. # similar) without naming the tool. Common with MiniMax-style outputs.
+2 -2
View File
@@ -9,7 +9,7 @@ import ast
import json import json
import logging import logging
import re import re
from typing import List, Optional from typing import List, Optional, Tuple
from src.agent_tools import ToolBlock, TOOL_TAGS from src.agent_tools import ToolBlock, TOOL_TAGS
@@ -41,7 +41,7 @@ _TOOL_BLOCK_RE = re.compile(
_CODE_FENCE_TAGS = frozenset({"bash", "python"}) _CODE_FENCE_TAGS = frozenset({"bash", "python"})
def _fenced_tool_call(m) -> Optional[tuple]: def _fenced_tool_call(m) -> Optional[Tuple[str, str]]:
"""Classify a Pattern-1 fence match: (tag, content) when it is an """Classify a Pattern-1 fence match: (tag, content) when it is an
executable tool call, None when the fence must stay display text. executable tool call, None when the fence must stay display text.
+22
View File
@@ -184,6 +184,28 @@ def plan_mode_disabled_tools() -> Set[str]:
return (all_names | _PLAN_MODE_KNOWN_MUTATORS) - PLAN_MODE_READONLY_TOOLS return (all_names | _PLAN_MODE_KNOWN_MUTATORS) - PLAN_MODE_READONLY_TOOLS
def email_tool_policy_names(tool_name: str) -> frozenset:
"""All policy-equivalent spellings of a tool name.
A bare built-in email tool name and its MCP-qualified mcp__email__<name>
form dispatch to the same email server tool, but policy sources spell
them either way — plan mode and the MCP settings toggle write qualified
names into denylists, chat-level toggles write bare ones. Every gate must
match against the full alias set, or a call in one spelling slips past a
denylist entry written in the other. Non-email names alias only to
themselves.
"""
if not isinstance(tool_name, str):
return frozenset((tool_name,))
if tool_name in BUILTIN_EMAIL_TOOLS:
return frozenset((tool_name, f"mcp__email__{tool_name}"))
if tool_name.startswith("mcp__email__"):
bare = tool_name[len("mcp__email__"):]
if bare in BUILTIN_EMAIL_TOOLS:
return frozenset((tool_name, bare))
return frozenset((tool_name,))
def is_public_blocked_tool(tool_name: Optional[str]) -> bool: def is_public_blocked_tool(tool_name: Optional[str]) -> bool:
"""Return True when a non-admin/public user must not execute this tool. """Return True when a non-admin/public user must not execute this tool.
+20 -4
View File
@@ -116,7 +116,23 @@ def test_frontend_tool_selector_covers_email_tools():
def test_default_assistant_seed_covers_email_tools(): def test_default_assistant_seed_covers_email_tools():
"""The default Assistant crew seed grants the full email set.""" """The default Assistant crew seed grants the full email set."""
source = (_REPO_ROOT / "src" / "task_scheduler.py").read_text() from src.task_scheduler import DEFAULT_ASSISTANT_ENABLED_TOOLS
assert "*sorted(BUILTIN_EMAIL_TOOLS)" in source, (
"default assistant enabled_tools no longer derives from BUILTIN_EMAIL_TOOLS" assert BUILTIN_EMAIL_TOOLS <= set(DEFAULT_ASSISTANT_ENABLED_TOOLS)
)
def test_email_policy_name_aliases():
"""The alias rule every execution gate relies on."""
from src.tool_security import email_tool_policy_names
assert email_tool_policy_names("list_emails") == {
"list_emails", "mcp__email__list_emails",
}
assert email_tool_policy_names("mcp__email__delete_email") == {
"delete_email", "mcp__email__delete_email",
}
# Non-email names alias only to themselves — including mcp__email__
# spellings of tools the email server doesn't expose.
assert email_tool_policy_names("bash") == {"bash"}
assert email_tool_policy_names("mcp__email__not_a_tool") == {"mcp__email__not_a_tool"}
assert email_tool_policy_names("mcp__other__list_emails") == {"mcp__other__list_emails"}