mirror of
https://github.com/instructkr/claw-code.git
synced 2026-05-18 21:41:26 +08:00
Prevent plugin command aliases from becoming prompts
Added focused Python-port regressions for plugin, plugins, marketplace, and reload-plugins command routing, execution, stream, and turn-loop paths; narrowed runtime routing to honor explicit leading commands before fuzzy inventory matches. Constraint: Task 10 scope limited changes to tests/test_porting_workspace.py plus narrow src parser/runtime fixes only if required. Rejected: Test-only coverage without alias routing fix | route /plugin previously preferred fuzzy AddMarketplace over exact plugin command. Confidence: high Scope-risk: narrow Directive: Keep --no-plugin-commands excluding plugin source hints; do not reinterpret that intentional filter as fallthrough. Tested: python3 -m unittest tests.test_porting_workspace; python3 -m compileall src tests/test_porting_workspace.py; CLI route/turn-loop/filter smoke for /plugin, /plugins, /marketplace, /reload-plugins. Not-tested: Full repository non-Python/Rust suites.
This commit is contained in:
@@ -35,6 +35,11 @@ def load_command_snapshot() -> tuple[PortingModule, ...]:
|
|||||||
|
|
||||||
PORTED_COMMANDS = load_command_snapshot()
|
PORTED_COMMANDS = load_command_snapshot()
|
||||||
|
|
||||||
|
COMMAND_ALIASES = {
|
||||||
|
'plugins': 'plugin',
|
||||||
|
'marketplace': 'plugin',
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@lru_cache(maxsize=1)
|
@lru_cache(maxsize=1)
|
||||||
def built_in_command_names() -> frozenset[str]:
|
def built_in_command_names() -> frozenset[str]:
|
||||||
@@ -50,7 +55,7 @@ def command_names() -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
def get_command(name: str) -> PortingModule | None:
|
def get_command(name: str) -> PortingModule | None:
|
||||||
needle = name.lower()
|
needle = COMMAND_ALIASES.get(name.lower(), name.lower())
|
||||||
for module in PORTED_COMMANDS:
|
for module in PORTED_COMMANDS:
|
||||||
if module.name.lower() == needle:
|
if module.name.lower() == needle:
|
||||||
return module
|
return module
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ from __future__ import annotations
|
|||||||
|
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
from .commands import PORTED_COMMANDS
|
from .commands import PORTED_COMMANDS, get_command
|
||||||
from .context import PortContext, build_port_context, render_context
|
from .context import PortContext, build_port_context, render_context
|
||||||
from .history import HistoryLog
|
from .history import HistoryLog
|
||||||
from .models import PermissionDenial, PortingModule
|
from .models import PermissionDenial, PortingModule
|
||||||
@@ -88,6 +88,7 @@ class RuntimeSession:
|
|||||||
|
|
||||||
class PortRuntime:
|
class PortRuntime:
|
||||||
def route_prompt(self, prompt: str, limit: int = 5) -> list[RoutedMatch]:
|
def route_prompt(self, prompt: str, limit: int = 5) -> list[RoutedMatch]:
|
||||||
|
explicit_command = self._explicit_command_match(prompt)
|
||||||
tokens = {token.lower() for token in prompt.replace('/', ' ').replace('-', ' ').split() if token}
|
tokens = {token.lower() for token in prompt.replace('/', ' ').replace('-', ' ').split() if token}
|
||||||
by_kind = {
|
by_kind = {
|
||||||
'command': self._collect_matches(tokens, PORTED_COMMANDS, 'command'),
|
'command': self._collect_matches(tokens, PORTED_COMMANDS, 'command'),
|
||||||
@@ -95,6 +96,16 @@ class PortRuntime:
|
|||||||
}
|
}
|
||||||
|
|
||||||
selected: list[RoutedMatch] = []
|
selected: list[RoutedMatch] = []
|
||||||
|
if explicit_command is not None:
|
||||||
|
selected.append(explicit_command)
|
||||||
|
by_kind['command'] = [
|
||||||
|
match
|
||||||
|
for match in by_kind['command']
|
||||||
|
if not (
|
||||||
|
match.name == explicit_command.name
|
||||||
|
and match.source_hint == explicit_command.source_hint
|
||||||
|
)
|
||||||
|
]
|
||||||
for kind in ('command', 'tool'):
|
for kind in ('command', 'tool'):
|
||||||
if by_kind[kind]:
|
if by_kind[kind]:
|
||||||
selected.append(by_kind[kind].pop(0))
|
selected.append(by_kind[kind].pop(0))
|
||||||
@@ -106,6 +117,22 @@ class PortRuntime:
|
|||||||
selected.extend(leftovers[: max(0, limit - len(selected))])
|
selected.extend(leftovers[: max(0, limit - len(selected))])
|
||||||
return selected[:limit]
|
return selected[:limit]
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _explicit_command_match(prompt: str) -> RoutedMatch | None:
|
||||||
|
first_token = prompt.strip().split(maxsplit=1)[0] if prompt.strip() else ''
|
||||||
|
command_name = first_token.removeprefix('/')
|
||||||
|
if not command_name:
|
||||||
|
return None
|
||||||
|
module = get_command(command_name)
|
||||||
|
if module is None:
|
||||||
|
return None
|
||||||
|
return RoutedMatch(
|
||||||
|
kind='command',
|
||||||
|
name=module.name,
|
||||||
|
source_hint=module.source_hint,
|
||||||
|
score=100,
|
||||||
|
)
|
||||||
|
|
||||||
def bootstrap_session(self, prompt: str, limit: int = 5) -> RuntimeSession:
|
def bootstrap_session(self, prompt: str, limit: int = 5) -> RuntimeSession:
|
||||||
context = build_port_context()
|
context = build_port_context()
|
||||||
setup_report = run_setup(trusted=True)
|
setup_report = run_setup(trusted=True)
|
||||||
|
|||||||
@@ -159,6 +159,80 @@ class PortingWorkspaceTests(unittest.TestCase):
|
|||||||
self.assertIn('Command entries:', command_result.stdout)
|
self.assertIn('Command entries:', command_result.stdout)
|
||||||
self.assertIn('Tool entries:', tool_result.stdout)
|
self.assertIn('Tool entries:', tool_result.stdout)
|
||||||
|
|
||||||
|
def test_plugin_command_filter_excludes_plugin_sources(self) -> None:
|
||||||
|
from src.commands import get_commands
|
||||||
|
|
||||||
|
all_commands = get_commands()
|
||||||
|
filtered_commands = get_commands(include_plugin_commands=False)
|
||||||
|
|
||||||
|
self.assertGreater(len(all_commands), len(filtered_commands))
|
||||||
|
self.assertFalse(
|
||||||
|
any('plugin' in command.source_hint.lower() for command in filtered_commands)
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_plugin_command_aliases_execute_as_local_commands(self) -> None:
|
||||||
|
for alias in ('plugin', 'plugins', 'marketplace'):
|
||||||
|
with self.subTest(alias=alias):
|
||||||
|
result = subprocess.run(
|
||||||
|
[sys.executable, '-m', 'src.main', 'exec-command', alias, f'{alias} list'],
|
||||||
|
check=True,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertIn("Mirrored command 'plugin'", result.stdout)
|
||||||
|
self.assertNotIn('Unknown mirrored command', result.stdout)
|
||||||
|
|
||||||
|
def test_route_plugin_slash_commands_match_commands(self) -> None:
|
||||||
|
prompts = ('/plugin list', '/plugins list', '/marketplace browse', '/reload-plugins')
|
||||||
|
for prompt in prompts:
|
||||||
|
with self.subTest(prompt=prompt):
|
||||||
|
result = subprocess.run(
|
||||||
|
[sys.executable, '-m', 'src.main', 'route', prompt, '--limit', '5'],
|
||||||
|
check=True,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
first_line = result.stdout.splitlines()[0]
|
||||||
|
self.assertTrue(first_line.startswith('command\t'), result.stdout)
|
||||||
|
self.assertRegex(first_line, r'\t(plugin|reload-plugins)\t')
|
||||||
|
|
||||||
|
def test_plugin_command_stream_emits_command_match(self) -> None:
|
||||||
|
from src.runtime import PortRuntime
|
||||||
|
|
||||||
|
for prompt in ('/plugin list', '/plugins list', '/marketplace browse', '/reload-plugins'):
|
||||||
|
with self.subTest(prompt=prompt):
|
||||||
|
session = PortRuntime().bootstrap_session(prompt, limit=5)
|
||||||
|
command_events = [
|
||||||
|
event for event in session.stream_events if event['type'] == 'command_match'
|
||||||
|
]
|
||||||
|
|
||||||
|
self.assertTrue(command_events, session.as_markdown())
|
||||||
|
self.assertNotIn('Matched commands: none', session.turn_result.output)
|
||||||
|
|
||||||
|
def test_turn_loop_plugin_commands_are_not_prompt_only(self) -> None:
|
||||||
|
for prompt in ('/plugin list', '/plugins list', '/marketplace browse', '/reload-plugins'):
|
||||||
|
with self.subTest(prompt=prompt):
|
||||||
|
result = subprocess.run(
|
||||||
|
[
|
||||||
|
sys.executable,
|
||||||
|
'-m',
|
||||||
|
'src.main',
|
||||||
|
'turn-loop',
|
||||||
|
prompt,
|
||||||
|
'--max-turns',
|
||||||
|
'1',
|
||||||
|
'--structured-output',
|
||||||
|
],
|
||||||
|
check=True,
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertIn('"Matched commands:', result.stdout)
|
||||||
|
self.assertNotIn('Matched commands: none', result.stdout)
|
||||||
|
|
||||||
def test_load_session_cli_runs(self) -> None:
|
def test_load_session_cli_runs(self) -> None:
|
||||||
from src.runtime import PortRuntime
|
from src.runtime import PortRuntime
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user