KrustyPlanet/hermes-agent
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(memory): add manual OpenViking setup path

Browse source
This commit is contained in:
Hao Zhe 2026-05-26 12:33:01 +08:00
parent 7f76cf7195
commit 70f53f36cb
2 changed files with 365 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

139
plugins/memory/openviking/__init__.py
View file

@ -79,6 +79,8 @@ _MEMORY_WRITE_TARGET_SUBDIR_MAP = {
"user": "preferences",
"memory": "patterns",
}
_LOCAL_OPENVIKING_HOSTS = {"localhost", "127.0.0.1", "::1"}
_SETUP_CANCELLED = object()
# ---------------------------------------------------------------------------
@ -451,6 +453,16 @@ def _connection_values_from_ovcli(data: dict) -> dict:
}
def _is_local_openviking_url(value: str) -> bool:
candidate = _clean_config_value(value)
if not candidate:
return False
if "://" not in candidate:
candidate = f"//{candidate}"
parsed = urlparse(candidate)
return (parsed.hostname or "").lower() in _LOCAL_OPENVIKING_HOSTS
def _load_hermes_openviking_config() -> dict:
try:
from hermes_cli.config import load_config
@ -552,11 +564,17 @@ def _remember_ovcli_path(provider_config: dict, ovcli_path: Path) -> None:
def _ovcli_data_from_connection_values(values: dict) -> dict:
data = {"url": _clean_config_value(values.get("endpoint")) or _DEFAULT_ENDPOINT}
api_key = _clean_config_value(values.get("api_key"))
api_key_type = _clean_config_value(values.get("api_key_type"))
root_api_key = _clean_config_value(values.get("root_api_key"))
account = _clean_config_value(values.get("account"))
user = _clean_config_value(values.get("user"))
agent = _clean_config_value(values.get("agent")) or _DEFAULT_AGENT
if api_key:
data["api_key"] = api_key
if root_api_key:
data["root_api_key"] = root_api_key
elif api_key and api_key_type == "root":
data["root_api_key"] = api_key
if account:
data["account"] = account
if user:
@ -572,6 +590,58 @@ def _write_ovcli_config(path: Path, values: dict) -> None:
_restrict_secret_file_permissions(path)
def _prompt_manual_connection_values(prompt, select, cancelled):
endpoint = _clean_config_value(
prompt("OpenViking server URL", default=_DEFAULT_ENDPOINT)
) or _DEFAULT_ENDPOINT
is_local = _is_local_openviking_url(endpoint)
api_key_label = (
"OpenViking API key (optional for local)"
if is_local
else "OpenViking API key"
)
api_key = _clean_config_value(prompt(api_key_label, secret=True))
if not api_key and not is_local:
print("\n Remote OpenViking servers require an API key.")
print(" No changes saved.\n")
return None
values = {
"endpoint": endpoint,
"api_key": api_key,
"account": "",
"user": "",
"agent": "",
}
if api_key:
key_type = select(
" OpenViking API key type",
[
("User API key", "server derives account/user automatically"),
("Root API key", "requires account and user IDs"),
],
default=0,
cancel_returns=cancelled,
)
if key_type == cancelled:
return _SETUP_CANCELLED
if key_type == 1:
values["api_key_type"] = "root"
values["account"] = _clean_config_value(prompt("OpenViking account"))
values["user"] = _clean_config_value(prompt("OpenViking user"))
if not values["account"] or not values["user"]:
print("\n Root API keys require both OpenViking account and user.")
print(" No changes saved.\n")
return None
else:
values["api_key_type"] = "user"
values["agent"] = _clean_config_value(
prompt("OpenViking agent", default=_DEFAULT_AGENT)
) or _DEFAULT_AGENT
return values
# ---------------------------------------------------------------------------
# MemoryProvider implementation
# ---------------------------------------------------------------------------
@ -668,6 +738,7 @@ class OpenVikingMemoryProvider(MemoryProvider):
setup_options = [
("Link to ovcli.conf", "Hermes follows the active OpenViking CLI config"),
("Copy once", "Hermes won't follow future ovcli.conf changes"),
("Manual Setup", "Enter a new URL/API key"),
]
choice = _curses_select(
" OpenViking config source",
@ -691,18 +762,64 @@ class OpenVikingMemoryProvider(MemoryProvider):
print(" Start a new session to activate.\n")
return
provider_config["use_ovcli_config"] = False
provider_config.pop("ovcli_config_path", None)
config["memory"]["provider"] = "openviking"
config["memory"]["openviking"] = provider_config
save_config(config)
_write_env_vars(
env_path,
_env_writes_from_connection_values(ovcli_values),
remove_keys=_OPENVIKING_ENV_KEYS,
if choice == 1:
provider_config["use_ovcli_config"] = False
provider_config.pop("ovcli_config_path", None)
config["memory"]["provider"] = "openviking"
config["memory"]["openviking"] = provider_config
save_config(config)
_write_env_vars(
env_path,
_env_writes_from_connection_values(ovcli_values),
remove_keys=_OPENVIKING_ENV_KEYS,
)
print(f"\n Memory provider: openviking")
print(" Connection saved to .env")
print(" Start a new session to activate.\n")
return
values = _prompt_manual_connection_values(_prompt, _curses_select, _CANCELLED)
if values is _SETUP_CANCELLED:
_print_cancelled_setup()
return
if values is None:
return
save_choice = _curses_select(
" Save OpenViking config",
[
("Write ovcli.conf and link", "Hermes and ov use this config"),
("Keep within Hermes", "Write values only to Hermes .env"),
],
default=1,
cancel_returns=_CANCELLED,
)
print(f"\n Memory provider: openviking")
print(" Connection saved to .env")
if save_choice == _CANCELLED:
_print_cancelled_setup()
return
config["memory"]["provider"] = "openviking"
if save_choice == 0:
_write_ovcli_config(ovcli_path, values)
provider_config["use_ovcli_config"] = True
_remember_ovcli_path(provider_config, ovcli_path)
config["memory"]["openviking"] = provider_config
save_config(config)
_write_env_vars(env_path, {}, remove_keys=_OPENVIKING_ENV_KEYS)
print(f"\n Memory provider: openviking")
print(f" Updated config: {ovcli_path}")
else:
provider_config["use_ovcli_config"] = False
provider_config.pop("ovcli_config_path", None)
config["memory"]["openviking"] = provider_config
save_config(config)
_write_env_vars(
env_path,
_env_writes_from_connection_values(values),
remove_keys=_OPENVIKING_ENV_KEYS,
)
print(f"\n Memory provider: openviking")
print(" Connection saved to .env")
print(" Start a new session to activate.\n")
return

248
tests/plugins/memory/test_openviking_provider.py
View file

@ -22,6 +22,17 @@ def _clear_openviking_env(monkeypatch):
monkeypatch.delenv(key, raising=False)
def _prompt_from_values(values: dict[str, str], *, forbidden: set[str] | None = None):
forbidden = forbidden or set()
def _prompt(label, default=None, secret=False):
if label in forbidden:
raise AssertionError(f"{label} should not be prompted")
return values.get(label, default or "")
return _prompt
@pytest.mark.skipif(os.name == "nt", reason="POSIX file modes")
def test_openviking_env_writer_restricts_file_permissions(tmp_path):
env_path = tmp_path / ".env"
@ -133,7 +144,8 @@ def test_post_setup_link_existing_ovcli_clears_hermes_env(tmp_path, monkeypatch)
encoding="utf-8",
)
ovcli_path = tmp_path / "ovcli.conf"
ovcli_path.write_text(json.dumps({"url": "http://openviking.local"}), encoding="utf-8")
original_ovcli = json.dumps({"url": "http://openviking.local"})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
@ -150,6 +162,7 @@ def test_post_setup_link_existing_ovcli_clears_hermes_env(tmp_path, monkeypatch)
env_text = env_path.read_text(encoding="utf-8")
assert "OPENVIKING_" not in env_text
assert "OTHER_KEY=keep" in env_text
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
def test_post_setup_copy_existing_ovcli_writes_hermes_env(tmp_path, monkeypatch):
@ -157,16 +170,14 @@ def test_post_setup_copy_existing_ovcli_writes_hermes_env(tmp_path, monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
ovcli_path = tmp_path / "ovcli.conf"
ovcli_path.write_text(
json.dumps({
"url": "http://openviking.local",
"api_key": "test-key",
"account": "acct",
"user": "alice",
"agent_id": "agent",
}),
encoding="utf-8",
)
original_ovcli = json.dumps({
"url": "http://openviking.local",
"api_key": "test-key",
"account": "acct",
"user": "alice",
"agent_id": "agent",
})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
@ -185,6 +196,221 @@ def test_post_setup_copy_existing_ovcli_writes_hermes_env(tmp_path, monkeypatch)
assert "OPENVIKING_ACCOUNT=acct" in env_text
assert "OPENVIKING_USER=alice" in env_text
assert "OPENVIKING_AGENT=agent" in env_text
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
def test_post_setup_manual_remote_root_writes_ovcli_and_links(tmp_path, monkeypatch):
_clear_openviking_env(monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
env_path = hermes_home / ".env"
env_path.write_text("OPENVIKING_ENDPOINT=http://old.local\n", encoding="utf-8")
ovcli_path = tmp_path / "ovcli.conf"
ovcli_path.write_text(json.dumps({"url": "http://old.local"}), encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
from hermes_cli import memory_setup
choices = iter([2, 1, 0])
monkeypatch.setattr(
memory_setup,
"_curses_select",
lambda *args, **kwargs: next(choices),
)
monkeypatch.setattr(
memory_setup,
"_prompt",
_prompt_from_values({
"OpenViking server URL": "https://openviking.example",
"OpenViking API key": "root-secret",
"OpenViking account": "acct",
"OpenViking user": "alice",
"OpenViking agent": "agent",
}),
)
config = {"memory": {}}
OpenVikingMemoryProvider().post_setup(str(hermes_home), config)
assert config["memory"]["provider"] == "openviking"
assert config["memory"]["openviking"]["use_ovcli_config"] is True
assert config["memory"]["openviking"]["ovcli_config_path"] == str(ovcli_path)
assert env_path.read_text(encoding="utf-8") == ""
data = json.loads(ovcli_path.read_text(encoding="utf-8"))
assert data == {
"url": "https://openviking.example",
"api_key": "root-secret",
"root_api_key": "root-secret",
"account": "acct",
"user": "alice",
"agent_id": "agent",
}
def test_post_setup_manual_remote_user_keeps_only_hermes_env(tmp_path, monkeypatch):
_clear_openviking_env(monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
ovcli_path = tmp_path / "ovcli.conf"
original_ovcli = json.dumps({"url": "http://old.local"})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
from hermes_cli import memory_setup
choices = iter([2, 0, 1])
monkeypatch.setattr(
memory_setup,
"_curses_select",
lambda *args, **kwargs: next(choices),
)
monkeypatch.setattr(
memory_setup,
"_prompt",
_prompt_from_values(
{
"OpenViking server URL": "https://openviking.example",
"OpenViking API key": "user-secret",
"OpenViking agent": "agent",
},
forbidden={"OpenViking account", "OpenViking user"},
),
)
config = {"memory": {}}
OpenVikingMemoryProvider().post_setup(str(hermes_home), config)
assert config["memory"]["provider"] == "openviking"
assert config["memory"]["openviking"]["use_ovcli_config"] is False
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
env_text = (hermes_home / ".env").read_text(encoding="utf-8")
assert "OPENVIKING_ENDPOINT=https://openviking.example" in env_text
assert "OPENVIKING_API_KEY=user-secret" in env_text
assert "OPENVIKING_AGENT=agent" in env_text
assert "OPENVIKING_ACCOUNT" not in env_text
assert "OPENVIKING_USER" not in env_text
def test_post_setup_manual_remote_requires_api_key(tmp_path, monkeypatch):
_clear_openviking_env(monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
ovcli_path = tmp_path / "ovcli.conf"
original_ovcli = json.dumps({"url": "http://old.local"})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
from hermes_cli import config as hermes_config
from hermes_cli import memory_setup
save_config = MagicMock()
monkeypatch.setattr(hermes_config, "save_config", save_config)
monkeypatch.setattr(memory_setup, "_curses_select", lambda *args, **kwargs: 2)
monkeypatch.setattr(
memory_setup,
"_prompt",
_prompt_from_values({
"OpenViking server URL": "https://openviking.example",
"OpenViking API key": "",
}),
)
config = {"memory": {"provider": "builtin"}}
OpenVikingMemoryProvider().post_setup(str(hermes_home), config)
save_config.assert_not_called()
assert config == {"memory": {"provider": "builtin"}}
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
assert not (hermes_home / ".env").exists()
def test_post_setup_manual_root_requires_account_and_user(tmp_path, monkeypatch):
_clear_openviking_env(monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
ovcli_path = tmp_path / "ovcli.conf"
original_ovcli = json.dumps({"url": "http://old.local"})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
from hermes_cli import config as hermes_config
from hermes_cli import memory_setup
save_config = MagicMock()
choices = iter([2, 1])
monkeypatch.setattr(hermes_config, "save_config", save_config)
monkeypatch.setattr(
memory_setup,
"_curses_select",
lambda *args, **kwargs: next(choices),
)
monkeypatch.setattr(
memory_setup,
"_prompt",
_prompt_from_values({
"OpenViking server URL": "https://openviking.example",
"OpenViking API key": "root-secret",
"OpenViking account": "",
"OpenViking user": "alice",
}),
)
config = {"memory": {"provider": "builtin"}}
OpenVikingMemoryProvider().post_setup(str(hermes_home), config)
save_config.assert_not_called()
assert config == {"memory": {"provider": "builtin"}}
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
assert not (hermes_home / ".env").exists()
def test_post_setup_manual_local_allows_blank_api_key(tmp_path, monkeypatch):
_clear_openviking_env(monkeypatch)
hermes_home = tmp_path / "hermes"
hermes_home.mkdir()
ovcli_path = tmp_path / "ovcli.conf"
original_ovcli = json.dumps({"url": "http://old.local"})
ovcli_path.write_text(original_ovcli, encoding="utf-8")
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.setenv("OPENVIKING_CLI_CONFIG_FILE", str(ovcli_path))
from hermes_cli import memory_setup
choices = iter([2, 1])
monkeypatch.setattr(
memory_setup,
"_curses_select",
lambda *args, **kwargs: next(choices),
)
monkeypatch.setattr(
memory_setup,
"_prompt",
_prompt_from_values(
{
"OpenViking server URL": "http://localhost:1933",
"OpenViking API key": "",
"OpenViking agent": "agent",
},
forbidden={"OpenViking account", "OpenViking user"},
),
)
config = {"memory": {}}
OpenVikingMemoryProvider().post_setup(str(hermes_home), config)
assert config["memory"]["provider"] == "openviking"
assert config["memory"]["openviking"]["use_ovcli_config"] is False
assert ovcli_path.read_text(encoding="utf-8") == original_ovcli
env_text = (hermes_home / ".env").read_text(encoding="utf-8")
assert "OPENVIKING_ENDPOINT=http://localhost:1933" in env_text
assert "OPENVIKING_AGENT=agent" in env_text
assert "OPENVIKING_API_KEY" not in env_text
assert "OPENVIKING_ACCOUNT" not in env_text
assert "OPENVIKING_USER" not in env_text
def test_post_setup_cancel_existing_ovcli_writes_nothing(tmp_path, monkeypatch):