Resolve Leo database tools from the active profile (#185)

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] = {

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."