hermes-agent/tests/plugins/memory/test_openviking_optional_peer.py

"""Optional peer identity must agree across setup, requests and memory writes."""

import json
import os
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import pytest
import hermes_yaml as yaml

import plugins.memory.openviking as ov


@pytest.fixture(autouse=True)
def isolated_config(tmp_path, monkeypatch):
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
    for key in (*ov._OPENVIKING_ENV_KEYS, "OPENVIKING_CLI_CONFIG_FILE"):
        # Track absent keys too, so setup's direct environment writes are undone.
        monkeypatch.setenv(key, "")
        monkeypatch.delenv(key, raising=False)


@pytest.mark.parametrize("source", ["env", "yaml", "actor_peer_id", "agent_id"])
@pytest.mark.parametrize("peer", ["", "hermes", "work-assistant"])
def test_configured_peer_routing_is_preserved(tmp_path, monkeypatch, source, peer):
    config = {}
    if source == "env":
        monkeypatch.setenv("OPENVIKING_AGENT", peer)
    elif source == "yaml":
        config["agent"] = peer
    else:
        path = tmp_path / "ovcli.conf"
        path.write_text(
            json.dumps({"url": "http://localhost:1933", source: peer}), encoding="utf-8"
        )
        config = {"use_ovcli_config": True, "ovcli_config_path": str(path)}

    settings = ov._resolve_connection_settings(config)
    client = ov._VikingClient("http://localhost:1933", agent=settings["agent"])
    monkeypatch.setattr(client, "get", lambda *a, **kw: {"result": {"user": "alice"}})
    provider = ov.OpenVikingMemoryProvider()
    uri = provider._build_memory_uri("preferences", client=client)

    assert settings["agent"] == peer
    assert client._headers().get("X-OpenViking-Actor-Peer", "") == peer
    prefix = f"peers/{peer}/" if peer else ""
    assert uri.startswith(f"viking://user/alice/{prefix}memories/preferences/mem_")


def test_unconfigured_client_and_schema_do_not_supply_a_peer():
    settings = ov._resolve_connection_settings({})
    client = ov._VikingClient("http://localhost:1933")
    schema = {
        field["key"]: field
        for field in ov.OpenVikingMemoryProvider().get_config_schema()
    }

    assert settings["agent"] == ""
    assert schema["agent"]["default"] == ""
    assert "X-OpenViking-Actor-Peer" not in client._headers()
    assert "X-OpenViking-Actor-Peer" not in client._multipart_headers()


@pytest.mark.parametrize("peer", ["", "hermes"])
def test_linked_profile_status_only_shows_a_configured_peer(tmp_path, peer):
    path = tmp_path / "ovcli.conf"
    path.write_text(
        json.dumps({"url": "http://localhost:1933", "actor_peer_id": peer}),
        encoding="utf-8",
    )
    display = ov.OpenVikingMemoryProvider().get_status_config({
        "use_ovcli_config": True,
        "ovcli_config_path": str(path),
    })

    if peer:
        assert display["agent"] == peer
    else:
        assert "agent" not in display


@pytest.mark.parametrize("peer", ["", "hermes"])
def test_memory_uri_uses_captured_peer_even_when_empty(monkeypatch, peer):
    client = ov._VikingClient("http://localhost:1933", agent=peer)
    monkeypatch.setattr(client, "get", lambda *a, **kw: {"result": {"user": "alice"}})
    provider = ov.OpenVikingMemoryProvider()
    provider._agent = "later-peer"

    uri = provider._build_memory_uri("preferences", client=client)

    prefix = f"peers/{peer}/" if peer else ""
    assert uri.startswith(f"viking://user/alice/{prefix}memories/preferences/mem_")
    assert "later-peer" not in uri


