"""Tests for slash command handlers.""" from __future__ import annotations import subprocess from pathlib import Path import pytest import openharness.commands.registry as registry_module from openharness.commands.registry import ( CommandContext, MemoryCommandBackend, create_default_command_registry, lookup_skill_slash_command, ) from openharness.autopilot import RepoVerificationStep from openharness.config.paths import get_feedback_log_path, get_project_issue_file, get_project_pr_comments_file from openharness.config.settings import load_settings, save_settings, Settings from openharness.engine.messages import ConversationMessage, TextBlock from openharness.engine.query_engine import QueryEngine from openharness.memory.paths import get_project_memory_dir from openharness.mcp.types import McpHttpServerConfig, McpStdioServerConfig from openharness.permissions import PermissionChecker from openharness.plugins.types import PluginCommandDefinition from openharness.state import AppState, AppStateStore from openharness.tasks import get_task_manager from openharness.tools import create_default_tool_registry class FakeApiClient: async def stream_message(self, request): del request raise AssertionError("stream_message should not be called in command tests") def _make_engine(tmp_path: Path) -> QueryEngine: return QueryEngine( api_client=FakeApiClient(), tool_registry=create_default_tool_registry(), permission_checker=PermissionChecker(load_settings().permission), cwd=tmp_path, model="claude-test", system_prompt="system", ) def _make_context(tmp_path: Path) -> CommandContext: tool_registry = create_default_tool_registry() return CommandContext( engine=QueryEngine( api_client=FakeApiClient(), tool_registry=tool_registry, permission_checker=PermissionChecker(load_settings().permission), cwd=tmp_path, model="claude-test", system_prompt="system", ), cwd=str(tmp_path), tool_registry=tool_registry, app_state=AppStateStore( AppState( model="claude-test", permission_mode="default", theme="default", keybindings={}, ) ), ) @pytest.mark.asyncio async def test_permissions_command_persists(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/permissions set full_auto") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "Auto" in result.message assert load_settings().permission.mode == "full_auto" @pytest.mark.asyncio async def test_permissions_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/permissions set full_auto") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_permissions_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/permissions set full_auto") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_stop_command_explains_interrupt_paths(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/stop") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "/stop" in result.message assert "Esc/Ctrl+C" in result.message @pytest.mark.asyncio async def test_plugin_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/plugin list") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_plugin_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/plugin list") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_reload_plugins_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/reload-plugins") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_reload_plugins_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/reload-plugins") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_bridge_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/bridge spawn id") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_bridge_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/bridge spawn id") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_autopilot_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/autopilot run-next") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_autopilot_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/autopilot run-next") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_diff_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/diff full") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_diff_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/diff full") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_project_context_commands_are_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() for payload in ( "/issue set Remote supplied issue :: marker", "/pr_comments add src/app.py:1 :: marker", ): command, _ = registry.lookup(payload) assert command is not None assert command.remote_invocable is False, payload @pytest.mark.asyncio async def test_project_context_commands_support_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() for payload in ( "/issue set Remote supplied issue :: marker", "/pr_comments add src/app.py:1 :: marker", ): command, _ = registry.lookup(payload) assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True, payload @pytest.mark.asyncio async def test_tasks_command_is_marked_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/tasks run id") assert command is not None assert command.remote_invocable is False @pytest.mark.asyncio async def test_tasks_command_supports_explicit_remote_admin_opt_in(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, _ = registry.lookup("/tasks run id") assert command is not None assert getattr(command, "remote_admin_opt_in", False) is True @pytest.mark.asyncio async def test_sensitive_control_plane_commands_are_local_only(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() for payload in ( "/config show", "/login TEST_KEY", "/logout", "/mcp", "/provider", "/model show", "/commit remote requested commit", "/ship", "/resume", "/resume session-from-another-sender", "/summary 10", ): command, _ = registry.lookup(payload) assert command is not None assert command.remote_invocable is False, payload assert command.remote_admin_opt_in is True, payload @pytest.mark.asyncio async def test_config_show_redacts_nested_mcp_and_vision_secrets(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) settings = Settings( api_key="TOP_LEVEL_FAKE_SECRET", mcp_servers={ "internal-http": McpHttpServerConfig( url="https://mcp.internal", headers={ "Authorization": "Bearer MCP_FAKE_SECRET", "X-Token": "RAW_FAKE_TOKEN", "X-Public": "non-secret-value", }, ), "local-stdio": McpStdioServerConfig( command="server", env={ "MCP_AUTH_TOKEN": "STDIO_FAKE_SECRET", "SAFE_SETTING": "visible", }, ), }, vision={ "model": "vision-test", "api_key": "VISION_FAKE_SECRET", "base_url": "https://vision.example", }, ) monkeypatch.setattr(registry_module, "load_settings", lambda: settings) registry = create_default_command_registry() command, args = registry.lookup("/config show") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) for secret in ( "TOP_LEVEL_FAKE_SECRET", "MCP_FAKE_SECRET", "RAW_FAKE_TOKEN", "STDIO_FAKE_SECRET", "VISION_FAKE_SECRET", ): assert secret not in result.message assert "[REDACTED]" in result.message assert "non-secret-value" in result.message assert "visible" in result.message @pytest.mark.asyncio async def test_memory_show_rejects_path_traversal(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() command, args = registry.lookup("/memory show ../../../../../../etc/hosts") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert result.message == "Memory entry path must stay within the configured memory directory." @pytest.mark.asyncio async def test_memory_show_reads_normal_entries_with_md_fallback(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() add_command, add_args = registry.lookup("/memory add Notes :: hello world") assert add_command is not None await add_command.handler(add_args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) show_command, show_args = registry.lookup("/memory show Notes") result = await show_command.handler(show_args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "hello world" in result.message @pytest.mark.asyncio async def test_model_command_persists(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/model opus") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "opus" in result.message assert load_settings().resolve_profile()[1].last_model == "opus" assert load_settings().model == "claude-opus-4-6" @pytest.mark.asyncio async def test_model_command_accepts_direct_value(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/model gpt-5.4") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "gpt-5.4" in result.message assert load_settings().model == "gpt-5.4" @pytest.mark.asyncio async def test_model_command_lists_profile_model_allowlist(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "local-llm", "provider": "openai", "api_format": "openai", "base_url": "http://localhost:8000/v1", "model": "deepseek-chat", "profiles": { "local-llm": { "label": "Local LLM", "provider": "openai", "api_format": "openai", "auth_source": "openai_api_key", "default_model": "deepseek-chat", "last_model": "deepseek-chat", "base_url": "http://localhost:8000/v1", "allowed_models": ["deepseek-chat", "qwen-vl"], } }, } ) ) registry = create_default_command_registry() command, args = registry.lookup("/model list") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "Switchable models for profile 'local-llm'" in result.message assert "- deepseek-chat" in result.message assert "- qwen-vl" in result.message @pytest.mark.asyncio async def test_model_command_adds_model_to_profile_allowlist(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "local-llm", "provider": "openai", "api_format": "openai", "base_url": "http://localhost:8000/v1", "model": "deepseek-chat", "profiles": { "local-llm": { "label": "Local LLM", "provider": "openai", "api_format": "openai", "auth_source": "openai_api_key", "default_model": "deepseek-chat", "last_model": "deepseek-chat", "base_url": "http://localhost:8000/v1", "allowed_models": ["deepseek-chat"], } }, } ) ) registry = create_default_command_registry() command, args = registry.lookup("/model add qwen-vl") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert result.refresh_runtime is True assert load_settings().resolve_profile()[1].allowed_models == ["deepseek-chat", "qwen-vl"] @pytest.mark.asyncio async def test_model_command_remove_current_model_resets_to_default(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "local-llm", "provider": "openai", "api_format": "openai", "base_url": "http://localhost:8000/v1", "model": "qwen-vl", "profiles": { "local-llm": { "label": "Local LLM", "provider": "openai", "api_format": "openai", "auth_source": "openai_api_key", "default_model": "deepseek-chat", "last_model": "qwen-vl", "base_url": "http://localhost:8000/v1", "allowed_models": ["deepseek-chat", "qwen-vl"], } }, } ) ) registry = create_default_command_registry() context = _make_context(tmp_path) command, args = registry.lookup("/model remove qwen-vl") assert command is not None result = await command.handler(args, context) profile = load_settings().resolve_profile()[1] assert result.refresh_runtime is True assert profile.allowed_models == ["deepseek-chat"] assert profile.last_model == "" assert context.engine.model == "deepseek-chat" @pytest.mark.asyncio async def test_model_command_clear_removes_profile_allowlist(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "local-llm", "provider": "openai", "api_format": "openai", "base_url": "http://localhost:8000/v1", "model": "deepseek-chat", "profiles": { "local-llm": { "label": "Local LLM", "provider": "openai", "api_format": "openai", "auth_source": "openai_api_key", "default_model": "deepseek-chat", "last_model": "deepseek-chat", "base_url": "http://localhost:8000/v1", "allowed_models": ["deepseek-chat", "qwen-vl"], } }, } ) ) registry = create_default_command_registry() command, args = registry.lookup("/model clear") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert result.refresh_runtime is True assert load_settings().resolve_profile()[1].allowed_models == [] @pytest.mark.asyncio async def test_model_command_default_clears_profile_override(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "claude-api", "profiles": { "claude-api": { "label": "Claude API", "provider": "anthropic", "api_format": "anthropic", "auth_source": "anthropic_api_key", "default_model": "sonnet", "last_model": "opus", } }, } ) ) registry = create_default_command_registry() command, args = registry.lookup("/model default") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "reset to default" in result.message assert load_settings().resolve_profile()[1].last_model == "" assert load_settings().model == "claude-sonnet-4-6" @pytest.mark.asyncio async def test_turns_show_reports_unlimited_engine_when_session_is_unbounded(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) context.engine.set_max_turns(None) command, args = registry.lookup("/turns show") assert command is not None result = await command.handler(args, context) assert "Max turns (engine): unlimited" in result.message @pytest.mark.asyncio async def test_turns_command_accepts_unlimited(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) command, args = registry.lookup("/turns unlimited") assert command is not None result = await command.handler(args, context) assert "unlimited for this session" in result.message assert context.engine.max_turns is None @pytest.mark.asyncio async def test_provider_command_switches_profile_and_requests_runtime_refresh(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "profiles": { "kimi-anthropic": { "label": "Kimi Anthropic", "provider": "anthropic", "api_format": "anthropic", "auth_source": "anthropic_api_key", "default_model": "kimi-k2.5", "last_model": "kimi-k2.5", "base_url": "https://api.moonshot.cn/anthropic", "allowed_models": ["kimi-k2.5"], } } } ) ) registry = create_default_command_registry() context = _make_context(tmp_path) command, args = registry.lookup("/provider kimi-anthropic") assert command is not None result = await command.handler(args, context) loaded = load_settings() assert result.refresh_runtime is True assert loaded.active_profile == "kimi-anthropic" assert loaded.base_url == "https://api.moonshot.cn/anthropic" assert loaded.model == "kimi-k2.5" @pytest.mark.asyncio async def test_autopilot_command_add_list_and_complete(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) add_command, add_args = registry.lookup("/autopilot add idea Build unified queue :: intake from issues and prs") assert add_command is not None add_result = await add_command.handler(add_args, context) assert "Queued autopilot card" in add_result.message list_command, list_args = registry.lookup("/autopilot list") assert list_command is not None list_result = await list_command.handler(list_args, context) assert "Build unified queue" in list_result.message next_command, next_args = registry.lookup("/autopilot next") next_result = await next_command.handler(next_args, context) assert "Build unified queue" in next_result.message card_id = next_result.message.split()[0] complete_command, complete_args = registry.lookup(f"/autopilot complete {card_id} implemented") complete_result = await complete_command.handler(complete_args, context) assert "-> completed" in complete_result.message status_command, status_args = registry.lookup("/autopilot status") status_result = await status_command.handler(status_args, context) assert "- completed: 1" in status_result.message @pytest.mark.asyncio async def test_autopilot_command_export_dashboard(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) command, args = registry.lookup("/autopilot export-dashboard") assert command is not None result = await command.handler(args, context) assert "Exported autopilot dashboard:" in result.message output_dir = Path(result.message.split(": ", 1)[1]) assert (output_dir / "index.html").exists() assert (output_dir / "snapshot.json").exists() @pytest.mark.asyncio async def test_ship_command_queues_and_executes_card(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) async def fake_run_agent_prompt(self, prompt: str, *, model, max_turns, permission_mode, cwd=None): return "Implemented the requested feature." def fake_run_verification_steps(self, policies, *, cwd=None): return [RepoVerificationStep(command="uv run pytest -q", returncode=0, status="success")] monkeypatch.setattr( "openharness.autopilot.service.RepoAutopilotStore._run_agent_prompt", fake_run_agent_prompt, ) monkeypatch.setattr( "openharness.autopilot.service.RepoAutopilotStore._run_verification_steps", fake_run_verification_steps, ) command, args = registry.lookup("/ship Build autopilot tick :: end-to-end automation") assert command is not None result = await command.handler(args, context) assert "-> completed" in result.message assert "run report:" in result.message assert "verification report:" in result.message @pytest.mark.asyncio async def test_plugin_command_registers_and_submits_prompt(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry( plugin_commands=[ PluginCommandDefinition( name="fixture:ops:restart", description="Restart services safely", content="Base workflow\n\n$ARGUMENTS", user_invocable=True, ) ] ) command, args = registry.lookup("/fixture:ops:restart api") assert command is not None result = await command.handler(args, _make_context(tmp_path)) assert result.submit_prompt == "Base workflow\n\napi" @pytest.mark.asyncio async def test_bundled_user_invocable_skill_registers_as_slash_command(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/skill-creator create a deployment skill") assert command is not None assert command.name == "skill-creator" result = await command.handler(args, _make_context(tmp_path)) assert result.submit_prompt is not None assert "Base directory for this skill:" in result.submit_prompt assert "# skill-creator" in result.submit_prompt assert "Arguments: create a deployment skill" in result.submit_prompt @pytest.mark.asyncio async def test_context_skill_slash_command_uses_folder_name(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) extra_root = tmp_path / "ohmo-skills" skill_dir = extra_root / "pikastream-video-meeting" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\n" "name: PikaStream Video Meeting\n" "description: Join and summarize video meetings.\n" "---\n\n" "# PikaStream Video Meeting\n\n" "Use the meeting workflow.\n", encoding="utf-8", ) context = _make_context(tmp_path) context.extra_skill_dirs = (extra_root,) parsed = lookup_skill_slash_command("/pikastream-video-meeting room 123", context) assert parsed is not None command, args = parsed result = await command.handler(args, context) assert command.name == "pikastream-video-meeting" assert result.submit_prompt is not None assert f"Base directory for this skill: {skill_dir.resolve()}" in result.submit_prompt assert "PikaStream Video Meeting" in result.submit_prompt assert "Arguments: room 123" in result.submit_prompt @pytest.mark.asyncio async def test_user_invocable_false_skill_is_not_slash_resolved(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) extra_root = tmp_path / "skills" skill_dir = extra_root / "hidden" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\n" "description: Model-only helper.\n" "user-invocable: false\n" "---\n\n" "# Hidden\n", encoding="utf-8", ) context = _make_context(tmp_path) context.extra_skill_dirs = (extra_root,) assert lookup_skill_slash_command("/hidden", context) is None @pytest.mark.asyncio async def test_project_skill_registers_as_context_slash_command(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) repo = tmp_path / "repo" repo.mkdir() skill_dir = repo / ".claude" / "skills" / "shipit" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\n" "description: Project ship workflow.\n" "---\n\n" "# Shipit\n\nShip this repo.\n", encoding="utf-8", ) context = _make_context(repo) parsed = lookup_skill_slash_command("/shipit now", context) assert parsed is not None command, args = parsed result = await command.handler(args, context) assert command.name == "shipit" assert result.submit_prompt is not None assert f"Base directory for this skill: {skill_dir.resolve()}" in result.submit_prompt assert "Arguments: now" in result.submit_prompt @pytest.mark.asyncio async def test_skills_command_lists_project_skill_path(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) repo = tmp_path / "repo" repo.mkdir() skill_dir = repo / ".agents" / "skills" / "triage" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text("# Triage\nTriage workflow.\n", encoding="utf-8") registry = create_default_command_registry() command, args = registry.lookup("/skills") assert command is not None result = await command.handler(args, _make_context(repo)) assert "triage (Triage) [project]" in result.message assert str((skill_dir / "SKILL.md").resolve()) in result.message @pytest.mark.asyncio async def test_disable_model_invocation_skill_still_allows_user_slash(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) extra_root = tmp_path / "skills" skill_dir = extra_root / "deploy" skill_dir.mkdir(parents=True) (skill_dir / "SKILL.md").write_text( "---\n" "description: User-triggered deploy workflow.\n" "disable-model-invocation: true\n" "model: gpt-5.4\n" "---\n\n" "# Deploy\n\n$ARGUMENTS\n", encoding="utf-8", ) context = _make_context(tmp_path) context.extra_skill_dirs = (extra_root,) parsed = lookup_skill_slash_command("/deploy staging", context) assert parsed is not None command, args = parsed result = await command.handler(args, context) assert result.submit_prompt is not None assert result.submit_model == "gpt-5.4" assert "staging" in result.submit_prompt @pytest.mark.asyncio async def test_model_command_rejects_values_outside_profile_allowlist(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) save_settings( Settings().model_copy( update={ "active_profile": "kimi-anthropic", "profiles": { "kimi-anthropic": { "label": "Kimi Anthropic", "provider": "anthropic", "api_format": "anthropic", "auth_source": "anthropic_api_key", "default_model": "kimi-k2.5", "last_model": "kimi-k2.5", "base_url": "https://api.moonshot.cn/anthropic", "allowed_models": ["kimi-k2.5"], } }, } ) ) registry = create_default_command_registry() command, args = registry.lookup("/model claude-opus-4-6") assert command is not None result = await command.handler(args, CommandContext(engine=_make_engine(tmp_path), cwd=str(tmp_path))) assert "is not allowed for profile 'kimi-anthropic'" in result.message @pytest.mark.asyncio async def test_doctor_command_reports_context(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() command, args = registry.lookup("/doctor") assert command is not None result = await command.handler( args, CommandContext( engine=_make_engine(tmp_path), cwd=str(tmp_path), plugin_summary="Plugins:\n- demo [enabled] Example", mcp_summary="No MCP servers configured.", ), ) assert "Doctor summary:" in result.message assert str(tmp_path) in result.message @pytest.mark.asyncio async def test_memory_command_manages_entries(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) add_command, add_args = registry.lookup("/memory add Pytest Tips :: use fixtures") add_result = await add_command.handler(add_args, context) assert "Added memory entry" in add_result.message list_command, list_args = registry.lookup("/memory list") list_result = await list_command.handler(list_args, context) assert "pytest_tips.md" in list_result.message show_command, show_args = registry.lookup("/memory show pytest_tips") show_result = await show_command.handler(show_args, context) assert "use fixtures" in show_result.message remove_command, remove_args = registry.lookup("/memory remove pytest_tips") remove_result = await remove_command.handler(remove_args, context) assert "Removed memory entry" in remove_result.message list_after_remove = await list_command.handler(list_args, context) assert "pytest_tips.md" not in list_after_remove.message show_after_remove = await show_command.handler(show_args, context) assert show_after_remove.message == "Memory entry not found: pytest_tips" @pytest.mark.asyncio async def test_memory_command_migrates_entries(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) memory_dir = get_project_memory_dir(tmp_path) legacy = memory_dir / "legacy.md" legacy.write_text("legacy command note\n", encoding="utf-8") registry = create_default_command_registry() context = _make_context(tmp_path) dry_command, dry_args = registry.lookup("/memory migrate --dry-run") dry_result = await dry_command.handler(dry_args, context) assert "Memory migration dry run." in dry_result.message assert "Changed: 1" in dry_result.message assert "schema_version" not in legacy.read_text(encoding="utf-8") apply_command, apply_args = registry.lookup("/memory migrate --apply") apply_result = await apply_command.handler(apply_args, context) assert "Memory migration applied." in apply_result.message assert "Backup:" in apply_result.message assert "schema_version: 1" in legacy.read_text(encoding="utf-8") @pytest.mark.asyncio async def test_memory_migrate_uses_backend_defaults_not_label(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) memory_dir = tmp_path / "custom-memory" memory_dir.mkdir() legacy = memory_dir / "legacy.md" legacy.write_text("personal preference note\n", encoding="utf-8") registry = create_default_command_registry() context = _make_context(tmp_path) context.memory_backend = MemoryCommandBackend( label="renamed user notes", default_type="personal", default_category="preference", get_memory_dir=lambda: memory_dir, get_entrypoint=lambda: memory_dir / "MEMORY.md", list_files=lambda: [], add_entry=lambda title, content: memory_dir / f"{title}.md", remove_entry=lambda name: False, ) apply_command, apply_args = registry.lookup("/memory migrate --apply") result = await apply_command.handler(apply_args, context) migrated = legacy.read_text(encoding="utf-8") assert "Memory migration applied." in result.message assert 'type: "personal"' in migrated assert 'category: "preference"' in migrated @pytest.mark.asyncio async def test_compact_summary_and_usage_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) context.engine.load_messages( [ ConversationMessage(role="user", content=[TextBlock(text="alpha request")]), ConversationMessage(role="assistant", content=[TextBlock(text="alpha reply")]), ConversationMessage(role="user", content=[TextBlock(text="beta request")]), ConversationMessage(role="assistant", content=[TextBlock(text="beta reply")]), ] ) summary_command, summary_args = registry.lookup("/summary 3") summary_result = await summary_command.handler(summary_args, context) assert "assistant: alpha reply" in summary_result.message or "user: beta request" in summary_result.message compact_command, compact_args = registry.lookup("/compact 2") compact_result = await compact_command.handler(compact_args, context) assert "Compacted conversation" in compact_result.message assert len(context.engine.messages) == 3 usage_command, usage_args = registry.lookup("/usage") usage_result = await usage_command.handler(usage_args, context) assert "Estimated conversation tokens" in usage_result.message stats_command, stats_args = registry.lookup("/stats") stats_result = await stats_command.handler(stats_args, context) assert "Session stats:" in stats_result.message @pytest.mark.asyncio async def test_ui_mode_commands_persist_and_update_state(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) registry = create_default_command_registry() context = _make_context(tmp_path) config_command, config_args = registry.lookup("/config set verbose true") config_result = await config_command.handler(config_args, context) assert "Updated verbose" in config_result.message assert load_settings().verbose is True output_command, output_args = registry.lookup("/output-style set minimal") output_result = await output_command.handler(output_args, context) assert "minimal" in output_result.message assert context.app_state.get().output_style == "minimal" keybindings_command, keybindings_args = registry.lookup("/keybindings") keybindings_result = await keybindings_command.handler(keybindings_args, context) assert "ctrl+l" in keybindings_result.message vim_command, vim_args = registry.lookup("/vim toggle") vim_result = await vim_command.handler(vim_args, context) assert "enabled" in vim_result.message assert context.app_state.get().vim_enabled is True voice_command, voice_args = registry.lookup("/voice keyterms Shipping pytest fixtures") voice_result = await voice_command.handler(voice_args, context) assert "pytest" in voice_result.message plan_command, plan_args = registry.lookup("/plan on") plan_result = await plan_command.handler(plan_args, context) assert "enabled" in plan_result.message assert load_settings().permission.mode == "plan" fast_command, fast_args = registry.lookup("/fast on") fast_result = await fast_command.handler(fast_args, context) assert "enabled" in fast_result.message assert load_settings().fast_mode is True assert context.app_state.get().fast_mode is True effort_command, effort_args = registry.lookup("/effort high") effort_result = await effort_command.handler(effort_args, context) assert "high" in effort_result.message assert load_settings().effort == "high" assert context.app_state.get().effort == "high" effort_command, effort_args = registry.lookup("/effort xhigh") effort_result = await effort_command.handler(effort_args, context) assert "xhigh" in effort_result.message assert load_settings().effort == "xhigh" assert context.app_state.get().effort == "xhigh" passes_command, passes_args = registry.lookup("/passes 3") passes_result = await passes_command.handler(passes_args, context) assert "3" in passes_result.message assert load_settings().passes == 3 assert context.app_state.get().passes == 3 @pytest.mark.asyncio async def test_version_context_and_share_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) version_command, version_args = registry.lookup("/version") version_result = await version_command.handler(version_args, context) assert "OpenHarness" in version_result.message context_command, context_args = registry.lookup("/context") context_result = await context_command.handler(context_args, context) assert "OpenHarness" in context_result.message or "interactive agent" in context_result.message share_command, share_args = registry.lookup("/share") share_result = await share_command.handler(share_args, context) assert "shareable transcript snapshot" in share_result.message @pytest.mark.asyncio async def test_auth_feedback_and_project_context_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) # Prevent env var leakage from overriding the configured api_key monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) registry = create_default_command_registry() context = _make_context(tmp_path) login_command, login_args = registry.lookup("/login sk-test-123456") login_result = await login_command.handler(login_args, context) assert "Stored API key" in login_result.message assert load_settings().api_key == "sk-test-123456" issue_command, issue_args = registry.lookup("/issue set Fix CI :: The CI flakes on task retry") issue_result = await issue_command.handler(issue_args, context) assert "Saved issue context" in issue_result.message assert "Fix CI" in get_project_issue_file(tmp_path).read_text(encoding="utf-8") pr_command, pr_args = registry.lookup("/pr_comments add src/app.py:12 :: simplify this branch") pr_result = await pr_command.handler(pr_args, context) assert "Added PR comment" in pr_result.message assert "simplify this branch" in get_project_pr_comments_file(tmp_path).read_text(encoding="utf-8") feedback_command, feedback_args = registry.lookup("/feedback this workflow feels good") feedback_result = await feedback_command.handler(feedback_args, context) assert "Saved feedback" in feedback_result.message assert "this workflow feels good" in get_feedback_log_path().read_text(encoding="utf-8") logout_command, logout_args = registry.lookup("/logout") logout_result = await logout_command.handler(logout_args, context) assert "Cleared stored API key" in logout_result.message assert load_settings().api_key == "" @pytest.mark.asyncio async def test_agents_session_files_and_reload_plugins_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) (tmp_path / "src").mkdir() (tmp_path / "src" / "app.py").write_text("print('hi')\n", encoding="utf-8") session_command, session_args = registry.lookup("/session") context.session_id = "session-smoke" session_result = await session_command.handler(session_args, context) assert "Session ID: session-smoke" in session_result.message assert "Session directory:" in session_result.message session_path_command, session_path_args = registry.lookup("/session path") session_path_result = await session_path_command.handler(session_path_args, context) assert "sessions" in session_path_result.message session_tag_command, session_tag_args = registry.lookup("/session tag smoke") session_tag_result = await session_tag_command.handler(session_tag_args, context) assert "smoke.json" in session_tag_result.message assert "smoke.md" in session_tag_result.message tag_command, tag_args = registry.lookup("/tag alias-smoke") tag_result = await tag_command.handler(tag_args, context) assert "alias-smoke.json" in tag_result.message assert "alias-smoke.md" in tag_result.message files_command, files_args = registry.lookup("/files app.py") files_result = await files_command.handler(files_args, context) assert "src/app.py" in files_result.message.replace("\\", "/") files_dirs_command, files_dirs_args = registry.lookup("/files dirs") files_dirs_result = await files_dirs_command.handler(files_dirs_args, context) assert "src" in files_dirs_result.message plugin_root = tmp_path / "config" / "plugins" / "fixture-plugin" (plugin_root / "skills").mkdir(parents=True) (plugin_root / "plugin.json").write_text( '{"name":"fixture-plugin","version":"1.0.0","description":"Fixture plugin"}', encoding="utf-8", ) reload_command, reload_args = registry.lookup("/reload-plugins") reload_result = await reload_command.handler(reload_args, context) assert "fixture-plugin" in reload_result.message manager = get_task_manager() task = await manager.create_agent_task( prompt="ready", description="test agent", cwd=tmp_path, command="python -u -c \"import sys; print(sys.stdin.readline().strip())\"", ) agents_command, agents_args = registry.lookup("/agents") agents_result = await agents_command.handler(agents_args, context) assert task.id in agents_result.message agent_show_command, agent_show_args = registry.lookup(f"/agents show {task.id}") agent_show_result = await agent_show_command.handler(agent_show_args, context) assert "test agent" in agent_show_result.message @pytest.mark.asyncio async def test_agents_help_and_subagents_alias(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) agents_help_command, agents_help_args = registry.lookup("/agents help") assert agents_help_command is not None agents_help_result = await agents_help_command.handler(agents_help_args, context) assert "Subagent guide:" in agents_help_result.message assert 'subagent_type="worker"' in agents_help_result.message subagents_command, subagents_args = registry.lookup("/subagents") assert subagents_command is not None subagents_result = await subagents_command.handler(subagents_args, context) assert "Subagent guide:" in subagents_result.message agents_command, agents_args = registry.lookup("/agents") assert agents_command is not None agents_result = await agents_command.handler(agents_args, context) assert "Subagent guide:" in agents_result.message @pytest.mark.asyncio async def test_init_and_bridge_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) init_command, init_args = registry.lookup("/init") init_result = await init_command.handler(init_args, context) assert "Initialized project files" in init_result.message or "already initialized" in init_result.message assert (tmp_path / "CLAUDE.md").exists() assert (tmp_path / ".openharness" / "memory" / "MEMORY.md").exists() bridge_show_command, bridge_show_args = registry.lookup("/bridge show") bridge_show_result = await bridge_show_command.handler(bridge_show_args, context) assert "Bridge summary:" in bridge_show_result.message bridge_encode_command, bridge_encode_args = registry.lookup("/bridge encode https://api.example.com token123") bridge_encode_result = await bridge_encode_command.handler(bridge_encode_args, context) assert bridge_encode_result.message bridge_decode_command, bridge_decode_args = registry.lookup(f"/bridge decode {bridge_encode_result.message}") bridge_decode_result = await bridge_decode_command.handler(bridge_decode_args, context) assert "api.example.com" in bridge_decode_result.message bridge_sdk_command, bridge_sdk_args = registry.lookup("/bridge sdk https://api.example.com session123") bridge_sdk_result = await bridge_sdk_command.handler(bridge_sdk_args, context) assert "session123" in bridge_sdk_result.message bridge_spawn_command, bridge_spawn_args = registry.lookup("/bridge spawn printf bridge-ok") bridge_spawn_result = await bridge_spawn_command.handler(bridge_spawn_args, context) assert "Spawned bridge session" in bridge_spawn_result.message session_id = bridge_spawn_result.message.split()[3] bridge_list_command, bridge_list_args = registry.lookup("/bridge list") bridge_list_result = await bridge_list_command.handler(bridge_list_args, context) assert session_id in bridge_list_result.message bridge_output_command, bridge_output_args = registry.lookup(f"/bridge output {session_id}") bridge_output_result = await bridge_output_command.handler(bridge_output_args, context) assert "bridge-ok" in bridge_output_result.message or bridge_output_result.message == "(no output)" bridge_stop_command, bridge_stop_args = registry.lookup(f"/bridge stop {session_id}") bridge_stop_result = await bridge_stop_command.handler(bridge_stop_args, context) assert f"Stopped bridge session {session_id}" in bridge_stop_result.message @pytest.mark.asyncio async def test_copy_rewind_and_meta_commands(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) registry = create_default_command_registry() context = _make_context(tmp_path) context.engine.load_messages( [ ConversationMessage.from_user_text("first prompt"), ConversationMessage(role="assistant", content=[TextBlock(text="first answer")]), ConversationMessage.from_user_text("second prompt"), ConversationMessage(role="assistant", content=[TextBlock(text="second answer")]), ] ) copied: list[str] = [] def _fake_copy(text: str) -> None: copied.append(text) monkeypatch.setattr(registry_module.pyperclip, "copy", _fake_copy) copy_command, copy_args = registry.lookup("/copy") copy_result = await copy_command.handler(copy_args, context) assert "Copied" in copy_result.message assert copied == ["second answer"] rewind_command, rewind_args = registry.lookup("/rewind 1") rewind_result = await rewind_command.handler(rewind_args, context) assert "removed 2 message(s)" in rewind_result.message assert len(context.engine.messages) == 2 privacy_command, privacy_args = registry.lookup("/privacy-settings") privacy_result = await privacy_command.handler(privacy_args, context) assert "user_config_dir" in privacy_result.message rate_command, rate_args = registry.lookup("/rate-limit-options") rate_result = await rate_command.handler(rate_args, context) assert "Rate limit options:" in rate_result.message release_command, release_args = registry.lookup("/release-notes") release_result = await release_command.handler(release_args, context) assert "Release Notes" in release_result.message upgrade_command, upgrade_args = registry.lookup("/upgrade") upgrade_result = await upgrade_command.handler(upgrade_args, context) assert "Upgrade instructions:" in upgrade_result.message @pytest.mark.asyncio async def test_mcp_and_voice_commands_report_richer_state(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data")) settings = Settings( mcp_servers={ "http-demo": McpHttpServerConfig(url="https://example.com/mcp"), "stdio-demo": McpStdioServerConfig(command="python", args=["-m", "demo"]), } ) save_settings(settings) registry = create_default_command_registry() context = _make_context(tmp_path) mcp_http_command, mcp_http_args = registry.lookup("/mcp auth http-demo secret-token") mcp_http_result = await mcp_http_command.handler(mcp_http_args, context) assert "Saved MCP auth for http-demo" in mcp_http_result.message assert load_settings().mcp_servers["http-demo"].headers["Authorization"] == "Bearer secret-token" mcp_stdio_command, mcp_stdio_args = registry.lookup("/mcp auth stdio-demo env DEMO_TOKEN") mcp_stdio_result = await mcp_stdio_command.handler(mcp_stdio_args, context) assert "Saved MCP auth for stdio-demo" in mcp_stdio_result.message assert load_settings().mcp_servers["stdio-demo"].env["MCP_AUTH_TOKEN"] == "DEMO_TOKEN" voice_command, voice_args = registry.lookup("/voice show") voice_result = await voice_command.handler(voice_args, context) assert "Voice mode:" in voice_result.message assert "Available:" in voice_result.message assert "Reason:" in voice_result.message @pytest.mark.asyncio async def test_git_commands_report_repository_state(tmp_path: Path, monkeypatch): monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config")) subprocess.run(["git", "init"], cwd=tmp_path, check=True, capture_output=True, text=True) subprocess.run( ["git", "config", "user.email", "openharness@example.com"], cwd=tmp_path, check=True, capture_output=True, text=True, ) subprocess.run( ["git", "config", "user.name", "OpenHarness Tests"], cwd=tmp_path, check=True, capture_output=True, text=True, ) (tmp_path / "demo.txt").write_text("hello\n", encoding="utf-8") registry = create_default_command_registry() context = _make_context(tmp_path) branch_command, branch_args = registry.lookup("/branch show") branch_result = await branch_command.handler(branch_args, context) assert "Current branch" in branch_result.message diff_command, diff_args = registry.lookup("/diff") diff_result = await diff_command.handler(diff_args, context) assert "demo.txt" in diff_result.message or "(no diff)" in diff_result.message commit_command, commit_args = registry.lookup("/commit initial commit") commit_result = await commit_command.handler(commit_args, context) assert "commit" in commit_result.message.lower() def test_quit_is_alias_for_exit(): """Regression for #183: /quit should resolve to the same handler as /exit.""" reg = create_default_command_registry() exit_cmd, _ = reg.lookup("/exit") quit_cmd, _ = reg.lookup("/quit") assert exit_cmd is quit_cmd assert quit_cmd.name == "exit" def test_help_and_list_do_not_duplicate_aliases(): """Aliases share a SlashCommand object; help/listing must not repeat it.""" reg = create_default_command_registry() names = [cmd.name for cmd in reg.list_commands()] assert names.count("exit") == 1 assert "quit" not in names # alias is resolvable via lookup, not listed help_text = reg.help_text() assert help_text.count("/exit ") == 1 assert "/quit" not in help_text