From a6569214a378db4584395e5bda98a2ae25d699df Mon Sep 17 00:00:00 2001
From: twentyOne2x <contact@attn.market>
Date: Sat, 18 Jul 2026 01:02:12 +0200
Subject: [PATCH] Resolve Leo database tools from the active profile (#185)

---
 .../vps/leo-db-context/__init__.py            | 20 +++++++++-
 .../test_hermes_leoclean_db_context_plugin.py | 38 +++++++++++++++++++
 2 files changed, 57 insertions(+), 1 deletion(-)

diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py
index eae9191..763390b 100644
--- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py
+++ b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py
@@ -556,6 +556,24 @@ def _take_snapshot(session_id: str, query_sha256: str) -> dict[str, Any] | None:
         return _PENDING_SNAPSHOTS.pop(_snapshot_key(session_id, query_sha256), None)
 
 
+def _resolve_hermes_home(explicit_home: Path | None) -> Path:
+    if explicit_home is not None:
+        return explicit_home
+
+    default_home = Path.home() / ".hermes"
+    candidates: list[Path] = []
+    environment_home = os.getenv("HERMES_HOME", "").strip()
+    if environment_home:
+        candidates.append(Path(environment_home))
+
+    # In an installed profile this file lives at <profile>/plugins/leo-db-context/__init__.py.
+    candidates.extend((Path(__file__).resolve().parents[2], default_home))
+    for candidate in candidates:
+        if (candidate / "bin" / "kb_tool.py").is_file():
+            return candidate
+    return candidates[0] if candidates else default_home
+
+
 def _load_database_snapshot(
     user_message: str,
     *,
@@ -567,7 +585,7 @@ def _load_database_snapshot(
 
     query = str(user_message or "")[:MAX_QUERY_CHARS]
     retrieval_query, retrieval_anchors = _contextual_retrieval_query(query, conversation_history)
-    home = hermes_home or Path(os.getenv("HERMES_HOME", str(Path.home() / ".hermes")))
+    home = _resolve_hermes_home(hermes_home)
     kb_tool = home / "bin" / "kb_tool.py"
     query_sha256 = hashlib.sha256(query.encode("utf-8")).hexdigest()
     base_record: dict[str, Any] = {
diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py
index 1c2adae..af111d7 100644
--- a/tests/test_hermes_leoclean_db_context_plugin.py
+++ b/tests/test_hermes_leoclean_db_context_plugin.py
@@ -21,6 +21,14 @@ def load_plugin():
     return module
 
 
+def load_plugin_from(path: Path, module_name: str):
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
 def test_plugin_registers_generation_and_delivery_hooks() -> None:
     module = load_plugin()
     registered = []
@@ -28,6 +36,36 @@ def test_plugin_registers_generation_and_delivery_hooks() -> None:
     assert [name for name, _callback in registered] == ["pre_llm_call", "post_llm_call"]
 
 
+def test_plugin_resolves_kb_tool_from_installed_profile_when_environment_home_is_missing(
+    tmp_path: Path, monkeypatch
+) -> None:
+    profile = tmp_path / "profiles" / "leoclean"
+    plugin_path = profile / "plugins" / "leo-db-context" / "__init__.py"
+    plugin_path.parent.mkdir(parents=True)
+    plugin_path.write_text(PLUGIN.read_text(encoding="utf-8"), encoding="utf-8")
+    kb_tool = profile / "bin" / "kb_tool.py"
+    kb_tool.parent.mkdir(parents=True)
+    kb_tool.write_text("# fixture\n", encoding="utf-8")
+    module = load_plugin_from(plugin_path, "leo_db_context_deployed_layout")
+    monkeypatch.delenv("HERMES_HOME", raising=False)
+    calls = []
+
+    def fake_runner(command, **kwargs):
+        calls.append(command)
+        payload = {
+            "claims": [],
+            "context_rows": [],
+            "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}],
+            "compiled_response": None,
+        }
+        return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
+
+    snapshot = module._load_database_snapshot("Review this candidate.", runner=fake_runner)
+
+    assert snapshot["status"] == "ok"
+    assert calls[0][1] == str(kb_tool)
+
+
 def test_compiled_fallback_preserves_explicit_subject_label_once() -> None:
     module = load_plugin()
     prompt = "Audit the database. Name the subject exactly once as `partner readiness` in your answer."