@pytest.mark.parametrize("save_to_store", [False, True])
@pytest.mark.parametrize("credential", ["dev", "user", "root", "service"])
@pytest.mark.parametrize("stale_env", [False, True])
def test_new_setup_does_not_ask_for_or_save_peer(
    tmp_path,
    monkeypatch,
    save_to_store,
    credential,
    stale_env,
):
    from hermes_cli import memory_setup

    home = tmp_path / "hermes"
    home.mkdir()
    (home / ".env").write_text(
        "OPENVIKING_AGENT=old-peer\nOTHER_KEY=keep\n", encoding="utf-8"
    )
    config = {"memory": {"openviking": {"agent": "old-peer", "recall_limit": 9}}}
    if stale_env:
        for key in ov._OPENVIKING_ENV_KEYS:
            monkeypatch.setenv(
                key, "old-peer" if key == "OPENVIKING_AGENT" else "old-value"
            )
    monkeypatch.setenv("OTHER_KEY", "keep")
    validations = []

    def validate(values, **kwargs):
        validations.append(dict(values))
        role = (
            "root"
            if credential == "root"
            else "user"
            if values.get("api_key")
            else None
        )
        return True, "", role

    def prompt(label, default=None, secret=False):
        values = {
            "OpenViking server URL": "http://localhost:1933",
            "OpenViking user API key": "test-user-key",
            "OpenViking root API key": "test-root-key",
            "OpenViking API key": "test-service-key",
            "OpenViking account": "account",
            "OpenViking user": "alice",
            "OpenViking profile name": "personal",
        }
        assert label in values, f"Unexpected setup question: {label}"
        return values[label]

    def select(title, options, **kwargs):
        choices = {
            "  OpenViking connection": 0 if credential == "service" else 1,
            "  OpenViking credential": {"dev": 2, "user": 0, "root": 1}.get(
                credential, 0
            ),
            "  Save OpenViking config": int(save_to_store),
        }
        assert title in choices, f"Unexpected setup menu: {title}"
        return choices[title]

    monkeypatch.setattr(memory_setup, "_prompt", prompt)
    monkeypatch.setattr(memory_setup, "_curses_select", select)
    monkeypatch.setattr(ov, "_validate_openviking_reachability", lambda *a: (True, ""))
    monkeypatch.setattr(ov, "_validate_openviking_setup_values", validate)

    ov.OpenVikingMemoryProvider().post_setup(str(home), config)

    assert validations
    assert all(values["agent"] == "" for values in validations)
    assert "OPENVIKING_AGENT" not in (home / ".env").read_text(encoding="utf-8")
    assert "OTHER_KEY=keep" in (home / ".env").read_text(encoding="utf-8")
    saved_config = ov._load_hermes_openviking_config()
    assert saved_config["recall_limit"] == 9
    settings = ov._resolve_connection_settings(saved_config)
    assert settings["agent"] == ""
    assert settings == {
        key: validations[-1][key]
        for key in ("endpoint", "api_key", "account", "user", "agent")
    }
    assert "OPENVIKING_AGENT" not in os.environ
    assert os.environ["OTHER_KEY"] == "keep"
    if save_to_store:
        saved = json.loads(
            Path(saved_config["ovcli_config_path"]).read_text(encoding="utf-8")
        )
        assert "actor_peer_id" not in saved
        assert "agent_id" not in saved


@pytest.mark.parametrize("peer", ["", "work-assistant"])
def test_hermes_only_save_uses_the_same_clean_values_in_file_and_process(
    tmp_path, peer
):
    from dotenv import dotenv_values

    env_path = tmp_path / ".env"
    ov._setup._save_hermes_only_config(
        config={"memory": {}},
        provider_config={},
        env_path=env_path,
        values={
            "endpoint": "http://localhost:29333",
            "api_key": "test\r\n-key\x00",
            "agent": peer,
        },
    )

    expected = {
        "OPENVIKING_ENDPOINT": "http://localhost:29333",
        "OPENVIKING_API_KEY": "test-key",
    }
    if peer:
        expected["OPENVIKING_AGENT"] = peer
    assert dict(dotenv_values(env_path)) == expected
    assert {
        key: os.environ[key] for key in ov._OPENVIKING_ENV_KEYS if key in os.environ
    } == expected


