diff --git a/cecli/commands/__init__.py b/cecli/commands/__init__.py index 4e77ec64099..db5aac58604 100644 --- a/cecli/commands/__init__.py +++ b/cecli/commands/__init__.py @@ -65,6 +65,7 @@ from .save_session import SaveSessionCommand from .settings import SettingsCommand from .spawn_agent import SpawnAgentCommand +from .switch_agent import SwitchAgentCommand from .terminal_setup import TerminalSetupCommand from .test import TestCommand from .think_tokens import ThinkTokensCommand @@ -118,6 +119,7 @@ CommandRegistry.register(InvokeAgentCommand) CommandRegistry.register(ReapAgentCommand) CommandRegistry.register(SpawnAgentCommand) +CommandRegistry.register(SwitchAgentCommand) CommandRegistry.register(IncludeSkillCommand) CommandRegistry.register(LintCommand) CommandRegistry.register(ListSessionsCommand) @@ -199,6 +201,7 @@ "InvokeAgentCommand", "ReapAgentCommand", "SpawnAgentCommand", + "SwitchAgentCommand", "LintCommand", "ListSessionsCommand", "ListSkillsCommand", diff --git a/cecli/commands/invoke_agent.py b/cecli/commands/invoke_agent.py index 6b14c8caf91..d1211cf31d7 100644 --- a/cecli/commands/invoke_agent.py +++ b/cecli/commands/invoke_agent.py @@ -25,9 +25,11 @@ async def execute(cls, io, coder, args, **kwargs): summary = await agent_service.invoke(name, prompt, blocking=True) if summary: from cecli.helpers.conversation.service import ConversationService + from cecli.helpers.conversation.tags import MessageTag ConversationService.get_manager(coder).add_message( message_dict=dict(role="user", content=summary), + tag=MessageTag.CUR, ) io.tool_output(f"Sub-agent '{name}' completed:\n{summary}") else: diff --git a/cecli/commands/switch_agent.py b/cecli/commands/switch_agent.py new file mode 100644 index 00000000000..4840f1581d4 --- /dev/null +++ b/cecli/commands/switch_agent.py @@ -0,0 +1,114 @@ +from typing import List + +from cecli.commands.utils.base_command import BaseCommand +from cecli.commands.utils.helpers import format_command_result +from cecli.helpers.agents.service import AgentService + + +class SwitchAgentCommand(BaseCommand): + NORM_NAME = "switch-agent" + DESCRIPTION = "Switch to a specific agent by name" + + @classmethod + async def execute(cls, io, coder, args, **kwargs): + """Execute the switch-agent command.""" + agent_name = args.strip() + if not agent_name: + io.tool_error("Usage: /switch-agent ") + return 1 + + try: + agent_service = AgentService.get_instance(coder) + except Exception as e: + io.tool_error(f"Could not get agent service: {e}") + return 1 + + agent_uuid = None + + if agent_name == "primary": + agent_uuid = str(coder.uuid) + else: + if agent_service and agent_service.sub_agents: + # Try parsing "name (uuid)" format + if agent_name.endswith(")") and " (" in agent_name: + try: + # Extract uuid prefix from "name (prefix)" + uuid_prefix = agent_name.rsplit(" (", 1)[1][:-1] + for uuid, info in agent_service.sub_agents.items(): + if uuid.startswith(uuid_prefix): + agent_uuid = uuid + break + except IndexError: + pass # Not the format we expected + + # If not found via "name (uuid)", try matching by name directly + if agent_uuid is None: + for uuid, sub_agent_info in agent_service.sub_agents.items(): + if sub_agent_info.name == agent_name: + agent_uuid = uuid + break + + # If still not found, try matching by uuid prefix directly + if agent_uuid is None: + for uuid, sub_agent_info in agent_service.sub_agents.items(): + if uuid.startswith(agent_name): + agent_uuid = uuid + break + + if agent_uuid is None: + io.tool_error(f"Error: Agent '{agent_name}' not found.") + return 1 + + if hasattr(io, "output_queue") and io.output_queue: + io.output_queue.put({"type": "switch_agent", "uuid": agent_uuid}) + else: + # Non-TUI mode + if agent_uuid == str(coder.uuid): + agent_service.foreground_uuid = None + else: + agent_service.foreground_uuid = agent_uuid + io.tool_output(f"Switched to agent: {agent_name}") + + return format_command_result(io, "switch-agent", f"Switched to agent '{agent_name}'") + + @classmethod + def get_completions(cls, io, coder, args) -> List[str]: + """Get completion options for switch-agent command.""" + try: + agent_service = AgentService.get_instance(coder) + names = [] + + # Determine current foreground agent + foreground_uuid = agent_service.foreground_uuid + + # Add "primary" only if not already on primary + if foreground_uuid is not None: + names.append("primary") + + # Add sub-agent names, excluding the currently active one + if agent_service and agent_service.sub_agents: + for uuid, sub_agent_info in agent_service.sub_agents.items(): + if uuid != foreground_uuid: + name = sub_agent_info.name + # Always include UUID prefix for sub-agents + names.append(f"{name} ({uuid[:3]})") + + current_arg = args.strip().lower() + if current_arg: + return [name for name in names if name.lower().startswith(current_arg)] + else: + return names + except Exception: + return ["primary"] + + @classmethod + def get_help(cls) -> str: + """Get help text for the switch-agent command.""" + help_text = super().get_help() + help_text += "\nUsage:\n" + help_text += " /switch-agent # Switch to a specific agent\n" + help_text += "\nExamples:\n" + help_text += " /switch-agent primary\n" + help_text += " /switch-agent reviewer\n" + help_text += "\nUse tab for auto-completion of agent names.\n" + return help_text diff --git a/cecli/tui/app.py b/cecli/tui/app.py index 375e03b3cc1..d3cd0eb736b 100644 --- a/cecli/tui/app.py +++ b/cecli/tui/app.py @@ -554,6 +554,14 @@ def handle_output_message(self, msg): footer = self.query_one(MainFooter) footer.update_mode(msg.get("mode", "code")) + elif msg_type == "switch_agent": + target_uuid = msg["uuid"] + # Ensure the target container exists before switching + primary_uuid = str(self.worker.coder.uuid) + if target_uuid != primary_uuid and target_uuid not in self._sub_agent_containers: + self.show_error("Agent container not found. Cannot switch.") + else: + self._switch_to_container(target_uuid) def add_output(self, text, task_id=None): """Add output to the output container.""" @@ -678,6 +686,8 @@ def on_input_area_text_changed(self, message: InputArea.TextChanged): def on_input_area_submit(self, message: InputArea.Submit): """Handle input submission.""" + from cecli.helpers.agents.service import AgentService + user_input = message.value if not user_input.strip(): @@ -703,6 +713,63 @@ def on_input_area_submit(self, message: InputArea.Submit): self._open_editor_suspended(initial_content) return + # Intercept /switch-agent command to handle immediately without LLM processing + if stripped.startswith("/switch-agent"): + parts = stripped.split(maxsplit=1) + agent_name = parts[1].strip() if len(parts) > 1 else "" + + input_area = self.query_one("#input", InputArea) + input_area.value = "" + + if not agent_name: + self.show_error("Usage: /switch-agent ") + return + + # Resolve agent name to UUID + agent_service = AgentService.get_instance(self.worker.coder) + primary_uuid = str(self.worker.coder.uuid) + + target_uuid = None + if agent_name == "primary": + target_uuid = primary_uuid + else: + # Try parsing "name (uuid)" format + if agent_name.endswith(")") and " (" in agent_name: + try: + # Extract uuid prefix from "name (prefix)" + uuid_prefix = agent_name.rsplit(" (", 1)[1][:-1] + for uuid, info in agent_service.sub_agents.items(): + if uuid.startswith(uuid_prefix): + target_uuid = uuid + break + except IndexError: + pass # Not the format we expected + + # If not found via "name (uuid)", try matching by name directly + if target_uuid is None: + for uuid, info in agent_service.sub_agents.items(): + if info.name == agent_name: + target_uuid = uuid + break + + # If still not found, try matching by uuid prefix directly + if target_uuid is None: + for uuid, info in agent_service.sub_agents.items(): + if uuid.startswith(agent_name): + target_uuid = uuid + break + + if target_uuid is None: + self.show_error(f"Agent '{agent_name}' not found.") + return + + if target_uuid != primary_uuid and target_uuid not in self._sub_agent_containers: + self.show_error(f"Agent container for '{agent_name}' not found.") + return + + self._switch_to_container(target_uuid) + return + # Save to history before clearing input_area = self.query_one("#input", InputArea) input_area.save_to_history(user_input) @@ -976,6 +1043,12 @@ def _switch_to_container(self, uuid: str) -> None: agent_service = AgentService.get_instance(self.worker.coder) primary_uuid = str(self.worker.coder.uuid) + # Check if the target container exists + if uuid != primary_uuid and uuid not in self._sub_agent_containers: + # Sub-agent container not found, fall back to primary + self.show_error(f"Agent container for UUID {uuid} not found. Switching to primary.") + uuid = primary_uuid + if uuid == primary_uuid: # Switch to primary agent agent_service.foreground_uuid = None diff --git a/cecli/tui/widgets/input_container.py b/cecli/tui/widgets/input_container.py index 850404b3f1b..d4b7a8fa3f7 100644 --- a/cecli/tui/widgets/input_container.py +++ b/cecli/tui/widgets/input_container.py @@ -40,7 +40,7 @@ def update_mode(self, mode: str): sub_agents = self._get_sub_agents() if sub_agents: pills_text = self._format_sub_agent_pills(sub_agents, self.show_squares) - self.border_title = f"{mode}: {pills_text}" + self.border_title = f"agent: {pills_text}" else: self.border_title = mode self.refresh() @@ -49,7 +49,7 @@ def _get_sub_agents(self) -> list: """Query AgentService via self.app to build sub-agent pill data. Returns: - List of dicts with ``name``, ``active``, and ``generating`` keys, + List of dicts with ``name``, ``uuid``, ``active``, and ``generating`` keys, or empty list. """ try: @@ -61,13 +61,14 @@ def _get_sub_agents(self) -> list: agent_service = AgentService.get_instance(coder) sub_agents = [] - primary_uuid = agent_service.coder.uuid + primary_uuid = str(agent_service.coder.uuid) active_uuid = agent_service.foreground_uuid or primary_uuid # Primary is never "generating" in the sub-agent sense sub_agents.append( { "name": "primary", + "uuid": primary_uuid, "active": active_uuid == primary_uuid, "generating": is_active(getattr(coder.io, "output_task", None)), } @@ -78,6 +79,7 @@ def _get_sub_agents(self) -> list: sub_agents.append( { "name": info.name, + "uuid": coder_uuid, "active": coder_uuid == active_uuid, "generating": is_active(info.generate_task), } @@ -101,13 +103,14 @@ def _format_sub_agent_pills(sub_agents: list, show_squares: bool = False) -> str - ◆/■ (generating, active) — alternates for animation Args: - sub_agents: List of dicts with ``name``, ``active``, and ``generating`` keys. + sub_agents: List of dicts with ``name``, ``uuid``, ``active``, and ``generating`` keys. show_squares: If True, use square icons (□/■) instead of diamonds (◇/◆) for generating agents. Returns: - A string like ``"◍ primary ◆ reviewer"``. + A string like ``"◍ primary ◆ reviewer (a6b)"``. """ parts = [] + for sa in sub_agents: active = sa.get("active", False) gen = sa.get("generating", False) @@ -118,7 +121,13 @@ def _format_sub_agent_pills(sub_agents: list, show_squares: bool = False) -> str icon = "◆" if active else "◇" else: icon = "●" if active else "○" - parts.append(f"{icon} {sa['name']}") + + name = sa["name"] + display_name = name + if name != "primary": + display_name = f"{name} ({sa['uuid'][:3]})" + + parts.append(f"{icon} {display_name}") return " ".join(parts) def update_cost(self, cost_text: str): diff --git a/cecli/website/docs/usage/commands.md b/cecli/website/docs/usage/commands.md index 2cf365f15b1..10d06994b65 100644 --- a/cecli/website/docs/usage/commands.md +++ b/cecli/website/docs/usage/commands.md @@ -59,6 +59,7 @@ cog.out(get_help_md()) | **/run** | Run a shell command and optionally add the output to the chat (alias: !) | | **/save** | Save commands to a file that can reconstruct the current chat session's files | | **/settings** | Print out the current settings | +| **/switch-agent** | Switch to a specific agent by name | | **/test** | Run a shell command and add the output to the chat on non-zero exit code | | **/think-tokens** | Set the thinking token budget, eg: 8096, 8k, 10.5k, 0.5M, or 0 to disable. | | **/tokens** | Report on the number of tokens used by the current chat context | diff --git a/tests/commands/test_switch_agent.py b/tests/commands/test_switch_agent.py new file mode 100644 index 00000000000..ae08db69f7d --- /dev/null +++ b/tests/commands/test_switch_agent.py @@ -0,0 +1,104 @@ +from unittest.mock import MagicMock, patch + +import pytest + +from cecli.commands.switch_agent import SwitchAgentCommand + + +@pytest.fixture +def mock_coder(): + coder = MagicMock() + coder.uuid = "primary-uuid" + return coder + + +@pytest.fixture +def mock_io(): + io = MagicMock() + io.output_queue = MagicMock() + return io + + +@pytest.fixture +def mock_agent_service(mock_coder): + with patch("cecli.commands.switch_agent.AgentService") as MockAgentService: + agent_service_instance = MockAgentService.get_instance.return_value + agent_service_instance.sub_agents = { + "sub-uuid-1": MagicMock(name="reviewer"), + } + agent_service_instance.foreground_uuid = None + yield agent_service_instance + + +class TestSwitchAgentCommand: + @pytest.mark.asyncio + async def test_execute_switch_to_sub_agent_tui(self, mock_coder, mock_io, mock_agent_service): + """Test switching to a sub-agent in TUI mode.""" + mock_io.output_queue.put = MagicMock() + + with patch("cecli.commands.switch_agent.hasattr", return_value=True): + await SwitchAgentCommand.execute(mock_io, mock_coder, "reviewer") + + mock_io.output_queue.put.assert_called_once_with( + {"type": "switch_agent", "uuid": "sub-uuid-1"} + ) + + @pytest.mark.asyncio + async def test_execute_switch_to_primary_tui(self, mock_coder, mock_io, mock_agent_service): + """Test switching back to the primary agent in TUI mode.""" + mock_agent_service.foreground_uuid = "sub-uuid-1" + mock_io.output_queue.put = MagicMock() + + with patch("cecli.commands.switch_agent.hasattr", return_value=True): + await SwitchAgentCommand.execute(mock_io, mock_coder, "primary") + + mock_io.output_queue.put.assert_called_once_with( + {"type": "switch_agent", "uuid": "primary-uuid"} + ) + + @pytest.mark.asyncio + async def test_execute_agent_not_found(self, mock_coder, mock_io, mock_agent_service): + """Test error handling when agent is not found.""" + await SwitchAgentCommand.execute(mock_io, mock_coder, "non-existent-agent") + mock_io.tool_error.assert_called_once_with("Error: Agent 'non-existent-agent' not found.") + + @pytest.mark.asyncio + async def test_execute_switch_by_uuid_prefix_tui(self, mock_coder, mock_io, mock_agent_service): + """Test switching to a sub-agent by first 3 UUID chars in TUI mode.""" + mock_io.output_queue.put = MagicMock() + + with patch("cecli.commands.switch_agent.hasattr", return_value=True): + await SwitchAgentCommand.execute(mock_io, mock_coder, "sub") + + mock_io.output_queue.put.assert_called_once_with( + {"type": "switch_agent", "uuid": "sub-uuid-1"} + ) + + def test_get_completions_on_primary(self, mock_coder, mock_io, mock_agent_service): + """Test completions when the primary agent is active.""" + mock_agent_service.foreground_uuid = None + completions = SwitchAgentCommand.get_completions(mock_io, mock_coder, "") + assert "reviewer" in completions + assert "primary" not in completions + + def test_get_completions_on_sub_agent(self, mock_coder, mock_io, mock_agent_service): + """Test completions when a sub-agent is active.""" + mock_agent_service.foreground_uuid = "sub-uuid-1" + completions = SwitchAgentCommand.get_completions(mock_io, mock_coder, "") + assert "primary" in completions + assert "reviewer" not in completions + + def test_get_completions_with_partial_arg(self, mock_coder, mock_io, mock_agent_service): + """Test completions with a partial argument.""" + mock_agent_service.foreground_uuid = None + completions = SwitchAgentCommand.get_completions(mock_io, mock_coder, "rev") + assert completions == ["reviewer"] + + def test_get_completions_with_duplicate_names(self, mock_coder, mock_io, mock_agent_service): + """Test completions include UUID prefixes when there are duplicate names.""" + # Add a second sub-agent with the same name + mock_agent_service.sub_agents["sub-uuid-2"] = MagicMock(name="reviewer") + mock_agent_service.foreground_uuid = None + completions = SwitchAgentCommand.get_completions(mock_io, mock_coder, "") + assert "reviewer (sub)" in completions + assert len([c for c in completions if c.startswith("reviewer")]) == 2