Command execution engine.
| 93 | |
| 94 | @dataclass |
| 95 | class CommandEngine: |
| 96 | """Command execution engine.""" |
| 97 | |
| 98 | registry: CommandRegistry |
| 99 | workspace_root: Path |
| 100 | context: CommandContext |
| 101 | _command_hooks: list[Callable[[str, CommandResult], None]] = field( |
| 102 | default_factory=list |
| 103 | ) |
| 104 | |
| 105 | async def execute( |
| 106 | self, |
| 107 | command_input: str, |
| 108 | ) -> CommandResult: |
| 109 | """ |
| 110 | Execute a command. |
| 111 | |
| 112 | Args: |
| 113 | command_input: Command input (e.g., "/help args") |
| 114 | |
| 115 | Returns: |
| 116 | CommandResult with the execution result |
| 117 | """ |
| 118 | # Parse command and args |
| 119 | if not command_input.startswith("/"): |
| 120 | return CommandResult.error( |
| 121 | "", |
| 122 | "Commands must start with '/'", |
| 123 | ) |
| 124 | |
| 125 | parts = command_input[1:].split(maxsplit=1) |
| 126 | command_name = parts[0].strip() |
| 127 | args = parts[1].strip() if len(parts) > 1 else "" |
| 128 | |
| 129 | # Get command |
| 130 | command = self.registry.get(command_name) |
| 131 | if command is None: |
| 132 | return CommandResult.error( |
| 133 | command_name, |
| 134 | f"Unknown command: {command_name}", |
| 135 | ) |
| 136 | |
| 137 | # Check if command is enabled |
| 138 | if not command.is_enabled(): |
| 139 | return CommandResult.error( |
| 140 | command_name, |
| 141 | f"Command {command_name} is disabled", |
| 142 | ) |
| 143 | |
| 144 | # Execute based on type |
| 145 | result: CommandResult |
| 146 | if command.command_type == CommandType.LOCAL: |
| 147 | result = await self._execute_local(command, args) |
| 148 | elif command.command_type == CommandType.PROMPT: |
| 149 | result = await self._execute_prompt(command, args) |
| 150 | elif command.command_type == CommandType.INTERACTIVE: |
| 151 | result = await self._execute_interactive(command, args) |
| 152 | else: |
no outgoing calls