diff --git a/cecli/coders/agent_coder.py b/cecli/coders/agent_coder.py
index 8ad410dd529..b9622fd856a 100644
--- a/cecli/coders/agent_coder.py
+++ b/cecli/coders/agent_coder.py
@@ -17,6 +17,7 @@
 from cecli.change_tracker import ChangeTracker
 from cecli.helpers import nested, responses
 from cecli.helpers.agents.service import AgentService
+from cecli.helpers.agents.tree import AgentTree
 from cecli.helpers.background_commands import BackgroundCommandManager
 from cecli.helpers.conversation import ConversationService, MessageTag
 from cecli.helpers.similarity import (
@@ -62,7 +63,7 @@ def __init__(self, *args, **kwargs):
         self.tool_similarity_threshold = 0.95
         self.max_tool_vector_history = 20
         self.read_tools = {
-            "command",
+            "shell",
             "explorecode",
             "readfile",
             "grep",
@@ -70,7 +71,7 @@ def __init__(self, *args, **kwargs):
             "searchfacts",
         }
         self.write_tools = {
-            "command",
+            "shell",
             "editfile",
             "undochange",
             "replacefacts",
@@ -233,7 +234,7 @@ def _get_agent_config(self, config=None):
             config,
             ["tools_excludelist", "tools_blacklist"],
             [
-                "create",
+                "manageextensions",
                 "gitbranch",
                 "gitdiff",
                 "gitlog",
@@ -782,7 +783,7 @@ def get_context_summary(self):
                 result += f" ({percentage:.1f}% of limit)"
                 if percentage > 80:
                     result += "\n\n⚠ **Context is getting full!**\n"
-                    result += "- Remove non-essential files, skills and tool servers via the `ResourceManager` tool.\n"
+                    result += "- Remove non-essential files, skills and tool servers via the `ManageResources` tool.\n"
                     result += "- Keep only essential files, skills, and MCP servers in context for best performance"
             result += "\n</context>"
             if not hasattr(self, "context_blocks_cache"):
@@ -821,7 +822,10 @@ def get_environment_info(self):
             primary_uuid = str(service.coder.uuid)
             self_name = service.get_agent_name(self) or "primary"
             result += f"- Agent: {self_name} ({self_uuid}) [self]\n"
-            result += f"- Primary agent: primary ({primary_uuid})\n"
+            # Workflow agents run in an isolated tree and must not see the
+            # primary agent (nor it them).
+            if AgentTree.is_visible(self_uuid, primary_uuid):
+                result += f"- Primary agent: primary ({primary_uuid})\n"
             if self.repo:
                 try:
                     rel_repo_dir = self.repo.get_rel_repo_dir()
@@ -1515,7 +1519,7 @@ async def check_for_file_mentions(self, content):
 
         Override parent's method to disable implicit file mention handling in agent mode.
         Files should only be added via explicit tool commands
-        (`ResourceManager`).
+        (`ManageResources`).
         """
         pass
 
@@ -1812,6 +1816,11 @@ def get_sub_agents_context(self):
             return None
         try:
             registry = AgentService.get_registry()
+            # Workflow agents live in isolated trees; the primary must not see
+            # them as delegatable sub-agent types.
+            registry = {
+                name: config for name, config in registry.items() if not name.startswith("wf:")
+            }
             if not registry:
                 return None
 
@@ -1842,9 +1851,11 @@ def get_sub_agent_states(self):
         independent sub-agents. The independent ones are those reachable via
         the ``Broadcast`` tool.
 
-        Finished/errored sub-agents are excluded. Returns None if there are no
-        active sub-agents, if enhanced context is disabled, or if the caller is
-        a sub-agent without nested delegation enabled.
+        Finished/errored sub-agents are excluded. Agents inside a workflow tree
+        see every active agent in that tree rather than only their own children.
+        Returns None if there are no active sub-agents, if enhanced context is
+        disabled, or if the caller is a sub-agent without nested delegation
+        enabled.
 
         The primary agent and calling agent's own identity are deliberately not
         included here — they live in ``get_environment_info()`` (a static block)
@@ -1868,34 +1879,50 @@ def get_sub_agent_states(self):
             from cecli.helpers.agents.service import SubAgentStatus
 
             originator_uuid = str(self.uuid)
-
-            # Dependent children are the coder's direct children.
-            children = service.get_children(self)
-            dependent_children = [
-                info
-                for info in children
-                if not info.independent
-                and not info.detached
-                and info.status not in (SubAgentStatus.FINISHED, SubAgentStatus.ERROR)
-            ]
-
-            # Independent agents aren't all direct children, so iterate over
-            # every active sub-agent reachable via the Broadcast tool and dedupe
-            # against the dependent children already captured above.
-            dependent_uuids = {info.coder.uuid for info in dependent_children}
-            independent_agents = []
-            seen = set()
-            for info in service.sub_agents.values():
-                if (
-                    info.independent
+            visible_uuids = AgentTree.get_visible_uuids(originator_uuid)
+            workflow_root = AgentTree.get_workflow_root(originator_uuid)
+
+            if workflow_root is not None:
+                # Inside a workflow tree, every active agent in that tree is
+                # visible: the workflow root, its children and their descendants.
+                dependent_children = [
+                    info
+                    for info in service.sub_agents.values()
+                    if str(info.coder.uuid) != originator_uuid
+                    and str(info.coder.uuid) in visible_uuids
                     and not info.detached
-                    and str(info.coder.uuid) != originator_uuid
-                    and info.status not in (SubAgentStatus.ERROR,)
-                    and info.coder.uuid not in dependent_uuids
-                    and info.coder.uuid not in seen
-                ):
-                    seen.add(info.coder.uuid)
-                    independent_agents.append(info)
+                    and info.status not in (SubAgentStatus.FINISHED, SubAgentStatus.ERROR)
+                ]
+                independent_agents = []
+            else:
+                # Outside a workflow tree, dependent agents are the coder's
+                # direct children; independent agents are the others reachable
+                # through Broadcast in the same visibility domain.
+                children = service.get_children(self)
+                dependent_children = [
+                    info
+                    for info in children
+                    if not info.independent
+                    and not info.detached
+                    and str(info.coder.uuid) in visible_uuids
+                    and info.status not in (SubAgentStatus.FINISHED, SubAgentStatus.ERROR)
+                ]
+
+                dependent_uuids = {info.coder.uuid for info in dependent_children}
+                independent_agents = []
+                seen = set()
+                for info in service.sub_agents.values():
+                    if (
+                        info.independent
+                        and not info.detached
+                        and str(info.coder.uuid) != originator_uuid
+                        and str(info.coder.uuid) in visible_uuids
+                        and info.status not in (SubAgentStatus.ERROR,)
+                        and info.coder.uuid not in dependent_uuids
+                        and info.coder.uuid not in seen
+                    ):
+                        seen.add(info.coder.uuid)
+                        independent_agents.append(info)
 
             if not dependent_children and not independent_agents:
                 return None
@@ -1937,7 +1964,7 @@ def get_background_command_output(self):
 
         Emits a roster of active commands (keeping command keys in context)
         plus any new incremental output. Output that has been paged to disk is
-        advertised by command key and read on demand through ``ResourceManager``
+        advertised by command key and read on demand through ``ManageResources``
         paging.
         """
         command_info = BackgroundCommandManager.list_background_commands()
@@ -1970,7 +1997,7 @@ def get_background_command_output(self):
             output += f"\nNew output ({command_key}):\n{cmd_output}\n"
 
         if paged_keys:
-            output += "\nPaged output is available via `ResourceManager` (up to 3 pages):\n"
+            output += "\nPaged output is available via `ManageResources` (up to 3 pages):\n"
             for command_key in paged_keys:
                 output += f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}\n'
 
diff --git a/cecli/coders/base_coder.py b/cecli/coders/base_coder.py
index e9d6db300b7..ed1cc94fd88 100755
--- a/cecli/coders/base_coder.py
+++ b/cecli/coders/base_coder.py
@@ -1054,8 +1054,11 @@ def get_announcements(self):
             from cecli.helpers.agents.service import AgentService
 
             registry = AgentService.get_registry()
-            if registry:
-                capabilities["Subagents"] = sorted(registry.keys())
+            # Workflow agents belong to isolated trees, so they are not
+            # advertised as general-purpose subagent types.
+            subagents = sorted(name for name in registry if not name.startswith("wf:"))
+            if subagents:
+                capabilities["Subagents"] = subagents
         except Exception:
             pass
 
@@ -2297,13 +2300,13 @@ async def compact_context_if_needed(self, force=False, message=""):
         is_min_size = all_tokens >= self.context_compaction_max_tokens
 
         # Condition 2: Percentage check (is chat history a significant part of the context?)
-        is_worth_by_percentage = all_tokens > 0 and (compactable_tokens / all_tokens) >= 0.20
+        is_worth_by_percentage = all_tokens > 0 and (compactable_tokens / all_tokens) >= 0.15
 
         # Condition 3: Absolute savings check (would compacting save enough tokens to fit?)
         tokens_over_limit = all_tokens - (self.context_compaction_max_tokens or all_tokens)
         potential_savings = compactable_tokens * 0.90
-        is_worth_by_absolute_savings = (
-            tokens_over_limit > 0 and potential_savings >= tokens_over_limit
+        is_worth_by_absolute_savings = tokens_over_limit > 0 and potential_savings >= (
+            tokens_over_limit * 0.50
         )
 
         if not force:
@@ -2787,10 +2790,6 @@ async def check_tokens(self, messages):
                 self.enable_context_compaction
                 and input_tokens >= self.context_compaction_max_tokens * 0.95
             ):
-                self.io.tool_output(
-                    f"Estimated chat context of {input_tokens:,} tokens exceeds the"
-                    f" {max_input_tokens:,} token limit. Attempting to compact..."
-                )
                 await self.compact_context_if_needed()
 
                 # After compaction, re-format messages and re-check tokens
@@ -2804,13 +2803,12 @@ async def check_tokens(self, messages):
                 if getattr(self, "_last_compaction_warning_time", 0) + 300 < time.time():
                     self._last_compaction_warning_time = time.time()
                     self.io.tool_error(
-                        f"Your estimated chat context of {input_tokens:,} tokens still exceeds the"
-                        f" {max_input_tokens:,} token limit for {self.get_active_model().name}!"
+                        f"Your estimated chat context of {input_tokens:,} tokens exceeds the"
+                        f" {max_input_tokens:,} token limit for {self.get_active_model().name}"
                     )
                     self.io.tool_output("To reduce the chat context:")
                     self.io.tool_output("- Use /drop to remove unneeded files from the chat")
                     self.io.tool_output("- Use /clear to clear the chat history")
-                    self.io.tool_output("- Break your code into smaller files")
                     self.io.tool_output(
                         "It's probably safe to try and send the request, most providers won't charge if"
                         " the context limit is exceeded."
diff --git a/cecli/commands/agent_tree.py b/cecli/commands/agent_tree.py
index 9e8cd96dfdc..951dc890a71 100644
--- a/cecli/commands/agent_tree.py
+++ b/cecli/commands/agent_tree.py
@@ -2,6 +2,7 @@
 
 from cecli.commands.utils.base_command import BaseCommand
 from cecli.helpers.agents.service import AgentService, SubAgentInfo, SubAgentStatus
+from cecli.helpers.agents.tree import AgentTree
 
 
 class TreeNode:
@@ -30,17 +31,15 @@ async def execute(cls, io, coder, args, **kwargs):
             return
 
         # Collect all agents: primary coder + sub-agents from sub_agents dict
-        all_infos: List[SubAgentInfo] = []
-
-        # Primary agent as a synthetic SubAgentInfo
         primary_coder = agent_service.coder
+        primary_uuid = str(primary_coder.uuid)
         primary_info = SubAgentInfo(
             name="primary",
             coder=primary_coder,
             parent_uuid=str(primary_coder.parent_uuid or ""),
             status=SubAgentStatus.RUNNING,
         )
-        all_infos.append(primary_info)
+        all_infos: List[SubAgentInfo] = [primary_info]
 
         # Sub-agents from the service's sub_agents dict
         for info in agent_service.sub_agents.values():
@@ -51,26 +50,38 @@ async def execute(cls, io, coder, args, **kwargs):
             io.tool_output("No active sub-agents found.")
             return
 
-        root_nodes = cls._build_tree(all_infos)
+        # The shared tree is process-wide, so structure the display from the
+        # primary's subtree only.
+        infos_by_uuid = {str(info.coder.uuid): info for info in all_infos}
+        subtree = AgentTree.get_subtree(primary_uuid)
+        if not subtree:
+            subtree = list(infos_by_uuid)
+
+        root_nodes = cls._build_tree(primary_uuid, infos_by_uuid, subtree)
         tree_output = cls._render_tree(root_nodes)
         io.tool_output(tree_output)
 
     @classmethod
-    def _build_tree(cls, all_agents: List[SubAgentInfo]) -> List[TreeNode]:
-        """Build the agent tree from a flat list of agents."""
-        nodes = {str(agent.coder.uuid): TreeNode(agent) for agent in all_agents}
-        root_nodes = []
-
-        for agent in all_agents:
-            parent_uuid = str(agent.coder.parent_uuid)
-            if parent_uuid and parent_uuid in nodes:
-                parent_node = nodes[parent_uuid]
-                child_node = nodes[str(agent.coder.uuid)]
-                parent_node.add_child(child_node)
+    def _build_tree(cls, root_uuid, infos, subtree_uuids) -> List[TreeNode]:
+        """Build the display tree from the shared :class:`AgentTree`."""
+        nodes = {uuid: TreeNode(infos[uuid]) for uuid in subtree_uuids if uuid in infos}
+        root = nodes.get(root_uuid)
+        if root is None:
+            return []
+
+        for uuid, node in nodes.items():
+            if uuid == root_uuid:
+                continue
+
+            parent = nodes.get(AgentTree.get_parent(uuid))
+            if parent is not None:
+                parent.add_child(node)
             else:
-                root_nodes.append(nodes[str(agent.coder.uuid)])
+                # Keep a node reachable even if its parent fell outside the
+                # session subtree (e.g. a spliced node left over).
+                root.add_child(node)
 
-        return root_nodes
+        return [root]
 
     @classmethod
     def _render_tree(cls, nodes: List[TreeNode]) -> str:
diff --git a/cecli/helpers/agents/defaults/creator.md b/cecli/helpers/agents/defaults/creator.md
index 269e9fa8dba..a4a655ebf0e 100644
--- a/cecli/helpers/agents/defaults/creator.md
+++ b/cecli/helpers/agents/defaults/creator.md
@@ -39,7 +39,7 @@ You create and maintain cecli extensions and configuration:
 - **custom tools** (`Tool` classes loaded from tools directories)
 - **custom commands** (`CustomCommand` classes loaded via `custom.command-paths`)
 - **workflows** (YAML workflow definitions)
-- **project/global configuration** (persisted to `creator.yml` through the `Create` tool)
+- **project/global configuration** (persisted to `creator.yml` through the `ManageExtensions` tool)
 
 Do not implement general application features, fix unrelated bugs, or debug the
 user's project.  If a request is outside this scope, explain that briefly and
@@ -64,11 +64,11 @@ Common locations:
 - Workflows: `.cecli/workflows/*.yml`, `~/.cecli/workflows/*.yml`
 - Custom commands: paths listed under `custom.command-paths`
 
-These are read/reference only — the `Create` tool owns all writes.
+These are read/reference only — the `ManageExtensions` tool owns all writes.
 
-## Persistence is the Create tool's job
+## Persistence is the ManageExtensions tool's job
 
-All extension and config writes go through `Create` — never edit `creator.yml`
+All extension and config writes go through `ManageExtensions` — never edit `creator.yml`
 or user config files (`.cecli.conf.yml`, `conf.yml`) yourself.  Each call
 writes the backing files, records the needed `creator.yml` entry (global
 `~/.cecli/creator.yml` or project-local `.cecli/creator.yml`, deep-merged with
@@ -91,7 +91,7 @@ non-extension settings use `type: config` with a dotted key path as `name`, and
 1. Clarify the goal with `Broadcast` when the request depends on prior context.
 2. Consult the documentation before choosing a format or config key.
 3. Inspect existing files and config before changing anything so you preserve
-   user work; use `Create(action: "list")` to see what is already persisted.
+   user work; use `ManageExtensions(action: "list")` to see what is already persisted.
 4. Make the smallest change that fully implements the request.
 5. Validate: run `python -m py_compile` for Python files, parse YAML/JSON, and
    re-read edited files to confirm the result.
diff --git a/cecli/helpers/agents/overrides.py b/cecli/helpers/agents/overrides.py
index a04932a617a..48973cb044b 100644
--- a/cecli/helpers/agents/overrides.py
+++ b/cecli/helpers/agents/overrides.py
@@ -80,7 +80,7 @@ def load_agent_overrides(primary_root: Optional[str] = None) -> Dict[str, Any]:
 def refresh_agent_overrides(primary_root: Optional[str] = None) -> Dict[str, Any]:
     """Re-read the ``creator.yml`` files and refresh the spawn-time index.
 
-    Both startup (``main_async``) and the Create tool use this single reader
+    Both startup (``main_async``) and the ManageExtensions tool use this single reader
     path: it re-reads the global and local ``creator.yml`` files, updates
     ``AgentService``'s override index, and re-registers/removes dynamic
     ``creator:<name>`` definitions. Call it after any mutation of a
diff --git a/cecli/helpers/agents/service.py b/cecli/helpers/agents/service.py
index 12827eb70f6..a920690b189 100644
--- a/cecli/helpers/agents/service.py
+++ b/cecli/helpers/agents/service.py
@@ -16,6 +16,7 @@
 
 import cecli.models as models
 from cecli.helpers import nested
+from cecli.helpers.agents.tree import AgentTree
 from cecli.helpers.coroutines import fire_and_forget
 
 logger = logging.getLogger(__name__)
@@ -351,6 +352,11 @@ def __init__(self, coder) -> None:
         # Register the primary coder in the global uuid map
         if hasattr(coder, "uuid"):
             self._uuid_coder_map[str(coder.uuid)] = weakref.ref(coder)
+            AgentTree.register(
+                str(coder.uuid),
+                parent_uuid=getattr(coder, "parent_uuid", None) or None,
+                name="primary",
+            )
         # uuid -> SubAgentInfo
         self.sub_agents: Dict[str, SubAgentInfo] = {}
         # Ordered list of sub-agent UUIDs for LRU reap
@@ -581,6 +587,7 @@ def _cleanup_sub_agent(self, agent_uuid: str) -> None:
         # not the parent's service instance. The parent's instance is cleaned
         # up separately in cleanup_all_for_parent().
         self._uuid_coder_map.pop(info.coder.uuid, None)
+        AgentTree.unregister(info.coder.uuid)
         self.destroy_instance(info.coder.uuid)
 
     def _check_max_sub_agents(self) -> None:
@@ -786,8 +793,13 @@ async def _create_sub_agent_coder(
         with self._get_lock(self._spawn_locks, parent_coder.uuid):
             self._check_max_sub_agents()
 
-            # Register in global coder lookup
+            # Register in global coder lookup and the visibility tree
             self._uuid_coder_map[new_uuid] = weakref.ref(new_coder)
+            AgentTree.register(
+                str(new_coder.uuid),
+                parent_uuid=str(parent_coder.uuid),
+                name=name,
+            )
 
             info = SubAgentInfo(
                 name=name,
@@ -882,6 +894,7 @@ async def switch_sub_agent_coder(self, info: SubAgentInfo, switch, preproc=True)
         kwargs["summarize_from_coder"] = False
 
         new_coder = await Coder.create(**kwargs)
+        AgentTree.replace(old_coder.uuid, new_coder.uuid)
 
         new_coder.prompt_queue = list(old_coder.prompt_queue)
         new_coder._queue_counter = old_coder._queue_counter
@@ -1389,6 +1402,7 @@ def cleanup_all_for_parent(self) -> None:
         """Clean up all sub-agents when the parent session ends."""
         for uuid in list(self.sub_agents.keys()):
             self._cleanup_sub_agent(uuid)
+        AgentTree.unregister(self.coder.uuid)
         # Clean up lock pools to prevent memory leaks
         self._spawn_locks.pop(self.coder.uuid, None)
         self._conversation_locks.pop(self.coder.uuid, None)
diff --git a/cecli/helpers/agents/tree.py b/cecli/helpers/agents/tree.py
new file mode 100644
index 00000000000..d7dc862bbed
--- /dev/null
+++ b/cecli/helpers/agents/tree.py
@@ -0,0 +1,271 @@
+"""Parent/child tree of tracked agent UUIDs.
+
+The tree is the single source of truth for agent visibility. A workflow agent
+(``wf:``) roots an isolated subtree: it cannot see the primary agent or any
+agent outside its own subtree, and the primary cannot see it. All other agents
+share the primary's visibility domain.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict, List, Optional, Set
+
+_PARENT = "parent"
+_CHILDREN = "children"
+_NAME = "name"
+_WORKFLOW = "workflow"
+
+
+class AgentTree:
+    """Process-wide tree of agent UUIDs used to resolve visibility.
+
+    Nodes are registered when a coder is created and removed when it is reaped.
+    Every lookup tolerates unknown UUIDs so callers never need to guard calls.
+    """
+
+    _nodes: Dict[str, Dict[str, Any]] = {}
+    _lock = threading.RLock()
+
+    # ------------------------------------------------------------------ #
+    # Mutation
+    # ------------------------------------------------------------------ #
+
+    @classmethod
+    def register(
+        cls,
+        agent_uuid: str,
+        parent_uuid: Optional[str] = None,
+        name: Optional[str] = None,
+    ) -> None:
+        """Add or reparent *agent_uuid*, inheriting its parent's workflow domain.
+
+        A ``wf:`` name starts a new workflow domain rooted at the node itself.
+        """
+        agent_key = str(agent_uuid)
+        parent_key = str(parent_uuid) if parent_uuid else None
+        if parent_key == agent_key:
+            parent_key = None
+
+        with cls._lock:
+            node = cls._ensure_node(agent_key)
+            if name is not None:
+                node[_NAME] = name
+
+            workflow = agent_key if cls._is_workflow_name(node[_NAME]) else None
+            if workflow is None and parent_key:
+                parent = cls._nodes.get(parent_key)
+                if parent is not None:
+                    workflow = parent[_WORKFLOW]
+
+            if node[_PARENT] != parent_key:
+                cls._detach_locked(agent_key)
+                node[_PARENT] = parent_key
+                if parent_key:
+                    cls._ensure_node(parent_key)[_CHILDREN].add(agent_key)
+
+            cls._set_workflow_locked(agent_key, workflow)
+
+    @classmethod
+    def unregister(cls, agent_uuid: str) -> List[str]:
+        """Remove *agent_uuid*, splicing its children onto its parent.
+
+        Returns the UUIDs that were removed (currently just the node itself).
+        """
+        agent_key = str(agent_uuid)
+        with cls._lock:
+            node = cls._nodes.pop(agent_key, None)
+            if node is None:
+                return []
+
+            parent_key = node[_PARENT]
+            parent = cls._nodes.get(parent_key) if parent_key else None
+            if parent is not None:
+                parent[_CHILDREN].discard(agent_key)
+
+            for child in list(node[_CHILDREN]):
+                child_node = cls._nodes.get(child)
+                if child_node is None:
+                    continue
+                child_node[_PARENT] = parent_key
+                if parent is not None:
+                    parent[_CHILDREN].add(child)
+
+            return [agent_key]
+
+    @classmethod
+    def replace(cls, old_uuid: str, new_uuid: str) -> None:
+        """Move *old_uuid*'s position and children onto *new_uuid*."""
+        old_key = str(old_uuid)
+        new_key = str(new_uuid)
+        if old_key == new_key:
+            return
+
+        with cls._lock:
+            old_node = cls._nodes.pop(old_key, None)
+            if old_node is None:
+                cls.register(new_key)
+                return
+
+            cls._detach_locked(new_key)
+            new_node = cls._ensure_node(new_key)
+            new_node[_NAME] = old_node[_NAME]
+            new_node[_PARENT] = old_node[_PARENT]
+
+            parent_key = old_node[_PARENT]
+            if parent_key:
+                parent = cls._nodes.get(parent_key)
+                if parent is not None:
+                    parent[_CHILDREN].discard(old_key)
+                    parent[_CHILDREN].add(new_key)
+
+            for child in list(old_node[_CHILDREN]):
+                child_node = cls._nodes.get(child)
+                if child_node is not None:
+                    child_node[_PARENT] = new_key
+                    new_node[_CHILDREN].add(child)
+
+            cls._set_workflow_locked(new_key, old_node[_WORKFLOW])
+
+    @classmethod
+    def clear(cls) -> None:
+        """Drop every node (used by tests and full session teardown)."""
+        with cls._lock:
+            cls._nodes.clear()
+
+    # ------------------------------------------------------------------ #
+    # Lookups
+    # ------------------------------------------------------------------ #
+
+    @classmethod
+    def get_parent(cls, agent_uuid: str) -> Optional[str]:
+        """Return the parent UUID of *agent_uuid*, or None if it is a root."""
+        with cls._lock:
+            node = cls._nodes.get(str(agent_uuid))
+            return node[_PARENT] if node is not None else None
+
+    @classmethod
+    def get_children(cls, agent_uuid: str) -> List[str]:
+        """Return the direct children of *agent_uuid*."""
+        with cls._lock:
+            node = cls._nodes.get(str(agent_uuid))
+            return list(node[_CHILDREN]) if node is not None else []
+
+    @classmethod
+    def get_descendants(cls, agent_uuid: str) -> List[str]:
+        """Return every descendant of *agent_uuid*, excluding itself."""
+        with cls._lock:
+            return cls._descendants_locked(str(agent_uuid), set())
+
+    @classmethod
+    def get_subtree(cls, agent_uuid: str) -> List[str]:
+        """Return *agent_uuid* and all of its descendants."""
+        with cls._lock:
+            agent_key = str(agent_uuid)
+            if agent_key not in cls._nodes:
+                return []
+            return [agent_key] + cls._descendants_locked(agent_key, {agent_key})
+
+    @classmethod
+    def get_workflow_root(cls, agent_uuid: str) -> Optional[str]:
+        """Return the UUID of the workflow tree containing *agent_uuid*.
+
+        Returns None for agents outside any workflow tree.
+        """
+        with cls._lock:
+            node = cls._nodes.get(str(agent_uuid))
+            return node[_WORKFLOW] if node is not None else None
+
+    @classmethod
+    def is_workflow_agent(cls, agent_uuid: str) -> bool:
+        """Return whether *agent_uuid* is itself a workflow (``wf:``) agent."""
+        with cls._lock:
+            node = cls._nodes.get(str(agent_uuid))
+            return node is not None and cls._is_workflow_name(node[_NAME])
+
+    @classmethod
+    def get_visible_uuids(cls, observer_uuid: str) -> Set[str]:
+        """Return the UUIDs that *observer_uuid* is allowed to see.
+
+        Agents outside a workflow domain see every non-workflow agent; agents
+        inside one see only the members of that same workflow tree.
+        """
+        observer_key = str(observer_uuid)
+        with cls._lock:
+            node = cls._nodes.get(observer_key)
+            workflow = node[_WORKFLOW] if node is not None else None
+
+            if workflow is not None:
+                visible = {
+                    key for key, candidate in cls._nodes.items() if candidate[_WORKFLOW] == workflow
+                }
+                if workflow in cls._nodes:
+                    visible.add(workflow)
+                return visible
+
+            return {key for key, candidate in cls._nodes.items() if candidate[_WORKFLOW] is None}
+
+    @classmethod
+    def is_visible(cls, observer_uuid: str, target_uuid: str) -> bool:
+        """Return whether *observer_uuid* may see *target_uuid*."""
+        return str(target_uuid) in cls.get_visible_uuids(observer_uuid)
+
+    # ------------------------------------------------------------------ #
+    # Internal helpers
+    # ------------------------------------------------------------------ #
+
+    @classmethod
+    def _ensure_node(cls, agent_key: str) -> Dict[str, Any]:
+        node = cls._nodes.get(agent_key)
+        if node is None:
+            node = {_PARENT: None, _CHILDREN: set(), _NAME: None, _WORKFLOW: None}
+            cls._nodes[agent_key] = node
+        return node
+
+    @classmethod
+    def _detach_locked(cls, agent_key: str) -> None:
+        node = cls._nodes.get(agent_key)
+        if node is None:
+            return
+
+        parent_key = node[_PARENT]
+        if not parent_key:
+            return
+
+        parent = cls._nodes.get(parent_key)
+        if parent is not None:
+            parent[_CHILDREN].discard(agent_key)
+        node[_PARENT] = None
+
+    @classmethod
+    def _set_workflow_locked(cls, agent_key: str, workflow: Optional[str]) -> None:
+        node = cls._nodes.get(agent_key)
+        if node is None:
+            return
+
+        if cls._is_workflow_name(node[_NAME]):
+            workflow = agent_key
+
+        node[_WORKFLOW] = workflow
+        for child in list(node[_CHILDREN]):
+            cls._set_workflow_locked(child, workflow)
+
+    @classmethod
+    def _descendants_locked(cls, agent_key: str, seen: Set[str]) -> List[str]:
+        node = cls._nodes.get(agent_key)
+        if node is None:
+            return []
+
+        result = []
+        for child in list(node[_CHILDREN]):
+            if child in seen:
+                continue
+            seen.add(child)
+            result.append(child)
+            result.extend(cls._descendants_locked(child, seen))
+
+        return result
+
+    @staticmethod
+    def _is_workflow_name(name: object) -> bool:
+        return isinstance(name, str) and name.startswith("wf:")
diff --git a/cecli/helpers/orchestration/agent_proxy.py b/cecli/helpers/orchestration/agent_proxy.py
index c753d075d43..a016d8d4ff9 100644
--- a/cecli/helpers/orchestration/agent_proxy.py
+++ b/cecli/helpers/orchestration/agent_proxy.py
@@ -566,7 +566,7 @@ class _HelpfulBuiltins(dict):
     """Custom __builtins__ dict that provides helpful hints for missing functions."""
 
     _HINTS: dict[str, str] = {
-        "open": "Filesystem access is not available. Use the Command tool instead.",
+        "open": "Filesystem access is not available. Use the Shell tool instead.",
         "eval": "eval() is disabled for security.",
         "exec": "exec() is disabled for security.",
         "__import__": "Imports are disabled. Use only the primitives provided.",
diff --git a/cecli/helpers/orchestration/safe_methods.py b/cecli/helpers/orchestration/safe_methods.py
index ef54efb96bb..3755999baff 100644
--- a/cecli/helpers/orchestration/safe_methods.py
+++ b/cecli/helpers/orchestration/safe_methods.py
@@ -50,7 +50,7 @@ def __getattr__(self, name: str):
 
             raise SecurityError(
                 f"Cannot call '{name}' on Path objects in the sandbox. "
-                f"Filesystem I/O is restricted to the Command tool."
+                f"Filesystem I/O is restricted to the Shell tool."
             )
         import pathlib as _pathlib
 
@@ -1045,7 +1045,7 @@ class _HelpfulBuiltins(dict):
     """Custom __builtins__ dict that provides helpful hints for missing functions."""
 
     _HINTS: dict[str, str] = {
-        "open": "Filesystem access is not available. Use the Command tool instead.",
+        "open": "Filesystem access is not available. Use the Shell tool instead.",
         "eval": "eval() is disabled for security.",
         "exec": "exec() is disabled for security.",
         "__import__": "Imports are disabled. Use only the primitives provided.",
diff --git a/cecli/helpers/server/acp/agent/updates.py b/cecli/helpers/server/acp/agent/updates.py
index 5e9cd35a894..7db2ad69c04 100644
--- a/cecli/helpers/server/acp/agent/updates.py
+++ b/cecli/helpers/server/acp/agent/updates.py
@@ -40,11 +40,12 @@
     "explorecode": "search",
     "searchfacts": "search",
     "editfile": "edit",
+    "manageextensions": "edit",
     "undochange": "edit",
     "replacefacts": "edit",
     # Not "execute": Zed treats execute calls as live terminals and hides its
     # progress spinner while one runs, but cecli runs commands itself
-    "command": "other",
+    "shell": "other",
 }
 
 # Argument keys worth showing in a tool call's title, in priority order
diff --git a/cecli/prompts/agent.yml b/cecli/prompts/agent.yml
index ec3a44ffaa6..900bed62f2e 100644
--- a/cecli/prompts/agent.yml
+++ b/cecli/prompts/agent.yml
@@ -57,7 +57,7 @@ main_system: |
   - **Scope**: No unrequested refactors. Avoid full-file rewrites. Only modify what you are asked to.
   - **Discover**: Use `ExploreCode` for symbols, `Grep` for text, and `ReadFile` to inspect matches.
   - **Plan**: Outline multi-step work with `UpdateTodoList`.
-  - **Execute**: Make read-only files editable with `ResourceManager` before editing, then review the diff and fix lint errors immediately.
+  - **Execute**: Make read-only files editable with `ManageResources` before editing, then review the diff and fix lint errors immediately.
   - **Recover**: Repair a bad edit in place; use `UndoChange` only when the corruption is beyond repair.
   - **Outputs**: Tool calls trigger turns. Yield when finished with the final user summary.
   - **Sandbox**: Perform all verification and temp logic in `.cecli/temp`.
diff --git a/cecli/prompts/subagent.yml b/cecli/prompts/subagent.yml
index 0348af624a3..b43afbf8fea 100644
--- a/cecli/prompts/subagent.yml
+++ b/cecli/prompts/subagent.yml
@@ -42,7 +42,7 @@ main_system: |
   - **Scope**: No unrequested refactors. Avoid full-file rewrites. Only modify what you are asked to.
   - **Discover**: Use `ExploreCode` for symbols, `Grep` for text, and `ReadFile` to inspect matches.
   - **Plan**: Outline multi-step work with `UpdateTodoList`.
-  - **Execute**: Make read-only files editable with `ResourceManager` before editing, then review the diff and fix lint errors immediately.
+  - **Execute**: Make read-only files editable with `ManageResources` before editing, then review the diff and fix lint errors immediately.
   - **Recover**: Repair a bad edit in place; use `UndoChange` only when the corruption is beyond repair.
   - **Outputs**: Tool calls trigger turns. Yield when finished with the final user summary.
   - **Sandbox**: Perform all verification and temp logic in `.cecli/temp`.
diff --git a/cecli/tools/__init__.py b/cecli/tools/__init__.py
index aabb5036024..10299be8aac 100644
--- a/cecli/tools/__init__.py
+++ b/cecli/tools/__init__.py
@@ -4,8 +4,6 @@
 from . import (
     _yield,
     broadcast,
-    command,
-    create,
     delegate,
     edit_file,
     explore_code,
@@ -16,20 +14,21 @@
     git_show,
     git_status,
     grep,
+    manage_extensions,
+    manage_resources,
     orchestrate,
     read_file,
     replace_facts,
-    resource_manager,
     search_docs,
     search_facts,
+    shell,
     undo_change,
     update_todo_list,
 )
 
 # List of all available tool modules for dynamic discovery
 TOOL_MODULES = [
-    command,
-    create,
+    shell,
     delegate,
     edit_file,
     explore_code,
@@ -42,10 +41,11 @@
     git_show,
     git_status,
     grep,
+    manage_extensions,
+    manage_resources,
     orchestrate,
     read_file,
     replace_facts,
-    resource_manager,
     search_docs,
     search_facts,
     undo_change,
diff --git a/cecli/tools/broadcast.py b/cecli/tools/broadcast.py
index e886af5a299..fdc513cdbac 100644
--- a/cecli/tools/broadcast.py
+++ b/cecli/tools/broadcast.py
@@ -142,18 +142,22 @@ def format_output(cls, coder, mcp_server, tool_response):
     def _collect_targets(cls, service, targets, originator_uuid):
         """Return ``(target_infos, errors)`` for the broadcast.
 
-        When ``targets`` is empty, returns every active sub-agent except the
-        originator. When ``targets`` is provided, each entry is resolved by
-        UUID, UUID prefix, or agent name (including ``primary`` for the primary
-        agent); unknown IDs produce an error.
+        When ``targets`` is empty, returns every active sub-agent visible to
+        the originator. When ``targets`` is provided, each entry is resolved by
+        UUID, UUID prefix, or agent name (including ``primary`` when visible);
+        unknown or invisible IDs produce an error.
         """
         from cecli.helpers.agents.service import SubAgentInfo, SubAgentStatus
+        from cecli.helpers.agents.tree import AgentTree
+
+        visible_uuids = AgentTree.get_visible_uuids(originator_uuid)
 
         if not targets:
             infos = [
                 info
                 for info in service.sub_agents.values()
                 if str(info.coder.uuid) != originator_uuid
+                and str(info.coder.uuid) in visible_uuids
                 and info.status not in (SubAgentStatus.ERROR,)
             ]
             return infos, []
@@ -161,7 +165,7 @@ def _collect_targets(cls, service, targets, originator_uuid):
         infos = []
         errors = []
         for target in targets:
-            info = cls._resolve_target(service, target)
+            info = cls._resolve_target(service, target, visible_uuids)
             if info is None:
                 errors.append(f"Unknown sub-agent target '{target}'.")
                 continue
@@ -172,31 +176,42 @@ def _collect_targets(cls, service, targets, originator_uuid):
         return infos, errors
 
     @staticmethod
-    def _resolve_target(service, target):
+    def _resolve_target(service, target, visible_uuids=None):
         """Resolve a target sub-agent by UUID, UUID prefix, or agent name.
 
         The primary agent is also resolvable by its name (``primary``) or UUID
-        so sub-agents can broadcast a message back to their parent.
+        so sub-agents can broadcast a message back to their parent. When
+        *visible_uuids* is provided, candidates outside that set are ignored.
         """
         target = str(target).strip()
         if not target:
             return None
 
+        def _visible(candidate):
+            if visible_uuids is None:
+                return True
+            return str(candidate.coder.uuid) in visible_uuids
+
         info = service.sub_agents.get(target)
-        if info is not None:
+        if info is not None and _visible(info):
             return info
 
         for uuid, candidate in service.sub_agents.items():
-            if uuid.startswith(target):
+            if uuid.startswith(target) and _visible(candidate):
                 return candidate
 
         for candidate in service.sub_agents.values():
-            if candidate.name == target:
+            if candidate.name == target and _visible(candidate):
                 return candidate
 
         primary_coder = service.coder
         primary_uuid = str(getattr(primary_coder, "uuid", ""))
-        if primary_coder is not None and (target == "primary" or target == primary_uuid):
+        primary_visible = visible_uuids is None or primary_uuid in visible_uuids
+        if (
+            primary_coder is not None
+            and primary_visible
+            and (target == "primary" or target == primary_uuid)
+        ):
             return primary_coder
 
         return None
diff --git a/cecli/tools/delegate.py b/cecli/tools/delegate.py
index cfe734f94d1..e5650f1b8f8 100644
--- a/cecli/tools/delegate.py
+++ b/cecli/tools/delegate.py
@@ -91,6 +91,14 @@ async def execute(cls, coder, **kwargs):
             if "prompt" not in d or not d["prompt"]:
                 response.append_error(f"delegations[{i}] is missing a 'prompt'.")
                 return response
+            # Workflow agents run in isolated trees and are launched by the
+            # workflow runner, never delegated to directly.
+            if str(d["name"]).startswith("wf:"):
+                response.append_error(
+                    f"delegations[{i}] targets a workflow agent; "
+                    "run workflows with the /workflows command instead."
+                )
+                return response
 
         from cecli.helpers.agents.service import AgentService
 
diff --git a/cecli/tools/create.py b/cecli/tools/manage_extensions.py
similarity index 96%
rename from cecli/tools/create.py
rename to cecli/tools/manage_extensions.py
index a5a7f745b7e..e330c205ba0 100644
--- a/cecli/tools/create.py
+++ b/cecli/tools/manage_extensions.py
@@ -1,4 +1,4 @@
-"""Create tool - persists cecli extensions into the session's creator.yml files."""
+"""ManageExtensions tool - persists cecli extensions into the session's creator.yml files."""
 
 import json
 import shutil
@@ -34,15 +34,16 @@
 
 
 class Tool(BaseTool):
-    NORM_NAME = "create"
+    NORM_NAME = "manageextensions"
     RESULT_TYPE = "list"
+    TRACK_INVOCATIONS = False
     SCHEMA = {
         "type": "function",
         "function": {
-            "name": "Create",
+            "name": "ManageExtensions",
             "description": (
-                "Persist a cecli extension or config change into creator.yml so it "
-                "survives sessions. One action per call: add/update entries of a given "
+                "Manage persisted cecli extensions and config in creator.yml so they "
+                "survive sessions. One action per call: add/update entries of a given "
                 "type into the global (~/.cecli/creator.yml) or local "
                 "(<root>/.cecli/creator.yml) file, remove entries, or list current "
                 "contents. Scoped to the creator sub-agent."
@@ -149,7 +150,7 @@ async def execute(
         """
         if not is_creator_coder(coder):
             raise ToolError(
-                "The Create tool is scoped to the creator sub-agent. Delegate the "
+                "The ManageExtensions tool is scoped to the creator sub-agent. Delegate the "
                 "request to the creator agent instead."
             )
 
@@ -224,7 +225,7 @@ async def execute(
 
     @classmethod
     def format_output(cls, coder, mcp_server, tool_response):
-        """Format output for Create tool - preview the requested change."""
+        """Format output for ManageExtensions tool - preview the requested change."""
         color_start, color_end = color_markers(coder)
 
         tool_header(coder=coder, mcp_server=mcp_server, tool_response=tool_response)
diff --git a/cecli/tools/resource_manager.py b/cecli/tools/manage_resources.py
similarity index 92%
rename from cecli/tools/resource_manager.py
rename to cecli/tools/manage_resources.py
index 84fbd68e2c6..e9e2ffbd8e6 100644
--- a/cecli/tools/resource_manager.py
+++ b/cecli/tools/manage_resources.py
@@ -8,7 +8,6 @@
     update_server_registration,
 )
 from cecli.decoding import safe_open
-from cecli.helpers.background_commands import BackgroundCommandManager
 from cecli.tools.utils.base_tool import BaseTool
 from cecli.tools.utils.helpers import ToolError, parse_arg_as_list
 from cecli.tools.utils.output import color_markers, tool_footer, tool_header
@@ -17,17 +16,17 @@
 
 
 class Tool(BaseTool):
-    NORM_NAME = "resourcemanager"
+    NORM_NAME = "manageresources"
     RESULT_TYPE = "list"
     SCHEMA = {
         "type": "function",
         "function": {
-            "name": "ResourceManager",
+            "name": "ManageResources",
             "description": (
-                "Manage files, long running commands, skills, and MCP servers"
-                " in the chat context: add, read_only, create, remove files;"
+                "Manage files, skills, and MCP servers in the chat context:"
+                " add, read_only, create, remove files;"
                 " view up to 3 pages of command output with paging;"
-                " stop background commands; load/remove skills and load/remove MCP servers."
+                " load/remove skills and load/remove MCP servers."
             ),
             "parameters": {
                 "type": "object",
@@ -52,11 +51,6 @@ class Tool(BaseTool):
                         "items": {"type": "string"},
                         "description": "List of file paths to remove from context.",
                     },
-                    "stop": {
-                        "type": "array",
-                        "items": {"type": "string"},
-                        "description": "List of command keys to stop background commands for.",
-                    },
                     "load_skill": {
                         "type": "array",
                         "items": {"type": "string"},
@@ -102,7 +96,7 @@ class Tool(BaseTool):
                                 "target": {
                                     "type": "string",
                                     "description": (
-                                        "Command key returned by Command (e.g., bg_1_1234), "
+                                        "Command key returned by Shell (e.g., bg_1_1234), "
                                         "matching ^bg_[0-9]+_[0-9]+$."
                                     ),
                                 },
@@ -127,7 +121,6 @@ async def execute(
         add=None,
         read_only=None,
         create=None,
-        stop=None,
         load_skill=None,
         remove_skill=None,
         load_mcp=None,
@@ -150,8 +143,6 @@ async def execute(
             Files to add as read-only (not command output aliases).
         create: list[str] | None
             Files to create and make editable.
-        stop: list[str] | None
-            Command keys to stop background commands for.
         load_skill: list[str] | None
             Skill names to load.
         remove_skill: list[str] | None
@@ -170,7 +161,6 @@ async def execute(
         editable_files = sorted(parse_arg_as_list(add), key=cls._natural_sort_key)
         view_files = sorted(parse_arg_as_list(read_only), key=cls._natural_sort_key)
         create_files = sorted(parse_arg_as_list(create), key=cls._natural_sort_key)
-        stop_keys = sorted(parse_arg_as_list(stop), key=cls._natural_sort_key)
         load_skill_names = sorted(parse_arg_as_list(load_skill), key=cls._natural_sort_key)
         remove_skill_names = sorted(parse_arg_as_list(remove_skill), key=cls._natural_sort_key)
         load_mcp_servers = sorted(parse_arg_as_list(load_mcp), key=cls._natural_sort_key)
@@ -192,7 +182,6 @@ async def execute(
             and not editable_files
             and not view_files
             and not create_files
-            and not stop_keys
             and not load_skill_names
             and not remove_skill_names
             and not load_mcp_servers
@@ -201,7 +190,7 @@ async def execute(
             and paging is None
         ):
             raise ToolError(
-                "You must specify at least one of: remove, add, read_only, create, stop, "
+                "You must specify at least one of: remove, add, read_only, create, "
                 "load_skill, remove_skill, load_mcp, remove_mcp, actions, or paging"
             )
 
@@ -270,8 +259,6 @@ async def execute(
                 response.append_result(cls._create(coder, f))
             else:
                 response.append_result(cls._editable(coder, f))
-        for key in stop_keys:
-            response.append_result(cls._stop_command(coder, key))
         for skill_name in load_skill_names:
             response.append_result(cls._load_skill(coder, skill_name))
         for skill_name in remove_skill_names:
@@ -298,7 +285,7 @@ async def execute(
 
     @classmethod
     def format_output(cls, coder, mcp_server, tool_response):
-        """Format output for ResourceManager tool."""
+        """Format output for ManageResources tool."""
         color_start, color_end = color_markers(coder)
 
         # Output header
@@ -318,7 +305,6 @@ def format_output(cls, coder, mcp_server, tool_response):
             "remove": "remove",
             "view": "view",
             "editable": "editable",
-            "stop": "stop",
             "load_skill": "load_skill",
             "remove_skill": "remove_skill",
             "load_mcp": "load_mcp",
@@ -375,32 +361,6 @@ def _remove(cls, coder, file_path):
             coder.io.tool_error(f"Error removing file '{file_path}': {str(e)}")
             return f"Error removing {file_path}: {e}"
 
-    @classmethod
-    def _stop_command(cls, coder, command_key):
-        """Stop a background command by its command key."""
-        try:
-            success, output, exit_code = BackgroundCommandManager.stop_background_command(
-                command_key
-            )
-            if success:
-                coder.io.tool_output(
-                    f"✗ Stopped background command '{command_key}'", type="tool-result"
-                )
-                return (
-                    f"Background command stopped: {command_key}\n"
-                    f"Exit code: {exit_code}\n"
-                    f"Final output:\n{output}"
-                )
-            else:
-                coder.io.tool_output(
-                    f"⚠ Background command '{command_key}' not found or not running",
-                    type="tool-result",
-                )
-                return f"Command not found or not running: {command_key}"
-        except Exception as e:
-            coder.io.tool_error(f"Error stopping command '{command_key}': {str(e)}")
-            return f"Error stopping {command_key}: {e}"
-
     @classmethod
     def _editable(cls, coder, file_path):
         """Make a file editable in the coder's context."""
diff --git a/cecli/tools/read_file.py b/cecli/tools/read_file.py
index e7c67ebcbfe..f65aba80c12 100644
--- a/cecli/tools/read_file.py
+++ b/cecli/tools/read_file.py
@@ -250,7 +250,7 @@ def execute(cls, coder, read, **kwargs):
                             "outline": "",
                             "note": (
                                 f"File {rel_path} is empty. Next: use EditFile with start_line @000 and"
-                                " end_line @000 to write content, or ResourceManager to scaffold —"
+                                " end_line @000 to write content, or ManageResources to scaffold —"
                                 " do not call ReadFile again on this empty file."
                             ),
                         }
diff --git a/cecli/tools/command.py b/cecli/tools/shell.py
similarity index 99%
rename from cecli/tools/command.py
rename to cecli/tools/shell.py
index 1eea71819c4..fca3f48a54e 100644
--- a/cecli/tools/command.py
+++ b/cecli/tools/shell.py
@@ -59,13 +59,13 @@
 
 
 class Tool(BaseTool):
-    NORM_NAME = "command"
+    NORM_NAME = "shell"
     TRACK_INVOCATIONS = False
     ALLOWED_SESSION_COMMANDS = {}
     SCHEMA = {
         "type": "function",
         "function": {
-            "name": "Command",
+            "name": "Shell",
             "description": "Execute a shell command or interact with background processes.",
             "parameters": {
                 "type": "object",
@@ -177,7 +177,7 @@ async def execute(
         Commands run with timeout from agent_config['command_timeout'] (default: 30 seconds),
         """
         response = ToolResponse(
-            "command",
+            "shell",
             result_type=cls.RESULT_TYPE if hasattr(cls, "RESULT_TYPE") else "str",
         )
         # Handle interactions with an existing background command
@@ -295,7 +295,7 @@ async def _get_confirmation(cls, coder, command_string, background):
             prompt,
             explicit_yes_required=True,
             allow_never=True,
-            group_response="Command Tool",
+            group_response="Shell Tool",
         )
 
         if confirmed:
@@ -597,7 +597,7 @@ async def _execute_foreground(cls, coder, command_string):
                 f"Output saved in {len(file_list)} pages.]\n"
                 f"Command key: {fg_key}\n"
                 f"Pages: 1-{len(file_list)}\n"
-                "Use `ResourceManager` to view up to 3 pages at a time:\n"
+                "Use `ManageResources` to view up to 3 pages at a time:\n"
                 f'{{"paging": [{{"target": "{fg_key}", "page": 1}}]}}\n'
                 "Change page or add entries to read other pages (maximum 3 entries). "
                 "Do not use add, read_only, or standard CLI tools to view command output "
@@ -764,7 +764,7 @@ async def _tail_background_command(cls, coder, command_key):
             f"Output so far: {info.get('total_chars', 0):,} chars",
         ]
         if pages:
-            lines.append(f"Paged output: pages 1-{pages}. Read with ResourceManager paging.")
+            lines.append(f"Paged output: pages 1-{pages}. Read with ManageResources paging.")
             lines.append(f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}')
 
         if output.strip():
@@ -787,7 +787,7 @@ def _pages_notice(command_key, page_count):
             f"[Output paged to disk: {page_count} page(s).]\n"
             f"Command key: {command_key}\n"
             f"Pages: 1-{page_count}\n"
-            "Use `ResourceManager` to view up to 3 pages at a time:\n"
+            "Use `ManageResources` to view up to 3 pages at a time:\n"
             f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}\n'
             "Change the page number to read other pages (maximum 3 entries). "
             "Do not use add, read_only, or standard CLI tools to view command output "
@@ -796,7 +796,7 @@ def _pages_notice(command_key, page_count):
 
     @classmethod
     def format_output(cls, coder, mcp_server, tool_response):
-        """Format output for Command tool."""
+        """Format output for Shell tool."""
         color_start, color_end = color_markers(coder)
 
         # Output header
diff --git a/cecli/tools/utils/helpers.py b/cecli/tools/utils/helpers.py
index efca476cdfe..ba749024d17 100644
--- a/cecli/tools/utils/helpers.py
+++ b/cecli/tools/utils/helpers.py
@@ -92,12 +92,12 @@ def validate_file_for_edit(coder, file_path):
     if abs_path not in coder.abs_fnames:
         if abs_path in coder.abs_read_only_fnames:
             raise ToolError(
-                f"File '{file_path}' is read-only. Make editable with `ResourceManager` first."
+                f"File '{file_path}' is read-only. Make editable with `ManageResources` first."
             )
         # else:
         #    # File exists but is not in context at all
         #    raise ToolError(
-        #        f"File '{file_path}' not in context. Make editable with `ResourceManager` first."
+        #        f"File '{file_path}' not in context. Make editable with `ManageResources` first."
         #    )
 
     # Reread content immediately before potential modification
diff --git a/cecli/tui/app.py b/cecli/tui/app.py
index f685c9d46d0..d51505184cd 100644
--- a/cecli/tui/app.py
+++ b/cecli/tui/app.py
@@ -45,6 +45,8 @@
 if platform.system() == "Windows":
     IS_WINDOWS = True
 
+MODIFIER_ORDER = ("alt", "ctrl", "shift", "super")
+
 logger = logging.getLogger(__name__)
 
 
@@ -122,116 +124,39 @@ def __init__(self, coder_worker, output_queue, input_queue, args):
         if other.get("use_terminal_background", True):
             patch_textual_strip_render_with_cache()
 
-        self.bind(
-            self._get_binding_keys("newline"),
-            "noop",
-            description="New Line",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("submit")), "noop", description="Submit", show=True
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("cycle_forward")),
-            "noop",
-            description="Cycle Forward",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("cycle_backward")),
-            "noop",
-            description="Cycle Backward",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("prev_agent")),
-            "switch_prev_agent",
-            description="Previous Agent",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("next_agent")),
-            "switch_next_agent",
-            description="Next Agent",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("main_agent")),
-            "switch_to_primary",
-            description="Main Agent",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("panel_up")),
-            "scroll_panel_up",
-            description="Scroll Panels Up",
-            show=False,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("panel_down")),
-            "scroll_panel_down",
-            description="Scroll Panels Down",
-            show=False,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("cancel")),
-            "interrupt",
-            description="Cancel",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("editor")),
-            "open_editor",
-            description="Editor",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("history")),
-            "history_search",
-            description="History Search",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("focus")),
-            "focus_input",
-            description="Focus Input",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("stop")),
-            "interrupt",
-            description="Interrupt",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("clear")),
-            "clear_output",
-            description="Clear",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("quit")), "quit", description="Quit", show=True
-        )
-
-        self.bind(
-            self._encode_keys(self.get_keys_for("voice")),
-            "start_voice",
-            description="Record Voice",
-            show=True,
-        )
-        self.bind(
-            self._encode_keys(self.get_keys_for("paste")),
-            "paste_clipboard",
-            description="Paste Clipboard",
-            show=True,
+        app_bindings = (
+            # Input-context keys stay non-priority so InputArea can keep
+            # handling selection, completion, and submit behavior.
+            ("newline", "noop", "New Line", False, True),
+            ("submit", "noop", "Submit", False, True),
+            ("cycle_forward", "noop", "Cycle Forward", False, True),
+            ("cycle_backward", "noop", "Cycle Backward", False, True),
+            ("cancel", "interrupt", "Cancel", False, True),
+            ("stop", "interrupt", "Interrupt", False, True),
+            # These actions must work regardless of which widget has focus.
+            ("prev_agent", "switch_prev_agent", "Previous Agent", True, True),
+            ("next_agent", "switch_next_agent", "Next Agent", True, True),
+            ("main_agent", "switch_to_primary", "Main Agent", True, True),
+            ("panel_up", "scroll_panel_up", "Scroll Panels Up", True, False),
+            ("panel_down", "scroll_panel_down", "Scroll Panels Down", True, False),
+            ("editor", "open_editor", "Editor", True, True),
+            ("history", "history_search", "History Search", True, True),
+            ("focus", "focus_input", "Focus Input", True, True),
+            ("clear", "clear_output", "Clear", True, True),
+            ("quit", "quit", "Quit", True, True),
+            ("voice", "start_voice", "Record Voice", True, True),
+            ("paste", "paste_clipboard", "Paste Clipboard", True, True),
+            ("panels", "toggle_floating_panels", "Toggle Panels", True, False),
         )
 
-        self.bind(
-            self._encode_keys(self.get_keys_for("panels")),
-            "toggle_floating_panels",
-            description="Toggle Panels",
-            show=False,
-        )
+        for binding_type, action, description, priority, show in app_bindings:
+            self._bind_key(
+                binding_type,
+                action,
+                description=description,
+                priority=priority,
+                show=show,
+            )
 
         self.register_theme(BASE_THEME)
         self.theme = "cecli"
@@ -1774,10 +1699,28 @@ def _get_binding_keys(self, type):
 
         return keys if encoded == keys else f"{keys},{encoded}"
 
+    def _bind_key(self, type, action, description="", priority=True, show=True):
+        """Register a configured key binding on the app.
+
+        Args:
+            type: The key binding name in ``tui_config``.
+            action: The Textual action to invoke.
+            description: Short description for the binding.
+            priority: Check the binding before the focused widget.
+            show: Whether the binding is shown in the footer.
+        """
+        self._bindings.bind(
+            self._get_binding_keys(type),
+            action,
+            description=description,
+            show=show,
+            priority=priority,
+        )
+
     def _encode_keys(self, key):
         key = key.replace("shift+enter", "ctrl+j")
 
-        return key
+        return self._canonicalize_key_list(key)
 
     def _decode_keys(self, key):
         key = key.replace("ctrl+j", "shift+enter")
@@ -1786,10 +1729,10 @@ def _decode_keys(self, key):
 
     def is_key_for(self, type, key):
         allowed_keys = self.tui_config["key_bindings"][type].split(",")
-        normalized_key = self._decode_keys(key)
+        normalized_key = self._encode_keys(key)
 
         for allowed_key in allowed_keys:
-            if normalized_key == self._decode_keys(allowed_key):
+            if normalized_key == self._encode_keys(allowed_key):
                 return True
 
         return False
@@ -1798,6 +1741,23 @@ def get_keys_for(self, type):
         allowed_keys = self.tui_config["key_bindings"][type]
         return self._decode_keys(allowed_keys)
 
+    def _canonicalize_key_list(self, keys):
+        """Return comma-separated keys in Textual's canonical modifier order."""
+        return ",".join(self._canonicalize_key(key.strip()) for key in keys.split(","))
+
+    def _canonicalize_key(self, key):
+        """Return a key with modifiers ordered as Textual emits them."""
+        if "+" not in key:
+            return key
+
+        *modifiers, name = key.split("+")
+        if not modifiers or any(modifier not in MODIFIER_ORDER for modifier in modifiers):
+            return key
+
+        modifiers = sorted(set(modifiers), key=MODIFIER_ORDER.index)
+
+        return "+".join([*modifiers, name])
+
     def _do_quit(self):
         """Perform the actual quit after UI updates."""
         self.worker.stop()
diff --git a/cecli/tui/widgets/floating_panels.py b/cecli/tui/widgets/floating_panels.py
index fedad000077..0d11c37b556 100644
--- a/cecli/tui/widgets/floating_panels.py
+++ b/cecli/tui/widgets/floating_panels.py
@@ -335,8 +335,18 @@ def _collect_agents(self):
             from cecli.helpers.agents.service import AgentService
 
             service = AgentService.get_instance(primary)
-            registered = set(AgentService.get_registry().keys())
-            live = {info.name for info in service.sub_agents.values() if info is not None}
+            # Workflow agents run in isolated trees and are not spawnable from
+            # this panel.
+            registered = {
+                name
+                for name in AgentService.get_registry().keys()
+                if not str(name).startswith("wf:")
+            }
+            live = {
+                info.name
+                for info in service.sub_agents.values()
+                if info is not None and not str(info.name).startswith("wf:")
+            }
             active_name = None
             foreground_uuid = getattr(service, "foreground_uuid", None)
             if foreground_uuid:
diff --git a/cecli/tui/widgets/input_area.py b/cecli/tui/widgets/input_area.py
index 52b92c33266..198dd70dd84 100644
--- a/cecli/tui/widgets/input_area.py
+++ b/cecli/tui/widgets/input_area.py
@@ -239,36 +239,6 @@ def on_key(self, event) -> None:
         if self.disabled:
             return
 
-        # Route hotkey bindings directly and stop TextArea from inserting a
-        # printable character (e.g. alt+shift+h would otherwise type an "h").
-        if self.app.is_key_for("voice", event.key):
-            event.stop()
-            event.prevent_default()
-            self.app.action_start_voice()
-            return
-
-        if self.app.is_key_for("paste", event.key):
-            event.stop()
-            event.prevent_default()
-            self.app.action_paste_clipboard()
-
-        if self.app.is_key_for("panels", event.key):
-            event.stop()
-            event.prevent_default()
-            self.app.action_toggle_floating_panels()
-            return
-
-        if self.app.is_key_for("history", event.key):
-            event.stop()
-            event.prevent_default()
-            self.app.action_history_search()
-            return
-
-        # Reserve any other alt+shift combination for hotkeys (no character input).
-        if event.key and "alt+shift" in event.key:
-            event.stop()
-            event.prevent_default()
-            return
         # Reset cycling if not a cycle command
         is_cycle = self.app.is_key_for("cycle_forward", event.key) or self.app.is_key_for(
             "cycle_backward", event.key
diff --git a/cecli/website/docs/config/agent-mode.md b/cecli/website/docs/config/agent-mode.md
index cf490bb1602..26ea45c5b26 100644
--- a/cecli/website/docs/config/agent-mode.md
+++ b/cecli/website/docs/config/agent-mode.md
@@ -47,8 +47,8 @@ This loop continues automatically until the `Yield` tool is called, or the maxim
 Agent Mode uses a centralized local tool registry. The standard tools include:
 
 - **Discovery and inspection**: `ExploreCode`, `Grep`, and `ReadFile`
-- **Context and editing**: `ResourceManager`, `EditFile`, `UndoChange`, and `UpdateTodoList`
-- **Execution and orchestration**: `Command` and `Orchestrate`
+- **Context and editing**: `ManageResources`, `EditFile`, `UndoChange`, and `UpdateTodoList`
+- **Execution and orchestration**: `Shell` and `Orchestrate`
 - **Sub-agent coordination**: `Delegate` and `Yield`
 - **Optional memory tools**: `SearchFacts` and `ReplaceFacts`
 
@@ -93,7 +93,7 @@ Tool Call: Grep
 Arguments: {"searches": [{"pattern": "function_name", "directory": "."}]}
 ```
 
-`ExploreCode` uses a codebase index when available. `ReadFile` returns content with virtual identifiers for precise follow-up edits. Files discovered through agent tools can be added as read-only context by using `ResourceManager` to change context membership.
+`ExploreCode` uses a codebase index when available. `ReadFile` returns content with virtual identifiers for precise follow-up edits. Files discovered through agent tools can be added as read-only context by using `ManageResources` to change context membership.
 
 #### 2. Planning Phase
 
@@ -106,17 +106,17 @@ Arguments: {"content": "## Task: Add new feature\n- [ ] Analyze existing code\n-
 
 #### 3. Execution Phase
 
-Files are added to context and made editable with `ResourceManager`, then modifications are applied with `EditFile`:
+Files are added to context and made editable with `ManageResources`, then modifications are applied with `EditFile`:
 
 ```text
-Tool Call: ResourceManager
+Tool Call: ManageResources
 Arguments: {"add": ["main.py"]}
 
 Tool Call: EditFile
 Arguments: {"edits": [{"file_path": "main.py", "operation": "replace", "start_line": "——def old_function():", "end_line": "——    return old_value", "text": "def new_function():\n    return new_value"}]}
 ```
 
-`EditFile` accepts an `edits` array containing `replace`  or `delete` operations. Its line targets come from the latest `ReadFile` result. `ResourceManager` can add files as editable or read-only and remove them from context.
+`EditFile` accepts an `edits` array containing `replace`  or `delete` operations. Its line targets come from the latest `ReadFile` result. `ManageResources` can add files as editable or read-only and remove them from context.
 
 #### 4. Verification Phase
 
@@ -221,9 +221,9 @@ agent-config:
 
 #### Essential Tools
 
-Only `Yield` is protected as an essential registry tool. Other tools, including `ResourceManager` and `ReadFile`, can be restricted by includelist/excludelist settings.
+Only `Yield` is protected as an essential registry tool. Other tools, including `ManageResources` and `ReadFile`, can be restricted by includelist/excludelist settings.
 
-- `ResourceManager` - Add, drop, and make files editable in context
+- `ManageResources` - Add, drop, and make files editable in context
 - `ReadFile` - Read file contents with virtual content IDs
 - `Yield` - Complete the task and wait for active child agents
 
@@ -333,8 +333,8 @@ agent: true
 # Agent Mode configuration
 agent-config:
   # Tool configuration
-  tools_includelist: ["resourcemanager", "readfile", "yield"]
-  tools_excludelist: ["command"]
+  tools_includelist: ["manageresources", "readfile", "yield"]
+  tools_excludelist: ["shell"]
   tools_paths: ["./custom-tools", "~/my-tools"]
 
   # Server configuration
diff --git a/cecli/website/docs/config/creator.md b/cecli/website/docs/config/creator.md
index 240a98670e6..87a3e0671d5 100644
--- a/cecli/website/docs/config/creator.md
+++ b/cecli/website/docs/config/creator.md
@@ -106,18 +106,18 @@ Backing files are left on disk by default so you can inspect them; say "delete t
 
 ## Safety Notes
 
-- The creator never hand-edits `.cecli.conf.yml`, `conf.yml`, or `creator.yml`. All writes go through its Create tool, which merges rather than replaces existing content.
+- The creator never hand-edits `.cecli.conf.yml`, `conf.yml`, or `creator.yml`. All writes go through its ManageExtensions tool, which merges rather than replaces existing content.
 - It inspects existing files before changing them, but it is designed to run unattended, so review its summary and use `list` when you want to confirm what was persisted.
 - Removing an entry prunes references but keeps backing files unless you ask to delete them.
 - Settings that only apply at startup (for example some model/provider options) may need `/hot-reload` or a restart; the creator tells you when that is the case.
 
 ## Advanced Reference
 
-Everything below describes the underlying Create tool. You normally do not need it — ask the creator instead.
+Everything below describes the underlying ManageExtensions tool. You normally do not need it — ask the creator instead.
 
-### Create tool
+### ManageExtensions tool
 
-The Create tool is scoped to the creator sub-agent (`creator_mode: true`). The primary agent and other sub-agents cannot call it directly; delegate the request to the creator. One action runs per call.
+The ManageExtensions tool is scoped to the creator sub-agent (`creator_mode: true`). The primary agent and other sub-agents cannot call it directly; delegate the request to the creator. One action runs per call.
 
 | Action | Behavior |
 |--------|----------|
@@ -181,7 +181,7 @@ See [Hooks](hooks.html) for the hook payload schema.
 
 #### `skill`
 
-Writes `<scope>/skills/<name>/SKILL.md` from `content`. If `content` is omitted and the file does not exist, Create scaffolds a minimal `SKILL.md` with `name` and `description` front matter for the author to fill in. The skill directory is recorded in `agent-config.skills_paths`; when a live coder already runs with a non-empty `skills_includelist`, the new name is added there as well.
+Writes `<scope>/skills/<name>/SKILL.md` from `content`. If `content` is omitted and the file does not exist, ManageExtensions scaffolds a minimal `SKILL.md` with `name` and `description` front matter for the author to fill in. The skill directory is recorded in `agent-config.skills_paths`; when a live coder already runs with a non-empty `skills_includelist`, the new name is added there as well.
 
 ```yaml
 agent-config:
@@ -195,7 +195,7 @@ After an `add`, the creator activates the skill for the targeted coders (or ever
 
 #### `custom_tool`
 
-Writes `<scope>/tools/<name>.py` from `content` and records the scope's tools directory in `agent-config.tools_paths`. `~/.cecli/tools` and `<root>/.cecli/tools` are scanned by default, but recording the path keeps the directory explicit. If a live coder's `tools_excludelist` names the tool, Create notes that the tool stays hidden there.
+Writes `<scope>/tools/<name>.py` from `content` and records the scope's tools directory in `agent-config.tools_paths`. `~/.cecli/tools` and `<root>/.cecli/tools` are scanned by default, but recording the path keeps the directory explicit. If a live coder's `tools_excludelist` names the tool, ManageExtensions notes that the tool stays hidden there.
 
 ```yaml
 agent-config:
@@ -215,7 +215,7 @@ custom:
 
 #### `workflow`
 
-Writes `<scope>/workflows/<name>.yml` from `content` or from a dict `config` (a `name` is added when missing). Workflows are discovered by scanning the workflows directories, so Create records **no fragment** in `creator.yml`. If the workflow has `enabled: false`, Create warns that no `wf:<name>` sub-agent will register.
+Writes `<scope>/workflows/<name>.yml` from `content` or from a dict `config` (a `name` is added when missing). Workflows are discovered by scanning the workflows directories, so ManageExtensions records **no fragment** in `creator.yml`. If the workflow has `enabled: false`, ManageExtensions warns that no `wf:<name>` sub-agent will register.
 
 > **Important:** Workflows are never recorded as a top-level list of dicts in `creator.yml`. A `workflows:` list in that file would bypass the normal workflow file discovery and precedence rules.
 
@@ -251,7 +251,7 @@ agents:
         - github
 ```
 
-`agents:` is read **only** from `creator.yml` (global then project-local); entries in `.cecli.conf.yml` or `.cecli/conf.yml` are ignored. Create refreshes the index after every successful change, so sub-agents spawned later in the same session pick up the override without a restart.
+`agents:` is read **only** from `creator.yml` (global then project-local); entries in `.cecli.conf.yml` or `.cecli/conf.yml` are ignored. ManageExtensions refreshes the index after every successful change, so sub-agents spawned later in the same session pick up the override without a restart.
 
 **Names and patterns.** Targets are exact registry names or `fnmatch` patterns (`"*"`, `"wf:*"`, `"ws:*"`). When a sub-agent spawns, matching overrides merge most-specific-last:
 
diff --git a/cecli/website/docs/config/custom-system-prompts.md b/cecli/website/docs/config/custom-system-prompts.md
index d94f00dfdf2..76dd3fcffec 100644
--- a/cecli/website/docs/config/custom-system-prompts.md
+++ b/cecli/website/docs/config/custom-system-prompts.md
@@ -83,7 +83,7 @@ main_system: |
   <context name="role_and_directives">
   ## Core Directives
   - **Role**: Act as an expert software engineer.
-  - **Act Proactively**: Autonomously use file discovery and context management tools (`ExploreCode`, `Grep`, `ReadFile`, `ResourceManager`) to gather information and fulfill the user's request. Chain tool calls across multiple turns to continue exploration.
+  - **Act Proactively**: Autonomously use file discovery and context management tools (`ExploreCode`, `Grep`, `ReadFile`, `ManageResources`) to gather information and fulfill the user's request. Chain tool calls across multiple turns to continue exploration.
   - **Be Decisive**: Trust that your initial findings are valid. Refrain from asking the same question or searching for the same term in multiple similar ways.
   - **Be Concise**: Keep all responses brief and direct (1-3 sentences). Avoid preamble, postamble, and unnecessary explanations. Do not repeat yourself.
   - **Be Careful**: Break updates down into smaller, more manageable chunks. Focus on one thing at a time.
diff --git a/cecli/website/docs/config/security.md b/cecli/website/docs/config/security.md
index 53c148cf7de..7c51c6f4e0a 100644
--- a/cecli/website/docs/config/security.md
+++ b/cecli/website/docs/config/security.md
@@ -167,7 +167,7 @@ When using Agent Mode, security configuration works alongside other security fea
 ```yaml
 agent: true
 agent-config:
-  tools_excludelist: ["command"]
+  tools_excludelist: ["shell"]
 security-config:
   allowed-domains:
     - github.com
diff --git a/tests/basic/test_background_commands.py b/tests/basic/test_background_commands.py
index 1f7f079a274..cdf9abb7e82 100644
--- a/tests/basic/test_background_commands.py
+++ b/tests/basic/test_background_commands.py
@@ -294,7 +294,7 @@ def test_tail_background_command_reports_output_and_pages(monkeypatch):
     """The tail action reports status, page roster, and new output."""
     import asyncio
 
-    from cecli.tools.command import Tool as CommandTool
+    from cecli.tools.shell import Tool as ShellTool
 
     monkeypatch.setattr(
         BackgroundCommandManager,
@@ -312,7 +312,7 @@ def test_tail_background_command_reports_output_and_pages(monkeypatch):
         BackgroundCommandManager, "get_new_command_output", lambda key: "new line\n"
     )
 
-    response = asyncio.run(CommandTool._tail_background_command(object(), "bg_1_1234"))
+    response = asyncio.run(ShellTool._tail_background_command(object(), "bg_1_1234"))
     content = response.to_dict()["result"][0]["content"]
 
     assert "bg_1_1234" in content
@@ -325,11 +325,11 @@ def test_tail_background_command_reports_output_and_pages(monkeypatch):
 def test_tail_background_command_missing_key(monkeypatch):
     import asyncio
 
-    from cecli.tools.command import Tool as CommandTool
+    from cecli.tools.shell import Tool as ShellTool
 
     monkeypatch.setattr(BackgroundCommandManager, "list_background_commands", lambda: {})
 
-    response = asyncio.run(CommandTool._tail_background_command(object(), "bg_9_9999"))
+    response = asyncio.run(ShellTool._tail_background_command(object(), "bg_9_9999"))
 
     assert response.to_dict()["errors"]
 
diff --git a/tests/basic/test_create_apply_to.py b/tests/basic/test_create_apply_to.py
index bdf53544514..633b34240df 100644
--- a/tests/basic/test_create_apply_to.py
+++ b/tests/basic/test_create_apply_to.py
@@ -1,4 +1,4 @@
-"""Tests for per-agent targeting (``Create --apply_to``)."""
+"""Tests for per-agent targeting (``ManageExtensions --apply_to``)."""
 
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
@@ -10,7 +10,7 @@
 from cecli.helpers.agents.config import SubAgentConfig
 from cecli.helpers.agents.service import AgentService
 from cecli.helpers.creator import utils as creator_utils
-from cecli.tools.create import Tool
+from cecli.tools.manage_extensions import Tool
 from cecli.tools.utils.helpers import ToolError
 
 
diff --git a/tests/basic/test_create_tool.py b/tests/basic/test_create_tool.py
index c70513ae9be..b260a133e19 100644
--- a/tests/basic/test_create_tool.py
+++ b/tests/basic/test_create_tool.py
@@ -1,4 +1,4 @@
-"""Tests for the creator-scoped Create tool."""
+"""Tests for the creator-scoped ManageExtensions tool."""
 
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
@@ -7,8 +7,8 @@
 import yaml
 
 from cecli.helpers.creator import utils as creator_utils
-from cecli.tools import create as create_module
-from cecli.tools.create import Tool
+from cecli.tools import manage_extensions
+from cecli.tools.manage_extensions import Tool
 from cecli.tools.utils.helpers import ToolError
 
 
@@ -124,7 +124,7 @@ async def test_add_skill_writes_file_and_path(coder, creator_paths, reloads):
 
 @pytest.mark.asyncio
 async def test_non_creator_coder_rejected(coder, creator_paths, reloads, monkeypatch):
-    monkeypatch.setattr(create_module, "is_creator_coder", lambda coder: False)
+    monkeypatch.setattr(manage_extensions, "is_creator_coder", lambda coder: False)
 
     with pytest.raises(ToolError):
         await Tool.execute(coder, action="add", type="config", name="x", config=1)
@@ -404,7 +404,7 @@ def registry(monkeypatch):
 
 @pytest.fixture(autouse=True)
 def stub_override_refresh(monkeypatch):
-    """Keep Create's post-write override refresh out of the real config files."""
+    """Keep ManageExtensions' post-write override refresh out of the real config files."""
     from cecli.helpers.agents import overrides
 
     monkeypatch.setattr(overrides, "refresh_agent_overrides", Mock(return_value={}))
diff --git a/tests/commands/test_agent_tree.py b/tests/commands/test_agent_tree.py
new file mode 100644
index 00000000000..ad29e029925
--- /dev/null
+++ b/tests/commands/test_agent_tree.py
@@ -0,0 +1,83 @@
+"""Tests for the /agent-tree command."""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cecli.commands.agent_tree import AgentTreeCommand
+from cecli.helpers.agents.service import AgentService, SubAgentStatus
+from cecli.helpers.agents.tree import AgentTree
+
+
+@pytest.fixture(autouse=True)
+def clean_tree():
+    """Start and end every test with an empty shared tree."""
+    AgentTree.clear()
+    yield
+    AgentTree.clear()
+
+
+def _info(name, coder, parent_uuid):
+    return SimpleNamespace(
+        name=name,
+        coder=coder,
+        parent_uuid=parent_uuid,
+        status=SubAgentStatus.RUNNING,
+        error=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_agent_tree_reads_shared_tree_and_scopes_to_primary_subtree():
+    """The command renders the primary's AgentTree subtree and nothing else."""
+    primary = SimpleNamespace(uuid="primary", parent_uuid="")
+    worker = SimpleNamespace(uuid="worker", parent_uuid="primary")
+    other_worker = SimpleNamespace(uuid="other-worker", parent_uuid="other-primary")
+
+    service = SimpleNamespace(
+        coder=primary,
+        sub_agents={
+            "worker": _info("worker", worker, "primary"),
+            "other-worker": _info("other-worker", other_worker, "other-primary"),
+        },
+    )
+
+    AgentTree.register("primary", name="primary")
+    AgentTree.register("worker", parent_uuid="primary", name="worker")
+    AgentTree.register("other-primary", name="primary")
+    AgentTree.register("other-worker", parent_uuid="other-primary", name="worker")
+
+    io = MagicMock()
+    with patch.object(AgentService, "get_instance", return_value=service):
+        await AgentTreeCommand.execute(io, primary, "")
+
+    output = "\n".join(call.args[0] for call in io.tool_output.call_args_list)
+
+    assert "primary" in output
+    assert "worker" in output
+    assert "other-worker" not in output
+
+
+@pytest.mark.asyncio
+async def test_agent_tree_uses_tree_structure_not_coder_parent_uuid():
+    """The rendered hierarchy follows AgentTree parent links."""
+    primary = SimpleNamespace(uuid="primary", parent_uuid="")
+    worker = SimpleNamespace(uuid="worker", parent_uuid="")
+
+    service = SimpleNamespace(
+        coder=primary,
+        sub_agents={"worker": _info("worker", worker, "")},
+    )
+
+    AgentTree.register("primary", name="primary")
+    AgentTree.register("worker", parent_uuid="primary", name="worker")
+
+    io = MagicMock()
+    with patch.object(AgentService, "get_instance", return_value=service):
+        await AgentTreeCommand.execute(io, primary, "")
+
+    output = "\n".join(call.args[0] for call in io.tool_output.call_args_list)
+
+    assert "└── primary" in output
+    assert "└── worker" in output
diff --git a/tests/integration/test_agent_mcp_management.py b/tests/integration/test_agent_mcp_management.py
index ff5ef3b7a94..30ee35c2d80 100644
--- a/tests/integration/test_agent_mcp_management.py
+++ b/tests/integration/test_agent_mcp_management.py
@@ -7,7 +7,7 @@
 
 from cecli.coders.agent_coder import AgentCoder
 from cecli.coders.sub_agent_coder import SubAgentCoder
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 @pytest.fixture
@@ -104,7 +104,7 @@ async def test_agent_can_load_mcp_server(agent_coder, mock_mcp_manager):
     """Verify an agent can load an MCP server."""
     server_name = "test_server"
 
-    await ResourceManagerTool.execute(agent_coder, load_mcp=[server_name])
+    await ManageResourcesTool.execute(agent_coder, load_mcp=[server_name])
 
     mock_mcp_manager.connect_server.assert_called_once_with(server_name)
     assert server_name in mock_mcp_manager.connected_servers
@@ -116,7 +116,7 @@ async def test_agent_can_remove_mcp_server(agent_coder, mock_mcp_manager):
     server_name = "test_server"
     mock_mcp_manager.connected_servers[server_name] = "connected"
 
-    await ResourceManagerTool.execute(agent_coder, remove_mcp=[server_name])
+    await ManageResourcesTool.execute(agent_coder, remove_mcp=[server_name])
 
     mock_mcp_manager.disconnect_server.assert_called_once_with(server_name)
     assert server_name not in mock_mcp_manager.connected_servers
@@ -127,7 +127,7 @@ async def test_sub_agent_can_load_mcp_server(sub_agent_coder, mock_mcp_manager):
     """Verify a subagent can load an MCP server."""
     server_name = "sub_test_server"
 
-    await ResourceManagerTool.execute(sub_agent_coder, load_mcp=[server_name])
+    await ManageResourcesTool.execute(sub_agent_coder, load_mcp=[server_name])
 
     mock_mcp_manager.connect_server.assert_called_once_with(server_name)
     assert server_name in mock_mcp_manager.connected_servers
@@ -139,7 +139,7 @@ async def test_sub_agent_can_remove_mcp_server(sub_agent_coder, mock_mcp_manager
     server_name = "sub_test_server"
     mock_mcp_manager.connected_servers[server_name] = "connected"
 
-    await ResourceManagerTool.execute(sub_agent_coder, remove_mcp=[server_name])
+    await ManageResourcesTool.execute(sub_agent_coder, remove_mcp=[server_name])
 
     mock_mcp_manager.disconnect_server.assert_called_once_with(server_name)
     assert server_name not in mock_mcp_manager.connected_servers
diff --git a/tests/integration/test_mcp_management.py b/tests/integration/test_mcp_management.py
index 0d5f86e4404..c2d3136766d 100644
--- a/tests/integration/test_mcp_management.py
+++ b/tests/integration/test_mcp_management.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class CoderMock:
@@ -58,14 +58,14 @@ async def test_integration_load_and_remove_server(coder):
     coder.coroutines.interruptible = coder.mock_interruptible
 
     # Load the server
-    load_result = await ResourceManagerTool.execute(coder, load_mcp=["integration-test-server"])
+    load_result = await ManageResourcesTool.execute(coder, load_mcp=["integration-test-server"])
     assert "Loaded server: integration-test-server" in str(load_result)
 
     # Mock the connected server for the remove tool
     coder.mcp_manager.connected_servers = {"integration-test-server": coder.mcp_manager.servers[0]}
 
     # Remove the server
-    remove_result = await ResourceManagerTool.execute(coder, remove_mcp=["integration-test-server"])
+    remove_result = await ManageResourcesTool.execute(coder, remove_mcp=["integration-test-server"])
     assert "Removed server: integration-test-server" in str(remove_result)
 
 
@@ -78,7 +78,7 @@ async def test_integration_wildcard_load_and_remove(coder):
     coder.coroutines.interruptible = coder.mock_interruptible
 
     # Load all enabled servers
-    load_result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+    load_result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
     assert "Loaded server: server1" in str(load_result)
     assert "Loaded server: server2" in str(load_result)
     # Non-enabled servers are filtered out silently by wildcard expansion
@@ -90,6 +90,6 @@ async def test_integration_wildcard_load_and_remove(coder):
     }
 
     # Remove all connected servers
-    remove_result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+    remove_result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
     assert "Removed server: server1" in str(remove_result)
     assert "Removed server: server2" in str(remove_result)
diff --git a/tests/mcp/test_manager_retry.py b/tests/mcp/test_manager_retry.py
index 976a7eac2d2..1e5bce8a2e1 100644
--- a/tests/mcp/test_manager_retry.py
+++ b/tests/mcp/test_manager_retry.py
@@ -687,9 +687,9 @@ async def test_load_session_benefits_from_retry(mock_server, mock_io, mock_tools
 
 
 @pytest.mark.asyncio
-async def test_resource_manager_benefits_from_retry(mock_server, mock_io, mock_tools, mock_session):
-    """TC-021: ResourceManager _load_mcp uses connect_server which now retries."""
-    from cecli.tools.resource_manager import Tool as ResourceManagerTool
+async def test_manage_resources_benefits_from_retry(mock_server, mock_io, mock_tools, mock_session):
+    """TC-021: ManageResources _load_mcp uses connect_server which now retries."""
+    from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
     coder = _make_mock_coder(mock_server, mock_io)
 
@@ -707,9 +707,9 @@ async def _connect_with_retry(name):
     # Mock the context block check
     coder.agent_config = {"include_context_blocks": {"servers"}, "exclude_context_blocks": set()}
 
-    with patch("cecli.tools.resource_manager.iter_all_coders", return_value=[coder]):
-        with patch("cecli.tools.resource_manager.update_server_registration"):
-            result = await ResourceManagerTool._load_mcp(coder, "test-server")
+    with patch("cecli.tools.manage_resources.iter_all_coders", return_value=[coder]):
+        with patch("cecli.tools.manage_resources.update_server_registration"):
+            result = await ManageResourcesTool._load_mcp(coder, "test-server")
 
     assert "Loaded server: test-server" in result
 
diff --git a/tests/subagents/test_tree.py b/tests/subagents/test_tree.py
new file mode 100644
index 00000000000..36e601244ef
--- /dev/null
+++ b/tests/subagents/test_tree.py
@@ -0,0 +1,112 @@
+"""Tests for cecli/helpers/agents/tree.py — agent visibility domains."""
+
+import pytest
+
+from cecli.helpers.agents.tree import AgentTree
+
+
+@pytest.fixture(autouse=True)
+def clean_tree():
+    """Start and end every test with an empty tree."""
+    AgentTree.clear()
+    yield
+    AgentTree.clear()
+
+
+@pytest.fixture
+def tree():
+    """A primary/worker/workflow hierarchy.
+
+    primary
+    ├── worker
+    │   └── grandchild
+    └── wf:demo
+        └── wf-child
+    """
+    AgentTree.register("primary", name="primary")
+    AgentTree.register("worker", parent_uuid="primary", name="worker")
+    AgentTree.register("grandchild", parent_uuid="worker", name="worker")
+    AgentTree.register("wf:demo", parent_uuid="primary", name="wf:demo")
+    AgentTree.register("wf-child", parent_uuid="wf:demo", name="worker")
+    return AgentTree
+
+
+class TestVisibility:
+    """Workflow trees are isolated from the rest of the agent hierarchy."""
+
+    def test_primary_sees_only_non_workflow_agents(self, tree):
+        visible = tree.get_visible_uuids("primary")
+
+        assert visible == {"primary", "worker", "grandchild"}
+        assert not tree.is_visible("primary", "wf:demo")
+        assert not tree.is_visible("primary", "wf-child")
+
+    def test_workflow_root_keeps_primary_as_structural_parent(self, tree):
+        """Isolation is visibility-only; the parent link is still the primary."""
+        assert tree.get_parent("wf:demo") == "primary"
+        assert "wf:demo" in tree.get_children("primary")
+        assert not tree.is_visible("primary", "wf:demo")
+
+    def test_regular_sub_agent_shares_primary_domain(self, tree):
+        visible = tree.get_visible_uuids("worker")
+
+        assert visible == {"primary", "worker", "grandchild"}
+        assert not tree.is_visible("worker", "wf:demo")
+
+    def test_workflow_root_sees_own_tree_only(self, tree):
+        visible = tree.get_visible_uuids("wf:demo")
+
+        assert visible == {"wf:demo", "wf-child"}
+        assert not tree.is_visible("wf:demo", "primary")
+        assert not tree.is_visible("wf:demo", "worker")
+
+    def test_workflow_descendant_sees_same_tree(self, tree):
+        visible = tree.get_visible_uuids("wf-child")
+
+        assert visible == {"wf:demo", "wf-child"}
+        assert not tree.is_visible("wf-child", "primary")
+
+    def test_nested_workflow_is_a_separate_domain(self):
+        AgentTree.register("primary", name="primary")
+        AgentTree.register("wf:outer", parent_uuid="primary", name="wf:outer")
+        AgentTree.register("outer-child", parent_uuid="wf:outer", name="worker")
+        AgentTree.register("wf:inner", parent_uuid="outer-child", name="wf:inner")
+        AgentTree.register("inner-child", parent_uuid="wf:inner", name="worker")
+
+        assert AgentTree.get_visible_uuids("wf:outer") == {"wf:outer", "outer-child"}
+        assert AgentTree.get_visible_uuids("outer-child") == {"wf:outer", "outer-child"}
+        assert AgentTree.get_visible_uuids("wf:inner") == {"wf:inner", "inner-child"}
+
+
+class TestMutations:
+    """Registration, removal and replacement maintain the hierarchy."""
+
+    def test_children_and_descendants(self, tree):
+        assert set(tree.get_children("primary")) == {"worker", "wf:demo"}
+        assert set(tree.get_descendants("primary")) == {
+            "worker",
+            "grandchild",
+            "wf:demo",
+            "wf-child",
+        }
+
+    def test_unregister_splices_children_onto_parent(self, tree):
+        tree.unregister("wf:demo")
+
+        assert tree.get_parent("wf-child") == "primary"
+        assert tree.get_workflow_root("wf-child") == "wf:demo"
+        assert tree.get_visible_uuids("wf-child") == {"wf-child"}
+
+    def test_replace_moves_children_to_new_uuid(self, tree):
+        tree.replace("worker", "worker-2")
+
+        assert tree.get_parent("worker-2") == "primary"
+        assert tree.get_parent("grandchild") == "worker-2"
+        assert tree.get_parent("worker") is None
+        assert tree.get_visible_uuids("worker-2") == {"primary", "worker-2", "grandchild"}
+
+    def test_unknown_uuid_is_tolerated(self, tree):
+        assert tree.get_parent("missing") is None
+        assert tree.get_children("missing") == []
+        assert tree.get_workflow_root("missing") is None
+        assert tree.get_visible_uuids("missing") == {"primary", "worker", "grandchild"}
diff --git a/tests/tools/test_create_format_output.py b/tests/tools/test_create_format_output.py
index 7a72850e7cc..28092014e01 100644
--- a/tests/tools/test_create_format_output.py
+++ b/tests/tools/test_create_format_output.py
@@ -1,9 +1,9 @@
-"""Tests for Create.format_output rendering."""
+"""Tests for ManageExtensions.format_output rendering."""
 
 import json
 from types import SimpleNamespace
 
-from cecli.tools import create
+from cecli.tools import manage_extensions
 
 
 class DummyIO:
@@ -31,14 +31,14 @@ def make_tool_response(args):
         id="test-id",
         type="function",
         function=SimpleNamespace(
-            name="Create",
+            name="ManageExtensions",
             arguments=json.dumps(args),
         ),
     )
 
 
 def render(coder, args):
-    create.Tool.format_output(
+    manage_extensions.Tool.format_output(
         coder,
         mcp_server=SimpleNamespace(name="Local"),
         tool_response=make_tool_response(args),
@@ -96,10 +96,10 @@ def test_format_output_reports_invalid_arguments():
     tool_response = SimpleNamespace(
         id="test-id",
         type="function",
-        function=SimpleNamespace(name="Create", arguments="{not json"),
+        function=SimpleNamespace(name="ManageExtensions", arguments="{not json"),
     )
 
-    create.Tool.format_output(
+    manage_extensions.Tool.format_output(
         coder,
         mcp_server=SimpleNamespace(name="Local"),
         tool_response=tool_response,
diff --git a/tests/tools/test_resource_manager_paging.py b/tests/tools/test_manage_resources_paging.py
similarity index 89%
rename from tests/tools/test_resource_manager_paging.py
rename to tests/tools/test_manage_resources_paging.py
index 1237b29dc6e..d5b86312004 100644
--- a/tests/tools/test_resource_manager_paging.py
+++ b/tests/tools/test_manage_resources_paging.py
@@ -1,4 +1,4 @@
-"""ResourceManager paging returns command output without adding context files."""
+"""ManageResources paging returns command output without adding context files."""
 
 import json
 from pathlib import Path
@@ -7,7 +7,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 from cecli.tools.utils.helpers import ToolError
 from cecli.tools.utils.responses import ToolResponse
 
@@ -40,7 +40,6 @@ def operations(monkeypatch):
         "_remove",
         "_view",
         "_editable",
-        "_stop_command",
         "_load_skill",
         "_remove_skill",
         "_load_mcp",
@@ -49,7 +48,7 @@ def operations(monkeypatch):
     ):
         mock_type = AsyncMock if name in ("_load_mcp", "_remove_mcp", "_list_mcp_servers") else Mock
         mocks[name] = mock_type(return_value="operation completed")
-        monkeypatch.setattr(ResourceManagerTool, name, mocks[name])
+        monkeypatch.setattr(ManageResourcesTool, name, mocks[name])
 
     return mocks
 
@@ -66,7 +65,7 @@ async def test_paging_returns_page_contents_without_adding_context(coder, target
     coder.abs_root_path.reset_mock()
     coder.local_agent_folder.reset_mock()
 
-    response = await ResourceManagerTool.execute(coder, paging=[{"target": target, "page": page}])
+    response = await ManageResourcesTool.execute(coder, paging=[{"target": target, "page": page}])
 
     assert isinstance(response, ToolResponse)
     result = response.to_dict()
@@ -85,7 +84,7 @@ async def test_missing_page_appends_response_error(coder):
     editable_before = coder.abs_fnames.copy()
     read_only_before = coder.abs_read_only_fnames.copy()
 
-    response = await ResourceManagerTool.execute(coder, paging=[{"target": "bg_1_1234", "page": 1}])
+    response = await ManageResourcesTool.execute(coder, paging=[{"target": "bg_1_1234", "page": 1}])
 
     assert isinstance(response, ToolResponse)
     result = response.to_dict()
@@ -116,7 +115,7 @@ async def test_missing_page_appends_response_error(coder):
 )
 async def test_invalid_paging_object_raises_before_operations(coder, operations, paging):
     with pytest.raises(ToolError):
-        await ResourceManagerTool.execute(coder, paging=paging, create=["untouched.txt"])
+        await ManageResourcesTool.execute(coder, paging=paging, create=["untouched.txt"])
 
     for operation in operations.values():
         operation.assert_not_called()
@@ -144,7 +143,7 @@ async def test_invalid_paging_object_raises_before_operations(coder, operations,
 )
 async def test_invalid_paging_target_raises_before_operations(coder, operations, target):
     with pytest.raises(ToolError):
-        await ResourceManagerTool.execute(
+        await ManageResourcesTool.execute(
             coder, paging=[{"target": target, "page": 1}], create=["untouched.txt"]
         )
 
@@ -156,7 +155,7 @@ async def test_invalid_paging_target_raises_before_operations(coder, operations,
 @pytest.mark.parametrize("page", [0, -1, 1.0, 1.5, "1", True, False, None, [], {}])
 async def test_invalid_page_number_raises_before_operations(coder, operations, page):
     with pytest.raises(ToolError):
-        await ResourceManagerTool.execute(
+        await ManageResourcesTool.execute(
             coder, paging=[{"target": "bg_1_1234", "page": page}], create=["untouched.txt"]
         )
 
@@ -183,7 +182,7 @@ async def test_command_key_alias_rejected_before_any_operations(coder, operation
     kwargs[action].append(alias)
 
     with pytest.raises(ToolError):
-        await ResourceManagerTool.execute(coder, **kwargs)
+        await ManageResourcesTool.execute(coder, **kwargs)
 
     for operation in operations.values():
         operation.assert_not_called()
@@ -192,7 +191,7 @@ async def test_command_key_alias_rejected_before_any_operations(coder, operation
 
 
 def test_paging_schema_requires_exact_target_and_positive_integer_page():
-    parameters = ResourceManagerTool.SCHEMA["function"]["parameters"]
+    parameters = ManageResourcesTool.SCHEMA["function"]["parameters"]
     paging = parameters["properties"]["paging"]
 
     assert paging["type"] == "array"
@@ -214,12 +213,12 @@ def test_format_output_shows_paging_target_and_page(coder):
         id="test-paging",
         type="function",
         function=SimpleNamespace(
-            name="ResourceManager",
+            name="ManageResources",
             arguments=json.dumps({"paging": [{"target": "bg_1_1234", "page": 7}]}),
         ),
     )
 
-    ResourceManagerTool.format_output(coder, SimpleNamespace(name="Local"), tool_response)
+    ManageResourcesTool.format_output(coder, SimpleNamespace(name="Local"), tool_response)
 
     output = "\n".join(
         str(call.args[0]) for call in coder.io.tool_output.call_args_list if call.args
@@ -241,7 +240,7 @@ async def test_three_pages_return_in_order_without_context_changes(coder):
 
     editable_before = coder.abs_fnames.copy()
     read_only_before = coder.abs_read_only_fnames.copy()
-    response = await ResourceManagerTool.execute(coder, paging=paging)
+    response = await ManageResourcesTool.execute(coder, paging=paging)
     result = response.to_dict()
 
     assert result["errors"] == []
@@ -259,7 +258,7 @@ async def test_invalid_later_entry_validated_before_reading_or_operations(coder,
     paging = [{"target": "bg_1_1234", "page": 1}, {"target": "bg_1_1234", "page": False}]
 
     with pytest.raises(ToolError):
-        await ResourceManagerTool.execute(coder, paging=paging, create=["untouched.txt"])
+        await ManageResourcesTool.execute(coder, paging=paging, create=["untouched.txt"])
 
     coder.local_agent_folder.assert_not_called()
     coder.abs_root_path.assert_not_called()
@@ -275,13 +274,13 @@ async def test_command_large_output_guidance_uses_paging_array(
     import asyncio
 
     from cecli.helpers import background_commands
-    from cecli.tools import command
+    from cecli.tools import shell
 
     coder.root = str(tmp_path)
     coder.large_file_token_threshold = 10
     coder.context_management_enabled = True
     coder.interrupt_event = asyncio.Event()
-    manager = command.BackgroundCommandManager
+    manager = shell.BackgroundCommandManager
     target = "bg_1_1234"
     output = "large command output\n" * 50
     monkeypatch.setattr(manager, "_generate_command_key", Mock(return_value=target))
@@ -289,8 +288,8 @@ async def test_command_large_output_guidance_uses_paging_array(
     monkeypatch.setattr(manager, "save_paginated_output", save)
 
     if execution_path == "foreground":
-        monkeypatch.setattr(command, "run_cmd_subprocess", Mock(return_value=(0, output)))
-        response = await command.Tool._execute_foreground(coder, "echo test")
+        monkeypatch.setattr(shell, "run_cmd_subprocess", Mock(return_value=(0, output)))
+        response = await shell.Tool._execute_foreground(coder, "echo test")
     else:
         process = Mock()
         process.wait.return_value = 0
@@ -306,14 +305,14 @@ def make_buffer(page_size=4096, pages_dir=None):
             return buffer
 
         monkeypatch.setattr(background_commands, "PagedOutputBuffer", make_buffer)
-        response = await command.Tool._execute_with_timeout(coder, "echo test", 30, use_pty=False)
+        response = await shell.Tool._execute_with_timeout(coder, "echo test", 30, use_pty=False)
 
     result = response.to_dict()
     assert result["errors"] == []
     content = result["result"][0]["content"]
     example = next(line for line in content.splitlines() if line.startswith('{"paging"'))
     assert json.loads(example) == {"paging": [{"target": target, "page": 1}]}
-    assert "ResourceManager" in content
+    assert "ManageResources" in content
     assert "command_key::" not in content
     assert "not added to file context" in content
 
diff --git a/tests/tools/test_registry.py b/tests/tools/test_registry.py
index 80985f5fbfe..27ee66602af 100644
--- a/tests/tools/test_registry.py
+++ b/tests/tools/test_registry.py
@@ -30,7 +30,7 @@ def test_registry_initialization(self):
         assert len(tools) > 0, "Registry should have tools after initialization"
 
         # Check that essential tools are registered
-        essential_tools = {"resourcemanager", "readfile", "yield"}
+        essential_tools = {"manageresources", "readfile", "yield"}
         for tool in essential_tools:
             assert tool in tools, f"Essential tool {tool} should be registered"
 
@@ -47,10 +47,10 @@ def test_ls_tool_removed(self):
     def test_get_tool(self):
         """Test getting individual tools by name"""
         # Get existing tool
-        tool_class = ToolRegistry.get_tool("resourcemanager")
-        assert tool_class is not None, "Should get resourcemanager tool"
+        tool_class = ToolRegistry.get_tool("manageresources")
+        assert tool_class is not None, "Should get manageresources tool"
         assert hasattr(tool_class, "NORM_NAME"), "Tool class should have NORM_NAME"
-        assert tool_class.NORM_NAME == "resourcemanager", "Tool name should match"
+        assert tool_class.NORM_NAME == "manageresources", "Tool name should match"
 
         # Get non-existent tool
         non_existent = ToolRegistry.get_tool("nonexistenttool")
@@ -64,107 +64,107 @@ def test_build_registry_empty_config(self):
         assert len(registry) > 0, "Should return tools with empty config"
 
         # Essential tools should always be included
-        assert "resourcemanager" in registry, "Essential tool should be included"
+        assert "manageresources" in registry, "Essential tool should be included"
         assert "readfile" in registry, "Essential tool should be included"
         assert "yield" in registry, "Essential tool should be included"
 
     def test_build_registry_with_includelist(self):
         """Test filtering with tools_includelist"""
-        config = {"tools_includelist": ["resourcemanager", "editfile"]}
+        config = {"tools_includelist": ["manageresources", "editfile"]}
         registry = ToolRegistry.build_registry(config)
 
         # Should only include tools from includelist, plus essential tools
         assert len(registry) == 3, "Should include 2 from list + 1 essential"
-        assert "resourcemanager" in registry
+        assert "manageresources" in registry
         assert "editfile" in registry
         assert "yield" in registry  # Essential
-        assert "command" not in registry, "Should not include tools not in includelist"
+        assert "shell" not in registry, "Should not include tools not in includelist"
 
     def test_build_registry_with_excludelist(self):
         """Test filtering with tools_excludelist"""
-        config = {"tools_excludelist": ["command", "commandinteractive"]}
+        config = {"tools_excludelist": ["shell", "commandinteractive"]}
         registry = ToolRegistry.build_registry(config)
 
         # Should exclude specified tools (except essentials)
-        assert "command" not in registry, "Should exclude command"
+        assert "shell" not in registry, "Should exclude shell"
         assert "commandinteractive" not in registry, "Should exclude commandinteractive"
-        assert "resourcemanager" in registry, "Essential tool should still be included"
+        assert "manageresources" in registry, "Essential tool should still be included"
 
     def test_build_registry_exclude_essential(self):
         """Test that essential tools cannot be excluded"""
-        config = {"tools_excludelist": ["resourcemanager", "readfile", "finished", "command"]}
+        config = {"tools_excludelist": ["manageresources", "readfile", "finished", "shell"]}
         registry = ToolRegistry.build_registry(config)
 
         # Essential tools should still be included despite excludelist
-        assert "resourcemanager" not in registry, "Non-essential tool can be excluded"
+        assert "manageresources" not in registry, "Non-essential tool can be excluded"
         assert "readfile" not in registry, "Non-essential tool can be excluded"
         assert "yield" in registry, "Essential tool cannot be excluded"
-        assert "command" not in registry, "Non-essential tool should be excluded"
+        assert "shell" not in registry, "Non-essential tool should be excluded"
 
     def test_build_registry_combined_filters(self):
         """Test combined filtering with includelist and excludelist"""
         config = {
-            "tools_includelist": ["resourcemanager", "editfile", "command"],
+            "tools_includelist": ["manageresources", "editfile", "shell"],
             "tools_excludelist": ["commandinteractive"],
         }
         registry = ToolRegistry.build_registry(config)
 
         # Should respect all filters
         assert len(registry) == 4, "Should include exactly 4 tools (3 from list + yield)"
-        assert "resourcemanager" in registry
+        assert "manageresources" in registry
         assert "editfile" in registry
         assert "yield" in registry
-        assert "command" in registry
+        assert "shell" in registry
         assert "commandinteractive" not in registry
 
     def test_get_filtered_tools(self):
         """Test get_filtered_tools method"""
-        config = {"tools_includelist": ["resourcemanager", "editfile"]}
+        config = {"tools_includelist": ["manageresources", "editfile"]}
         ToolRegistry.build_registry(config)
         tool_names = ToolRegistry.get_registered_tools()
 
         # Should return list of tool names
         assert isinstance(tool_names, list)
-        # Should include resourcemanager, editfile, and yield (essentials)
+        # Should include manageresources, editfile, and yield (essentials)
         assert len(tool_names) == 3
-        assert "resourcemanager" in tool_names
+        assert "manageresources" in tool_names
         assert "editfile" in tool_names
         assert "yield" in tool_names  # Essential tool always included
 
     def test_legacy_config_names(self):
         """Test backward compatibility with legacy config names (whitelist/blacklist)"""
         config = {
-            "tools_whitelist": ["resourcemanager", "editfile"],
-            "tools_blacklist": ["command"],
+            "tools_whitelist": ["manageresources", "editfile"],
+            "tools_blacklist": ["shell"],
         }
         registry = ToolRegistry.build_registry(config)
 
         # Should work with legacy names
-        assert "resourcemanager" in registry
+        assert "manageresources" in registry
         assert "editfile" in registry
-        assert "command" not in registry
+        assert "shell" not in registry
 
     def test_config_precedence(self):
         """Test that new config names take precedence over legacy names"""
         config = {
-            "tools_includelist": ["resourcemanager"],
-            "tools_whitelist": ["command"],  # Should be ignored
+            "tools_includelist": ["manageresources"],
+            "tools_whitelist": ["shell"],  # Should be ignored
             "tools_excludelist": ["commandinteractive"],
             "tools_blacklist": ["finished"],  # Should be ignored for essential tool
         }
         registry = ToolRegistry.build_registry(config)
 
         # New names should take precedence
-        assert "resourcemanager" in registry, "Should use tools_includelist"
+        assert "manageresources" in registry, "Should use tools_includelist"
         assert (
-            "command" not in registry
+            "shell" not in registry
         ), "Should not use tools_whitelist when tools_includelist present"
         assert "commandinteractive" not in registry, "Should use tools_excludelist"
         assert "yield" in registry, "Essential tool cannot be excluded"
 
     def test_registry_consistency(self):
         """Test that registry methods return consistent results"""
-        config = {"tools_includelist": ["resourcemanager", "editfile"]}
+        config = {"tools_includelist": ["manageresources", "editfile"]}
 
         # build_registry should return consistent results
         registry = ToolRegistry.build_registry(config)
@@ -179,9 +179,9 @@ def test_skill_and_mcp_tools_in_context_manager(self):
         """Test that skill/MCP functionality is now part of context_manager."""
         # The individual load_skill, remove_skill, load_mcp, remove_mcp tools
         # have been merged into context_manager
-        ctx_tool = ToolRegistry.get_tool("resourcemanager")
-        assert ctx_tool is not None, "resourcemanager tool should be registered"
-        assert ctx_tool.NORM_NAME == "resourcemanager"
+        ctx_tool = ToolRegistry.get_tool("manageresources")
+        assert ctx_tool is not None, "manageresources tool should be registered"
+        assert ctx_tool.NORM_NAME == "manageresources"
 
         # Verify the merged tools no longer exist as separate entries
         assert ToolRegistry.get_tool("loadskill") is None
diff --git a/tests/tools/test_remove_mcp_tool.py b/tests/tools/test_remove_mcp_tool.py
index 2da2e85e37c..837d44a6dc4 100644
--- a/tests/tools/test_remove_mcp_tool.py
+++ b/tests/tools/test_remove_mcp_tool.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class DummyIO:
@@ -54,7 +54,7 @@ class TestRemoveMcpTool:
     def _patch_global_exclusion(self):
         """Patch is_server_globally_excluded to prevent test isolation issues with AgentService."""
         with patch(
-            "cecli.tools.resource_manager.is_server_globally_excluded",
+            "cecli.tools.manage_resources.is_server_globally_excluded",
             return_value=True,
         ):
             yield
@@ -63,7 +63,7 @@ def _patch_global_exclusion(self):
     async def test_no_configured_servers(self, coder):
         """Test when no MCP servers are configured at all."""
         coder.mcp_manager.servers = []
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["test"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["test"])
         assert result.to_dict()["result"][0]["content"] == "No MCP servers are configured."
 
     @pytest.mark.asyncio
@@ -72,7 +72,7 @@ async def test_server_not_found(self, coder, mock_server):
         coder.mcp_manager.servers = [mock_server]
         coder.mcp_manager.connected_servers = {"existing": "server"}
         coder.mcp_manager.get_server.return_value = None
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["nonexistent"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["nonexistent"])
         assert "MCP server nonexistent does not exist." in str(result)
 
     @pytest.mark.asyncio
@@ -81,7 +81,7 @@ async def test_all_servers_not_loaded(self, coder, mock_server):
         coder.mcp_manager.servers = [mock_server]
         coder.mcp_manager.connected_servers = {}
         coder.mcp_manager.get_server.return_value = mock_server
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
         assert "Server test-server is not currently connected." in str(result)
 
     @pytest.mark.asyncio
@@ -91,7 +91,7 @@ async def test_successful_removal(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {"test-server": mock_server}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (True, False)
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
         assert "Removed server: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -101,7 +101,7 @@ async def test_removal_interrupted(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {"test-server": mock_server}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (False, True)
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
         assert "Interrupted: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -111,7 +111,7 @@ async def test_removal_failed(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {"test-server": mock_server}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (False, False)
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
         assert "Unable to remove server: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -127,7 +127,7 @@ async def test_remove_all_servers(self, coder):
             (s for s in [server1, server2] if s.name == name), None
         )
         coder.coroutines.interruptible.return_value = (True, False)
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
         assert "Removed server: server1" in str(result)
         assert "Removed server: server2" in str(result)
 
@@ -152,6 +152,6 @@ async def mock_interruptible_func(*args, **kwargs):
             return result
 
         coder.coroutines.interruptible.side_effect = mock_interruptible_func
-        result = await ResourceManagerTool.execute(coder, remove_mcp=["server1", "server2"])
+        result = await ManageResourcesTool.execute(coder, remove_mcp=["server1", "server2"])
         assert "Removed server: server1" in str(result)
         assert "Unable to remove server: server2" in str(result)
diff --git a/tests/tools/test_command_format_output.py b/tests/tools/test_shell_format_output.py
similarity index 92%
rename from tests/tools/test_command_format_output.py
rename to tests/tools/test_shell_format_output.py
index 7e5b3423dd0..f91ecc28a37 100644
--- a/tests/tools/test_command_format_output.py
+++ b/tests/tools/test_shell_format_output.py
@@ -1,4 +1,4 @@
-"""Regression tests for Command.format_output rendering.
+"""Regression tests for Shell.format_output rendering.
 
 format_output() prints the "Tool Call:" header before execute() runs, so it
 must clearly communicate what execute() is about to do — including the case
@@ -14,7 +14,7 @@
 
 import pytest
 
-from cecli.tools import command
+from cecli.tools import shell
 
 
 class DummyIO:
@@ -45,14 +45,14 @@ def make_tool_response(args):
         id="test-id",
         type="function",
         function=SimpleNamespace(
-            name="Command",
+            name="Shell",
             arguments=json.dumps(args),
         ),
     )
 
 
 def render(coder, args):
-    command.Tool.format_output(
+    shell.Tool.format_output(
         coder,
         mcp_server=SimpleNamespace(name="Local"),
         tool_response=make_tool_response(args),
@@ -72,7 +72,7 @@ def test_format_output_empty_args_shows_explicit_placeholder():
     """Regression test: an empty-args call (no command, no background_key/action)
     must not render a header with a silently blank body.
 
-    Reproduces a real session where the model called `Command` with `{}` —
+    Reproduces a real session where the model called `Shell` with `{}` —
     execute() correctly rejects this ("'command' must be provided."), but the
     displayed panel showed nothing after the header, which was mistaken for
     a display bug (auto-approved commands failing to render their body).
@@ -111,7 +111,7 @@ def __init__(self):
             self.agent_config = {}
 
     coder = ExecCoder()
-    response = await command.Tool.execute(coder)
+    response = await shell.Tool.execute(coder)
     result = response.to_dict()
     assert result["result"] == []
     assert "'command' must be provided." in result["errors"]
diff --git a/tests/tools/test_command_timeout_paging.py b/tests/tools/test_shell_timeout_paging.py
similarity index 96%
rename from tests/tools/test_command_timeout_paging.py
rename to tests/tools/test_shell_timeout_paging.py
index c4fb654a12f..2abf4e42204 100644
--- a/tests/tools/test_command_timeout_paging.py
+++ b/tests/tools/test_shell_timeout_paging.py
@@ -10,8 +10,8 @@
 import pytest_asyncio
 
 from cecli.helpers import background_commands
-from cecli.tools.command import Tool as CommandTool
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
+from cecli.tools.shell import Tool as ShellTool
 
 
 @pytest_asyncio.fixture
@@ -70,7 +70,7 @@ async def execute(output, threshold=8, enabled=True):
         coder.large_file_token_threshold = threshold
         coder.context_management_enabled = enabled
         state["output"] = output
-        response = await CommandTool._execute_with_timeout(
+        response = await ShellTool._execute_with_timeout(
             coder, "pending command", 0.001, use_pty=False
         )
         result = response.to_dict()
@@ -116,7 +116,7 @@ async def test_elapsed_timeout_pages_output_and_exposes_command_keys(elapsed_com
     assert num_pages >= 1
     assert output not in content
     assert f"Output paged to disk: {num_pages} page(s)." in content
-    assert "ResourceManager" in content
+    assert "ManageResources" in content
     assert "not added to file context" in content
     assert "command_key::" not in content
     example = next(line for line in content.splitlines() if line.startswith('{"paging"'))
@@ -133,7 +133,7 @@ async def test_elapsed_timeout_pages_output_and_exposes_command_keys(elapsed_com
 
     editable_before = coder.abs_fnames.copy()
     read_only_before = coder.abs_read_only_fnames.copy()
-    response = await ResourceManagerTool.execute(coder, paging=[{"target": target, "page": 1}])
+    response = await ManageResourcesTool.execute(coder, paging=[{"target": target, "page": 1}])
     result = response.to_dict()
     assert result["errors"] == []
     assert result["result"][0]["content"].endswith(output[:page_size])
diff --git a/tests/tools/test_tools_load_mcp_tool.py b/tests/tools/test_tools_load_mcp_tool.py
index 6e9071cb711..2397bb2b2dd 100644
--- a/tests/tools/test_tools_load_mcp_tool.py
+++ b/tests/tools/test_tools_load_mcp_tool.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class DummyIO:
@@ -54,7 +54,7 @@ class TestLoadMcpTool:
     async def test_no_mcp_servers_found(self, coder):
         """Test when no MCP servers are configured."""
         coder.mcp_manager.servers = []
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test"])
         assert result.to_dict()["result"][0]["content"] == "No MCP servers found, nothing to load."
 
     @pytest.mark.asyncio
@@ -62,7 +62,7 @@ async def test_server_not_found(self, coder, mock_server):
         """Test when requested server doesn't exist."""
         coder.mcp_manager.servers = [mock_server]
         coder.mcp_manager.get_server.return_value = None
-        result = await ResourceManagerTool.execute(coder, load_mcp=["nonexistent"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["nonexistent"])
         assert "MCP server nonexistent does not exist." in str(result)
 
     @pytest.mark.asyncio
@@ -74,7 +74,7 @@ async def test_server_already_loaded(self, coder, mock_server):
         coder.mcp_manager.get_server.return_value = mock_server
         # Must return tuple (did_connect, interrupted)
         coder.coroutines.interruptible.return_value = (True, False)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
         assert "Server already loaded: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -83,7 +83,7 @@ async def test_server_not_enabled_by_default(self, coder, mock_server):
         mock_server.config = {"enabled": False}
         coder.mcp_manager.servers = [mock_server]
         coder.mcp_manager.get_server.return_value = mock_server
-        await ResourceManagerTool.execute(coder, load_mcp=["*"])
+        await ManageResourcesTool.execute(coder, load_mcp=["*"])
 
     @pytest.mark.asyncio
     async def test_successful_load(self, coder, mock_server):
@@ -92,7 +92,7 @@ async def test_successful_load(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (True, False)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
         assert "Loaded server: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -102,7 +102,7 @@ async def test_load_interrupted(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (False, True)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
         assert "Interrupted: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -112,7 +112,7 @@ async def test_load_failed(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {}
         coder.mcp_manager.get_server.return_value = mock_server
         coder.coroutines.interruptible.return_value = (False, False)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
         assert "Unable to load server: test-server" in str(result)
 
     @pytest.mark.asyncio
@@ -130,7 +130,7 @@ async def test_load_all_servers(self, coder):
             (s for s in [server1, server2] if s.name == name), None
         )
         coder.coroutines.interruptible.return_value = (True, False)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
         assert "Loaded server: server1" in str(result)
         assert "Loaded server: server2" in str(result)
 
@@ -160,7 +160,7 @@ async def mock_interruptible_func(*args, **kwargs):
                 return (False, False)
 
         coder.coroutines.interruptible.side_effect = mock_interruptible_func
-        result = await ResourceManagerTool.execute(coder, load_mcp=["server1", "server2"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["server1", "server2"])
         assert "Loaded server: server1" in str(result)
         assert "Unable to load server: server2" in str(result)
 
@@ -173,7 +173,7 @@ async def test_duplicate_iteration_bug_fix(self, coder, mock_server):
         coder.mcp_manager.connected_servers = {"test-server": mock_server}
         coder.mcp_manager.get_server.return_value = mock_server
 
-        result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
 
         # Should only report server already loaded once
         assert str(result).count("Server already loaded: test-server") == 1
@@ -210,7 +210,7 @@ async def mock_interruptible(coro, event):
 
         coder.mcp_manager.connect_server = AsyncMock(side_effect=mock_connect_server)
         coder.coroutines.interruptible = AsyncMock(side_effect=mock_interruptible)
-        result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+        result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
 
         # Should only attempt to load server2 (server1 should be skipped)
         # Wildcard expansion skips already-connected servers, so server1 is not reported
diff --git a/tests/tui/test_floating_panels.py b/tests/tui/test_floating_panels.py
index 983197229b3..33ade48386a 100644
--- a/tests/tui/test_floating_panels.py
+++ b/tests/tui/test_floating_panels.py
@@ -203,7 +203,10 @@ async def _run_panel_checks():
 
 def test_floating_panels_layout_scroll_and_toggle():
     agent_service = SimpleNamespace(
-        sub_agents={"u1": SimpleNamespace(name="worker")},
+        sub_agents={
+            "u1": SimpleNamespace(name="worker"),
+            "u2": SimpleNamespace(name="wf:demo"),
+        },
         foreground_uuid="u1",
         foreground_coder=None,
     )
@@ -214,7 +217,7 @@ def test_floating_panels_layout_scroll_and_toggle():
         ),
         patch(
             "cecli.helpers.agents.service.AgentService.get_registry",
-            return_value={"tester": None, "worker": None},
+            return_value={"tester": None, "worker": None, "wf:demo": None},
         ),
     ):
         asyncio.run(_run_panel_checks())
diff --git a/tests/tui/test_key_bindings.py b/tests/tui/test_key_bindings.py
new file mode 100644
index 00000000000..9c741f037d4
--- /dev/null
+++ b/tests/tui/test_key_bindings.py
@@ -0,0 +1,55 @@
+"""Tests for TUI key binding normalization and registration."""
+
+import pytest
+from textual.binding import BindingsMap
+
+from cecli.tui.app import TUI
+
+
+@pytest.fixture
+def tui_instance(monkeypatch):
+    """Create a TUI instance without running Textual's App init."""
+    monkeypatch.setattr("cecli.tui.app.TUI.__init__", lambda *args, **kwargs: None)
+    tui = TUI(coder_worker=None, output_queue=None, input_queue=None, args=None)
+    tui._bindings = BindingsMap()
+    tui.tui_config = {"key_bindings": {}}
+    return tui
+
+
+def test_encode_keys_canonicalizes_modifier_order(tui_instance):
+    assert tui_instance._encode_keys("ctrl+alt+right") == "alt+ctrl+right"
+    assert tui_instance._encode_keys("shift+ctrl+alt+right") == "alt+ctrl+shift+right"
+    assert tui_instance._encode_keys("alt+ctrl+right") == "alt+ctrl+right"
+
+
+def test_encode_keys_preserves_unknown_modifiers(tui_instance):
+    assert tui_instance._encode_keys("hyper+right") == "hyper+right"
+
+
+def test_is_key_for_accepts_any_modifier_order(tui_instance):
+    tui_instance.tui_config["key_bindings"]["next_agent"] = tui_instance._encode_keys(
+        "ctrl+alt+right"
+    )
+    assert tui_instance.is_key_for("next_agent", "alt+ctrl+right")
+    assert tui_instance.is_key_for("next_agent", "ctrl+alt+right")
+    assert not tui_instance.is_key_for("next_agent", "ctrl+alt+left")
+
+
+def test_bind_key_registers_priority_binding(tui_instance):
+    tui_instance.tui_config["key_bindings"]["next_agent"] = tui_instance._encode_keys(
+        "ctrl+alt+right"
+    )
+    tui_instance._bind_key(
+        "next_agent", "switch_next_agent", description="Next Agent", priority=True
+    )
+    bindings = tui_instance._bindings.key_to_bindings["alt+ctrl+right"]
+    assert bindings
+    assert all(binding.priority for binding in bindings)
+
+
+def test_bind_key_can_register_non_priority_binding(tui_instance):
+    tui_instance.tui_config["key_bindings"]["submit"] = tui_instance._encode_keys("enter")
+    tui_instance._bind_key("submit", "noop", description="Submit", priority=False)
+    bindings = tui_instance._bindings.key_to_bindings["enter"]
+    assert bindings
+    assert not any(binding.priority for binding in bindings)
diff --git a/tests/unit/test_load_mcp.py b/tests/unit/test_load_mcp.py
index 17380a33252..e5c536a40db 100644
--- a/tests/unit/test_load_mcp.py
+++ b/tests/unit/test_load_mcp.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class DummyIO:
@@ -50,7 +50,7 @@ def mock_server():
 async def test_no_mcp_servers_found(coder):
     """Test when no MCP servers are configured."""
     coder.mcp_manager.servers = []
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test"])
     assert result.to_dict()["result"][0]["content"] == "No MCP servers found, nothing to load."
 
 
@@ -59,7 +59,7 @@ async def test_server_not_found(coder, mock_server):
     """Test when requested server doesn't exist."""
     coder.mcp_manager.servers = [mock_server]
     coder.mcp_manager.get_server.return_value = None
-    result = await ResourceManagerTool.execute(coder, load_mcp=["nonexistent"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["nonexistent"])
     assert "MCP server nonexistent does not exist." in str(result)
 
 
@@ -77,7 +77,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
     assert "Server already loaded: test-server" in str(result)
     # connect_server should not have been called since it was already loaded
     coder.mcp_manager.connect_server.assert_not_called()
@@ -91,7 +91,7 @@ async def test_server_not_enabled_by_default(coder, mock_server):
     coder.mcp_manager.connected_servers = {}
     coder.mcp_manager.get_server.return_value = mock_server
     coder.mcp_manager.connect_server = AsyncMock()
-    result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
     # Non-enabled servers are silently filtered by wildcard expansion
     # Result is an empty list since nothing was loaded
     assert result.to_dict()["result"] == []
@@ -115,7 +115,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
     assert "Loaded server: test-server" in str(result)
 
 
@@ -137,7 +137,7 @@ async def mock_interruptible(coro, event):
         return False, True
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
     assert "Interrupted: test-server" in str(result)
 
 
@@ -156,7 +156,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
     assert "Unable to load server: test-server" in str(result)
 
 
@@ -186,7 +186,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
     assert "Loaded server: server1" in str(result)
     assert "Loaded server: server2" in str(result)
 
@@ -221,7 +221,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["server1", "server2"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["server1", "server2"])
     assert "Loaded server: server1" in str(result)
     assert "Unable to load server: server2" in str(result)
 
@@ -236,7 +236,7 @@ async def test_duplicate_iteration_bug_fix(coder, mock_server):
     coder.mcp_manager.get_server.return_value = mock_server
     # Set up connect_server as AsyncMock
     coder.mcp_manager.connect_server = AsyncMock()
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
     # Should only report server already loaded once
     assert str(result).count("Server already loaded: test-server") == 1
     # connect_server should not have been called since it was already loaded
@@ -273,7 +273,7 @@ async def mock_interruptible(coro, event):
         return await coro, False
 
     coder.coroutines.interruptible = mock_interruptible
-    result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
     # Should only attempt to load server2 (server1 should be skipped)
     # server1 is already connected so it's skipped silently by wildcard expansion
     assert "Server already loaded: server1" not in str(result)
diff --git a/tests/unit/test_remove_mcp.py b/tests/unit/test_remove_mcp.py
index 557baf36bf7..d2b63e9b5c6 100644
--- a/tests/unit/test_remove_mcp.py
+++ b/tests/unit/test_remove_mcp.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class DummyIO:
@@ -70,7 +70,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
     # Execute
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
     # Assertions
     assert "Removed server: test-server" in str(result)
     coder.mcp_manager.disconnect_server.assert_awaited_once_with("test-server")
@@ -90,7 +90,7 @@ async def test_remove_mcp_tool_non_existent():
     # But the one we're looking for doesn't exist
     coder.mcp_manager.get_server.return_value = None
     # Execute
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["non-existent-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["non-existent-server"])
     # Assertions
     assert "MCP server non-existent-server does not exist." in str(result)
 
@@ -106,7 +106,7 @@ async def test_remove_mcp_tool_not_connected():
     coder.mcp_manager.servers = [server]
     coder.mcp_manager.get_server.return_value = server
     coder.mcp_manager.connected_servers = {}
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
     assert "Server test-server is not currently connected." in str(result)
 
 
@@ -139,7 +139,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines = MagicMock()
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
     assert "Removed server: server1" in str(result)
     assert "Removed server: server2" in str(result)
 
@@ -166,7 +166,7 @@ async def mock_interruptible(coro, event):
 
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
     assert "Interrupted: test-server" in str(result)
 
 
@@ -192,7 +192,7 @@ async def mock_interruptible(coro, event):
 
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
     assert "Unable to remove server: test-server" in str(result)
 
 
@@ -203,7 +203,7 @@ async def test_remove_mcp_tool_no_servers_configured():
     coder.agent_config = {"include_context_blocks": {"servers"}, "exclude_context_blocks": set()}
     coder.mcp_manager = MagicMock()
     coder.mcp_manager.servers = []
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test"])
     assert result.to_dict()["result"][0]["content"] == "No MCP servers are configured."
 
 
@@ -237,7 +237,7 @@ async def mock_interruptible(coro, event):
 
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["server1", "server2"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["server1", "server2"])
     assert "Removed server: server1" in str(result)
     assert "Unable to remove server: server2" in str(result)
 
@@ -268,7 +268,7 @@ async def mock_interruptible(coro, event):
 
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
     # Should successfully remove both servers using dictionary keys
     assert "Removed server: server1" in str(result)
     assert "Removed server: server2" in str(result)
diff --git a/tests/unit/test_unit_load_mcp_tool.py b/tests/unit/test_unit_load_mcp_tool.py
index 0376ac79284..689d09d4257 100644
--- a/tests/unit/test_unit_load_mcp_tool.py
+++ b/tests/unit/test_unit_load_mcp_tool.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 @pytest.fixture
@@ -73,7 +73,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible.side_effect = mock_interruptible
     coder.interrupt_event = MagicMock()
     coder.registered_servers = {"included": set(), "excluded": set()}
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
 
     assert "Loaded server: test-server" in str(result)
     mock_mcp_manager.connect_server.assert_awaited_once_with("test-server")
@@ -88,7 +88,7 @@ async def test_load_mcp_tool_non_existent(mock_mcp_manager):
     coder.mcp_manager = mock_mcp_manager
     coder.registered_servers = {"included": set(), "excluded": set()}
 
-    result = await ResourceManagerTool.execute(coder, load_mcp=["non-existent-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["non-existent-server"])
 
     assert "MCP server non-existent-server does not exist." in str(result)
     mock_mcp_manager.connect_server.assert_not_awaited()
@@ -105,7 +105,7 @@ async def test_load_mcp_tool_already_loaded(mock_mcp_manager):
     server = mock_mcp_manager.get_server("test-server")
     coder.mcp_manager.connected_servers = {"test-server": server}
 
-    result = await ResourceManagerTool.execute(coder, load_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["test-server"])
 
     assert "Server already loaded: test-server" in str(result)
     mock_mcp_manager.connect_server.assert_not_awaited()
@@ -132,7 +132,7 @@ async def mock_interruptible(coro, event):
     server1 = mock_mcp_manager.get_server("test-server")
     coder.mcp_manager.connected_servers = {"test-server": server1}
 
-    result = await ResourceManagerTool.execute(coder, load_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, load_mcp=["*"])
 
     # Check results
     # Wildcard expansion skips already-connected servers; no "already loaded" message is produced
diff --git a/tests/unit/test_unit_remove_mcp_tool.py b/tests/unit/test_unit_remove_mcp_tool.py
index aab2f93e500..ce7b1256870 100644
--- a/tests/unit/test_unit_remove_mcp_tool.py
+++ b/tests/unit/test_unit_remove_mcp_tool.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from cecli.tools.resource_manager import Tool as ResourceManagerTool
+from cecli.tools.manage_resources import Tool as ManageResourcesTool
 
 
 class DummyIO:
@@ -72,7 +72,7 @@ async def mock_interruptible(coro, event):
     coder.interrupt_event = MagicMock()
 
     # Execute
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
 
     # Assertions
     assert "Removed server: test-server" in str(result)
@@ -95,7 +95,7 @@ async def test_remove_mcp_tool_non_existent():
     coder.mcp_manager.get_server.return_value = None
 
     # Execute
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["non-existent-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["non-existent-server"])
 
     # Assertions
     assert "MCP server non-existent-server does not exist." in str(result)
@@ -114,7 +114,7 @@ async def test_remove_mcp_tool_not_connected():
     coder.mcp_manager.get_server.return_value = server
     coder.mcp_manager.connected_servers = {}
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
 
     assert "Server test-server is not currently connected." in str(result)
 
@@ -150,7 +150,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
 
     assert "Removed server: server1" in str(result)
     assert "Removed server: server2" in str(result)
@@ -180,7 +180,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
 
     assert "Interrupted: test-server" in str(result)
 
@@ -209,7 +209,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test-server"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test-server"])
 
     assert "Unable to remove server: test-server" in str(result)
 
@@ -223,7 +223,7 @@ async def test_remove_mcp_tool_no_servers_configured():
     coder.registered_servers = {"included": set(), "excluded": set()}
     coder.mcp_manager.servers = []
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["test"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["test"])
 
     assert result.to_dict()["result"][0]["content"] == "No MCP servers are configured."
 
@@ -261,7 +261,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["server1", "server2"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["server1", "server2"])
 
     assert "Removed server: server1" in str(result)
     assert "Unable to remove server: server2" in str(result)
@@ -295,7 +295,7 @@ async def mock_interruptible(coro, event):
     coder.coroutines.interruptible = mock_interruptible
     coder.interrupt_event = MagicMock()
 
-    result = await ResourceManagerTool.execute(coder, remove_mcp=["*"])
+    result = await ManageResourcesTool.execute(coder, remove_mcp=["*"])
 
     # Should successfully remove both servers using dictionary keys
     assert "Removed server: server1" in str(result)