def test_hermes_only_save_failure_leaves_process_environment_unchanged(
    tmp_path, monkeypatch
):
    for key in ov._OPENVIKING_ENV_KEYS:
        monkeypatch.setenv(key, "old-value")

    def fail_write(*args, **kwargs):
        raise OSError("test write failure")

    monkeypatch.setattr(ov, "_write_env_vars", fail_write)
    with pytest.raises(OSError, match="test write failure"):
        ov._setup._save_hermes_only_config(
            config={"memory": {}},
            provider_config={},
            env_path=tmp_path / ".env",
            values={"endpoint": "http://localhost:29333", "api_key": "test-key"},
        )

    assert all(os.environ[key] == "old-value" for key in ov._OPENVIKING_ENV_KEYS)


@pytest.mark.parametrize("peer", ["", "hermes"])
def test_wire_requests_keep_writes_and_session_messages_in_the_selected_scope(
    tmp_path,
    monkeypatch,
    peer,
):
    records = []

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def respond(self, payload):
            body = json.dumps(payload).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            if self.path == "/health":
                self.respond({"status": "ok", "healthy": True, "version": "test"})
            elif self.path == "/api/v1/system/status":
                self.respond({"result": {"user": "alice"}})
            else:
                self.send_error(404)

        def do_POST(self):
            payload = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
            records.append((self.path, dict(self.headers), payload))
            self.respond({"status": "ok", "result": {"written_bytes": 10}})

    server = HTTPServer(("127.0.0.1", 0), Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    home = tmp_path / "hermes"
    home.mkdir()
    provider_config = {"endpoint": f"http://127.0.0.1:{server.server_port}"}
    if peer:
        provider_config["agent"] = peer
    (home / "config.yaml").write_text(
        yaml.safe_dump({
            "memory": {"provider": "openviking", "openviking": provider_config}
        }),
        encoding="utf-8",
    )
    provider = ov.OpenVikingMemoryProvider()
    try:
        provider.initialize("peer-test", hermes_home=str(home))
        assert provider._client is not None
        result = json.loads(
            provider.handle_tool_call("viking_remember", {"content": "I like tea"})
        )
        assert result["status"] == "submitted"
        provider.on_memory_write("add", "user", "I like coffee")
        provider.sync_turn("hello", "hi", session_id="peer-test")
        assert provider._drain_writers("peer-test", timeout=5.0)
        provider.sync_turn(
            "next",
            "reply",
            session_id="peer-test",
            messages=[
                {"role": "user", "content": "next"},
                {"role": "assistant", "content": "reply"},
            ],
        )
        assert provider._drain_writers("peer-test", timeout=5.0)
        provider.on_session_end([])
    finally:
        provider.shutdown()
        server.shutdown()
        server.server_close()
        thread.join(timeout=3.0)

    assert records
    for _path, headers, _payload in records:
        if peer:
            assert headers["X-OpenViking-Actor-Peer"] == peer
        else:
            assert "X-OpenViking-Actor-Peer" not in headers
    writes = [
        payload for path, _, payload in records if path == "/api/v1/content/write"
    ]
    prefix = f"peers/{peer}/" if peer else ""
    assert {write["content"] for write in writes} == {"I like coffee"}
    assert all(
        write["uri"].startswith(f"viking://user/alice/{prefix}memories/")
        for write in writes
    )
    remember_messages = [
        (path, payload)
        for path, _, payload in records
        if path.startswith("/api/v1/sessions/hermes-remember-")
        and path.endswith("/messages")
    ]
    assert len(remember_messages) == 1
    remember_path, remember_message = remember_messages[0]
    remember_session = remember_path.removesuffix("/messages")
    assert remember_message == {
        "role": "user",
        "parts": [{"type": "text", "text": "I like tea"}],
    }
    assert any(path == f"{remember_session}/commit" for path, _, _ in records)
    batches = [
        payload["messages"]
        for path, _, payload in records
        if path.endswith("/messages/batch")
    ]
    assert len(batches) == 2
    for batch in batches:
        assert "peer_id" not in batch[0]
        if peer:
            assert batch[1]["peer_id"] == peer
        else:
            assert "peer_id" not in batch[1]
    assert any(path.endswith("/commit") for path, _, _ in records)