mirrors/hkuds--openharness
1
0
Fork 0
Code Issues Pull Requests Actions 2 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

feat(model): manage profile model allowlists

Browse Source 
Add /model list/add/remove/clear so a single provider profile can expose multiple switchable models in the TUI selector. Also add regression coverage for invalid grep regexes in the Python fallback.\n\nFixes #222\nRefs #218
This commit is contained in:
tjb-tech
2026-05-01 12:38:48 +00:00
parent d9002c3cb2
commit c835d7cf2e
4 changed files with 304 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/openharness/commands/registry.py
+89 -3
View File
@@ -56,6 +56,7 @@ from openharness.tasks import get_task_manager
from openharness.plugins.types import PluginCommandDefinition
if TYPE_CHECKING:
from openharness.config.settings import ProviderProfile
from openharness.state import AppStateStore
from openharness.tools.base import ToolRegistry
@@ -1150,6 +1151,36 @@ def create_default_command_registry(
return CommandResult(message="Plan mode disabled.", refresh_runtime=True)
return CommandResult(message="Usage: /plan [on|off]")
def _dedupe_model_values(values: Iterable[str]) -> list[str]:
models: list[str] = []
seen: set[str] = set()
for value in values:
model = value.strip()
if not model or model in seen:
continue
models.append(model)
seen.add(model)
return models
def _seed_model_values(profile: "ProviderProfile") -> list[str]:
existing = _dedupe_model_values(profile.allowed_models)
if existing:
return existing
return _dedupe_model_values([display_model_setting(profile)])
def _format_model_status(active_profile: str, profile: "ProviderProfile") -> str:
lines = [
f"Model: {display_model_setting(profile)}",
f"Profile: {active_profile}",
]
if profile.allowed_models:
lines.append("Available models:")
lines.extend(f"- {model}" for model in profile.allowed_models)
else:
lines.append("Available models: unrestricted for this profile")
lines.append("Use /model add MODEL to pin switchable models for the TUI selector.")
return "\n".join(lines)
async def _model_handler(args: str, context: CommandContext) -> CommandResult:
settings = load_settings()
manager = AuthManager(settings)
@@ -1157,7 +1188,62 @@ def create_default_command_registry(
_, profile = settings.resolve_profile(active_profile)
tokens = args.split(maxsplit=1)
if not tokens or tokens[0] == "show":
return CommandResult(message=f"Model: {display_model_setting(profile)}\nProfile: {active_profile}")
return CommandResult(message=_format_model_status(active_profile, profile))
if tokens[0] == "list":
if profile.allowed_models:
return CommandResult(
message=(
f"Switchable models for profile '{active_profile}':\n"
+ "\n".join(f"- {model}" for model in profile.allowed_models)
)
)
return CommandResult(
message=(
f"Profile '{active_profile}' has no pinned model list. "
"Any model value is accepted. Use /model add MODEL to add one."
)
)
if tokens[0] == "add" and len(tokens) == 2:
model_name = tokens[1].strip()
models = _dedupe_model_values([*_seed_model_values(profile), model_name])
if not model_name:
return CommandResult(message="Usage: /model add MODEL")
manager.update_profile(active_profile, allowed_models=models)
return CommandResult(
message=f"Added model '{model_name}' to profile '{active_profile}'.",
refresh_runtime=True,
)
if tokens[0] == "add":
return CommandResult(message="Usage: /model add MODEL")
if tokens[0] in {"remove", "rm"} and len(tokens) == 2:
model_name = tokens[1].strip()
models = [model for model in _dedupe_model_values(profile.allowed_models) if model != model_name]
if len(models) == len(_dedupe_model_values(profile.allowed_models)):
return CommandResult(message=f"Model '{model_name}' is not pinned for profile '{active_profile}'.")
reset_current = (profile.last_model or "").strip() == model_name
manager.update_profile(
active_profile,
allowed_models=models,
last_model="" if reset_current else None,
)
updated = load_settings()
if reset_current:
context.engine.set_model(updated.model)
if context.app_state is not None:
updated_profile = updated.resolve_profile()[1]
context.app_state.set(model=display_model_setting(updated_profile))
return CommandResult(
message=f"Removed model '{model_name}' from profile '{active_profile}'.",
refresh_runtime=True,
)
if tokens[0] in {"remove", "rm"}:
return CommandResult(message="Usage: /model remove MODEL")
if tokens[0] == "clear":
manager.update_profile(active_profile, allowed_models=[])
return CommandResult(
message=f"Cleared pinned models for profile '{active_profile}'. Any model value is now accepted.",
refresh_runtime=True,
)
if tokens[0] == "set" and len(tokens) == 2:
model_name = tokens[1].strip()
elif args.strip():
@@ -1180,7 +1266,7 @@ def create_default_command_registry(
updated_profile = updated.resolve_profile()[1]
context.app_state.set(model=display_model_setting(updated_profile))
return CommandResult(message=message, refresh_runtime=True)
return CommandResult(message="Usage: /model [show|MODEL]")
return CommandResult(message="Usage: /model [show|list|add MODEL|remove MODEL|clear|MODEL]")
async def _provider_handler(args: str, context: CommandContext) -> CommandResult:
manager = AuthManager()
@@ -1943,7 +2029,7 @@ def create_default_command_registry(
registry.register(SlashCommand("turns", "Show or update maximum agentic turn count", _turns_handler))
registry.register(SlashCommand("continue", "Continue the previous tool loop if it was interrupted", _continue_handler))
registry.register(SlashCommand("provider", "Show or switch provider profiles", _provider_handler))
registry.register(SlashCommand("model", "Show or update the default model", _model_handler))
registry.register(SlashCommand("model", "Show, switch, or manage profile models", _model_handler))
registry.register(SlashCommand("theme", "List, set, show or preview TUI themes", _theme_handler))
registry.register(SlashCommand("output-style", "Show or update output style", _output_style_handler))
registry.register(SlashCommand("keybindings", "Show resolved keybindings", _keybindings_handler))
tests/test_commands/test_registry.py
+149
View File
@@ -204,6 +204,155 @@ async def test_model_command_accepts_direct_value(tmp_path: Path, monkeypatch):
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"))
tests/test_tools/test_grep_tool.py
+17
View File
@@ -146,3 +146,20 @@ async def test_grep_tool_discards_rg_stderr_for_file_search(monkeypatch, tmp_pat
assert result.is_error is False
assert seen_kwargs["stderr"] is asyncio.subprocess.DEVNULL
@pytest.mark.asyncio
async def test_grep_tool_python_fallback_reports_invalid_regex(monkeypatch, tmp_path: Path):
tool = GrepTool()
monkeypatch.setattr("openharness.tools.grep_tool.shutil.which", lambda _: None)
file_path = tmp_path / "notes.txt"
file_path.write_text("hello\n", encoding="utf-8")
result = await tool.execute(
GrepToolInput(pattern="hello(", root=str(file_path)),
type("Ctx", (), {"cwd": tmp_path})(),
)
assert result.is_error is False
assert "invalid regex pattern 'hello('" in result.output
assert "unterminated subpattern" in result.output
tests/test_ui/test_react_backend.py
+49
View File
@@ -12,6 +12,7 @@ from openharness.api.client import ApiMessageCompleteEvent
from openharness.api.usage import UsageSnapshot
from openharness.engine.stream_events import CompactProgressEvent
from openharness.engine.messages import ConversationMessage, TextBlock
from openharness.config.settings import Settings, save_settings
from openharness.ui.backend_host import BackendHostConfig, ReactBackendHost, run_backend_host
from openharness.ui.protocol import BackendEvent
from openharness.ui.runtime import build_runtime, close_runtime, start_runtime
@@ -446,6 +447,54 @@ async def test_backend_host_emits_model_select_request(tmp_path, monkeypatch):
assert any(option["value"] == "default" for option in event.select_options)
@pytest.mark.asyncio
async def test_backend_host_model_selector_uses_profile_model_allowlist(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
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"],
}
},
}
)
)
host = ReactBackendHost(BackendHostConfig(api_client=StaticApiClient("unused")))
host._bundle = await build_runtime(api_client=StaticApiClient("unused"))
events = []
async def _emit(event):
events.append(event)
host._emit = _emit # type: ignore[method-assign]
await start_runtime(host._bundle)
try:
await host._handle_select_command("model")
finally:
await close_runtime(host._bundle)
event = next(item for item in events if item.type == "select_request")
assert [option["value"] for option in event.select_options] == ["deepseek-chat", "qwen-vl"]
assert any(option["value"] == "qwen-vl" and option.get("active") for option in event.select_options)
@pytest.mark.asyncio
async def test_backend_host_emits_theme_select_request(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)